[Postgres] Add checks for RLS affecting replication

.changeset/eight-turtles-design.md

@@ -0,0 +1,6 @@
+---
+'@powersync/service-core': patch
+'@powersync/service-image': patch
+---
+
+Add checks for RLS affecting replication.

.changeset/mean-foxes-hug.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-jpgwire': patch
+---
+
+Add types to pgwireRows.

.changeset/ten-readers-happen.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-module-postgres': minor
+---
+
+Add checks for RLS affecting replication.

modules/module-postgres/src/replication/WalStream.ts

@@ -15,7 +15,7 @@ import * as pg_utils from '../utils/pgwire_utils.js';
 
 import { PgManager } from './PgManager.js';
 import { getPgOutputRelation, getRelId } from './PgRelation.js';
-import { checkSourceConfiguration, getReplicationIdentityColumns } from './replication-utils.js';
+import { checkSourceConfiguration, checkTableRls, getReplicationIdentityColumns } from './replication-utils.js';
 import { ReplicationMetric } from '@powersync/service-types';
 
 export interface WalStreamOptions {
@@ -198,6 +198,17 @@ export class WalStream {
         continue;
       }
 
+      try {
+        const result = await checkTableRls(db, relid);
+        if (!result.canRead) {
+          // We log the message, then continue anyway, since the check does not cover all cases.
+          logger.warn(result.message!);
+        }
+      } catch (e) {
+        // It's possible that we just don't have permission to access pg_roles - log the error and continue.
+        logger.warn(`Could not check RLS access for ${tablePattern.schema}.${name}`, e);
+      }
+
       const cresult = await getReplicationIdentityColumns(db, relid);
 
       const table = await this.handleRelation(

modules/module-postgres/src/replication/replication-utils.ts

@@ -1,7 +1,7 @@
 import * as pgwire from '@powersync/service-jpgwire';
 
 import * as lib_postgres from '@powersync/lib-service-postgres';
-import { ErrorCode, logger, ServiceError } from '@powersync/lib-services-framework';
+import { ErrorCode, logger, ServiceAssertionError, ServiceError } from '@powersync/lib-services-framework';
 import { PatternResult, storage } from '@powersync/service-core';
 import * as sync_rules from '@powersync/service-sync-rules';
 import * as service_types from '@powersync/service-types';
@@ -136,6 +136,61 @@ $$ LANGUAGE plpgsql;`
   }
 }
 
+export async function checkTableRls(
+  db: pgwire.PgClient,
+  relationId: number
+): Promise<{ canRead: boolean; message?: string }> {
+  const rs = await lib_postgres.retriedQuery(db, {
+    statement: `
+WITH user_info AS (
+    SELECT 
+        current_user as username,
+        r.rolsuper,
+        r.rolbypassrls
+    FROM pg_roles r 
+    WHERE r.rolname = current_user
+)
+SELECT
+    c.relname as tablename,
+    c.relrowsecurity as rls_enabled,
+    u.username as username,
+    u.rolsuper as is_superuser,
+    u.rolbypassrls as bypasses_rls
+FROM pg_class c
+CROSS JOIN user_info u
+WHERE c.oid = $1::oid;
+`,
+    params: [{ type: 'int4', value: relationId }]
+  });
+
+  const rows = pgwire.pgwireRows<{
+    rls_enabled: boolean;
+    tablename: string;
+    username: string;
+    is_superuser: boolean;
+    bypasses_rls: boolean;
+  }>(rs);
+  if (rows.length == 0) {
+    // Not expected, since we already got the oid
+    throw new ServiceAssertionError(`Table with OID ${relationId} does not exist.`);
+  }
+  const row = rows[0];
+  if (row.is_superuser || row.bypasses_rls) {
+    // Bypasses RLS automatically.
+    return { canRead: true };
+  }
+
+  if (row.rls_enabled) {
+    // Don't skip, since we _may_ still be able to get results.
+    return {
+      canRead: false,
+      message: `[${ErrorCode.PSYNC_S1145}] Row Level Security is enabled on table "${row.tablename}". To make sure that ${row.username} can read the table, run: 'ALTER ROLE ${row.username} BYPASSRLS'.`
+    };
+  }
+
+  return { canRead: true };
+}
+
 export interface GetDebugTablesInfoOptions {
   db: pgwire.PgClient;
   publicationName: string;
@@ -309,14 +364,17 @@ export async function getDebugTableInfo(options: GetDebugTableInfoOptions): Prom
     };
   }
 
+  const rlsCheck = await checkTableRls(db, relationId);
+  const rlsError = rlsCheck.canRead ? null : { message: rlsCheck.message!, level: 'warning' };
+
   return {
     schema: schema,
     name: name,
     pattern: tablePattern.isWildcard ? tablePattern.tablePattern : undefined,
     replication_id: id_columns.map((c) => c.name),
     data_queries: syncData,
     parameter_queries: syncParameters,
-    errors: [id_columns_error, selectError, replicateError].filter(
+    errors: [id_columns_error, selectError, replicateError, rlsError].filter(
       (error) => error != null
     ) as service_types.ReplicationError[]
   };

packages/jpgwire/src/util.ts

@@ -300,13 +300,13 @@ export function dateToSqlite(source?: string) {
  *
  * This converts it to objects.
  */
-export function pgwireRows(rs: pgwire.PgResult): Record<string, any>[] {
+export function pgwireRows<T = Record<string, any>>(rs: pgwire.PgResult): T[] {
   const columns = rs.columns;
   return rs.rows.map((row) => {
-    let r: Record<string, any> = {};
+    let r: T = {} as any;
     for (let i = 0; i < columns.length; i++) {
       const c = columns[i];
-      r[c.name] = row[i];
+      (r as any)[c.name] = row[i];
     }
     return r;
   });

packages/service-errors/src/codes.ts

@@ -168,6 +168,18 @@ export enum ErrorCode {
    */
   PSYNC_S1144 = 'PSYNC_S1144',
 
+  /**
+   * Table has RLS enabled, but the replication role does not have the BYPASSRLS attribute.
+   *
+   * We recommend using a dedicated replication role with the BYPASSRLS attribute for replication:
+   *
+   *     ALTER ROLE powersync_role BYPASSRLS
+   *
+   * An alternative is to create explicit policies for the replication role. If you have done that,
+   * you may ignore this warning.
+   */
+  PSYNC_S1145 = 'PSYNC_S1145',
+
   // ## PSYNC_S12xx: MySQL replication issues
 
   // ## PSYNC_S13xx: MongoDB replication issues