fix(security): RLS für special_assignments nachziehen (DSGVO)
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
This commit is contained in:
@@ -0,0 +1,44 @@
|
||||
"""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")
|
||||
Reference in New Issue
Block a user