sql/019_document_credit_rls.sql

@@ -0,0 +1,25 @@
+-- The GUC is transaction-local and supplied only by trusted server code.
+-- RLS is a defense against missing tenant predicates, not authentication.
+ALTER TABLE opt_credit_accounts ENABLE ROW LEVEL SECURITY;
+ALTER TABLE opt_credit_accounts FORCE ROW LEVEL SECURITY;
+CREATE POLICY opt_credit_accounts_tenant ON opt_credit_accounts
+  USING (tenant_id = current_setting('app.opt_credit_tenant_id', true))
+  WITH CHECK (tenant_id = current_setting('app.opt_credit_tenant_id', true));
+
+ALTER TABLE opt_credit_reservations ENABLE ROW LEVEL SECURITY;
+ALTER TABLE opt_credit_reservations FORCE ROW LEVEL SECURITY;
+CREATE POLICY opt_credit_reservations_tenant ON opt_credit_reservations
+  USING (tenant_id = current_setting('app.opt_credit_tenant_id', true))
+  WITH CHECK (tenant_id = current_setting('app.opt_credit_tenant_id', true));
+
+ALTER TABLE opt_credit_events ENABLE ROW LEVEL SECURITY;
+ALTER TABLE opt_credit_events FORCE ROW LEVEL SECURITY;
+CREATE POLICY opt_credit_events_tenant ON opt_credit_events
+  USING (tenant_id = current_setting('app.opt_credit_tenant_id', true))
+  WITH CHECK (tenant_id = current_setting('app.opt_credit_tenant_id', true));
+
+ALTER TABLE opt_credit_purchases ENABLE ROW LEVEL SECURITY;
+ALTER TABLE opt_credit_purchases FORCE ROW LEVEL SECURITY;
+CREATE POLICY opt_credit_purchases_tenant ON opt_credit_purchases
+  USING (tenant_id = current_setting('app.opt_credit_tenant_id', true))
+  WITH CHECK (tenant_id = current_setting('app.opt_credit_tenant_id', true));

src/document-workbench/credit-ledger.ts

@@ -1,4 +1,5 @@
 import type { RepositorySql } from '../repositories/db';
+import { withCreditTenant } from './credit-tenant-scope';
 
 export type CreditOperation = 'verified_payment' | 'reserve' | 'settle' | 'release';
 export interface CreditCommand {
@@ -37,7 +38,7 @@ export class PostgresCreditLedger {
     )
       throw new Error('Invalid credit command');
 
-    return this.sql.begin(async (tx) => {
+    return withCreditTenant(this.sql, tenantId, async (tx) => {
       await tx`INSERT INTO opt_credit_accounts (tenant_id) VALUES (${tenantId}) ON CONFLICT DO NOTHING`;
       const accounts = await tx<{ available: string; reserved: string }[]>`
         SELECT available, reserved FROM opt_credit_accounts WHERE tenant_id = ${tenantId} FOR UPDATE`;

src/document-workbench/credit-purchase-store.ts

@@ -1,5 +1,6 @@
 import type { RepositorySql } from '../repositories/db';
 import type { CreditPurchaseSnapshot } from './pay0-credit-verification';
+import { withCreditTenant } from './credit-tenant-scope';
 
 interface PurchaseRow {
   tenant_id: string;
@@ -37,28 +38,28 @@ export class PostgresCreditPurchaseStore {
     ) {
       throw new Error('Invalid credit purchase snapshot');
     }
-    const rows = await this.sql<PurchaseRow[]>`
-      INSERT INTO opt_credit_purchases (tenant_id, purchase_id, checkout_session_id, amount_minor, currency, credits)
-      VALUES (${snapshot.tenantId}, ${snapshot.purchaseId}, ${snapshot.checkoutSessionId}, ${snapshot.amountMinor}, ${snapshot.currency}, ${snapshot.credits})
-      ON CONFLICT (tenant_id, purchase_id) DO NOTHING RETURNING *`;
-    if (rows[0]) return fromRow(rows[0]);
-    const existing = await this.get(snapshot.tenantId, snapshot.purchaseId);
-    if (
-      !existing ||
-      existing.checkoutSessionId !== snapshot.checkoutSessionId ||
-      existing.amountMinor !== snapshot.amountMinor ||
-      existing.currency !== snapshot.currency ||
-      existing.credits !== snapshot.credits
-    ) {
-      throw new Error('Purchase id reused for a different price snapshot');
-    }
-    return existing;
+    return withCreditTenant(this.sql, snapshot.tenantId, async (tx) => {
+      const rows = await tx<PurchaseRow[]>`
+        INSERT INTO opt_credit_purchases (tenant_id, purchase_id, checkout_session_id, amount_minor, currency, credits)
+        VALUES (${snapshot.tenantId}, ${snapshot.purchaseId}, ${snapshot.checkoutSessionId}, ${snapshot.amountMinor}, ${snapshot.currency}, ${snapshot.credits})
+        ON CONFLICT (tenant_id, purchase_id) DO NOTHING RETURNING *`;
+      if (rows[0]) return fromRow(rows[0]);
+      const existingRows = await tx<PurchaseRow[]>`
+        SELECT * FROM opt_credit_purchases WHERE tenant_id = ${snapshot.tenantId} AND purchase_id = ${snapshot.purchaseId}`;
+      const existing = existingRows[0] ? fromRow(existingRows[0]) : undefined;
+      if (!existing || existing.checkoutSessionId !== snapshot.checkoutSessionId || existing.amountMinor !== snapshot.amountMinor
+        || existing.currency !== snapshot.currency || existing.credits !== snapshot.credits) {
+        throw new Error('Purchase id reused for a different price snapshot');
+      }
+      return existing;
+    });
   }
 
   async get(tenantId: string, purchaseId: string): Promise<CreditPurchaseSnapshot | undefined> {
-    const rows = await this.sql<
-      PurchaseRow[]
-    >`SELECT * FROM opt_credit_purchases WHERE tenant_id = ${tenantId} AND purchase_id = ${purchaseId}`;
-    return rows[0] ? fromRow(rows[0]) : undefined;
+    if (typeof purchaseId !== 'string' || !purchaseId.trim()) throw new Error('Invalid credit purchase id');
+    return withCreditTenant(this.sql, tenantId, async (tx) => {
+      const rows = await tx<PurchaseRow[]>`SELECT * FROM opt_credit_purchases WHERE tenant_id = ${tenantId} AND purchase_id = ${purchaseId}`;
+      return rows[0] ? fromRow(rows[0]) : undefined;
+    });
   }
 }

src/document-workbench/credit-tenant-scope.ts

@@ -0,0 +1,33 @@
+import type postgres from 'postgres';
+import type { RepositorySql } from '../repositories/db';
+
+/** A scoped transaction is not a user authorization or payment verification. */
+export async function withCreditTenant<T>(
+  sql: RepositorySql,
+  tenantId: string,
+  action: (tx: postgres.TransactionSql) => Promise<T>,
+): Promise<T> {
+  if (typeof tenantId !== 'string' || !tenantId.trim() || tenantId.length > 256) throw new Error('Invalid credit tenant');
+  return sql.begin(async (tx) => {
+    const roles = await tx<{ rolsuper: boolean; rolbypassrls: boolean }[]>`
+      SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user`;
+    if (!roles[0] || roles[0].rolsuper || roles[0].rolbypassrls) throw new Error('Credits require an RLS-enforced database role');
+    // Do not silently run against pre-migration credit tables. Policy bodies
+    // themselves remain a trusted migration/database-administrator boundary.
+    const gates = await tx<{ enabled: boolean }[]>`
+      SELECT COUNT(*) = 4 AS enabled
+      FROM (VALUES
+        ('opt_credit_accounts', 'opt_credit_accounts_tenant'),
+        ('opt_credit_reservations', 'opt_credit_reservations_tenant'),
+        ('opt_credit_events', 'opt_credit_events_tenant'),
+        ('opt_credit_purchases', 'opt_credit_purchases_tenant')
+      ) AS expected(table_name, policy_name)
+      JOIN pg_class AS c ON c.oid = to_regclass(expected.table_name)
+      WHERE c.relrowsecurity AND c.relforcerowsecurity
+        AND (SELECT COUNT(*) FROM pg_policy AS p WHERE p.polrelid = c.oid) = 1
+        AND EXISTS (SELECT 1 FROM pg_policy AS p WHERE p.polrelid = c.oid AND p.polname = expected.policy_name)`;
+    if (gates[0]?.enabled !== true) throw new Error('Credit tenant isolation migration is not active');
+    await tx`SELECT set_config('app.opt_credit_tenant_id', ${tenantId}, true)`;
+    return action(tx);
+  }) as Promise<T>;
+}

tests/document-workbench/credit-ledger.realinfra.test.ts

@@ -8,31 +8,55 @@ import { reconcileCreditPurchase } from '../../src/document-workbench/reconcile-
 
 const url = process.env.CREDIT_LEDGER_TEST_DATABASE_URL;
 const schema = `credit_test_${process.pid}_${Date.now()}`;
+const role = `credit_role_${process.pid}_${Date.now()}`;
+let admin: ReturnType<typeof postgres>;
+let scopedUrl: string;
 let sql: ReturnType<typeof postgres>;
 let ledger: PostgresCreditLedger;
 
 describe.skipIf(!url)('credit ledger against isolated PostgreSQL schema', () => {
   afterEach(() => vi.unstubAllGlobals());
   beforeAll(async () => {
-    const admin = postgres(url!, { max: 1 });
-    try {
-      await admin.unsafe(`CREATE SCHEMA ${schema}`);
-    } finally {
-      await admin.end({ timeout: 2 });
+    if (process.env.CREDIT_LEDGER_TEST_ALLOW_LOCAL !== '1' || new URL(url!).hostname !== '127.0.0.1') {
+      throw new Error('Only an explicitly enabled isolated loopback database may be used');
     }
-    sql = postgres(url!, { max: 8, prepare: false, connection: { search_path: schema } });
-    await sql.unsafe(await fs.readFile('sql/017_document_credits.sql', 'utf8'));
+    admin = postgres(url!, { max: 1, prepare: false, connection: { search_path: schema } });
+    await admin.unsafe(`CREATE SCHEMA ${schema}`);
+    await admin.unsafe(`CREATE ROLE ${role} LOGIN NOSUPERUSER NOBYPASSRLS`);
+    await admin.unsafe(await fs.readFile('sql/017_document_credits.sql', 'utf8'));
+    await admin.unsafe(await fs.readFile('sql/019_document_credit_rls.sql', 'utf8'));
+    await admin.unsafe(`GRANT USAGE ON SCHEMA ${schema} TO ${role}`);
+    await admin.unsafe(`GRANT SELECT, INSERT, UPDATE ON ALL TABLES IN SCHEMA ${schema} TO ${role}`);
+    await admin.unsafe(`GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA ${schema} TO ${role}`);
+    const appUrl = new URL(url!);
+    appUrl.username = role;
+    scopedUrl = appUrl.toString();
+    sql = postgres(scopedUrl, { max: 8, prepare: false, connection: { search_path: schema } });
     ledger = new PostgresCreditLedger(sql);
   });
   afterAll(async () => {
     if (sql) await sql.end();
-    const admin = postgres(url!, { max: 1 });
-    try {
-      await admin.unsafe(`DROP SCHEMA IF EXISTS ${schema} CASCADE`);
-    } finally {
-      await admin.end({ timeout: 2 });
+    if (admin) {
+      try {
+        await admin.unsafe(`DROP SCHEMA IF EXISTS ${schema} CASCADE`);
+        await admin.unsafe(`DROP ROLE IF EXISTS ${role}`);
+      } finally { await admin.end({ timeout: 2 }); }
     }
   });
+  it('rejects superusers and does not expose rows without a tenant scope', async () => {
+    await expect(new PostgresCreditLedger(admin).apply({ tenantId: 'a', idempotencyKey: 'bad-role', kind: 'verified_payment', reference: 'bad-role', amount: 1 })).rejects.toThrow(/RLS-enforced/);
+    expect(await sql`SELECT * FROM opt_credit_accounts`).toEqual([]);
+    expect(await sql`SELECT * FROM opt_credit_purchases`).toEqual([]);
+  });
+
+  it('fails before writing when the isolation migration is not fully active', async () => {
+    await admin.unsafe('ALTER TABLE opt_credit_accounts DISABLE ROW LEVEL SECURITY');
+    try {
+      await expect(ledger.apply({ tenantId: 'unscoped', idempotencyKey: 'disabled-rls', kind: 'verified_payment', reference: 'disabled-rls', amount: 1 })).rejects.toThrow(/isolation migration/);
+    } finally { await admin.unsafe('ALTER TABLE opt_credit_accounts ENABLE ROW LEVEL SECURITY'); }
+    const rows = await admin`SELECT * FROM opt_credit_accounts WHERE tenant_id = 'unscoped'`;
+    expect(rows).toEqual([]);
+  });
 
   it('grants verified payments only once and rejects a payment reused across tenants', async () => {
     const payment = {
@@ -61,9 +85,10 @@ describe.skipIf(!url)('credit ledger against isolated PostgreSQL schema', () =>
       balance: { available: number; reserved: number };
     }>;
     expect(won.value.balance).toEqual({ available: 20, reserved: 80 });
-    const rows = await sql<
-      { reservation_id: string }[]
-    >`SELECT reservation_id FROM opt_credit_reservations WHERE tenant_id = 'a'`;
+    const rows = await sql.begin(async (tx) => {
+      await tx`SELECT set_config('app.opt_credit_tenant_id', 'a', true)`;
+      return tx<{ reservation_id: string }[]>`SELECT reservation_id FROM opt_credit_reservations`;
+    });
     const reservationId = rows[0]!.reservation_id;
     await expect(
       ledger.apply({
@@ -133,7 +158,7 @@ describe.skipIf(!url)('credit ledger against isolated PostgreSQL schema', () =>
     };
     const results = await Promise.all([ledger.apply(payment), ledger.apply(payment)]);
     expect(results.map((item) => item.replay).sort()).toEqual([false, true]);
-    const secondSql = postgres(url!, { max: 1, prepare: false, connection: { search_path: schema } });
+    const secondSql = postgres(scopedUrl, { max: 1, prepare: false, connection: { search_path: schema } });
     try {
       const replay = await new PostgresCreditLedger(secondSql).apply(payment);
       expect(replay).toEqual({ replay: true, balance: { available: 25, reserved: 0 } });
@@ -223,4 +248,20 @@ describe.skipIf(!url)('credit ledger against isolated PostgreSQL schema', () =>
       reconcileCreditPurchase(deps, 'd', 'purchase-d', { ...context, tenantId: 'other' }),
     ).rejects.toThrow(/scope/);
   });
+  it('enforces tenant filters even when a query omits its predicate', async () => {
+    const visible = await sql.begin(async (tx) => {
+      await tx`SELECT set_config('app.opt_credit_tenant_id', 'a', true)`;
+      const accounts = await tx<{ tenant_id: string }[]>`SELECT tenant_id FROM opt_credit_accounts`;
+      const purchases = await tx<{ tenant_id: string }[]>`SELECT tenant_id FROM opt_credit_purchases`;
+      return { accounts, purchases };
+    });
+    expect(visible.accounts).toEqual([{ tenant_id: 'a' }]);
+    expect(visible.purchases).toEqual([{ tenant_id: 'a' }]);
+    expect(await sql`SELECT * FROM opt_credit_accounts`).toEqual([]);
+    await expect(sql.begin(async (tx) => {
+      await tx`SELECT set_config('app.opt_credit_tenant_id', 'b', true)`;
+      await tx`INSERT INTO opt_credit_accounts (tenant_id) VALUES ('wrong-tenant')`;
+    })).rejects.toThrow();
+  });
+
 });