feat: prevent deleting storage schema rows via direct SQL query

migrations/tenant/0050-prevent-direct-deletes.sql

@@ -0,0 +1,30 @@
+-- Create a function that prevents direct deletes unless storage.allow_delete_query is set
+CREATE OR REPLACE FUNCTION storage.protect_delete()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    -- Check if storage.allow_delete_query is set to 'true'
+    IF COALESCE(current_setting('storage.allow_delete_query', true), 'false') != 'true' THEN
+        RAISE EXCEPTION 'Direct deletion from storage tables is not allowed. Use the Storage API instead.'
+            USING HINT = 'This prevents accidental data loss from orphaned objects.',
+                  ERRCODE = '42501';
+    END IF;
+    RETURN NULL;
+END;
+$$;
+
+CREATE TRIGGER protect_buckets_delete
+    BEFORE DELETE ON storage.buckets
+    FOR EACH STATEMENT
+    EXECUTE FUNCTION storage.protect_delete();
+
+CREATE TRIGGER protect_objects_delete
+    BEFORE DELETE ON storage.objects
+    FOR EACH STATEMENT
+    EXECUTE FUNCTION storage.protect_delete();
+
+CREATE TRIGGER protect_objects_delete
+    BEFORE DELETE ON storage.prefixes
+    FOR EACH STATEMENT
+    EXECUTE FUNCTION storage.protect_delete();

src/internal/database/connection.ts

@@ -134,7 +134,8 @@ export class TenantConnection {
           set_config('request.headers', ?, true),
           set_config('request.method', ?, true),
           set_config('request.path', ?, true),
-          set_config('storage.operation', ?, true);
+          set_config('storage.operation', ?, true),
+          set_config('storage.allow_delete_query', 'true', true);
     `,
       [
         this.role,

src/internal/database/migrations/types.ts

@@ -49,4 +49,5 @@ export const DBMigration = {
   'iceberg-table-metadata': 47,
   'iceberg-catalog-ids': 48,
   'buckets-objects-grants-postgres': 49,
+  'prevent-direct-deletes': 50,
 }

src/test/database-protection.test.ts

@@ -0,0 +1,199 @@
+'use strict'
+
+import { useStorage, withDeleteEnabled } from './utils/storage'
+import { DatabaseError } from 'pg'
+
+describe('Database Protection Triggers', () => {
+  const tHelper = useStorage()
+  const testBucketName = `test-db-protection-${Date.now()}`
+
+  beforeAll(async () => {
+    await tHelper.database.createBucket({
+      id: testBucketName,
+      name: testBucketName,
+    })
+  })
+
+  afterAll(async () => {
+    await tHelper.database.connection.dispose()
+  })
+
+  describe('Direct DELETE protection (migration 0050)', () => {
+    it('should prevent direct DELETE on storage.buckets without storage.allow_delete_query', async () => {
+      const db = tHelper.database.connection.pool.acquire()
+      const testBucket = `temp-bucket-${Date.now()}`
+
+      // Create a test bucket
+      await db.raw('INSERT INTO storage.buckets (id, name) VALUES (?, ?)', [testBucket, testBucket])
+
+      // Attempt to delete without setting storage.allow_delete_query
+      try {
+        await db.raw('DELETE FROM storage.buckets WHERE id = ?', [testBucket])
+        fail('Expected DELETE to be blocked by trigger')
+      } catch (error) {
+        const dbError = error as DatabaseError
+        expect(dbError.code).toBe('42501') // PostgreSQL error code for insufficient privilege
+        expect(dbError.message).toContain('Direct deletion from storage tables is not allowed')
+      }
+
+      // Verify bucket still exists
+      const result = await db.raw('SELECT id FROM storage.buckets WHERE id = ?', [testBucket])
+      expect(result.rows).toHaveLength(1)
+
+      // Cleanup: delete with proper config
+      await withDeleteEnabled(db, async (db) => {
+        await db.raw('DELETE FROM storage.buckets WHERE id = ?', [testBucket])
+      })
+    })
+
+    it('should prevent direct DELETE on storage.objects without storage.allow_delete_query', async () => {
+      const db = tHelper.database.connection.pool.acquire()
+      const testObjectName = `test-object-${Date.now()}.txt`
+
+      // Create a test object
+      await db.raw(
+        'INSERT INTO storage.objects (bucket_id, name, owner, version) VALUES (?, ?, ?, ?)',
+        [testBucketName, testObjectName, null, '1']
+      )
+
+      // Attempt to delete without setting storage.allow_delete_query
+      try {
+        await db.raw('DELETE FROM storage.objects WHERE bucket_id = ? AND name = ?', [
+          testBucketName,
+          testObjectName,
+        ])
+        fail('Expected DELETE to be blocked by trigger')
+      } catch (error) {
+        const dbError = error as DatabaseError
+        expect(dbError.code).toBe('42501')
+        expect(dbError.message).toContain('Direct deletion from storage tables is not allowed')
+      }
+
+      // Verify object still exists
+      const result = await db.raw(
+        'SELECT name FROM storage.objects WHERE bucket_id = ? AND name = ?',
+        [testBucketName, testObjectName]
+      )
+      expect(result.rows).toHaveLength(1)
+
+      // Cleanup: delete with proper config
+      await withDeleteEnabled(db, async (db) => {
+        await db.raw('DELETE FROM storage.objects WHERE bucket_id = ? AND name = ?', [
+          testBucketName,
+          testObjectName,
+        ])
+      })
+    })
+
+    it('should prevent direct DELETE on storage.prefixes without storage.allow_delete_query', async () => {
+      const db = tHelper.database.connection.pool.acquire()
+      const testPrefix = `test-prefix-${Date.now()}/`
+
+      // Create a test prefix (level is auto-generated)
+      await db.raw('INSERT INTO storage.prefixes (bucket_id, name) VALUES (?, ?)', [
+        testBucketName,
+        testPrefix,
+      ])
+
+      // Attempt to delete without setting storage.allow_delete_query
+      try {
+        await db.raw('DELETE FROM storage.prefixes WHERE bucket_id = ? AND name = ?', [
+          testBucketName,
+          testPrefix,
+        ])
+        fail('Expected DELETE to be blocked by trigger')
+      } catch (error) {
+        const dbError = error as DatabaseError
+        expect(dbError.code).toBe('42501')
+        expect(dbError.message).toContain('Direct deletion from storage tables is not allowed')
+      }
+
+      // Verify prefix still exists
+      const result = await db.raw(
+        'SELECT name FROM storage.prefixes WHERE bucket_id = ? AND name = ?',
+        [testBucketName, testPrefix]
+      )
+      expect(result.rows).toHaveLength(1)
+
+      // Cleanup: delete with proper config
+      await withDeleteEnabled(db, async (db) => {
+        await db.raw('DELETE FROM storage.prefixes WHERE bucket_id = ? AND name = ?', [
+          testBucketName,
+          testPrefix,
+        ])
+      })
+    })
+
+    it('should allow DELETE on storage.buckets when storage.allow_delete_query is set', async () => {
+      const db = tHelper.database.connection.pool.acquire()
+      const testBucket = `temp-bucket-allow-${Date.now()}`
+
+      await withDeleteEnabled(db, async (db) => {
+        // Create a test bucket
+        await db.raw('INSERT INTO storage.buckets (id, name) VALUES (?, ?)', [
+          testBucket,
+          testBucket,
+        ])
+
+        // Delete with proper config should succeed
+        await db.raw('DELETE FROM storage.buckets WHERE id = ?', [testBucket])
+
+        // Verify bucket is deleted
+        const result = await db.raw('SELECT id FROM storage.buckets WHERE id = ?', [testBucket])
+        expect(result.rows).toHaveLength(0)
+      })
+    })
+
+    it('should allow DELETE on storage.objects when storage.allow_delete_query is set', async () => {
+      const db = tHelper.database.connection.pool.acquire()
+      const testObjectName = `test-object-allow-${Date.now()}.txt`
+
+      await withDeleteEnabled(db, async (db) => {
+        // Create a test object
+        await db.raw(
+          'INSERT INTO storage.objects (bucket_id, name, owner, version) VALUES (?, ?, ?, ?)',
+          [testBucketName, testObjectName, null, '1']
+        )
+
+        // Delete with proper config should succeed
+        await db.raw('DELETE FROM storage.objects WHERE bucket_id = ? AND name = ?', [
+          testBucketName,
+          testObjectName,
+        ])
+
+        // Verify object is deleted
+        const result = await db.raw(
+          'SELECT name FROM storage.objects WHERE bucket_id = ? AND name = ?',
+          [testBucketName, testObjectName]
+        )
+        expect(result.rows).toHaveLength(0)
+      })
+    })
+
+    it('should allow DELETE on storage.prefixes when storage.allow_delete_query is set', async () => {
+      const db = tHelper.database.connection.pool.acquire()
+      const testPrefix = `test-prefix-allow-${Date.now()}/`
+
+      await withDeleteEnabled(db, async (db) => {
+        // Create a test prefix (level is auto-generated)
+        await db.raw('INSERT INTO storage.prefixes (bucket_id, name) VALUES (?, ?)', [
+          testBucketName,
+          testPrefix,
+        ])
+
+        // Delete with proper config should succeed
+        await db.raw('DELETE FROM storage.prefixes WHERE bucket_id = ? AND name = ?', [
+          testBucketName,
+          testPrefix,
+        ])
+
+        // Verify prefix is deleted
+        const result = await db.raw(
+          'SELECT name FROM storage.prefixes WHERE bucket_id = ? AND name = ?',
+          [testBucketName, testPrefix]
+        )
+        expect(result.rows).toHaveLength(0)
+      })
+    })
+  })
+})

src/test/object.test.ts

@@ -11,6 +11,7 @@ import { getServiceKeyUser, getPostgresConnection } from '@internal/database'
 import { Knex } from 'knex'
 import { ErrorCode, StorageBackendError } from '@internal/errors'
 import { FastifyInstance } from 'fastify'
+import { withDeleteEnabled } from './utils/storage'
 
 const { jwtSecret, serviceKeyAsync, tenantId } = getConfig()
 const anonKey = process.env.ANON_KEY || ''
@@ -1873,13 +1874,15 @@ describe('testing uploading with generated signed upload URL', () => {
     expect(objectResponse?.owner).toBe(owner)
 
     // remove row to not to break other tests
-    await db
-      .from<Obj>('objects')
-      .where({
-        name: OBJECT_NAME,
-        bucket_id: BUCKET_ID,
-      })
-      .delete()
+    await withDeleteEnabled(db, async (db) => {
+      await db
+        .from<Obj>('objects')
+        .where({
+          name: OBJECT_NAME,
+          bucket_id: BUCKET_ID,
+        })
+        .delete()
+    })
   })
 
   test('upload object without a token', async () => {