Agent Zero
104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
"""Simplify RLS to pure tenant isolation.
|
|
|
|
Per architecture review: RLS should be the "safety belt" (tenant isolation only),
|
|
NOT the "vehicle control" (business authorization). Business authorization
|
|
(owner_id, sharing, entity_permissions) belongs in the application layer
|
|
(visibility.py with Defense-in-Depth tenant_id filter).
|
|
|
|
Revision ID: 0069
|
|
Revises: 0068
|
|
"""
|
|
|
|
from alembic import op
|
|
from sqlalchemy import text
|
|
|
|
revision = "0069"
|
|
down_revision = "0068"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
RLS_TABLES = [
|
|
"contacts", "addresses", "attachments", "bank_accounts",
|
|
"contact_folders", "contact_folder_permissions", "entity_permissions",
|
|
"entity_policies", "event_outbox", "audit_log", "notifications",
|
|
"saved_filters", "saved_views", "webhooks", "workflow_instances",
|
|
"workflow_step_history", "sequences", "custom_field_definitions",
|
|
"custom_field_values", "guest_users", "guest_invitations",
|
|
"consumer_inbox", "tenant_plugin_activation", "permission_templates",
|
|
"permission_delegations", "dms_files", "dms_folders",
|
|
"calendar_events", "calendars", "tasks", "task_lists",
|
|
"messages", "channels", "entity_links", "tags", "tag_assignments",
|
|
"mail_accounts", "mail_messages", "mail_folders",
|
|
"report_templates", "report_generations", "ai_conversations",
|
|
"ai_messages", "automation_workflows", "automation_runs",
|
|
"mcp_server_configs", "mcp_client_configs", "system_notifications",
|
|
]
|
|
|
|
CONTACTS_POLICIES_TO_DROP = [
|
|
"contacts_admin_select", "contacts_owner_select",
|
|
"contacts_shared_select", "contacts_tenant_owned_select",
|
|
"contacts_delete_policy", "contacts_insert_policy",
|
|
"contacts_update_policy",
|
|
]
|
|
|
|
|
|
def upgrade() -> None:
|
|
conn = op.get_bind()
|
|
|
|
# 1. Drop all business-logic RLS policies on contacts
|
|
for policy in CONTACTS_POLICIES_TO_DROP:
|
|
op.execute(f"DROP POLICY IF EXISTS {policy} ON contacts")
|
|
|
|
# 2. Drop old tenant_isolation policy on contacts
|
|
op.execute("DROP POLICY IF EXISTS contacts_tenant_isolation ON contacts")
|
|
|
|
# 3. Create simple tenant isolation for ALL operations on contacts
|
|
op.execute(
|
|
"CREATE POLICY contacts_tenant_isolation ON contacts "
|
|
"FOR ALL "
|
|
"USING (tenant_id = current_setting('app.current_tenant_id', true)::uuid) "
|
|
"WITH CHECK (tenant_id = current_setting('app.current_tenant_id', true)::uuid)"
|
|
)
|
|
|
|
# 4. For all other RLS tables: drop existing policies, create simple tenant isolation
|
|
for table in RLS_TABLES:
|
|
if table == "contacts":
|
|
continue
|
|
|
|
# Check if table exists first
|
|
table_exists = conn.execute(
|
|
text(f"SELECT 1 FROM information_schema.tables WHERE table_name = '{table}'")
|
|
).fetchone() is not None
|
|
|
|
if not table_exists:
|
|
continue
|
|
|
|
# Get all existing policies on this table
|
|
result = conn.execute(
|
|
text(f"SELECT polname FROM pg_policy WHERE polrelid = '{table}'::regclass")
|
|
)
|
|
policies = [row[0] for row in result]
|
|
|
|
# Drop each policy
|
|
for policy in policies:
|
|
op.execute(f'DROP POLICY IF EXISTS "{policy}" ON {table}')
|
|
|
|
# Check if table has tenant_id column
|
|
col_result = conn.execute(
|
|
text(f"SELECT 1 FROM information_schema.columns "
|
|
f"WHERE table_name = '{table}' AND column_name = 'tenant_id'")
|
|
)
|
|
has_tenant_id = col_result.fetchone() is not None
|
|
|
|
if has_tenant_id:
|
|
op.execute(
|
|
f"CREATE POLICY {table}_tenant_isolation ON {table} "
|
|
"FOR ALL "
|
|
"USING (tenant_id = current_setting('app.current_tenant_id', true)::uuid) "
|
|
"WITH CHECK (tenant_id = current_setting('app.current_tenant_id', true)::uuid)"
|
|
)
|
|
|
|
|
|
def downgrade() -> None:
|
|
pass
|