fix(security): enforce defensive RLS for agent chat tables

Close Supabase advisor findings by enabling RLS and deny-by-default policies on new public agent-chat tables. Clarify backend RLS governance and incident runbook steps to prevent config-drift regressions.
2026-02-25 18:04:05 +08:00
parent a88e42babd
commit 443977be9b
3 changed files with 107 additions and 10 deletions
@@ -0,0 +1,58 @@
+"""enable_rls_for_agent_chat_tables
+
+Revision ID: 20260226_agent_chat_rls
+Revises: 20260226_agent_chat_core
+Create Date: 2026-02-26 18:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+
+revision: str = "20260226_agent_chat_rls"
+down_revision: Union[str, Sequence[str], None] = "20260226_agent_chat_core"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+TABLES = ("llm_factory", "llms", "sessions", "messages")
+
+
+def _enable_rls_and_deny_public(table: str) -> None:
+    op.execute(f"ALTER TABLE public.{table} ENABLE ROW LEVEL SECURITY")
+    op.execute(
+        f"CREATE POLICY {table}_deny_public_select ON public.{table} "
+        "FOR SELECT TO anon, authenticated USING (false)"
+    )
+    op.execute(
+        f"CREATE POLICY {table}_deny_public_insert ON public.{table} "
+        "FOR INSERT TO anon, authenticated WITH CHECK (false)"
+    )
+    op.execute(
+        f"CREATE POLICY {table}_deny_public_update ON public.{table} "
+        "FOR UPDATE TO anon, authenticated USING (false) WITH CHECK (false)"
+    )
+    op.execute(
+        f"CREATE POLICY {table}_deny_public_delete ON public.{table} "
+        "FOR DELETE TO anon, authenticated USING (false)"
+    )
+
+
+def _disable_rls_and_drop_policies(table: str) -> None:
+    op.execute(f"DROP POLICY IF EXISTS {table}_deny_public_select ON public.{table}")
+    op.execute(f"DROP POLICY IF EXISTS {table}_deny_public_insert ON public.{table}")
+    op.execute(f"DROP POLICY IF EXISTS {table}_deny_public_update ON public.{table}")
+    op.execute(f"DROP POLICY IF EXISTS {table}_deny_public_delete ON public.{table}")
+    op.execute(f"ALTER TABLE public.{table} DISABLE ROW LEVEL SECURITY")
+
+
+def upgrade() -> None:
+    for table in TABLES:
+        _enable_rls_and_deny_public(table)
+
+
+def downgrade() -> None:
+    for table in reversed(TABLES):
+        _disable_rls_and_drop_policies(table)