diff --git a/.github/workflows/seed-v2-database.yml b/.github/workflows/seed-v2-database.yml index 6a9206846..634352775 100644 --- a/.github/workflows/seed-v2-database.yml +++ b/.github/workflows/seed-v2-database.yml @@ -54,6 +54,14 @@ jobs: run: bash .github/scripts/migrate_v2_metadata_schema.sh env: V2_MIGRATION_DATABASE_URL: ${{ secrets.V2_MIGRATION_DATABASE_URL }} + - name: Grant declared v2 table privileges + run: uv run python scripts/reconcile_v2_privileges.py apply + env: + V2_MIGRATION_DATABASE_URL: ${{ secrets.V2_MIGRATION_DATABASE_URL }} + - name: Verify declared v2 table privileges + run: uv run python scripts/reconcile_v2_privileges.py verify + env: + V2_MIGRATION_DATABASE_URL: ${{ secrets.V2_MIGRATION_DATABASE_URL }} - name: Seed and validate the v2 metadata catalog run: uv run python scripts/publish_v2_metadata_catalog.py env: diff --git a/.github/workflows/v2-integration-check.yml b/.github/workflows/v2-integration-check.yml index 6d41bb5c1..3f402ae80 100644 --- a/.github/workflows/v2-integration-check.yml +++ b/.github/workflows/v2-integration-check.yml @@ -51,6 +51,8 @@ jobs: ALEMBIC_DATABASE_URL: mysql+pymysql://root:policyengine_test@127.0.0.1:3306/policyengine_alembic_test V2_MIGRATION_DATABASE_URL: postgresql+psycopg://postgres:policyengine_v2_test@127.0.0.1:5432/policyengine_v2_alembic_test V2_ALEMBIC_DISPOSABLE_TEST: "1" + # Names the Stage 12 role in the disposable database: policyengine_stage12_test. + DEPLOYMENT_ENVIRONMENT: test RUNTIME_CACHE_TEST_URL: redis://127.0.0.1:6379/0 steps: - name: Checkout repo @@ -67,6 +69,10 @@ jobs: run: uv run alembic -c alembic-v1.ini upgrade head - name: Prepare the disposable v2 schema run: uv run alembic -c alembic-v2.ini upgrade head + - name: Create the managed v2 roles and grant their declared privileges + run: uv run python scripts/reconcile_v2_privileges.py apply --create-missing-roles + - name: Verify the managed v2 roles hold their declared privileges + run: uv run python scripts/reconcile_v2_privileges.py verify - name: Verify the installed PolicyEngine.py catalog interface run: uv run coverage run --branch -m pytest -q tests/integration/test_v2_catalog_installed.py env: @@ -77,6 +83,8 @@ jobs: run: uv run coverage run -a --branch -m pytest -q tests/integration/test_v2_policy_persistence.py tests/integration/test_v1_policy_dual_write.py tests/integration/test_mysql_policy_dual_write.py tests/integration/test_v2_user_policy_mirroring.py tests/integration/test_v1_user_policy_dual_write.py - name: Test signed-in identity provisioning and user associations run: uv run coverage run -a --branch -m pytest -q tests/integration/test_v2_user_identity_persistence.py + - name: Test comparison-run persistence and privilege reconciliation as the managed roles + run: uv run coverage run -a --branch -m pytest -q tests/integration/test_v2_stage12_comparison_persistence.py tests/integration/test_v2_privilege_reconciliation.py - name: Test v2 household persistence and immediate v1 copying run: uv run coverage run -a --branch -m pytest -q tests/integration/test_v2_household_persistence.py tests/integration/test_v1_household_dual_write.py tests/integration/test_mysql_household_spm_dual_write.py - name: Qualify production-scale v2 metadata publication diff --git a/AGENTS.md b/AGENTS.md index f68e940c5..fd978b11c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -91,7 +91,7 @@ To add or change a rule, read `docs/engineering/skills/rules-registry.md`. ### Migration and schema -- Checked: `api.migration.alembic-v1-qualified`, `api.migration.alembic-v2-qualified`, `api.migration.bundle-supported-by-simulation-api`, `api.migration.contracts-consistent`, `api.migration.cross-database-integration` +- Checked: `api.migration.alembic-v1-qualified`, `api.migration.alembic-v2-qualified`, `api.migration.bundle-supported-by-simulation-api`, `api.migration.contracts-consistent`, `api.migration.cross-database-integration`, `api.v2.deployed-privileges-match`, `api.v2.runtime-privileges-declared`, `api.v2.runtime-privileges-tested-as-runtime-role` - `api.migration.alembic-autogenerated-only`: Alembic revisions are autogenerated against an explicitly selected v1 or v2 configuration; never hand-written DDL or data operations. (doc-only, not checked: follow it yourself) ### Code quality and types diff --git a/changelog.d/v2-table-privileges.added.md b/changelog.d/v2-table-privileges.added.md new file mode 100644 index 000000000..5100bd3cc --- /dev/null +++ b/changelog.d/v2-table-privileges.added.md @@ -0,0 +1 @@ +Declare the v2 database roles' table privileges in the repository (`policyengine_api/data/v2/privileges.py`), grant them after every v2 schema upgrade and check them before each deploy with `scripts/reconcile_v2_privileges.py`, and run the cross-database integration tests as those roles instead of the database superuser. diff --git a/changelog.d/v2-table-privileges.fixed.md b/changelog.d/v2-table-privileges.fixed.md new file mode 100644 index 000000000..c6ca83fc0 --- /dev/null +++ b/changelog.d/v2-table-privileges.fixed.md @@ -0,0 +1 @@ +The first deploy grants the Public API's production runtime role the privileges its v2 code needs and production lacked: INSERT on `policies`, `households` and `parameter_values`, and INSERT and UPDATE on `legacy_policy_mappings` and `legacy_household_mappings`. It also records in the repository the SELECT grants applied by hand to fix failed reform saves (#3977). diff --git a/docs/engineering/skills/alembic-migrations.md b/docs/engineering/skills/alembic-migrations.md index b8c79484b..d4a043d91 100644 --- a/docs/engineering/skills/alembic-migrations.md +++ b/docs/engineering/skills/alembic-migrations.md @@ -126,6 +126,68 @@ population tables. Review every autogenerated table change, and compare the live schema directly with the complete SQLModel metadata during the migration lifecycle. +## Table privileges + +The table privileges of the managed v2 database roles are declared in +`policyengine_api/data/v2/privileges.py`, not in Alembic revisions: revisions +stay generated-only, the simulation API's role name differs per environment, +and a revision runs once per database, so it can't repair drift. + +| Role | Used by | +| --- | --- | +| `policyengine_v2_runtime` | The Public API on Cloud Run | +| `policyengine_v2_catalog_writer` | `scripts/publish_v2_metadata_catalog.py` | +| `policyengine_stage12_` | The simulation API and its Modal worker, on the temporary comparison-run tables | + +Every v2 table has an entry for every role, so "no access" is a decision, not +an omission. The migration role owns every table and isn't listed. Three rules +check the declarations: + +- `api.v2.runtime-privileges-declared`: a unit test reads each connector + module and fails when it needs a privilege its role isn't declared to hold. +- `api.v2.runtime-privileges-tested-as-runtime-role`: the cross-database + integration tests run the v2 user services as `policyengine_v2_runtime` and + the comparison-run services as `policyengine_stage12_test`, holding only + their declared privileges. +- `api.v2.deployed-privileges-match`: every deploy runs + `scripts/reconcile_v2_privileges.py apply` right after the schema upgrade, + then `verify`, before the catalog is seeded. + +`apply` grants each declared privilege a role lacks. It never revokes: it +reports privileges the file doesn't declare, and the projects' default +privileges, in the job summary. It grants nothing when a declared role or +table is missing. The roles themselves are provisioned outside the +repository; `--create-missing-roles` is accepted only for the disposable test +database. + +The declarations include write privileges for code paths production doesn't +turn on: the dual-write copies into `legacy_policy_mappings` and +`legacy_household_mappings`, and the writes behind the native `/v2/policies` +and `/v2/households` routes. Narrow them when those tables are removed. + +When a pull request adds or changes a v2 table, or a connector's use of one: + +1. Update the table's entry in `privileges.py` for every role. +2. Run `make test`, which includes `tests/unit/v2/test_privilege_manifest.py`. +3. Run the integration tests as the managed roles against disposable + PostgreSQL: + + ```bash + export V2_MIGRATION_DATABASE_URL="postgresql+psycopg://.../policyengine_v2_alembic_test" \ + V2_ALEMBIC_DISPOSABLE_TEST=1 DEPLOYMENT_ENVIRONMENT=test + uv run alembic -c alembic-v2.ini upgrade head + uv run python scripts/reconcile_v2_privileges.py apply --create-missing-roles + uv run python scripts/reconcile_v2_privileges.py verify + uv run pytest -q tests/integration/test_v2_user_identity_persistence.py \ + tests/integration/test_v2_stage12_comparison_persistence.py \ + tests/integration/test_v2_privilege_reconciliation.py + ``` + +Never grant privileges by hand in staging or production; they reach those +databases only through the deploy step, after review. The unit test reads +model names in connector modules, so it misses raw `text()` SQL and SQL issued +outside `database_connectors/`; the integration tests catch those. + ## V2 application-data migrations Alembic is also the sole authority for intentional migrations of retained v2 diff --git a/docs/engineering/skills/v2-code-organization.md b/docs/engineering/skills/v2-code-organization.md index 5c60b2caa..bbe997b65 100644 --- a/docs/engineering/skills/v2-code-organization.md +++ b/docs/engineering/skills/v2-code-organization.md @@ -132,6 +132,15 @@ request-level or business validation, or convert rows into public response types. Connector modules do not call one another to sequence a workflow; the service makes that ordering explicit. +The module name also decides which table privileges the connector needs +(rule `api.v2.runtime-privileges-declared`): `reads` modules need SELECT; +`creates`, `updates` and `deletes` modules need INSERT, UPDATE or DELETE, plus +SELECT; and a function that locks rows with `with_for_update` needs UPDATE on +the tables it locks. A connector that starts using a table, or a new kind of +operation on one, needs that privilege declared in +`policyengine_api/data/v2/privileges.py` in the same pull request; see +[Table privileges](alembic-migrations.md#table-privileges). + Do not use `repository` as a generic synonym for SQL access. Reserve that term for an intentional Repository-pattern abstraction with a stable interface that hides interchangeable persistence implementations. API v2 currently uses diff --git a/docs/generated/rules.md b/docs/generated/rules.md index 60db81a67..af648867e 100644 --- a/docs/generated/rules.md +++ b/docs/generated/rules.md @@ -8,11 +8,11 @@ Every rule that governs `policyengine-api` is listed here with how it is enforce | Tier | enforced | ratcheted | advisory | doc-only | Total | | --- | ---: | ---: | ---: | ---: | ---: | -| mechanical | 47 | 0 | 1 | 12 | 60 | +| mechanical | 50 | 0 | 1 | 12 | 63 | | metric | 0 | 0 | 0 | 0 | 0 | | judgment | 0 | 0 | 0 | 1 | 1 | -**Blocking:** 47 of 61 rules. +**Blocking:** 50 of 64 rules. ## Gaps @@ -232,6 +232,9 @@ flowchart LR | `api.migration.bundle-supported-by-simulation-api` | A PolicyEngine bundle version change is supported by the simulation API before it merges. | mechanical | enforced | yes | CI job `ensure-policyengine-bundle-supported-by-simulation-api` | `pr.yml:ensure-policyengine-bundle-supported-by-simulation-api` | — | | `api.migration.contracts-consistent` | Migration route groups and workflow contracts are internally consistent, and their generated docs are current. | mechanical | enforced | yes | quality guard `migration-contracts` | `pr.yml:quality-guards` | **unproven** | | `api.migration.cross-database-integration` | Cross-database integration tests pass against real MySQL and Postgres. | mechanical | enforced | yes | CI job `v2-integration-check` | `pr.yml:v2-integration-check` | — | +| `api.v2.deployed-privileges-match` | Every v2 deploy grants the managed roles each privilege policyengine_api/data/v2/privileges.py declares right after the schema upgrade, reports privileges it doesn't declare without revoking them, and stops before the catalog is seeded if a declared privilege is still missing. | mechanical | enforced | yes | CI job `seed`, `catalog-runtime-integration` | `seed-v2-database.yml:seed`, `v2-integration-check.yml:catalog-runtime-integration` | yes | +| `api.v2.runtime-privileges-declared` | Every v2 table has an entry for each managed database role in policyengine_api/data/v2/privileges.py, and every privilege a v2 connector module needs (from its reads, creates, updates or deletes module name and its row locks) is declared for the role that runs it. | mechanical | enforced | yes | pytest `tests/unit/v2/test_privilege_manifest.py` | `pr.yml:test` | yes | +| `api.v2.runtime-privileges-tested-as-runtime-role` | The cross-database integration tests run the v2 user services as policyengine_v2_runtime and the comparison-run services as policyengine_stage12_, holding only their declared privileges, so a statement those privileges don't allow fails CI. | mechanical | enforced | yes | CI job `v2-integration-check`, `catalog-runtime-integration` | `pr.yml:v2-integration-check`, `v2-integration-check.yml:catalog-runtime-integration` | — | ```mermaid flowchart LR @@ -252,6 +255,15 @@ flowchart LR r_api_migration_cross_database_integration("api.migration.cross-database-integration (enforced)") e_CI_job_v2_integration_check["CI job v2-integration-check"] j_CI_pr_yml_v2_integration_check["CI pr.yml:v2-integration-check"] + r_api_v2_deployed_privileges_match("api.v2.deployed-privileges-match (enforced)") + e_CI_job_seed__catalog_runtime_integration["CI job seed, catalog-runtime-integration"] + j_CI_seed_v2_database_yml_seed["CI seed-v2-database.yml:seed"] + j_CI_v2_integration_check_yml_catalog_runtime_integration["CI v2-integration-check.yml:catalog-runtime-integration"] + r_api_v2_runtime_privileges_declared("api.v2.runtime-privileges-declared (enforced)") + e_pytest_tests_unit_v2_test_privilege_manifest_py["pytest tests/unit/v2/test_privilege_manifest.py"] + j_CI_pr_yml_test["CI pr.yml:test"] + r_api_v2_runtime_privileges_tested_as_runtime_role("api.v2.runtime-privileges-tested-as-runtime-role (enforced)") + e_CI_job_v2_integration_check__catalog_runtime_integration["CI job v2-integration-check, catalog-runtime-integration"] r_api_migration_alembic_autogenerated_only -.-> gap_unenforced r_api_migration_alembic_v1_qualified --> e_CI_job_alembic_v1_check e_CI_job_alembic_v1_check --> j_CI_pr_yml_alembic_v1_check @@ -263,6 +275,14 @@ flowchart LR e_quality_guard_migration_contracts --> j_CI_pr_yml_quality_guards r_api_migration_cross_database_integration --> e_CI_job_v2_integration_check e_CI_job_v2_integration_check --> j_CI_pr_yml_v2_integration_check + r_api_v2_deployed_privileges_match --> e_CI_job_seed__catalog_runtime_integration + e_CI_job_seed__catalog_runtime_integration --> j_CI_seed_v2_database_yml_seed + e_CI_job_seed__catalog_runtime_integration --> j_CI_v2_integration_check_yml_catalog_runtime_integration + r_api_v2_runtime_privileges_declared --> e_pytest_tests_unit_v2_test_privilege_manifest_py + e_pytest_tests_unit_v2_test_privilege_manifest_py --> j_CI_pr_yml_test + r_api_v2_runtime_privileges_tested_as_runtime_role --> e_CI_job_v2_integration_check__catalog_runtime_integration + e_CI_job_v2_integration_check__catalog_runtime_integration --> j_CI_pr_yml_v2_integration_check + e_CI_job_v2_integration_check__catalog_runtime_integration --> j_CI_v2_integration_check_yml_catalog_runtime_integration ``` ## Code quality and types @@ -271,7 +291,7 @@ flowchart LR | --- | --- | --- | --- | --- | --- | --- | --- | | `api.quality.alembic-revisions-unused-imports` | Ruff doesn't lint Alembic revision files (lint.exclude in pyproject.toml), because they must stay exactly as Alembic generates them. | mechanical | advisory | no | nothing | — | — | | `api.quality.ci-cd-scripts-ruff-exemptions` | CI/CD scripts under .github may print their results and fetch fixed URLs (T201, S310); every other ruff rule applies to them. | mechanical | enforced | yes | ruff per-file ignore of S310, T201 for `.github/**` | `pr.yml:lint` | — | -| `api.quality.command-line-modules-ruff-exemptions` | Application modules that also run as command-line programs (if __name__ == "__main__") may print their output (T201); every other ruff rule applies to them. | mechanical | enforced | yes | ruff per-file ignore of T201 for `policyengine_api/{data/v2/catalog/initialization,data/v2/household_migration_qualification,data/v2/policy_migration_qualification,data/v2/user_association_migration_qualification,worker_spm_release}.py` | `pr.yml:lint` | — | +| `api.quality.command-line-modules-ruff-exemptions` | Application modules that also run as command-line programs (if __name__ == "__main__") may print their output (T201); every other ruff rule applies to them. | mechanical | enforced | yes | ruff per-file ignore of T201 for `policyengine_api/{data/v2/catalog/initialization,data/v2/household_migration_qualification,data/v2/policy_migration_qualification,data/v2/privilege_reconciliation,data/v2/user_association_migration_qualification,worker_spm_release}.py` | `pr.yml:lint` | — | | `api.quality.command-line-scripts-ruff-exemptions` | Command-line scripts under scripts/ may print their output, run other programs, fetch fixed URLs, stamp today's date and pick random samples (T201, S603, S607, S310, DTZ011, S311); every other ruff rule applies to them. | mechanical | enforced | yes | ruff per-file ignore of DTZ011, S310, S311, S603, S607, T201 for `scripts/**` | `pr.yml:lint` | — | | `api.quality.environment-variables-load` | Required environment variables and credentials load in a deployed-like environment. | mechanical | enforced | yes | CI job `test_env_vars` | `pr.yml:test_env_vars` | — | | `api.quality.mypy-v2` | Shared query models, FastAPI dependencies, v2 routes and services, and the user sign-in and account-deletion libraries pass mypy with untyped definitions disallowed. | mechanical | enforced | yes | mypy | `pr.yml:typecheck-v2` | — | @@ -289,7 +309,7 @@ flowchart LR e_ruff_per_file_ignore_of_S310__T201_for__github___["ruff per-file ignore of S310, T201 for .github/**"] j_CI_pr_yml_lint["CI pr.yml:lint"] r_api_quality_command_line_modules_ruff_exemptions("api.quality.command-line-modules-ruff-exemptions (enforced)") - e_ruff_per_file_ignore_of_T201_for_policyengine_api__data_v2_catalog_initialization_data_v2_household_migration_qualification_data_v2_policy_migration_qualification_data_v2_user_association_migration_qualification_worker_spm_release__py["ruff per-file ignore of T201 for policyengine_api/{data/v2/catalog/initialization,data/v2/household_migration_qualification,data/v2/policy_migration_qualification,data/v2/user_association_migration_qualification,worker_spm_release}.py"] + e_ruff_per_file_ignore_of_T201_for_policyengine_api__data_v2_catalog_initialization_data_v2_household_migration_qualification_data_v2_policy_migration_qualification_data_v2_privilege_reconciliation_data_v2_user_association_migration_qualification_worker_spm_release__py["ruff per-file ignore of T201 for policyengine_api/{data/v2/catalog/initialization,data/v2/household_migration_qualification,data/v2/policy_migration_qualification,data/v2/privilege_reconciliation,data/v2/user_association_migration_qualification,worker_spm_release}.py"] r_api_quality_command_line_scripts_ruff_exemptions("api.quality.command-line-scripts-ruff-exemptions (enforced)") e_ruff_per_file_ignore_of_DTZ011__S310__S311__S603__S607__T201_for_scripts___["ruff per-file ignore of DTZ011, S310, S311, S603, S607, T201 for scripts/**"] r_api_quality_environment_variables_load("api.quality.environment-variables-load (enforced)") @@ -312,8 +332,8 @@ flowchart LR r_api_quality_alembic_revisions_unused_imports -.-> gap_unenforced r_api_quality_ci_cd_scripts_ruff_exemptions --> e_ruff_per_file_ignore_of_S310__T201_for__github___ e_ruff_per_file_ignore_of_S310__T201_for__github___ --> j_CI_pr_yml_lint - r_api_quality_command_line_modules_ruff_exemptions --> e_ruff_per_file_ignore_of_T201_for_policyengine_api__data_v2_catalog_initialization_data_v2_household_migration_qualification_data_v2_policy_migration_qualification_data_v2_user_association_migration_qualification_worker_spm_release__py - e_ruff_per_file_ignore_of_T201_for_policyengine_api__data_v2_catalog_initialization_data_v2_household_migration_qualification_data_v2_policy_migration_qualification_data_v2_user_association_migration_qualification_worker_spm_release__py --> j_CI_pr_yml_lint + r_api_quality_command_line_modules_ruff_exemptions --> e_ruff_per_file_ignore_of_T201_for_policyengine_api__data_v2_catalog_initialization_data_v2_household_migration_qualification_data_v2_policy_migration_qualification_data_v2_privilege_reconciliation_data_v2_user_association_migration_qualification_worker_spm_release__py + e_ruff_per_file_ignore_of_T201_for_policyengine_api__data_v2_catalog_initialization_data_v2_household_migration_qualification_data_v2_policy_migration_qualification_data_v2_privilege_reconciliation_data_v2_user_association_migration_qualification_worker_spm_release__py --> j_CI_pr_yml_lint r_api_quality_command_line_scripts_ruff_exemptions --> e_ruff_per_file_ignore_of_DTZ011__S310__S311__S603__S607__T201_for_scripts___ e_ruff_per_file_ignore_of_DTZ011__S310__S311__S603__S607__T201_for_scripts___ --> j_CI_pr_yml_lint r_api_quality_environment_variables_load --> e_CI_job_test_env_vars diff --git a/docs/migration/stage-8-supabase-target.md b/docs/migration/stage-8-supabase-target.md index 2ae9877cb..202f9b36f 100644 --- a/docs/migration/stage-8-supabase-target.md +++ b/docs/migration/stage-8-supabase-target.md @@ -88,9 +88,14 @@ repository records access classes and intended use only: | Generated v2 Alembic chain | Dedicated migration credential | Database connect and reviewed schema creation; no platform administration | | Future ordinary v2 persistence | Dedicated runtime credential | Ordinary application data access; no schema migration or platform administration | -The migration role owns the default privileges for objects it later creates; -ordinary table read/write and sequence use are granted to the runtime role. -Those grants do not create an application object or row. +Each project also has default privileges: tables the migration role creates +give the runtime role read/write access, and its sequences give sequence use. +They don't cover tables created before they were set, which is how the +production runtime role came to lack `legacy_policy_mappings` (issue #3977). +`policyengine_api/data/v2/privileges.py` is now the source of truth for every +managed role's table privileges, and each deploy grants and checks them +([Table privileges](../engineering/skills/alembic-migrations.md#table-privileges)). +None of these grants creates an application object or row. Runtime service accounts must not hold project-wide secret-access rights. Give each identity per-secret access only to the runtime values it needs, and grant diff --git a/policyengine_api/data/v2/privilege_reconciliation.py b/policyengine_api/data/v2/privilege_reconciliation.py new file mode 100644 index 000000000..29c74c624 --- /dev/null +++ b/policyengine_api/data/v2/privilege_reconciliation.py @@ -0,0 +1,520 @@ +"""Grant and check the table privileges the v2 manifest declares. + +``scripts/reconcile_v2_privileges.py`` runs this after every v2 schema upgrade, +connected as the migration role, which owns every table and so may grant on +it. Two modes: + +- ``apply`` reads the managed roles' current grants, grants every declared + privilege a role lacks, and reports privileges the manifest doesn't declare. + It never revokes, and it changes nothing when a declared role or table is + missing. Running it again grants nothing more. +- ``verify`` checks every declared privilege with ``has_table_privilege`` + (rule ``api.v2.deployed-privileges-match``) and fails on any that is missing. + +``--create-missing-roles`` creates the managed roles as ``NOLOGIN`` roles. It +is refused unless the target is the disposable test database +(``V2_ALEMBIC_DISPOSABLE_TEST=1``); in staging and production the roles are +provisioned outside the repository. + +The report goes to stdout and, when set, ``GITHUB_STEP_SUMMARY``. It names roles, +tables and privileges only, never a connection URL or credential. +""" + +from __future__ import annotations + +import argparse +import os +import re +import sys +from collections import defaultdict +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from sqlalchemy import bindparam, text +from sqlalchemy.pool import NullPool +from sqlmodel import create_engine + +from policyengine_api.data.v2.migration_target import ( + V2AlembicSettings, + load_v2_alembic_settings, + qualify_v2_connection, +) +from policyengine_api.data.v2.privileges import ( + PERSISTENT_ENVIRONMENTS, + TablePrivilege, + declared_privileges, +) +from policyengine_api.data.v2.settings import V2ConfigurationError + +if TYPE_CHECKING: + from collections.abc import Callable, Collection, Iterable, Mapping, Sequence + + from sqlalchemy import Connection, Engine + +DEPLOYMENT_ENVIRONMENT = "DEPLOYMENT_ENVIRONMENT" +GITHUB_STEP_SUMMARY = "GITHUB_STEP_SUMMARY" +SCHEMA = "public" +SAFE_IDENTIFIER = re.compile(r"^[a-z_][a-z0-9_]{0,62}$") + + +class PrivilegeReconciliationError(RuntimeError): + """Raised when privileges can't be reconciled or don't match the manifest.""" + + +class UnsafeIdentifierError(PrivilegeReconciliationError): + """Raised for a role or table name that isn't a plain lowercase identifier.""" + + +@dataclass(frozen=True, order=True) +class PrivilegeEntry: + """One privilege one role holds, or should hold, on one table.""" + + role: str + table: str + privilege: str + + +@dataclass(frozen=True) +class DefaultPrivilege: + """A default privilege PostgreSQL gives on objects a role creates later.""" + + creating_role: str + schema: str + object_type: str + grantee: str + privileges: tuple[str, ...] + + +@dataclass(frozen=True) +class ReconciliationPlan: + """What ``apply`` grants and reports for one database.""" + + grants: tuple[PrivilegeEntry, ...] + undeclared: tuple[PrivilegeEntry, ...] + missing_roles: tuple[str, ...] + missing_tables: tuple[str, ...] + + @property + def can_apply(self) -> bool: + """Whether every declared role and table exists, so grants may run.""" + return not self.missing_roles and not self.missing_tables + + +def plan_reconciliation( + declared: Mapping[str, Mapping[str, frozenset[TablePrivilege]]], + granted: Mapping[str, Mapping[str, Collection[str]]], + *, + existing_roles: Collection[str], + existing_tables: Collection[str], +) -> ReconciliationPlan: + """Compare declared privileges with what the database grants. + + ``granted`` holds the managed roles' direct grants, by role then table. + """ + missing_roles = tuple(sorted(set(declared) - set(existing_roles))) + declared_tables = {table for tables in declared.values() for table in tables} + missing_tables = tuple(sorted(declared_tables - set(existing_tables))) + grants = sorted( + PrivilegeEntry(role, table, privilege.value) + for role, tables in declared.items() + if role not in missing_roles + for table, privileges in tables.items() + if table not in missing_tables + for privilege in privileges + if privilege.value not in granted.get(role, {}).get(table, ()) + ) + undeclared = sorted( + PrivilegeEntry(role, table, privilege) + for role, tables in granted.items() + for table, privileges in tables.items() + for privilege in privileges + if privilege + not in { + declared_privilege.value + for declared_privilege in declared.get(role, {}).get(table, frozenset()) + } + ) + return ReconciliationPlan( + grants=tuple(grants), + undeclared=tuple(undeclared), + missing_roles=missing_roles, + missing_tables=missing_tables, + ) + + +def _safe(identifier: str) -> str: + if SAFE_IDENTIFIER.fullmatch(identifier) is None: + msg = f"refusing to use {identifier!r} as a role or table name" + raise UnsafeIdentifierError(msg) + return identifier + + +def grant_statements( + plan: ReconciliationPlan, quote_identifier: Callable[[str], str] +) -> list[str]: + """Return one GRANT per role and table, with its privileges in name order.""" + privileges_by_target: dict[tuple[str, str], set[str]] = defaultdict(set) + for entry in plan.grants: + privileges_by_target[(_safe(entry.role), _safe(entry.table))].add( + TablePrivilege(entry.privilege).value + ) + return [ + f"GRANT {', '.join(sorted(privileges))} ON TABLE " + f"{quote_identifier(SCHEMA)}.{quote_identifier(table)} " + f"TO {quote_identifier(role)}" + for (role, table), privileges in sorted( + privileges_by_target.items(), key=lambda item: (item[0][0], item[0][1]) + ) + ] + + +def verification_failures( + declared: Mapping[str, Mapping[str, frozenset[TablePrivilege]]], + has_privilege: Callable[[str, str, str], bool], +) -> list[PrivilegeEntry]: + """Return every declared privilege the role doesn't effectively hold.""" + return sorted( + PrivilegeEntry(role, table, privilege.value) + for role, tables in declared.items() + for table, privileges in tables.items() + for privilege in privileges + if not has_privilege(role, table, privilege.value) + ) + + +def _privilege_rows(entries: Iterable[PrivilegeEntry]) -> list[str]: + rows_by_target: dict[tuple[str, str], list[str]] = defaultdict(list) + for entry in entries: + rows_by_target[(entry.role, entry.table)].append(entry.privilege) + return [ + f"| `{role}` | `{table}` | {', '.join(sorted(privileges))} |" + for (role, table), privileges in sorted(rows_by_target.items()) + ] + + +def render_summary( + plan: ReconciliationPlan, + default_privileges: Sequence[DefaultPrivilege], + *, + deployment_environment: str, + applied: bool, +) -> str: + """Return the Markdown report for the job summary and the log.""" + lines = [f"## v2 table privileges ({deployment_environment})", ""] + if plan.missing_roles: + lines += [ + "Declared roles missing from the database (nothing was granted): " + + ", ".join(f"`{role}`" for role in plan.missing_roles), + "", + ] + if plan.missing_tables: + lines += [ + "Declared tables missing from the database (nothing was granted): " + + ", ".join(f"`{table}`" for table in plan.missing_tables), + "", + ] + grant_heading = "Granted" if applied else "Would grant" + lines += [f"### {grant_heading}", ""] + if plan.grants: + lines += ["| Role | Table | Privileges |", "| --- | --- | --- |"] + lines += _privilege_rows(plan.grants) + else: + lines.append("Nothing: every declared privilege was already granted.") + lines += [ + "", + "### Privileges the manifest doesn't declare (reported, not revoked)", + "", + ] + if plan.undeclared: + lines += ["| Role | Table | Privileges |", "| --- | --- | --- |"] + lines += _privilege_rows(plan.undeclared) + else: + lines.append("None.") + lines += ["", "### Default privileges for new tables", ""] + if default_privileges: + lines += [ + "| Creating role | Schema | Objects | Grantee | Privileges |", + "| --- | --- | --- | --- | --- |", + ] + lines += [ + f"| `{default.creating_role}` | `{default.schema}` | " + f"{default.object_type} | `{default.grantee}` | " + f"{', '.join(default.privileges)} |" + for default in default_privileges + ] + else: + lines.append("None.") + return "\n".join(lines) + "\n" + + +# --- Database access ------------------------------------------------------- + +_GRANTED_PRIVILEGES = text( + """ + SELECT grantee_role.rolname AS role_name, + table_class.relname AS table_name, + acl_entry.privilege_type AS privilege + FROM pg_class AS table_class + JOIN pg_namespace AS table_namespace + ON table_namespace.oid = table_class.relnamespace + CROSS JOIN LATERAL aclexplode(table_class.relacl) AS acl_entry + JOIN pg_roles AS grantee_role ON grantee_role.oid = acl_entry.grantee + WHERE table_namespace.nspname = :schema + AND table_class.relkind IN ('r', 'p', 'S', 'v', 'm') + AND grantee_role.rolname IN :roles + """ +).bindparams(bindparam("roles", expanding=True)) + +_EXISTING_ROLES = text( + "SELECT rolname FROM pg_roles WHERE rolname IN :roles" +).bindparams(bindparam("roles", expanding=True)) + +_PUBLIC_TABLES = text( + """ + SELECT table_class.relname + FROM pg_class AS table_class + JOIN pg_namespace AS table_namespace + ON table_namespace.oid = table_class.relnamespace + WHERE table_namespace.nspname = :schema AND table_class.relkind IN ('r', 'p') + """ +) + +_DEFAULT_PRIVILEGES = text( + """ + SELECT pg_get_userbyid(default_acl.defaclrole) AS creating_role, + coalesce(acl_namespace.nspname, '') AS schema_name, + CASE default_acl.defaclobjtype + WHEN 'r' THEN 'tables' WHEN 'S' THEN 'sequences' + WHEN 'f' THEN 'functions' WHEN 'T' THEN 'types' + WHEN 'n' THEN 'schemas' ELSE default_acl.defaclobjtype::text + END AS object_type, + coalesce(grantee_role.rolname, 'PUBLIC') AS grantee, + acl_entry.privilege_type AS privilege + FROM pg_default_acl AS default_acl + LEFT JOIN pg_namespace AS acl_namespace + ON acl_namespace.oid = default_acl.defaclnamespace + CROSS JOIN LATERAL aclexplode(default_acl.defaclacl) AS acl_entry + LEFT JOIN pg_roles AS grantee_role ON grantee_role.oid = acl_entry.grantee + WHERE coalesce(grantee_role.rolname, '') IN :roles + OR pg_get_userbyid(default_acl.defaclrole) IN :roles + """ +).bindparams(bindparam("roles", expanding=True)) + +_HAS_TABLE_PRIVILEGE = text( + "SELECT has_table_privilege(:role, :qualified_table, :privilege)" +) + + +def read_existing_roles(connection: Connection, roles: Collection[str]) -> set[str]: + """Return which of ``roles`` exist.""" + return set(connection.execute(_EXISTING_ROLES, {"roles": list(roles)}).scalars()) + + +def read_public_tables(connection: Connection) -> set[str]: + """Return every table in the public schema.""" + return set(connection.execute(_PUBLIC_TABLES, {"schema": SCHEMA}).scalars()) + + +def read_granted_privileges( + connection: Connection, roles: Collection[str] +) -> dict[str, dict[str, set[str]]]: + """Return the roles' direct grants on public tables, views and sequences.""" + granted: dict[str, dict[str, set[str]]] = defaultdict(lambda: defaultdict(set)) + for row in connection.execute( + _GRANTED_PRIVILEGES, {"schema": SCHEMA, "roles": list(roles)} + ): + granted[row.role_name][row.table_name].add(row.privilege) + return {role: dict(tables) for role, tables in granted.items()} + + +def read_default_privileges( + connection: Connection, roles: Collection[str] +) -> list[DefaultPrivilege]: + """Return default privileges that involve the managed roles.""" + grouped: dict[tuple[str, str, str, str], set[str]] = defaultdict(set) + for row in connection.execute(_DEFAULT_PRIVILEGES, {"roles": list(roles)}): + grouped[(row.creating_role, row.schema_name, row.object_type, row.grantee)].add( + row.privilege + ) + return [ + DefaultPrivilege( + creating_role=creating_role, + schema=schema, + object_type=object_type, + grantee=grantee, + privileges=tuple(sorted(privileges)), + ) + for (creating_role, schema, object_type, grantee), privileges in sorted( + grouped.items() + ) + ] + + +def create_missing_roles(connection: Connection, roles: Iterable[str]) -> list[str]: + """Create each missing role as NOLOGIN; only for the disposable test database.""" + missing = sorted(set(roles) - read_existing_roles(connection, list(roles))) + quote = connection.dialect.identifier_preparer.quote + for role in missing: + connection.execute(text(f"CREATE ROLE {quote(_safe(role))} NOLOGIN")) + return missing + + +def apply_declared_privileges( + connection: Connection, + declared: Mapping[str, Mapping[str, frozenset[TablePrivilege]]], +) -> tuple[ReconciliationPlan, list[DefaultPrivilege]]: + """Grant every declared privilege a role lacks; never revoke. + + Grants nothing when a declared role or table is missing. + """ + roles = list(declared) + plan = plan_reconciliation( + declared, + read_granted_privileges(connection, roles), + existing_roles=read_existing_roles(connection, roles), + existing_tables=read_public_tables(connection), + ) + if plan.can_apply: + for statement in grant_statements( + plan, connection.dialect.identifier_preparer.quote + ): + connection.execute(text(statement)) + return plan, read_default_privileges(connection, roles) + + +def check_declared_privileges( + connection: Connection, + declared: Mapping[str, Mapping[str, frozenset[TablePrivilege]]], +) -> list[PrivilegeEntry]: + """Return every declared privilege a role doesn't effectively hold. + + A missing role or table counts as missing every privilege declared for it. + """ + existing_roles = read_existing_roles(connection, list(declared)) + existing_tables = read_public_tables(connection) + + def has_privilege(role: str, table: str, privilege: str) -> bool: + if role not in existing_roles or table not in existing_tables: + return False + return bool( + connection.execute( + _HAS_TABLE_PRIVILEGE, + { + "role": role, + "qualified_table": f"{SCHEMA}.{_safe(table)}", + "privilege": privilege, + }, + ).scalar_one() + ) + + return verification_failures(declared, has_privilege) + + +# --- Command-line interface ----------------------------------------------- + + +def _deployment_environment( + environ: Mapping[str, str], settings: V2AlembicSettings +) -> str: + environment = environ.get(DEPLOYMENT_ENVIRONMENT, "").strip() + if not environment: + msg = f"{DEPLOYMENT_ENVIRONMENT} is required" + raise V2ConfigurationError(msg) + if not settings.disposable_test and environment not in PERSISTENT_ENVIRONMENTS: + msg = ( + f"{DEPLOYMENT_ENVIRONMENT} must be staging or production for a " + "persistent v2 database" + ) + raise V2ConfigurationError(msg) + return environment + + +def _write_summary(environ: Mapping[str, str], summary: str) -> None: + print(summary) + summary_path = environ.get(GITHUB_STEP_SUMMARY, "").strip() + if summary_path: + with open(summary_path, "a", encoding="utf-8") as summary_file: + summary_file.write(summary) + + +def _build_engine(settings: V2AlembicSettings) -> Engine: + return create_engine(settings.url, poolclass=NullPool) + + +def run( + mode: str, + *, + create_roles: bool = False, + environ: Mapping[str, str] | None = None, + engine_builder: Callable[[V2AlembicSettings], Engine] = _build_engine, +) -> int: + """Run ``apply`` or ``verify`` against the configured v2 database.""" + values = os.environ if environ is None else environ + settings = load_v2_alembic_settings(values) + environment = _deployment_environment(values, settings) + if create_roles and not settings.disposable_test: + msg = "--create-missing-roles is only allowed for the disposable test database" + raise V2ConfigurationError(msg) + declared = declared_privileges(environment) + engine = engine_builder(settings) + try: + with engine.begin() as connection: + qualify_v2_connection(connection, settings) + if mode == "verify": + failures = check_declared_privileges(connection, declared) + if failures: + rows = "\n".join( + f"- `{failure.role}` lacks {failure.privilege} on " + f"`{failure.table}`" + for failure in failures + ) + _write_summary( + values, + f"## v2 table privileges ({environment})\n\n" + f"Declared privileges missing after reconciling:\n\n{rows}\n", + ) + return 1 + print(f"Every declared v2 table privilege is held ({environment}).") + return 0 + if create_roles: + created = create_missing_roles(connection, declared) + if created: + print("Created roles: " + ", ".join(created)) + plan, defaults = apply_declared_privileges(connection, declared) + _write_summary( + values, + render_summary( + plan, + defaults, + deployment_environment=environment, + applied=plan.can_apply, + ), + ) + return 0 if plan.can_apply else 1 + finally: + engine.dispose() + + +def main(argv: Sequence[str] | None = None) -> int: + """Parse the arguments and run; configuration errors exit with status 2.""" + parser = argparse.ArgumentParser( + description="Grant or check the v2 table privileges the manifest declares." + ) + parser.add_argument("mode", choices=("apply", "verify")) + parser.add_argument( + "--create-missing-roles", + action="store_true", + help="create the managed roles first (disposable test database only)", + ) + arguments = parser.parse_args(argv) + try: + return run(arguments.mode, create_roles=arguments.create_missing_roles) + except V2ConfigurationError as error: + print( + f"v2 privilege reconciliation is not configured: {error}", file=sys.stderr + ) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/policyengine_api/data/v2/privilege_requirements.py b/policyengine_api/data/v2/privilege_requirements.py new file mode 100644 index 000000000..154fd6404 --- /dev/null +++ b/policyengine_api/data/v2/privilege_requirements.py @@ -0,0 +1,234 @@ +"""Derive the table privileges v2 connector modules need from their source. + +The privilege manifest (``policyengine_api.data.v2.privileges``) declares what +each database role may do. This module reads connector source with ``ast`` and +works out what the code needs, so a test can compare the two without a +database (rule ``api.v2.runtime-privileges-declared``): + +- ``reads*.py`` modules need SELECT on every v2 table they use. +- ``creates*.py``, ``updates*.py`` and ``deletes*.py`` modules need INSERT, + UPDATE or DELETE, plus SELECT: PostgreSQL authorizes the columns an + ``INSERT ... RETURNING``, an ORM refresh, or an ``UPDATE``/``DELETE ... WHERE`` + reads with SELECT. +- A function that locks rows (``with_for_update``, including ``read=True`` + for ``FOR SHARE``) needs UPDATE on each table it selects, because PostgreSQL + requires UPDATE privilege for ``SELECT ... FOR UPDATE``/``FOR SHARE``. + +A table counts as used when its model class, imported at runtime from +``policyengine_api.data.v2.models``, appears outside annotations. Imports under +``if TYPE_CHECKING:`` and names that appear only in annotations don't count. +""" + +from __future__ import annotations + +import ast +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from policyengine_api.data.v2.privileges import TablePrivilege + +if TYPE_CHECKING: + from collections.abc import Iterable, Iterator, Mapping + from pathlib import Path + +V2_MODELS_MODULE = "policyengine_api.data.v2.models" + +PRIVILEGES_BY_MODULE_KIND: dict[str, frozenset[TablePrivilege]] = { + "reads": frozenset({TablePrivilege.SELECT}), + "creates": frozenset({TablePrivilege.INSERT, TablePrivilege.SELECT}), + "updates": frozenset({TablePrivilege.UPDATE, TablePrivilege.SELECT}), + "deletes": frozenset({TablePrivilege.DELETE, TablePrivilege.SELECT}), +} + + +class UnclassifiedConnectorError(ValueError): + """Raised for a connector module not named reads, creates, updates or deletes.""" + + +@dataclass(frozen=True, order=True) +class RequiredPrivilege: + """One privilege a connector module needs on one table, and why.""" + + table: str + privilege: TablePrivilege + connector: str + reason: str + + +def connector_kind(path: Path) -> str: + """Return reads, creates, updates or deletes from the module's file name.""" + for kind in PRIVILEGES_BY_MODULE_KIND: + if path.stem == kind or path.stem.startswith(f"{kind}_"): + return kind + msg = f"{path.name} is not named reads, creates, updates or deletes" + raise UnclassifiedConnectorError(msg) + + +def _is_type_checking_block(node: ast.AST) -> bool: + if not isinstance(node, ast.If): + return False + test = node.test + return (isinstance(test, ast.Name) and test.id == "TYPE_CHECKING") or ( + isinstance(test, ast.Attribute) and test.attr == "TYPE_CHECKING" + ) + + +def _runtime_model_imports( + module: ast.Module, table_by_model: Mapping[str, str] +) -> dict[str, str]: + """Return local names bound to v2 model classes at runtime, mapped to tables.""" + tables_by_local_name: dict[str, str] = {} + for statement in module.body: + if _is_type_checking_block(statement): + continue + for node in ast.walk(statement): + if not isinstance(node, ast.ImportFrom) or node.module is None: + continue + if not node.module.startswith(V2_MODELS_MODULE): + continue + for alias in node.names: + table = table_by_model.get(alias.name) + if table is not None: + tables_by_local_name[alias.asname or alias.name] = table + return tables_by_local_name + + +def _annotation_nodes(module: ast.Module) -> set[int]: + """Return the ids of every node inside an annotation, which never runs SQL.""" + annotation_roots: list[ast.AST] = [] + for node in ast.walk(module): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + if node.returns is not None: + annotation_roots.append(node.returns) + arguments = node.args + for argument in [ + *arguments.posonlyargs, + *arguments.args, + *arguments.kwonlyargs, + *([arguments.vararg] if arguments.vararg else []), + *([arguments.kwarg] if arguments.kwarg else []), + ]: + if argument.annotation is not None: + annotation_roots.append(argument.annotation) + elif isinstance(node, ast.AnnAssign): + annotation_roots.append(node.annotation) + return {id(inner) for root in annotation_roots for inner in ast.walk(root)} + + +def _used_names( + nodes: Iterable[ast.AST], excluded_node_ids: set[int] +) -> Iterator[ast.Name]: + for node in nodes: + for inner in ast.walk(node): + if isinstance(inner, ast.Name) and id(inner) not in excluded_node_ids: + yield inner + + +def _locked_tables( + function: ast.FunctionDef | ast.AsyncFunctionDef, + tables_by_local_name: Mapping[str, str], +) -> set[str]: + """Return the tables a function locks with ``with_for_update``. + + With ``of=``, the named models are locked; otherwise every model selected or + joined in the function is, as PostgreSQL locks every table in the FROM list. + """ + lock_calls = [ + node + for node in ast.walk(function) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "with_for_update" + ] + if not lock_calls: + return set() + locked: set[str] = set() + for lock_call in lock_calls: + of_keyword = next( + (keyword for keyword in lock_call.keywords if keyword.arg == "of"), None + ) + if of_keyword is not None: + locked.update( + tables_by_local_name[name.id] + for name in _used_names([of_keyword.value], set()) + if name.id in tables_by_local_name + ) + continue + for node in ast.walk(function): + if not isinstance(node, ast.Call): + continue + called = node.func + called_name = ( + called.id + if isinstance(called, ast.Name) + else called.attr + if isinstance(called, ast.Attribute) + else None + ) + if called_name not in {"select", "join", "outerjoin", "select_from"}: + continue + locked.update( + tables_by_local_name[name.id] + for name in _used_names(node.args, set()) + if name.id in tables_by_local_name + ) + return locked + + +def required_privileges( + path: Path, + table_by_model: Mapping[str, str], + *, + connector: str | None = None, +) -> set[RequiredPrivilege]: + """Return every privilege the connector module at ``path`` needs, with reasons.""" + label = connector or path.name + kind = connector_kind(path) + module = ast.parse(path.read_text(encoding="utf-8")) + tables_by_local_name = _runtime_model_imports(module, table_by_model) + annotation_ids = _annotation_nodes(module) + import_ids = { + id(inner) + for node in ast.walk(module) + if isinstance(node, (ast.Import, ast.ImportFrom)) + for inner in ast.walk(node) + } + used_tables = { + tables_by_local_name[name.id] + for name in _used_names(module.body, annotation_ids | import_ids) + if name.id in tables_by_local_name + } + requirements = { + RequiredPrivilege( + table=table, + privilege=privilege, + connector=label, + reason=f"{kind} module uses the table", + ) + for table in used_tables + for privilege in PRIVILEGES_BY_MODULE_KIND[kind] + } + for node in ast.walk(module): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + requirements.update( + RequiredPrivilege( + table=table, + privilege=TablePrivilege.UPDATE, + connector=label, + reason=f"{node.name} locks rows with with_for_update", + ) + for table in _locked_tables(node, tables_by_local_name) + ) + return requirements + + +def undeclared_requirements( + requirements: Iterable[RequiredPrivilege], + declared: Mapping[str, frozenset[TablePrivilege]], +) -> list[RequiredPrivilege]: + """Return the requirements the declared table privileges don't cover, sorted.""" + return sorted( + requirement + for requirement in requirements + if requirement.privilege not in declared.get(requirement.table, frozenset()) + ) diff --git a/policyengine_api/data/v2/privileges.py b/policyengine_api/data/v2/privileges.py new file mode 100644 index 000000000..8f8c34515 --- /dev/null +++ b/policyengine_api/data/v2/privileges.py @@ -0,0 +1,178 @@ +"""The table privileges each managed v2 database role holds. + +This manifest is the source of truth for what the application roles may do +in the v2 Supabase database. ``scripts/reconcile_v2_privileges.py`` grants +whatever it declares after every schema upgrade and checks the result before +a deploy (rule ``api.v2.deployed-privileges-match``). A unit test compares it +with the privileges the connector code needs (rule +``api.v2.runtime-privileges-declared``), and the cross-database integration +tests run the services as these roles (rule +``api.v2.runtime-privileges-tested-as-runtime-role``). + +Every table in the v2 SQLModel metadata, plus ``alembic_version``, has an entry +for every managed role, so "no access" is a decision, not an omission. + +Roles and what decides their privileges: + +- ``policyengine_v2_runtime``: the Public API on Cloud Run. Every connector + module under ``policyengine_api/services/v2/`` except ``comparison_runs`` + runs as this role. +- ``policyengine_stage12_``: the simulation API and its Modal + worker. The ``comparison_runs`` connectors describe their operations on the + temporary Stage 12 tables (``docs/migration/stage-12-worker-contract.md``). +- ``policyengine_v2_catalog_writer``: the metadata catalog publication + (``scripts/publish_v2_metadata_catalog.py``). Its privileges follow what + publication writes and reads; its code isn't organized as connector modules, + so the deploy check, not the connector analysis, covers it. + +The migration role (``policyengine_v2_migrator``) owns every table, and +Supabase's ``postgres`` role administers the project; neither is managed here. + +Extra privileges a database holds beyond this manifest are reported, never +revoked. The v2 schema has no sequences (every key is a UUID); a test fails if +one appears, so the manifest can be extended to cover it. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import StrEnum +from typing import Final + +RUNTIME_ROLE: Final = "policyengine_v2_runtime" +CATALOG_WRITER_ROLE: Final = "policyengine_v2_catalog_writer" +STAGE12_ROLE_PREFIX: Final = "policyengine_stage12_" +PERSISTENT_ENVIRONMENTS: Final = frozenset({"staging", "production"}) +ENVIRONMENT_PATTERN: Final = re.compile(r"^[a-z][a-z0-9_]{0,30}$") + + +class TablePrivilege(StrEnum): + """A PostgreSQL table privilege the manifest can declare.""" + + SELECT = "SELECT" + INSERT = "INSERT" + UPDATE = "UPDATE" + DELETE = "DELETE" + + +NO_ACCESS: Final[frozenset[TablePrivilege]] = frozenset() +READ: Final = frozenset({TablePrivilege.SELECT}) +READ_INSERT: Final = frozenset({TablePrivilege.SELECT, TablePrivilege.INSERT}) +READ_INSERT_UPDATE: Final = frozenset( + {TablePrivilege.SELECT, TablePrivilege.INSERT, TablePrivilege.UPDATE} +) +READ_WRITE: Final = frozenset( + { + TablePrivilege.SELECT, + TablePrivilege.INSERT, + TablePrivilege.UPDATE, + TablePrivilege.DELETE, + } +) + + +@dataclass(frozen=True) +class TableAccess: + """What each managed role may do on one table.""" + + runtime: frozenset[TablePrivilege] + catalog_writer: frozenset[TablePrivilege] + stage12: frozenset[TablePrivilege] + + +def _access( + *, + runtime: frozenset[TablePrivilege] = NO_ACCESS, + catalog_writer: frozenset[TablePrivilege] = NO_ACCESS, + stage12: frozenset[TablePrivilege] = NO_ACCESS, +) -> TableAccess: + return TableAccess(runtime=runtime, catalog_writer=catalog_writer, stage12=stage12) + + +TABLE_ACCESS: Final[dict[str, TableAccess]] = { + # Alembic's revision record; the catalog writer checks the schema revision. + "alembic_version": _access(catalog_writer=READ), + # Simulation-domain result tables, unused by the managed roles. + "aggregates": _access(), + "budget_summary": _access(), + "change_aggregates": _access(), + "congressional_district_impacts": _access(), + "constituency_impacts": _access(), + "decile_impacts": _access(), + "dynamics": _access(), + "household_jobs": _access(), + "inequality": _access(), + "intra_decile_impacts": _access(), + "local_authority_impacts": _access(), + "poverty": _access(), + "program_statistics": _access(), + "report_runs": _access(catalog_writer=READ), + "reports": _access(catalog_writer=READ), + "simulations": _access(catalog_writer=READ), + # Metadata catalog: the catalog writer publishes it, the Public API reads it. + "dataset_versions": _access(catalog_writer=READ), + "datasets": _access(runtime=READ, catalog_writer=READ_INSERT_UPDATE), + "parameter_nodes": _access(runtime=READ, catalog_writer=READ_INSERT_UPDATE), + # The Public API also inserts a v2 policy's own parameter values. + "parameter_values": _access(runtime=READ_INSERT, catalog_writer=READ_INSERT_UPDATE), + "parameters": _access(runtime=READ, catalog_writer=READ_INSERT_UPDATE), + "regions": _access(runtime=READ, catalog_writer=READ_INSERT_UPDATE), + "tax_benefit_model_versions": _access( + runtime=READ, catalog_writer=READ_INSERT_UPDATE + ), + "tax_benefit_models": _access(runtime=READ, catalog_writer=READ_INSERT_UPDATE), + "variables": _access(runtime=READ, catalog_writer=READ_INSERT_UPDATE), + # v2 policies and households, created by the native routes and by Stage 10 + # dual-write. A copy locks its mapping row (FOR UPDATE), which needs UPDATE. + "households": _access(runtime=READ_INSERT), + "legacy_household_mappings": _access(runtime=READ_INSERT_UPDATE), + "legacy_policy_mappings": _access(runtime=READ_INSERT_UPDATE), + "policies": _access(runtime=READ_INSERT), + # Stage 10 saved-policy mirror. + "legacy_user_mappings": _access(runtime=READ_WRITE), + "legacy_user_policy_mappings": _access(runtime=READ_WRITE), + # Users and their saved work (Stage 12A). + "user_household_associations": _access(runtime=READ_WRITE), + "user_identities": _access(runtime=READ_WRITE), + "user_policies": _access(runtime=READ_WRITE), + "user_report_associations": _access(runtime=READ_WRITE), + "user_simulation_associations": _access(runtime=READ_WRITE), + "users": _access(runtime=READ_WRITE), + # Temporary Stage 12 comparison-run tables. + "stage12_evaluation_reports": _access(runtime=READ_WRITE, stage12=READ_WRITE), + "stage12_evaluation_simulations": _access(runtime=READ_WRITE, stage12=READ_WRITE), +} + + +class InvalidDeploymentEnvironmentError(ValueError): + """Raised for a deployment environment that can't name a Stage 12 role.""" + + +def stage12_role(deployment_environment: str) -> str: + """Return the Stage 12 role name for a deployment environment. + + Staging and production name their persistent roles; a disposable test + database may use any short lowercase name. + """ + if ENVIRONMENT_PATTERN.fullmatch(deployment_environment) is None: + msg = ( + "the deployment environment must be lowercase letters, digits or " + f"underscores, got {deployment_environment!r}" + ) + raise InvalidDeploymentEnvironmentError(msg) + return f"{STAGE12_ROLE_PREFIX}{deployment_environment}" + + +def declared_privileges( + deployment_environment: str, +) -> dict[str, dict[str, frozenset[TablePrivilege]]]: + """Return each managed role's declared privileges, keyed by role then table.""" + stage12 = stage12_role(deployment_environment) + return { + RUNTIME_ROLE: {table: access.runtime for table, access in TABLE_ACCESS.items()}, + CATALOG_WRITER_ROLE: { + table: access.catalog_writer for table, access in TABLE_ACCESS.items() + }, + stage12: {table: access.stage12 for table, access in TABLE_ACCESS.items()}, + } diff --git a/pyproject.toml b/pyproject.toml index a6c52ccf3..d711c81a0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -128,7 +128,7 @@ exclude = ["migrations/*/versions/*.py"] "DTZ011", "S310", "S311", "S603", "S607", "T201", ] # Modules that also run as command-line programs print their output. -"policyengine_api/{data/v2/catalog/initialization,data/v2/household_migration_qualification,data/v2/policy_migration_qualification,data/v2/user_association_migration_qualification,worker_spm_release}.py" = ["T201"] +"policyengine_api/{data/v2/catalog/initialization,data/v2/household_migration_qualification,data/v2/policy_migration_qualification,data/v2/privilege_reconciliation,data/v2/user_association_migration_qualification,worker_spm_release}.py" = ["T201"] [tool.mypy] python_version = "3.11" diff --git a/rules/fixtures/api.v2.deployed-privileges-match/bad_effective_privileges.json b/rules/fixtures/api.v2.deployed-privileges-match/bad_effective_privileges.json new file mode 100644 index 000000000..4a9aedf98 --- /dev/null +++ b/rules/fixtures/api.v2.deployed-privileges-match/bad_effective_privileges.json @@ -0,0 +1,6 @@ +[ + {"role": "policyengine_v2_runtime", "table": "users", "privilege": "SELECT"}, + {"role": "policyengine_v2_runtime", "table": "users", "privilege": "INSERT"}, + {"role": "policyengine_v2_runtime", "table": "users", "privilege": "UPDATE"}, + {"role": "policyengine_v2_runtime", "table": "users", "privilege": "DELETE"} +] diff --git a/rules/fixtures/api.v2.deployed-privileges-match/good_effective_privileges.json b/rules/fixtures/api.v2.deployed-privileges-match/good_effective_privileges.json new file mode 100644 index 000000000..53f134c2d --- /dev/null +++ b/rules/fixtures/api.v2.deployed-privileges-match/good_effective_privileges.json @@ -0,0 +1,7 @@ +[ + {"role": "policyengine_v2_runtime", "table": "users", "privilege": "SELECT"}, + {"role": "policyengine_v2_runtime", "table": "users", "privilege": "INSERT"}, + {"role": "policyengine_v2_runtime", "table": "users", "privilege": "UPDATE"}, + {"role": "policyengine_v2_runtime", "table": "users", "privilege": "DELETE"}, + {"role": "policyengine_v2_runtime", "table": "legacy_policy_mappings", "privilege": "SELECT"} +] diff --git a/rules/fixtures/api.v2.runtime-privileges-declared/reads_bad.py b/rules/fixtures/api.v2.runtime-privileges-declared/reads_bad.py new file mode 100644 index 000000000..0fa8f412b --- /dev/null +++ b/rules/fixtures/api.v2.runtime-privileges-declared/reads_bad.py @@ -0,0 +1,37 @@ +"""Bad fixture: a runtime read of a table the runtime role has no grant on. + +This is the mistake behind issue #3977. ``POST /v2/user-policies`` started +reading ``legacy_policy_mappings`` to fill ``policy_id``, but no one granted the +runtime role SELECT on that table, so every reform save returned 503. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from sqlmodel import Session, col, select + +from policyengine_api.data.v2.models import LegacyPolicyMapping, UserPolicy + +if TYPE_CHECKING: + from uuid import UUID + + +def read_mirrored_policy_ids( + session: Session, *, country_id: str, legacy_policy_ids: list[int] +) -> dict[int, UUID]: + """Return v2 policy UUIDs for v1 policies the Stage 10 mirror copied.""" + rows = session.exec( + select( + LegacyPolicyMapping.legacy_policy_id, LegacyPolicyMapping.policy_id + ).where( + LegacyPolicyMapping.country_id == country_id, + col(LegacyPolicyMapping.legacy_policy_id).in_(legacy_policy_ids), + ) + ).all() + return {int(legacy_policy_id): policy_id for legacy_policy_id, policy_id in rows} + + +def read_user_policy_row(session: Session, association_id: UUID) -> UserPolicy | None: + """Return the saved policy with this id, or None.""" + return session.get(UserPolicy, association_id) diff --git a/rules/fixtures/api.v2.runtime-privileges-declared/reads_good.py b/rules/fixtures/api.v2.runtime-privileges-declared/reads_good.py new file mode 100644 index 000000000..5437a95c0 --- /dev/null +++ b/rules/fixtures/api.v2.runtime-privileges-declared/reads_good.py @@ -0,0 +1,28 @@ +"""Good fixture: a runtime read of only the tables the runtime role is granted. + +The user-policy lookup reads ``user_policies``, which the runtime role holds +SELECT on, and nothing else. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from policyengine_api.data.v2.models import UserPolicy + +if TYPE_CHECKING: + from uuid import UUID + + from sqlmodel import Session + + from policyengine_api.data.v2.models import LegacyPolicyMapping + + +def read_user_policy_row(session: Session, association_id: UUID) -> UserPolicy | None: + """Return the saved policy with this id, or None.""" + return session.get(UserPolicy, association_id) + + +def describe_mapping(mapping: LegacyPolicyMapping) -> str: + """Name a mapping the caller already holds; only its annotation uses the model.""" + return f"{mapping.country_id}:{mapping.legacy_policy_id}" diff --git a/rules/registry.toml b/rules/registry.toml index 57efe03c3..639e3cd13 100644 --- a/rules/registry.toml +++ b/rules/registry.toml @@ -641,6 +641,79 @@ workflow = ".github/workflows/pr.yml" job = "v2-integration-check" invocation = "v2-integration-check.yml" +[[rule]] +id = "api.v2.runtime-privileges-declared" +category = "migration" +statement = "Every v2 table has an entry for each managed database role in policyengine_api/data/v2/privileges.py, and every privilege a v2 connector module needs (from its reads, creates, updates or deletes module name and its row locks) is declared for the role that runs it." +rationale = "docs/engineering/skills/alembic-migrations.md#table-privileges" +scope = ["policyengine_api/data/v2/privileges.py", "policyengine_api/data/v2/models/*.py", "policyengine_api/services/v2/*/database_connectors/*.py"] +tier = "mechanical" +status = "enforced" +autofix = false +[rule.enforcer] +kind = "pytest" +nodes = [ + "tests/unit/v2/test_privilege_manifest.py::test_every_v2_table_has_an_entry_for_every_managed_role", + "tests/unit/v2/test_privilege_manifest.py::test_v2_schema_has_no_sequences_the_manifest_would_need_to_cover", + "tests/unit/v2/test_privilege_manifest.py::test_runtime_connectors_need_only_declared_runtime_privileges", + "tests/unit/v2/test_privilege_manifest.py::test_stage12_connectors_need_only_declared_stage12_privileges", +] +[[rule.ci]] +workflow = ".github/workflows/pr.yml" +job = "test" +invocation = "make test" +[rule.proof] +bad_fixture = "rules/fixtures/api.v2.runtime-privileges-declared/reads_bad.py" +good_fixture = "rules/fixtures/api.v2.runtime-privileges-declared/reads_good.py" +test = "tests/unit/v2/test_privilege_manifest.py::test_rule_proof_runtime_privileges_declared" + +[[rule]] +id = "api.v2.runtime-privileges-tested-as-runtime-role" +category = "migration" +statement = "The cross-database integration tests run the v2 user services as policyengine_v2_runtime and the comparison-run services as policyengine_stage12_, holding only their declared privileges, so a statement those privileges don't allow fails CI." +rationale = "docs/engineering/skills/alembic-migrations.md#table-privileges" +scope = ["tests/integration/conftest.py", "tests/integration/test_v2_user_identity_persistence.py", "tests/integration/test_v2_stage12_comparison_persistence.py"] +tier = "mechanical" +status = "enforced" +autofix = false +[rule.enforcer] +kind = "workflow-job" +[[rule.ci]] +workflow = ".github/workflows/pr.yml" +job = "v2-integration-check" +invocation = "v2-integration-check.yml" +[[rule.ci]] +workflow = ".github/workflows/v2-integration-check.yml" +job = "catalog-runtime-integration" +invocation = "reconcile_v2_privileges.py apply --create-missing-roles" + +[[rule]] +id = "api.v2.deployed-privileges-match" +category = "migration" +statement = "Every v2 deploy grants the managed roles each privilege policyengine_api/data/v2/privileges.py declares right after the schema upgrade, reports privileges it doesn't declare without revoking them, and stops before the catalog is seeded if a declared privilege is still missing." +rationale = "docs/engineering/skills/alembic-migrations.md#table-privileges" +scope = ["policyengine_api/data/v2/privileges.py", "policyengine_api/data/v2/privilege_reconciliation.py", "scripts/reconcile_v2_privileges.py", ".github/workflows/seed-v2-database.yml"] +tier = "mechanical" +status = "enforced" +autofix = false +[rule.enforcer] +kind = "workflow-job" +[[rule.ci]] +workflow = ".github/workflows/seed-v2-database.yml" +job = "seed" +invocation = "reconcile_v2_privileges.py verify" +[[rule.ci]] +workflow = ".github/workflows/v2-integration-check.yml" +job = "catalog-runtime-integration" +invocation = "reconcile_v2_privileges.py verify" +[rule.proof] +bad_fixture = "rules/fixtures/api.v2.deployed-privileges-match/bad_effective_privileges.json" +good_fixture = "rules/fixtures/api.v2.deployed-privileges-match/good_effective_privileges.json" +test = "tests/unit/v2/test_privilege_reconciliation.py::test_rule_proof_deployed_privileges_match" +workflow = ".github/workflows/pr.yml" +job = "test" +invocation = "make test" + [[rule]] id = "api.migration.alembic-autogenerated-only" category = "migration" @@ -807,6 +880,7 @@ scope = [ "policyengine_api/data/v2/catalog/initialization.py", "policyengine_api/data/v2/household_migration_qualification.py", "policyengine_api/data/v2/policy_migration_qualification.py", + "policyengine_api/data/v2/privilege_reconciliation.py", "policyengine_api/data/v2/user_association_migration_qualification.py", "policyengine_api/worker_spm_release.py", ] @@ -816,7 +890,7 @@ autofix = false [rule.enforcer] kind = "ruff" mode = "per-file-ignore" -pattern = "policyengine_api/{data/v2/catalog/initialization,data/v2/household_migration_qualification,data/v2/policy_migration_qualification,data/v2/user_association_migration_qualification,worker_spm_release}.py" +pattern = "policyengine_api/{data/v2/catalog/initialization,data/v2/household_migration_qualification,data/v2/policy_migration_qualification,data/v2/privilege_reconciliation,data/v2/user_association_migration_qualification,worker_spm_release}.py" codes = ["T201"] [[rule.ci]] workflow = ".github/workflows/pr.yml" diff --git a/scripts/reconcile_v2_privileges.py b/scripts/reconcile_v2_privileges.py new file mode 100755 index 000000000..b5fa4d75a --- /dev/null +++ b/scripts/reconcile_v2_privileges.py @@ -0,0 +1,12 @@ +#!/usr/bin/env python3 +"""Grant or check the v2 table privileges the manifest declares. + +Usage: ``reconcile_v2_privileges.py apply [--create-missing-roles]`` or +``reconcile_v2_privileges.py verify``. See +``policyengine_api.data.v2.privilege_reconciliation``. +""" + +from policyengine_api.data.v2.privilege_reconciliation import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index c386efe41..b90efebd2 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -1,6 +1,7 @@ import os import time import uuid +from dataclasses import dataclass import httpx import pytest @@ -61,6 +62,85 @@ def disposable_v2_database_url() -> str: return settings.url.render_as_string(hide_password=False) +@dataclass(frozen=True) +class ManagedV2Roles: + """The managed v2 roles in the disposable database, by what runs as them.""" + + deployment_environment: str + runtime: str + stage12: str + + +@pytest.fixture(scope="session") +def managed_v2_roles(disposable_v2_database_url: str) -> ManagedV2Roles: + """Create the managed v2 roles if missing and grant their declared privileges. + + The cross-database integration job does the same first with + ``reconcile_v2_privileges.py apply --create-missing-roles``. Repeating it + here grants nothing more there, and lets the tests run locally on their own. + """ + from sqlalchemy import create_engine + from sqlalchemy.pool import NullPool + + from policyengine_api.data.v2.privilege_reconciliation import ( + apply_declared_privileges, + create_missing_roles, + ) + from policyengine_api.data.v2.privileges import ( + RUNTIME_ROLE, + declared_privileges, + stage12_role, + ) + + deployment_environment = ( + os.environ.get("DEPLOYMENT_ENVIRONMENT", "").strip() or "test" + ) + declared = declared_privileges(deployment_environment) + engine = create_engine(disposable_v2_database_url, poolclass=NullPool) + try: + with engine.begin() as connection: + create_missing_roles(connection, declared) + plan, _default_privileges = apply_declared_privileges(connection, declared) + finally: + engine.dispose() + if not plan.can_apply: + pytest.fail( + "the disposable v2 database lacks declared roles or tables: " + f"{plan.missing_roles} {plan.missing_tables}" + ) + return ManagedV2Roles( + deployment_environment=deployment_environment, + runtime=RUNTIME_ROLE, + stage12=stage12_role(deployment_environment), + ) + + +@pytest.fixture(scope="session") +def v2_engine_as_role(disposable_v2_database_url: str, managed_v2_roles): + """Return a builder of engines whose connections act as one managed role. + + Each new connection runs ``SET ROLE`` and commits it, so every statement on + it is checked against that role's privileges, as in staging and production, + instead of the disposable database's superuser. + """ + from psycopg import sql + from sqlalchemy import create_engine, event + + def build(role: str, **engine_options): + set_role = sql.SQL("SET ROLE {}").format(sql.Identifier(role)) + engine = create_engine(disposable_v2_database_url, **engine_options) + + @event.listens_for(engine, "connect") + def act_as_role(dbapi_connection, _connection_record) -> None: + with dbapi_connection.cursor() as cursor: + cursor.execute(set_role) + dbapi_connection.commit() + + return engine + + return build + + @pytest.fixture def api_base_url() -> str: base_url = os.environ.get("API_BASE_URL") diff --git a/tests/integration/test_v2_privilege_reconciliation.py b/tests/integration/test_v2_privilege_reconciliation.py new file mode 100644 index 000000000..f720c3fb9 --- /dev/null +++ b/tests/integration/test_v2_privilege_reconciliation.py @@ -0,0 +1,139 @@ +"""PostgreSQL tests for granting and checking the declared v2 table privileges. + +``scripts/reconcile_v2_privileges.py`` runs these operations after every v2 +schema upgrade (rule ``api.v2.deployed-privileges-match``). Tests that need a +privilege to be missing or extra use their own temporary roles, so the managed +roles other integration tests run as keep exactly their declared privileges. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.pool import NullPool + +from policyengine_api.data.v2.migration_target import V2_ALEMBIC_DISPOSABLE_TEST +from policyengine_api.data.v2.privilege_reconciliation import ( + DEPLOYMENT_ENVIRONMENT, + PrivilegeEntry, + apply_declared_privileges, + check_declared_privileges, + run, +) +from policyengine_api.data.v2.privileges import READ, TablePrivilege +from policyengine_api.data.v2.settings import V2_MIGRATION_DATABASE_URL + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + + from sqlalchemy import Engine + + +@pytest.fixture +def superuser_engine(disposable_v2_database_url) -> Iterator[Engine]: + engine = create_engine(disposable_v2_database_url, poolclass=NullPool) + try: + yield engine + finally: + engine.dispose() + + +@pytest.fixture +def temporary_role(superuser_engine) -> Iterator[Callable[[], str]]: + """Return a builder of NOLOGIN roles that are dropped after the test.""" + created_roles: list[str] = [] + + def create_role() -> str: + role = f"policyengine_privilege_test_{uuid4().hex[:12]}" + with superuser_engine.begin() as connection: + connection.execute(text(f'CREATE ROLE "{role}" NOLOGIN')) + created_roles.append(role) + return role + + try: + yield create_role + finally: + with superuser_engine.begin() as connection: + for role in created_roles: + connection.execute(text(f'DROP OWNED BY "{role}"')) + connection.execute(text(f'DROP ROLE "{role}"')) + + +def _has_privilege(engine: Engine, role: str, table: str, privilege: str) -> bool: + with engine.connect() as connection: + return bool( + connection.execute( + text("SELECT has_table_privilege(:role, :table, :privilege)"), + {"role": role, "table": f"public.{table}", "privilege": privilege}, + ).scalar_one() + ) + + +def test_apply_grants_nothing_the_second_time_and_verify_passes( + disposable_v2_database_url, managed_v2_roles, capsys +) -> None: + environ = { + V2_MIGRATION_DATABASE_URL: disposable_v2_database_url, + V2_ALEMBIC_DISPOSABLE_TEST: "1", + DEPLOYMENT_ENVIRONMENT: managed_v2_roles.deployment_environment, + } + + first_apply_status = run("apply", environ=environ) + capsys.readouterr() + second_apply_status = run("apply", environ=environ) + second_apply_report = capsys.readouterr().out + verify_status = run("verify", environ=environ) + verify_output = capsys.readouterr().out + + assert (first_apply_status, second_apply_status, verify_status) == (0, 0, 0) + assert "Nothing: every declared privilege was already granted." in ( + second_apply_report + ) + assert "Every declared v2 table privilege is held" in verify_output + assert disposable_v2_database_url not in second_apply_report + verify_output + + +def test_apply_reports_an_extra_privilege_and_does_not_revoke_it( + superuser_engine, temporary_role +) -> None: + role = temporary_role() + declared = {role: {"regions": READ}} + with superuser_engine.begin() as connection: + first_plan, _defaults = apply_declared_privileges(connection, declared) + connection.execute(text(f'GRANT DELETE ON TABLE public.regions TO "{role}"')) + with superuser_engine.begin() as connection: + second_plan, _defaults = apply_declared_privileges(connection, declared) + missing_after_apply = check_declared_privileges(connection, declared) + + assert first_plan.grants == (PrivilegeEntry(role, "regions", "SELECT"),) + assert second_plan.grants == () + assert second_plan.undeclared == (PrivilegeEntry(role, "regions", "DELETE"),) + assert missing_after_apply == [] + assert _has_privilege(superuser_engine, role, "regions", "DELETE") + + +def test_a_missing_role_blocks_every_grant_and_fails_verification( + superuser_engine, temporary_role +) -> None: + role = temporary_role() + absent_role = f"policyengine_privilege_test_absent_{uuid4().hex[:12]}" + declared = { + role: {"regions": READ}, + absent_role: {"regions": frozenset({TablePrivilege.SELECT})}, + } + with superuser_engine.begin() as connection: + plan, _defaults = apply_declared_privileges(connection, declared) + missing = check_declared_privileges(connection, declared) + + assert not plan.can_apply + assert plan.missing_roles == (absent_role,) + assert not _has_privilege(superuser_engine, role, "regions", "SELECT") + assert missing == sorted( + [ + PrivilegeEntry(absent_role, "regions", "SELECT"), + PrivilegeEntry(role, "regions", "SELECT"), + ] + ) diff --git a/tests/integration/test_v2_stage12_comparison_persistence.py b/tests/integration/test_v2_stage12_comparison_persistence.py new file mode 100644 index 000000000..bbd75a4d2 --- /dev/null +++ b/tests/integration/test_v2_stage12_comparison_persistence.py @@ -0,0 +1,165 @@ +"""PostgreSQL tests for comparison-run persistence as the Stage 12 role. + +The simulation API and its Modal worker write the temporary Stage 12 tables as +``policyengine_stage12_``. These tests run the comparison-run +services as that role in the disposable database, so a statement its declared +privileges don't allow fails here (rule +``api.v2.runtime-privileges-tested-as-runtime-role``). +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine, delete, func, select +from sqlalchemy.exc import ProgrammingError +from sqlalchemy.orm import sessionmaker +from sqlmodel import Session + +from policyengine_api.data.v2.models import ( + Stage12ComparisonReport, + Stage12ComparisonSimulation, + User, +) +from policyengine_api.services.v2.comparison_runs.database_session import ( + ComparisonRunDatabaseSession, +) +from policyengine_api.services.v2.comparison_runs.services import ( + V2ComparisonRunService, +) +from policyengine_api.services.v2.comparison_runs.types import ( + ComparisonRunLifecycleStatus, +) +from tests.unit.v2.comparison_run_support import ( + comparison_report_record, + comparison_simulation_record, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from policyengine_api.services.v2.comparison_runs.types import ( + ComparisonReportRecord, + ComparisonSimulationRecord, + ) + + +@pytest.fixture +def stage12_sessions( + disposable_v2_database_url, managed_v2_roles, v2_engine_as_role +) -> Iterator[sessionmaker[Session]]: + """Open sessions as the Stage 12 role; remove this test's rows as the superuser.""" + stage12_engine = v2_engine_as_role(managed_v2_roles.stage12) + cleanup_engine = create_engine(disposable_v2_database_url) + factory = sessionmaker(stage12_engine, class_=Session, expire_on_commit=False) + factory.kw["info"] = {"evaluation_ids": []} + try: + yield factory + finally: + evaluation_ids = factory.kw["info"]["evaluation_ids"] + if evaluation_ids: + with cleanup_engine.begin() as connection: + connection.execute( + delete(Stage12ComparisonSimulation).where( + Stage12ComparisonSimulation.evaluation_id.in_(evaluation_ids) + ) + ) + connection.execute( + delete(Stage12ComparisonReport).where( + Stage12ComparisonReport.evaluation_id.in_(evaluation_ids) + ) + ) + stage12_engine.dispose() + cleanup_engine.dispose() + + +def _new_report(stage12_sessions) -> ComparisonReportRecord: + """Return a report record with a fresh identity the fixture removes afterwards.""" + created_at = datetime.now(UTC) + evaluation_id = uuid4() + stage12_sessions.kw["info"]["evaluation_ids"].append(evaluation_id) + return comparison_report_record( + evaluation_id=evaluation_id, + production_identity=f"production-job-{evaluation_id.hex[:12]}", + incumbent_execution_id=f"production-job-{evaluation_id.hex[:12]}", + created_at=created_at, + updated_at=created_at, + retention_expires_at=created_at + timedelta(days=30), + ) + + +def _baseline_simulation(report: ComparisonReportRecord) -> ComparisonSimulationRecord: + return comparison_simulation_record( + simulation_execution_id=uuid4(), + evaluation_id=report.evaluation_id, + created_at=report.created_at, + updated_at=report.created_at, + retention_expires_at=report.retention_expires_at, + ) + + +def test_stage12_sessions_act_as_the_stage12_role( + stage12_sessions, managed_v2_roles +) -> None: + with stage12_sessions() as session: + current_role = session.execute(select(func.current_user())).scalar_one() + + assert current_role == managed_v2_roles.stage12 + + +def test_stage12_role_creates_and_resolves_reports_and_simulations( + stage12_sessions, +) -> None: + service = V2ComparisonRunService(ComparisonRunDatabaseSession(stage12_sessions)) + report = _new_report(stage12_sessions) + simulation = _baseline_simulation(report) + + created_report = service.create_or_resolve_report(report) + retried_report = service.create_or_resolve_report(report) + created_simulation = service.create_or_resolve_simulation(simulation) + retried_simulation = service.create_or_resolve_simulation(simulation) + + assert (created_report.created, retried_report.created) == (True, False) + assert retried_report.record.evaluation_id == report.evaluation_id + assert (created_simulation.created, retried_simulation.created) == (True, False) + assert service.get_report(report.evaluation_id).status is ( + ComparisonRunLifecycleStatus.PENDING + ) + + +def test_stage12_role_updates_lifecycles_under_row_locks(stage12_sessions) -> None: + service = V2ComparisonRunService(ComparisonRunDatabaseSession(stage12_sessions)) + report = _new_report(stage12_sessions) + simulation = _baseline_simulation(report) + service.create_or_resolve_report(report) + service.create_or_resolve_simulation(simulation) + started_at = datetime.now(UTC) + + running_report = service.replace_report_lifecycle( + type(report).model_validate( + {**report.model_dump(), "status": "running", "updated_at": started_at} + ) + ) + running_simulation = service.replace_simulation_lifecycle( + type(simulation).model_validate( + {**simulation.model_dump(), "status": "running", "updated_at": started_at} + ) + ) + + assert running_report.status is ComparisonRunLifecycleStatus.RUNNING + assert running_simulation.status is ComparisonRunLifecycleStatus.RUNNING + assert service.get_report(report.evaluation_id).status is ( + ComparisonRunLifecycleStatus.RUNNING + ) + + +def test_stage12_role_cannot_read_tables_outside_its_declared_privileges( + stage12_sessions, +) -> None: + with stage12_sessions() as session, pytest.raises(ProgrammingError) as refusal: + session.execute(select(func.count()).select_from(User)).scalar_one() + + assert "permission denied for table users" in str(refusal.value) diff --git a/tests/integration/test_v2_user_identity_persistence.py b/tests/integration/test_v2_user_identity_persistence.py index e0f0f694d..237b61130 100644 --- a/tests/integration/test_v2_user_identity_persistence.py +++ b/tests/integration/test_v2_user_identity_persistence.py @@ -119,15 +119,24 @@ def _disposable_url() -> str: @pytest.fixture -def sessions() -> Iterator[sessionmaker[Session]]: - engine = create_engine(_disposable_url(), pool_size=10, max_overflow=10) - factory = sessionmaker(engine, class_=Session, expire_on_commit=False) +def sessions(managed_v2_roles, v2_engine_as_role) -> Iterator[sessionmaker[Session]]: + """Open sessions as the Public API's runtime role; clean up as the superuser. + + Running the services as ``policyengine_v2_runtime`` proves its declared + privileges cover them (rule + ``api.v2.runtime-privileges-tested-as-runtime-role``). + """ + runtime_engine = v2_engine_as_role( + managed_v2_roles.runtime, pool_size=10, max_overflow=10 + ) + cleanup_engine = create_engine(_disposable_url()) + factory = sessionmaker(runtime_engine, class_=Session, expire_on_commit=False) issuer = f"https://stage12a-{uuid4().hex[:12]}.example.test/" factory.kw["info"] = {"issuer": issuer, "browser_user_ids": []} try: yield factory finally: - with engine.begin() as connection: + with cleanup_engine.begin() as connection: owned = select(UserIdentity.user_id).where(UserIdentity.issuer == issuer) user_ids = [ *connection.execute(owned).scalars(), @@ -140,7 +149,8 @@ def sessions() -> Iterator[sessionmaker[Session]]: ) ) connection.execute(delete(User).where(User.id.in_(user_ids))) - engine.dispose() + runtime_engine.dispose() + cleanup_engine.dispose() def _principal(sessions, subject: str) -> UserPrincipal: @@ -234,6 +244,13 @@ def _save_policy(services: Services, principal) -> UUID: ).id +def test_sessions_act_as_the_runtime_role(sessions, managed_v2_roles) -> None: + with sessions() as session: + current_role = session.execute(select(func.current_user())).scalar_one() + + assert current_role == managed_v2_roles.runtime + + def test_concurrent_first_requests_provision_exactly_one_user(sessions) -> None: principal = _principal(sessions, "google-oauth2|race") workers = 8 diff --git a/tests/unit/test_alembic_workflows.py b/tests/unit/test_alembic_workflows.py index 5e978e5a7..c8d195d62 100644 --- a/tests/unit/test_alembic_workflows.py +++ b/tests/unit/test_alembic_workflows.py @@ -227,6 +227,8 @@ def test_reusable_v2_integration_check_uses_disposable_databases_and_coverage(): assert "test_v2_user_policy_mirroring.py" in workflow assert "test_v1_user_policy_dual_write.py" in workflow assert "test_v2_user_identity_persistence.py" in workflow + assert "test_v2_stage12_comparison_persistence.py" in workflow + assert "test_v2_privilege_reconciliation.py" in workflow assert "test_v2_household_persistence.py" in workflow assert "test_v1_household_dual_write.py" in workflow assert "test_mysql_household_spm_dual_write.py" in workflow @@ -235,12 +237,26 @@ def test_reusable_v2_integration_check_uses_disposable_databases_and_coverage(): assert "test_runtime_cache_redis.py" in workflow assert "uv sync --frozen" in workflow assert workflow.count("coverage run --branch") == 1 - assert workflow.count("coverage run -a --branch") == 6 + assert workflow.count("coverage run -a --branch") == 7 assert "coverage xml -i -o coverage-v2.xml" in workflow assert "codecov/codecov-action@v5" in workflow assert "files: coverage-v2.xml" in workflow +def test_v2_integration_check_grants_and_verifies_privileges_before_tests(): + workflow = _workflow("v2-integration-check.yml") + + schema_upgrade = workflow.index("alembic -c alembic-v2.ini upgrade head") + grant = workflow.index( + "scripts/reconcile_v2_privileges.py apply --create-missing-roles" + ) + verify = workflow.index("scripts/reconcile_v2_privileges.py verify") + first_test = workflow.index("pytest") + + assert " DEPLOYMENT_ENVIRONMENT: test\n" in workflow + assert schema_upgrade < grant < verify < first_test + + def test_release_migration_fails_closed_before_tests_and_cloud_run_deploy(): workflow = _workflow("push.yml") orchestration_script = ( diff --git a/tests/unit/v2/comparison_run_support.py b/tests/unit/v2/comparison_run_support.py new file mode 100644 index 000000000..6cfff7ce1 --- /dev/null +++ b/tests/unit/v2/comparison_run_support.py @@ -0,0 +1,69 @@ +"""Valid temporary Stage 12 comparison-run records for service and database tests.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from uuid import UUID + +from policyengine_api.services.v2.comparison_runs.types import ( + ComparisonReportRecord, + ComparisonSimulationRecord, +) + +NOW = datetime(2026, 9, 14, tzinfo=UTC) +EVALUATION_ID = UUID("00000000-0000-4000-8000-000000000001") +SIMULATION_ID = UUID("00000000-0000-4000-8000-000000000002") +DIGEST_A = "a" * 64 +DIGEST_B = "b" * 64 + + +def comparison_report_record(**changes: object) -> ComparisonReportRecord: + """Return a pending comparison report, with ``changes`` applied.""" + fields: dict[str, object] = { + "evaluation_id": EVALUATION_ID, + "status": "pending", + "aggregation_status": "not_started", + "environment": "test", + "calculation_flow": "economy", + "originating_request_id": "request-1", + "production_identity": "production-job-1", + "incumbent_execution_id": "production-job-1", + "worker_version": "5.2.0", + "modal_application": "policyengine-v2-worker-5-2-0", + "report_coordinator_callable": "coordinate_report", + "version_manifest_sha256": DIGEST_A, + "policyengine_version": "5.2.0", + "country_package_name": "policyengine-us", + "country_package_version": "1.900.0", + "country": "us", + "dataset_identity": "populace_us_2024", + "dataset_uri": "hf://policyengine/populace-us/data.h5@revision", + "data_package_name": "policyengine-us-data", + "data_package_version": "1.0.0", + "data_artifact_revision": "revision", + "created_at": NOW, + "updated_at": NOW, + "retention_expires_at": NOW + timedelta(days=30), + } + fields.update(changes) + return ComparisonReportRecord.model_validate(fields) + + +def comparison_simulation_record(**changes: object) -> ComparisonSimulationRecord: + """Return a pending baseline simulation of the default report, with ``changes``.""" + fields: dict[str, object] = { + "simulation_execution_id": SIMULATION_ID, + "evaluation_id": EVALUATION_ID, + "role": "baseline", + "input_sha256": DIGEST_B, + "worker_version": "5.2.0", + "modal_application": "policyengine-v2-worker-5-2-0", + "simulation_callable": "run_simulation", + "version_manifest_sha256": DIGEST_A, + "status": "pending", + "created_at": NOW, + "updated_at": NOW, + "retention_expires_at": NOW + timedelta(days=30), + } + fields.update(changes) + return ComparisonSimulationRecord.model_validate(fields) diff --git a/tests/unit/v2/test_metadata_deployment.py b/tests/unit/v2/test_metadata_deployment.py index bb99da23d..c752b46c0 100644 --- a/tests/unit/v2/test_metadata_deployment.py +++ b/tests/unit/v2/test_metadata_deployment.py @@ -105,6 +105,42 @@ def test_user_association_qualification_precedes_schema_upgrade() -> None: assert household_qualification < user_association_qualification < upgrade +def test_declared_privileges_are_granted_and_verified_before_catalog_seeding() -> None: + workflow = _read(".github/workflows/seed-v2-database.yml") + + upgrade = workflow.index("migrate_v2_metadata_schema.sh") + grant = workflow.index("scripts/reconcile_v2_privileges.py apply") + verify = workflow.index("scripts/reconcile_v2_privileges.py verify") + catalog_seed = workflow.index("scripts/publish_v2_metadata_catalog.py") + + assert upgrade < grant < verify < catalog_seed + assert "--create-missing-roles" not in workflow + + +def test_privilege_steps_connect_with_the_migration_credential() -> None: + workflow = _read(".github/workflows/seed-v2-database.yml") + grant = _step( + workflow, + "Grant declared v2 table privileges", + "Verify declared v2 table privileges", + ) + verify = _step( + workflow, + "Verify declared v2 table privileges", + "Seed and validate the v2 metadata catalog", + ) + + for step in (grant, verify): + assert ( + "V2_MIGRATION_DATABASE_URL: ${{ secrets.V2_MIGRATION_DATABASE_URL }}" + in step + ) + assert "V2_DATA_WRITE_DATABASE_URL" not in step + assert "DEPLOYMENT_ENVIRONMENT: ${{ inputs.deployment_environment }}" in _job( + workflow, "seed" + ) + + def test_seeding_success_is_required_before_candidate_creation() -> None: workflow = _read(".github/workflows/push.yml") staging_seed = _job(workflow, "seed-v2-staging-database") diff --git a/tests/unit/v2/test_privilege_manifest.py b/tests/unit/v2/test_privilege_manifest.py new file mode 100644 index 000000000..4b9134015 --- /dev/null +++ b/tests/unit/v2/test_privilege_manifest.py @@ -0,0 +1,147 @@ +"""The v2 privilege manifest covers every table and every connector's needs. + +Rule ``api.v2.runtime-privileges-declared`` (rules/registry.toml). +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import sqlalchemy as sa + +from policyengine_api.data.v2 import models +from policyengine_api.data.v2.models import V2_METADATA +from policyengine_api.data.v2.privilege_requirements import ( + required_privileges, + undeclared_requirements, +) +from policyengine_api.data.v2.privileges import ( + CATALOG_WRITER_ROLE, + NO_ACCESS, + READ_WRITE, + RUNTIME_ROLE, + TABLE_ACCESS, + InvalidDeploymentEnvironmentError, + TablePrivilege, + declared_privileges, + stage12_role, +) + +PROJECT_ROOT = Path(__file__).parents[3] +SERVICES_ROOT = PROJECT_ROOT / "policyengine_api" / "services" / "v2" +PROOF_FIXTURES = PROJECT_ROOT / "rules/fixtures/api.v2.runtime-privileges-declared" +STAGE12_PACKAGES = frozenset({"comparison_runs"}) +TEST_ENVIRONMENT = "staging" + + +def _table_by_model() -> dict[str, str]: + return { + name: model.__tablename__ + for name in dir(models) + if isinstance(model := getattr(models, name), type) + and isinstance(getattr(model, "__tablename__", None), str) + } + + +def _connector_modules(*, stage12: bool) -> list[str]: + paths = sorted( + path.relative_to(SERVICES_ROOT).as_posix() + for path in SERVICES_ROOT.glob("*/database_connectors/*.py") + if path.name != "__init__.py" + and (path.parts[-3] in STAGE12_PACKAGES) is stage12 + ) + assert paths, "no connector modules found" + return paths + + +def _missing(relative_path: str, role: str) -> list[str]: + requirements = required_privileges( + SERVICES_ROOT / relative_path, _table_by_model(), connector=relative_path + ) + declared = declared_privileges(TEST_ENVIRONMENT)[role] + return [ + f"{missing.connector}: {missing.privilege} on {missing.table} " + f"({missing.reason})" + for missing in undeclared_requirements(requirements, declared) + ] + + +def test_every_v2_table_has_an_entry_for_every_managed_role() -> None: + metadata_tables = {table.name for table in V2_METADATA.tables.values()} + + assert set(TABLE_ACCESS) == metadata_tables | {"alembic_version"} + for role, privileges_by_table in declared_privileges(TEST_ENVIRONMENT).items(): + assert set(privileges_by_table) == set(TABLE_ACCESS), role + + +def test_v2_schema_has_no_sequences_the_manifest_would_need_to_cover() -> None: + sequence_columns = [ + f"{table.name}.{column.name}" + for table in V2_METADATA.tables.values() + for column in table.columns + if isinstance(column.default, sa.Sequence) + or column.identity is not None + or ( + column.primary_key + and len(table.primary_key.columns) == 1 + and isinstance(column.type, sa.Integer) + and column.autoincrement is not False + ) + ] + + assert sequence_columns == [], ( + "a v2 column now uses a sequence; declare its privileges in " + "policyengine_api/data/v2/privileges.py and extend the reconcile script" + ) + + +@pytest.mark.parametrize("relative_path", _connector_modules(stage12=False)) +def test_runtime_connectors_need_only_declared_runtime_privileges( + relative_path: str, +) -> None: + assert _missing(relative_path, RUNTIME_ROLE) == [] + + +@pytest.mark.parametrize("relative_path", _connector_modules(stage12=True)) +def test_stage12_connectors_need_only_declared_stage12_privileges( + relative_path: str, +) -> None: + assert _missing(relative_path, stage12_role(TEST_ENVIRONMENT)) == [] + + +def test_stage12_role_name_follows_the_deployment_environment() -> None: + assert stage12_role("staging") == "policyengine_stage12_staging" + assert stage12_role("production") == "policyengine_stage12_production" + for invalid in ("", "Production", "prod-1", "x" * 40, "staging; DROP"): + with pytest.raises(InvalidDeploymentEnvironmentError): + stage12_role(invalid) + + +def test_declared_privileges_name_the_three_managed_roles() -> None: + assert set(declared_privileges("production")) == { + RUNTIME_ROLE, + CATALOG_WRITER_ROLE, + "policyengine_stage12_production", + } + + +def test_rule_proof_runtime_privileges_declared() -> None: + """Proof for rule api.v2.runtime-privileges-declared (rules/registry.toml). + + The bad fixture repeats the mistake behind issue #3977: a runtime connector + reads ``legacy_policy_mappings``, which the runtime role had no grant on. + """ + runtime_before_3977 = {table: NO_ACCESS for table in TABLE_ACCESS} | { + "user_policies": READ_WRITE + } + table_by_model = _table_by_model() + + bad = required_privileges(PROOF_FIXTURES / "reads_bad.py", table_by_model) + good = required_privileges(PROOF_FIXTURES / "reads_good.py", table_by_model) + + assert [ + (missing.table, missing.privilege) + for missing in undeclared_requirements(bad, runtime_before_3977) + ] == [("legacy_policy_mappings", TablePrivilege.SELECT)] + assert undeclared_requirements(good, runtime_before_3977) == [] diff --git a/tests/unit/v2/test_privilege_reconciliation.py b/tests/unit/v2/test_privilege_reconciliation.py new file mode 100644 index 000000000..73ca30e19 --- /dev/null +++ b/tests/unit/v2/test_privilege_reconciliation.py @@ -0,0 +1,282 @@ +"""Planning, statements and reports for the v2 privilege reconcile script.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from policyengine_api.data.v2 import privilege_reconciliation +from policyengine_api.data.v2.privilege_reconciliation import ( + DefaultPrivilege, + PrivilegeEntry, + ReconciliationPlan, + grant_statements, + plan_reconciliation, + render_summary, + verification_failures, +) +from policyengine_api.data.v2.privileges import ( + READ, + READ_WRITE, + TablePrivilege, +) + +PROJECT_ROOT = Path(__file__).parents[3] +PROOF_FIXTURES = PROJECT_ROOT / "rules/fixtures/api.v2.deployed-privileges-match" +RUNTIME = "policyengine_v2_runtime" +STAGE12 = "policyengine_stage12_staging" + + +def _quote(identifier: str) -> str: + return f'"{identifier}"' + + +def _declared() -> dict[str, dict[str, frozenset[TablePrivilege]]]: + return { + RUNTIME: { + "users": READ_WRITE, + "legacy_policy_mappings": READ, + "reports": frozenset(), + }, + STAGE12: { + "users": frozenset(), + "legacy_policy_mappings": frozenset(), + "reports": frozenset(), + }, + } + + +def test_missing_declared_privileges_become_grants() -> None: + plan = plan_reconciliation( + _declared(), + granted={RUNTIME: {"users": {"SELECT", "INSERT"}}}, + existing_roles={RUNTIME, STAGE12}, + existing_tables={"users", "legacy_policy_mappings", "reports"}, + ) + + assert plan.grants == ( + PrivilegeEntry(RUNTIME, "legacy_policy_mappings", "SELECT"), + PrivilegeEntry(RUNTIME, "users", "DELETE"), + PrivilegeEntry(RUNTIME, "users", "UPDATE"), + ) + assert plan.undeclared == () + assert plan.missing_roles == () + assert plan.missing_tables == () + + +def test_privileges_beyond_the_manifest_are_reported_not_granted_or_revoked() -> None: + plan = plan_reconciliation( + _declared(), + granted={ + RUNTIME: { + "users": {"SELECT", "INSERT", "UPDATE", "DELETE", "TRUNCATE"}, + "legacy_policy_mappings": {"SELECT", "INSERT"}, + "reports": {"SELECT"}, + }, + STAGE12: {"users": {"SELECT"}}, + }, + existing_roles={RUNTIME, STAGE12}, + existing_tables={"users", "legacy_policy_mappings", "reports"}, + ) + + assert plan.grants == () + assert plan.undeclared == ( + PrivilegeEntry(STAGE12, "users", "SELECT"), + PrivilegeEntry(RUNTIME, "legacy_policy_mappings", "INSERT"), + PrivilegeEntry(RUNTIME, "reports", "SELECT"), + PrivilegeEntry(RUNTIME, "users", "TRUNCATE"), + ) + assert grant_statements(plan, _quote) == [] + + +def test_privileges_on_tables_outside_the_manifest_are_reported() -> None: + plan = plan_reconciliation( + _declared(), + granted={RUNTIME: {"scratch_table": {"SELECT"}}}, + existing_roles={RUNTIME, STAGE12}, + existing_tables={"users", "legacy_policy_mappings", "reports", "scratch_table"}, + ) + + assert PrivilegeEntry(RUNTIME, "scratch_table", "SELECT") in plan.undeclared + + +def test_missing_roles_and_tables_are_recorded_and_block_the_plan() -> None: + plan = plan_reconciliation( + _declared(), + granted={}, + existing_roles={RUNTIME}, + existing_tables={"users", "reports"}, + ) + + assert plan.missing_roles == (STAGE12,) + assert plan.missing_tables == ("legacy_policy_mappings",) + assert not plan.can_apply + # Nothing is granted on a missing table or to a missing role. + assert all( + entry.role != STAGE12 and entry.table != "legacy_policy_mappings" + for entry in plan.grants + ) + + +def test_grant_statements_group_privileges_per_role_and_table() -> None: + plan = ReconciliationPlan( + grants=( + PrivilegeEntry(RUNTIME, "users", "DELETE"), + PrivilegeEntry(RUNTIME, "users", "UPDATE"), + PrivilegeEntry(STAGE12, "reports", "SELECT"), + ), + undeclared=(), + missing_roles=(), + missing_tables=(), + ) + + assert grant_statements(plan, _quote) == [ + 'GRANT SELECT ON TABLE "public"."reports" TO "policyengine_stage12_staging"', + 'GRANT DELETE, UPDATE ON TABLE "public"."users" TO "policyengine_v2_runtime"', + ] + + +@pytest.mark.parametrize( + "unsafe_identifier", + ['users"; DROP TABLE users; --', "Users", "users table", ""], +) +def test_grant_statements_refuse_unsafe_identifiers(unsafe_identifier: str) -> None: + plan = ReconciliationPlan( + grants=(PrivilegeEntry(RUNTIME, unsafe_identifier, "SELECT"),), + undeclared=(), + missing_roles=(), + missing_tables=(), + ) + + with pytest.raises(privilege_reconciliation.UnsafeIdentifierError): + grant_statements(plan, _quote) + + +def test_verification_reports_each_declared_privilege_the_role_lacks() -> None: + effective = { + (RUNTIME, "users", "SELECT"), + (RUNTIME, "users", "INSERT"), + (RUNTIME, "users", "UPDATE"), + (RUNTIME, "legacy_policy_mappings", "SELECT"), + } + + failures = verification_failures( + _declared(), + lambda role, table, privilege: (role, table, privilege) in effective, + ) + + assert failures == [PrivilegeEntry(RUNTIME, "users", "DELETE")] + + +def test_summary_lists_grants_extras_and_default_privileges_without_secrets() -> None: + plan = ReconciliationPlan( + grants=(PrivilegeEntry(RUNTIME, "legacy_policy_mappings", "SELECT"),), + undeclared=(PrivilegeEntry(RUNTIME, "policies", "DELETE"),), + missing_roles=(), + missing_tables=(), + ) + defaults = ( + DefaultPrivilege( + creating_role="policyengine_v2_migrator", + schema="public", + object_type="tables", + grantee=RUNTIME, + privileges=("DELETE", "INSERT", "SELECT", "UPDATE"), + ), + ) + + summary = render_summary( + plan, defaults, deployment_environment="staging", applied=True + ) + + assert "## v2 table privileges (staging)" in summary + assert ( + "| `policyengine_v2_runtime` | `legacy_policy_mappings` | SELECT |" in summary + ) + assert "| `policyengine_v2_runtime` | `policies` | DELETE |" in summary + assert "reported, not revoked" in summary + assert "`policyengine_v2_migrator`" in summary + assert "postgresql" not in summary.lower().replace("postgresql's", "") + + +def _fixture_privileges(name: str) -> set[tuple[str, str, str]]: + entries = json.loads((PROOF_FIXTURES / name).read_text(encoding="utf-8")) + return {(entry["role"], entry["table"], entry["privilege"]) for entry in entries} + + +def test_rule_proof_deployed_privileges_match() -> None: + """Proof for rule api.v2.deployed-privileges-match (rules/registry.toml). + + The bad fixture is production's runtime role before the #3977 fix: it + lacked SELECT on ``legacy_policy_mappings``. The good fixture has it. + """ + declared = {RUNTIME: {"users": READ_WRITE, "legacy_policy_mappings": READ}} + bad = _fixture_privileges("bad_effective_privileges.json") + good = _fixture_privileges("good_effective_privileges.json") + + assert verification_failures( + declared, lambda role, table, privilege: (role, table, privilege) in bad + ) == [PrivilegeEntry(RUNTIME, "legacy_policy_mappings", "SELECT")] + assert ( + verification_failures( + declared, lambda role, table, privilege: (role, table, privilege) in good + ) + == [] + ) + + +PERSISTENT_ENVIRONMENT = { + "V2_MIGRATION_DATABASE_URL": ( + "postgresql+psycopg://policyengine_v2_migrator:secret-value@" + "db.abcdefghijklmnopqrst.supabase.co:5432/postgres?sslmode=require" + ), + "V2_SUPABASE_PROJECT_REF": "abcdefghijklmnopqrst", + "V2_SUPABASE_ENVIRONMENT": "staging", + "DEPLOYMENT_ENVIRONMENT": "staging", +} + + +def _engine_never_built(_settings: object) -> object: + msg = "the command must stop before connecting" + raise AssertionError(msg) + + +def test_creating_roles_is_refused_for_a_persistent_database() -> None: + with pytest.raises( + privilege_reconciliation.V2ConfigurationError, match="disposable" + ): + privilege_reconciliation.run( + "apply", + create_roles=True, + environ=PERSISTENT_ENVIRONMENT, + engine_builder=_engine_never_built, + ) + + +@pytest.mark.parametrize("environment", ["", "test", "prod"]) +def test_a_persistent_database_needs_staging_or_production(environment: str) -> None: + with pytest.raises( + privilege_reconciliation.V2ConfigurationError, match="DEPLOYMENT_ENVIRONMENT" + ): + privilege_reconciliation.run( + "verify", + environ=PERSISTENT_ENVIRONMENT | {"DEPLOYMENT_ENVIRONMENT": environment}, + engine_builder=_engine_never_built, + ) + + +def test_configuration_errors_exit_with_status_two_without_the_url( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + for name, value in PERSISTENT_ENVIRONMENT.items(): + monkeypatch.setenv(name, value) + monkeypatch.setenv("DEPLOYMENT_ENVIRONMENT", "test") + + status = privilege_reconciliation.main(["verify"]) + + captured = capsys.readouterr() + assert status == 2 + assert "DEPLOYMENT_ENVIRONMENT" in captured.err + assert "secret-value" not in captured.err + captured.out diff --git a/tests/unit/v2/test_privilege_requirements.py b/tests/unit/v2/test_privilege_requirements.py new file mode 100644 index 000000000..76ba0b5eb --- /dev/null +++ b/tests/unit/v2/test_privilege_requirements.py @@ -0,0 +1,173 @@ +"""How the connector analysis decides which privileges a module needs.""" + +from __future__ import annotations + +import textwrap +from typing import TYPE_CHECKING + +import pytest + +from policyengine_api.data.v2.privilege_requirements import ( + RequiredPrivilege, + UnclassifiedConnectorError, + connector_kind, + required_privileges, +) +from policyengine_api.data.v2.privileges import TablePrivilege + +if TYPE_CHECKING: + from pathlib import Path + +TABLE_BY_MODEL = { + "User": "users", + "UserPolicy": "user_policies", + "LegacyPolicyMapping": "legacy_policy_mappings", +} + + +def _connector(directory: Path, name: str, source: str) -> Path: + path = directory / name + path.write_text(textwrap.dedent(source), encoding="utf-8") + return path + + +def _needed(path: Path) -> set[tuple[str, TablePrivilege]]: + return { + (requirement.table, requirement.privilege) + for requirement in required_privileges(path, TABLE_BY_MODEL) + } + + +@pytest.mark.parametrize( + ("file_name", "kind"), + [ + ("reads.py", "reads"), + ("reads_regions.py", "reads"), + ("creates.py", "creates"), + ("updates.py", "updates"), + ("deletes.py", "deletes"), + ], +) +def test_module_kind_comes_from_the_file_name( + tmp_path: Path, file_name: str, kind: str +) -> None: + assert connector_kind(tmp_path / file_name) == kind + + +def test_an_unclassified_module_name_is_refused(tmp_path: Path) -> None: + with pytest.raises(UnclassifiedConnectorError): + connector_kind(tmp_path / "readers.py") + + +def test_mutation_modules_also_need_select(tmp_path: Path) -> None: + path = _connector( + tmp_path, + "creates.py", + """ + from policyengine_api.data.v2.models import UserPolicy + + def create(session, values): + session.add(UserPolicy(**values)) + """, + ) + + assert _needed(path) == { + ("user_policies", TablePrivilege.INSERT), + ("user_policies", TablePrivilege.SELECT), + } + + +def test_type_checking_imports_and_annotations_need_nothing(tmp_path: Path) -> None: + path = _connector( + tmp_path, + "reads.py", + """ + from __future__ import annotations + + from typing import TYPE_CHECKING + + from policyengine_api.data.v2.models import User, UserPolicy + + if TYPE_CHECKING: + from policyengine_api.data.v2.models import LegacyPolicyMapping + + def read(session, user_id) -> UserPolicy | None: + return session.get(User, user_id) + + def describe(mapping: LegacyPolicyMapping) -> str: + return str(mapping) + """, + ) + + assert _needed(path) == {("users", TablePrivilege.SELECT)} + + +def test_a_row_lock_needs_update_on_the_selected_table_only(tmp_path: Path) -> None: + path = _connector( + tmp_path, + "reads.py", + """ + from sqlmodel import select + + from policyengine_api.data.v2.models import LegacyPolicyMapping, User + + def read_mapping(session, *, lock): + statement = select(LegacyPolicyMapping) + if lock: + statement = statement.with_for_update() + return session.exec(statement).one_or_none() + + def read_user(session, user_id): + return session.get(User, user_id) + """, + ) + + requirements = required_privileges(path, TABLE_BY_MODEL) + + assert ( + RequiredPrivilege( + table="legacy_policy_mappings", + privilege=TablePrivilege.UPDATE, + connector="reads.py", + reason="read_mapping locks rows with with_for_update", + ) + in requirements + ) + assert ("users", TablePrivilege.UPDATE) not in _needed(path) + + +def test_a_share_lock_also_needs_update(tmp_path: Path) -> None: + path = _connector( + tmp_path, + "reads.py", + """ + from sqlmodel import select + + from policyengine_api.data.v2.models import User + + def read_user(session, user_id): + return session.exec(select(User).with_for_update(read=True)).one() + """, + ) + + assert ("users", TablePrivilege.UPDATE) in _needed(path) + + +def test_a_lock_with_of_needs_update_on_the_named_table_only(tmp_path: Path) -> None: + path = _connector( + tmp_path, + "reads.py", + """ + from sqlmodel import select + + from policyengine_api.data.v2.models import User, UserPolicy + + def read_with_owner(session): + return session.exec( + select(UserPolicy).join(User).with_for_update(of=UserPolicy) + ).all() + """, + ) + + assert ("user_policies", TablePrivilege.UPDATE) in _needed(path) + assert ("users", TablePrivilege.UPDATE) not in _needed(path) diff --git a/tests/unit/v2/test_stage12_comparison_service.py b/tests/unit/v2/test_stage12_comparison_service.py index 072b655f0..c112cd165 100644 --- a/tests/unit/v2/test_stage12_comparison_service.py +++ b/tests/unit/v2/test_stage12_comparison_service.py @@ -2,9 +2,9 @@ from __future__ import annotations -from datetime import UTC, datetime, timedelta +from datetime import timedelta from unittest.mock import Mock -from uuid import UUID, uuid4 +from uuid import uuid4 import pytest from sqlalchemy.dialects import postgresql @@ -24,8 +24,6 @@ simulation_record, ) from policyengine_api.services.v2.comparison_runs.types import ( - ComparisonReportRecord, - ComparisonSimulationRecord, ResultComparisonStatus, ) from policyengine_api.services.v2.comparison_runs.validators import ( @@ -36,74 +34,26 @@ require_report_identity, require_result_comparison_transition, ) - -NOW = datetime(2026, 9, 14, tzinfo=UTC) -EVALUATION_ID = UUID("00000000-0000-4000-8000-000000000001") -SIMULATION_ID = UUID("00000000-0000-4000-8000-000000000002") -DIGEST_A = "a" * 64 -DIGEST_B = "b" * 64 - - -def _report(**changes: object) -> ComparisonReportRecord: - fields: dict[str, object] = { - "evaluation_id": EVALUATION_ID, - "status": "pending", - "aggregation_status": "not_started", - "environment": "test", - "calculation_flow": "economy", - "originating_request_id": "request-1", - "production_identity": "production-job-1", - "incumbent_execution_id": "production-job-1", - "worker_version": "5.2.0", - "modal_application": "policyengine-v2-worker-5-2-0", - "report_coordinator_callable": "coordinate_report", - "version_manifest_sha256": DIGEST_A, - "policyengine_version": "5.2.0", - "country_package_name": "policyengine-us", - "country_package_version": "1.900.0", - "country": "us", - "dataset_identity": "populace_us_2024", - "dataset_uri": "hf://policyengine/populace-us/data.h5@revision", - "data_package_name": "policyengine-us-data", - "data_package_version": "1.0.0", - "data_artifact_revision": "revision", - "created_at": NOW, - "updated_at": NOW, - "retention_expires_at": NOW + timedelta(days=30), - } - fields.update(changes) - return ComparisonReportRecord.model_validate(fields) - - -def _simulation(**changes: object) -> ComparisonSimulationRecord: - fields: dict[str, object] = { - "simulation_execution_id": SIMULATION_ID, - "evaluation_id": EVALUATION_ID, - "role": "baseline", - "input_sha256": DIGEST_B, - "worker_version": "5.2.0", - "modal_application": "policyengine-v2-worker-5-2-0", - "simulation_callable": "run_simulation", - "version_manifest_sha256": DIGEST_A, - "status": "pending", - "created_at": NOW, - "updated_at": NOW, - "retention_expires_at": NOW + timedelta(days=30), - } - fields.update(changes) - return ComparisonSimulationRecord.model_validate(fields) +from tests.unit.v2.comparison_run_support import ( + DIGEST_A, + DIGEST_B, + EVALUATION_ID, + NOW, + comparison_report_record, + comparison_simulation_record, +) def test_insert_statements_use_the_parent_and_child_identities() -> None: report_statement = ( creates.insert(creates.Stage12ComparisonReport) - .values(**report_insert_values(_report())) + .values(**report_insert_values(comparison_report_record())) .on_conflict_do_nothing(constraint="uq_stage12_eval_reports_identity") .returning(creates.Stage12ComparisonReport.evaluation_id) ) simulation_statement = ( creates.insert(creates.Stage12ComparisonSimulation) - .values(**simulation_insert_values(_simulation())) + .values(**simulation_insert_values(comparison_simulation_record())) .on_conflict_do_nothing(constraint="uq_stage12_eval_simulations_identity") .returning(creates.Stage12ComparisonSimulation.simulation_execution_id) ) @@ -123,8 +73,8 @@ def test_insert_statements_use_the_parent_and_child_identities() -> None: def test_contract_records_round_trip_through_sqlmodel_rows() -> None: - report = _report() - simulation = _simulation() + report = comparison_report_record() + simulation = comparison_simulation_record() report_row = Stage12ComparisonReport(**report_insert_values(report)) simulation_row = Stage12ComparisonSimulation(**simulation_insert_values(simulation)) @@ -133,7 +83,7 @@ def test_contract_records_round_trip_through_sqlmodel_rows() -> None: def test_conflicting_report_creation_reuses_the_stored_identity(monkeypatch) -> None: - stored = _report() + stored = comparison_report_record() row = Stage12ComparisonReport(**report_insert_values(stored)) retried = stored.model_copy( update={ @@ -161,7 +111,7 @@ def test_conflicting_report_creation_reuses_the_stored_identity(monkeypatch) -> def test_conflicting_report_creation_rejects_different_bundle_provenance( monkeypatch, ) -> None: - stored = _report() + stored = comparison_report_record() row = Stage12ComparisonReport(**report_insert_values(stored)) candidate = stored.model_copy(update={"country_package_version": "1.901.0"}) monkeypatch.setattr(services, "create_comparison_report", lambda *_args: None) @@ -176,8 +126,10 @@ def test_conflicting_report_creation_rejects_different_bundle_provenance( def test_matching_successful_child_is_returned_for_a_retry(monkeypatch) -> None: - parent_row = Stage12ComparisonReport(**report_insert_values(_report())) - stored = _simulation( + parent_row = Stage12ComparisonReport( + **report_insert_values(comparison_report_record()) + ) + stored = comparison_simulation_record( status="succeeded", completed_at=NOW + timedelta(minutes=1), output_uri="gs://private-stage12/simulations/baseline.parquet", @@ -188,7 +140,7 @@ def test_matching_successful_child_is_returned_for_a_retry(monkeypatch) -> None: row_identity_sha256=DIGEST_B, ) row = Stage12ComparisonSimulation(**simulation_insert_values(stored)) - retried = _simulation(simulation_execution_id=uuid4()) + retried = comparison_simulation_record(simulation_execution_id=uuid4()) monkeypatch.setattr( services, "create_comparison_simulation", @@ -227,8 +179,10 @@ def test_child_creation_rejects_parent_provenance_or_retention_mismatch( field_name, other_value, ) -> None: - parent_row = Stage12ComparisonReport(**report_insert_values(_report())) - child = _simulation(**{field_name: other_value}) + parent_row = Stage12ComparisonReport( + **report_insert_values(comparison_report_record()) + ) + child = comparison_simulation_record(**{field_name: other_value}) create = Mock() monkeypatch.setattr( services, @@ -253,13 +207,13 @@ def test_child_creation_requires_an_existing_parent(monkeypatch) -> None: monkeypatch.setattr(services, "create_comparison_simulation", create) with pytest.raises(LookupError, match="does not exist"): - services.create_or_resolve_simulation(Mock(), _simulation()) + services.create_or_resolve_simulation(Mock(), comparison_simulation_record()) create.assert_not_called() def test_successful_report_cannot_be_overwritten(monkeypatch) -> None: - stored = _report( + stored = comparison_report_record( status="succeeded", aggregation_status="succeeded", aggregate_output_uri="gs://private-stage12/reports/aggregate.json", @@ -284,7 +238,7 @@ def test_successful_report_cannot_be_overwritten(monkeypatch) -> None: def test_successful_report_can_record_a_result_comparison(monkeypatch) -> None: - stored = _report( + stored = comparison_report_record( status="succeeded", aggregation_status="succeeded", aggregate_output_uri="gs://private-stage12/reports/aggregate.json", @@ -320,7 +274,7 @@ def test_successful_report_can_record_a_result_comparison(monkeypatch) -> None: def test_lifecycle_update_cannot_change_result_comparison(monkeypatch) -> None: - stored = _report(comparison_status="pending") + stored = comparison_report_record(comparison_status="pending") row = Stage12ComparisonReport(**report_insert_values(stored)) monkeypatch.setattr( services, @@ -341,14 +295,14 @@ def test_lifecycle_update_cannot_change_result_comparison(monkeypatch) -> None: def test_comparison_contract_rejects_completed_state_without_artifact() -> None: with pytest.raises(ValueError, match="requires an output artifact"): - _report( + comparison_report_record( comparison_status="different", comparison_completed_at=NOW, ) def test_successful_simulation_replay_is_read_only(monkeypatch) -> None: - stored = _simulation( + stored = comparison_simulation_record( status="succeeded", completed_at=NOW + timedelta(minutes=1), output_uri="gs://private-stage12/simulations/baseline.parquet", @@ -377,7 +331,7 @@ def test_successful_simulation_replay_is_read_only(monkeypatch) -> None: def test_successful_simulation_cannot_be_overwritten(monkeypatch) -> None: - stored = _simulation( + stored = comparison_simulation_record( status="succeeded", completed_at=NOW + timedelta(minutes=1), output_uri="gs://private-stage12/simulations/baseline.parquet", @@ -406,7 +360,7 @@ def test_successful_simulation_cannot_be_overwritten(monkeypatch) -> None: def test_lifecycle_updates_reject_identity_changes_and_state_reversal() -> None: - pending = _report() + pending = comparison_report_record() changed_identity = pending.model_copy(update={"worker_version": "5.3.0"}) with pytest.raises(ComparisonRunIdentityError, match="worker_version"): require_report_identity(pending, changed_identity) @@ -458,7 +412,7 @@ def test_lifecycle_updates_reject_identity_changes_and_state_reversal() -> None: @pytest.mark.parametrize( "record_factory", - [_report, _simulation], + [comparison_report_record, comparison_simulation_record], ) def test_contract_records_require_bounded_timezone_aware_retention( record_factory,