From 9b7597e1694f868350f6b4ab2d85d34c773a9c1f Mon Sep 17 00:00:00 2001 From: Gwyneth Pena-Siguenza Date: Tue, 26 May 2026 02:00:42 +0000 Subject: [PATCH] Fix verification database permissions Grant the verification Functions role the narrow requirements columns needed by the typed submission trigger. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../0038_grant_fn_requirement_kind_lookup.py | 69 +++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100644 api/alembic/versions/0038_grant_fn_requirement_kind_lookup.py diff --git a/api/alembic/versions/0038_grant_fn_requirement_kind_lookup.py b/api/alembic/versions/0038_grant_fn_requirement_kind_lookup.py new file mode 100644 index 00000000..bd168b4c --- /dev/null +++ b/api/alembic/versions/0038_grant_fn_requirement_kind_lookup.py @@ -0,0 +1,69 @@ +"""grant Functions role requirement kind lookup + +Revision ID: 0038_grant_fn_requirement_kind_lookup +Revises: 0037_validate_typed_submitted_value_constraints +Create Date: 2026-05-26 +""" + +import os +from collections.abc import Sequence + +from alembic import op + +revision: str = "0038_grant_fn_requirement_kind_lookup" +down_revision: str | None = "0037_validate_typed_submitted_value_constraints" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_REQUIRED_COLUMNS = "uuid, submission_value_kind" + + +def upgrade() -> None: + role = _verification_functions_role() + if not role: + return + + op.execute( + f""" + DO $$ + BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{role}') THEN + GRANT SELECT ({_REQUIRED_COLUMNS}) + ON requirements + TO "{role}"; + END IF; + END $$; + """ + ) + + +def downgrade() -> None: + role = _verification_functions_role() + if not role: + return + + op.execute( + f""" + DO $$ + BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{role}') THEN + REVOKE SELECT ({_REQUIRED_COLUMNS}) + ON requirements + FROM "{role}"; + END IF; + END $$; + """ + ) + + +def _verification_functions_role() -> str | None: + role = os.environ.get("POSTGRES_VERIFICATION_FUNCTIONS_ROLE") + if not role: + return None + if not (role[0].isalpha() or role[0] == "_") or not all( + c.isalnum() or c == "_" for c in role + ): + raise RuntimeError( + f"POSTGRES_VERIFICATION_FUNCTIONS_ROLE is not a valid identifier: {role!r}" + ) + return role