patrickandClaude Sonnet 5
456ca1526b
special_assignments (Migration 0029) hatte keine Row-Level-Security in Produktion, obwohl company_id/user_id geführt werden - Tests liefen trotzdem grün, da conftest.py die Policy bereits simuliert hatte (gefunden durch postgres-expert-Review). Analog 0039 (hours_payouts) nachgezogen. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Gis16MnuwkYcivLrSxK1pD
45 lines
1.9 KiB
Python
45 lines
1.9 KiB
Python
"""RLS für special_assignments nachziehen (Mandantentrennung / DSGVO)
|
|
|
|
Revision ID: 0043
|
|
Revises: 0042
|
|
Create Date: 2026-08-27
|
|
|
|
special_assignments (Migration 0029) hatte wie zuvor hours_payouts (0039) keine
|
|
Row-Level-Security, obwohl die Tabelle company_id + user_id führt. conftest.py
|
|
hatte die Policy testseitig bereits simuliert, wodurch die Lücke in der
|
|
Produktions-DB unbemerkt blieb. Hier nachgezogen, analog 0039.
|
|
"""
|
|
from alembic import op
|
|
from sqlalchemy import text
|
|
|
|
revision = "0043"
|
|
down_revision = "0042"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
_BYPASS = "COALESCE(current_setting('app.bypass_rls', true), 'off') = 'on'"
|
|
_CID = "company_id = NULLIF(current_setting('app.company_id', true), '')::uuid"
|
|
_USING = f"({_BYPASS} OR {_CID})"
|
|
|
|
|
|
def _exec(sql: str) -> None:
|
|
op.execute(text(sql))
|
|
|
|
|
|
def upgrade() -> None:
|
|
_exec("ALTER TABLE special_assignments ENABLE ROW LEVEL SECURITY")
|
|
_exec("ALTER TABLE special_assignments FORCE ROW LEVEL SECURITY")
|
|
for cmd in ("select", "insert", "update", "delete"):
|
|
_exec(f"DROP POLICY IF EXISTS rls_special_assignments_{cmd} ON special_assignments")
|
|
_exec(f"CREATE POLICY rls_special_assignments_select ON special_assignments FOR SELECT USING {_USING}")
|
|
_exec(f"CREATE POLICY rls_special_assignments_insert ON special_assignments FOR INSERT WITH CHECK {_USING}")
|
|
_exec(f"CREATE POLICY rls_special_assignments_update ON special_assignments FOR UPDATE USING {_USING} WITH CHECK {_USING}")
|
|
_exec(f"CREATE POLICY rls_special_assignments_delete ON special_assignments FOR DELETE USING {_USING}")
|
|
|
|
|
|
def downgrade() -> None:
|
|
for cmd in ("select", "insert", "update", "delete"):
|
|
_exec(f"DROP POLICY IF EXISTS rls_special_assignments_{cmd} ON special_assignments")
|
|
_exec("ALTER TABLE special_assignments NO FORCE ROW LEVEL SECURITY")
|
|
_exec("ALTER TABLE special_assignments DISABLE ROW LEVEL SECURITY")
|