diff --git a/.secrets.baseline b/.secrets.baseline index 5891e9cd..c4cc8d9a 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -1019,7 +1019,7 @@ "filename": "tests/iac/conftest.py", "hashed_secret": "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3", "is_verified": false, - "line_number": 835 + "line_number": 843 } ], "tests/iac/test_iac_aws_apply_engine_moto.py": [ @@ -1028,7 +1028,7 @@ "filename": "tests/iac/test_iac_aws_apply_engine_moto.py", "hashed_secret": "dc724af18fbdd4e59189f5fe768a5f8311527050", "is_verified": false, - "line_number": 103 + "line_number": 110 } ], "tests/iac/test_iac_aws_dbt_athena_e2e.py": [ @@ -2154,5 +2154,5 @@ } ] }, - "generated_at": "2026-10-02T17:10:21Z" + "generated_at": "2026-10-05T22:13:49Z" } diff --git a/docs/apply.md b/docs/apply.md index 5c4aeba9..dae8bdc9 100644 --- a/docs/apply.md +++ b/docs/apply.md @@ -232,7 +232,8 @@ must not land silently. When the first expose's binding is a GCP `bigquery_table`, the result is written as Parquet under `.fluid/staging//` and one load job moves it -into that table (`WRITE_TRUNCATE`, `CREATE_NEVER`, the table's own schema), +into that table (`WRITE_TRUNCATE_DATA`, which keeps the table's row access policies, +`CREATE_NEVER`, the table's own schema), the load the duckdb acquisition runner performs; a `gs://` `location.path` on the binding is never written to. A failed or short load fails the build. Any other landing this path cannot write, a `gs://` or other non-S3 URI, a GCS diff --git a/docs/governance-parity.md b/docs/governance-parity.md index 552f21ab..375efcd3 100644 --- a/docs/governance-parity.md +++ b/docs/governance-parity.md @@ -16,13 +16,20 @@ contract needs `fluidVersion: "0.7.6"` to validate with them. | **Retention**: `exposes[].lifecycle {retention, expire: true}` | One S3 lifecycle rule per expose, filtered to the binding's prefix, expiring objects `retention` after they are written (`aws_s3_bucket_lifecycle_configuration`). | Daily partitions that expire `retention` after their day ends: `google_bigquery_table.time_partitioning {type: DAY, expiration_ms}`, by ingestion time, or on `binding.location.partitionBy` when it names one date or timestamp column (see "Retention counts from the partition's date" below). Never a table expiration, which would delete the product. | AWS: the enabled prefix rule's `Expiration.Days`, and no rule that expires sooner. GCP: the table's partition type, field and `expirationMs`, and that the table itself has no `expirationTime`. | | **Encryption at rest**: `binding.encryption.kms` | `product`: a KMS key per bucket (`aws_kms_key`, rotation on, alias `alias/fluid//`) as the bucket's default SSE-KMS. `alias/...` or an ARN: that key. `none`: SSE-S3. | `product`: a key ring and key per dataset (`google_kms_key_ring` `fluid--`, `google_kms_crypto_key` `bigquery`, 90-day rotation) in the dataset's location, `roles/cloudkms.cryptoKeyEncrypterDecrypter` for the BigQuery service agent (`google_kms_crypto_key_iam_member`), used as the dataset's `default_encryption_configuration` and the table's `encryption_configuration`. `projects/.../cryptoKeys/...`: that key. `none`: Google-managed keys. Every table of one dataset must declare the same key (see "One dataset, one key"). | AWS: the key is `Enabled`, and the objects under the prefix are SSE-KMS with it. GCP: `kmsKeyName` of the table, and of the dataset's default when the product owns the dataset. | | **Column restrictions**: `exposes[].policy.authz.columnRestrictions` | Each Lake Formation grant with `SELECT` excludes the restricted columns its principal may not read (`aws_lakeformation_permissions.table_with_columns.excluded_column_names`, with `wildcard`). | A Data Catalog taxonomy per product and dataset with fine-grained access control (`google_data_catalog_taxonomy`), a policy tag per set of restricted columns that share their readers (`google_data_catalog_policy_tag`), attached through the table schema's `policyTags`, and `roles/datacatalog.categoryFineGrainedReader` for exactly the allowed readers (`google_data_catalog_policy_tag_iam_member`). The restrictions' `tags` and `labels` go into the policy tag's description. | AWS: `lakeformation:ListPermissions` on tables; no `SELECT` of a denied principal, of a read grantee, or of `IAM_ALLOWED_PRINCIPALS` reaches a column it may not read, including grants made outside the contract. GCP: every restricted column carries a tag, the tag's fine-grained readers are exactly the derived set (a denied principal, or one granted outside the contract, fails), and the taxonomy enforces fine-grained access control. | +| **Column masking**: `columnRestrictions[] {access: mask, mask: last_four \| first_four \| nullify}` | Lake Formation cannot mask. The masked principal's grant excludes the masked columns from the table, as a deny does, and a protected multi-dialect Glue Data Catalog view returns them masked (`aws_glue_catalog_table`, `table_type = VIRTUAL_VIEW`, `view_definition {is_protected, definer, representations {dialect = ATHENA, validation_connection}}`, named `_masked`): exactly the columns the principal's own grant reads raw (its `columns` list, or every column less its exclusions), the masked ones through the same rule in Athena SQL, and the principal's row filter as the view's `WHERE`. The view reads the table as its definer, so it selects nothing the grant does not. The definer gets `SELECT` with grant option on the table and `CREATE_TABLE` and `DESCRIBE` on the database (Lake Formation's view prerequisites), the principal `SELECT` and `DESCRIBE` on the view. Needs `governance.lakeFormation.maskedViews {definer, validationConnection}`. | The masked columns get a policy tag of their own with a BigQuery data policy on it (`google_bigquery_datapolicy_data_policy`, `DATA_MASKING_POLICY`, predefined `LAST_FOUR_CHARACTERS`, `FIRST_FOUR_CHARACTERS` or `ALWAYS_NULL`) and `roles/bigquerydatapolicy.maskedReader` on the data policy for the masked principals (`google_bigquery_datapolicy_data_policy_iam_member`). The tag's fine-grained readers read the clear value. | AWS: the masked principal's table grant excludes the masked columns (the column check above). GCP: the masked tag's fine-grained readers are exactly the derived set. Not checked yet: the data policy and its masked readers, and the view's SQL. | +| **Row filters**: `exposes[].policy.authz.rowFilters[] {principal, where, name?}` | A Lake Formation data cells filter per filter (`aws_lakeformation_data_cells_filter`, `row_filter.filter_expression` the predicate, `column_wildcard.excluded_column_names` the principal's excluded columns), and the principal's `SELECT` granted on the filter instead of the table. | A row access policy per filter (`google_bigquery_row_access_policy`, `filter_predicate` the predicate, `grantees` every identity of the principal, whatever it holds: a writer reads rows too), and one more, `fluid_all_rows` (`TRUE`), for every other principal the expose grants anything to, never for a filtered one, since BigQuery shows a principal no policy names no row at all and the union of those that do. | AWS: the column check reads a grant made on a data cells filter, so a filtered principal's excluded columns are still checked. Not checked yet: the filter expression on AWS, and the row access policies on GCP. | +| **Governance labels**: `labels`, `exposes[].labels`, `dataClassification`, `sovereignty {jurisdiction, regulatoryFramework, dataResidency}` | The same keys as AWS tags on the bucket and the key, and as parameters on the Glue table. | GCP labels on the dataset, the table and the key: the contract's and the expose's labels, and `fluid_classification`, `fluid_jurisdiction`, `fluid_regulation` (the frameworks joined), `fluid_residency`. | Not checked yet. | | **Access grants**: `accessPolicy.grants[]` | Not emitted: on AWS, access is the binding's `governance.lakeFormation.grants`. `fluid validate` warns only for an aws binding with no Lake Formation grants, where the contract's access intent is unenforced. | One non-authoritative `google_bigquery_dataset_iam_member` per role and member (and `google_storage_bucket_iam_member` for GCS), with the logical principal mapped to its GCP identity. | Not checked yet on either cloud. | A policy that a binding cannot apply is refused at `fluid validate`, `fluid plan` and `fluid apply`, never dropped: retention, a key or a column restriction on a GCP binding that is not a BigQuery table (GCS, Pub/Sub, Iceberg storage), a column restriction on an AWS binding with no Lake Formation grants or on a non-Glue format, -an AWS key reference on GCP and a Cloud KMS key name on AWS. +an AWS key reference on GCP and a Cloud KMS key name on AWS, a mask on an AWS binding +with no `maskedViews` (`column-mask-unenforceable`), a text mask on a column that is +not a string (`column-mask-type`), a row filter whose principal holds no Lake +Formation read grant (`row-filter-unenforceable`), and a row filter on a BigQuery view +(`row-filter-view`) or on a GCP binding that is not a BigQuery table. ## Logical principals and `binding.principals` @@ -85,7 +92,11 @@ The pattern follows ODCS v3, which declares `roles[]` once and binds them per se * A column named in any restriction is restricted. * `deny`: the principal may not read the columns. * `allow`: the columns are readable only by the principals an `allow` names. -* A deny beats an allow. A restriction never grants access: the readers are the +* `mask` (with `mask: last_four`, `first_four` or `nullify`): the principal reads the + columns masked, never clear. A masked principal is denied the raw column, so on + every check that counts readers it counts as denied. A rule needs `access: mask`, a + mask needs its rule, and a column has one rule. +* A deny beats an allow and a mask. A restriction never grants access: the readers are the expose's readers. On GCP they are the `accessPolicy` read grantees and the expose's own `policy.authz.readers` (whose table access is managed elsewhere, so only the fine-grained reader role on the tag is granted to them); on AWS, the Lake @@ -103,12 +114,77 @@ The pattern follows ODCS v3, which declares `roles[]` once and binds them per se (`column-restriction-conflict`). On GCP a denied principal gets an access error on the restricted columns; -`SELECT * EXCEPT (customer_id, msisdn)` still works for it. BigQuery dynamic data -masking (`google_bigquery_datapolicy_data_policy`, SHA-256 or nullify, for principals -holding `roles/bigquerydatapolicy.maskedReader`) is not emitted yet. It needs a third -principal set (who sees masked values rather than an error) that the contract has no -field for, and a column masked at landing (`policy.privacy.masking`) would be hashed -twice. +`SELECT * EXCEPT (customer_id, msisdn)` still works for it. + +## Masking: the platform's, and the landing's + +`access: mask` is the platform masking a column at read time, for its principal only; +the stored value is clear and its other readers read it. `policy.privacy.masking` is +the build hashing (or tokenizing) the value before it lands, so nobody reads it clear. +A contract can carry both on two columns of one value, as a column kept clear for its +stewards and masked for analysts, beside a salted hash every reader may join on. +Masking a column that `privacy.masking` already hashes would mask the hash. + +* **GCP**: BigQuery's predefined rules. `last_four` is `LAST_FOUR_CHARACTERS` + (`XXXXX` and the last four characters), `first_four` is `FIRST_FOUR_CHARACTERS`, + `nullify` is `ALWAYS_NULL`. The data policy is attached to the policy tag of the + masked columns, so the columns' fine-grained readers read the clear value and the + masked readers read the masked one. A principal with neither is refused, as for a + deny. BigQuery applies a new masked reader about a minute or two after the grant. + A data policy id longer than BigQuery takes (it refuses 200 characters or more) keeps + its first characters and ends in a hash of the whole. +* **AWS**: Lake Formation has no masking, so the masked value comes from a protected + Glue Data Catalog view (Lake Formation's multi-dialect views, run as their + definer). The view returns the rule's value in Athena SQL: + `concat('XXXXX', substr(c, -4))` for `last_four`, the first four and `XXXXX` for + `first_four`, `NULL` for `nullify`, and NULL stays NULL. A value of four characters + or fewer is returned as its SHA-256, since its last four characters would be all of + it. The principal keeps its grant on the table, without the masked columns. +* The view needs, per account: an Athena workgroup, a Glue connection of type + `VIEW_VALIDATION_ATHENA` naming it (`validationConnection`), and a definer role + that Glue and Lake Formation may assume (`sts:AssumeRole` and `sts:SetContext` for + `glue.amazonaws.com` and `lakeformation.amazonaws.com`). The identity running + `fluid apply` needs `iam:PassRole` on the definer and `glue:PassConnection` on the + connection. A validated view carries no storage descriptor. +* The view is created by terraform-provider-aws 6 (`aws_glue_catalog_table` + `view_definition`, which 5.x does not have), which forge-cli pins as `~> 6.0`. + +## Row filters + +`exposes[].policy.authz.rowFilters[]` names a principal and one SQL boolean +expression over the expose's columns; that principal reads only the rows it +selects. A reader no filter names reads every row, and a principal has one filter +per expose. The predicate is parsed (sqlglot) before it is written into a policy: a +statement separator, a comment, a subquery or a column the schema does not declare is +refused. Write it NULL-aware: `is_suppressed = false` drops the rows where the column +is NULL, which `(is_suppressed IS NULL OR is_suppressed = false)` keeps. + +```yaml +policy: + authz: + rowFilters: + - principal: group:analysts@northwind.example + name: analysts_consented + where: consent_data_analytics = true AND (is_suppressed IS NULL OR is_suppressed = false) +``` + +* **GCP**: a row access policy per filter, naming every identity of its principal, + whatever the principal holds (a row access policy never grants the table, and a + writer's `roles/bigquery.dataEditor` reads rows), and `fluid_all_rows` (`FILTER USING + (TRUE)`) for every other principal the expose grants anything to, the pipeline's + identity included, because BigQuery shows a principal that no row access policy + names no row at all. A filtered identity is never in `fluid_all_rows`: BigQuery + shows a principal the union of the policies naming it. +* **AWS**: a data cells filter per filter; the principal's `SELECT` is granted on the + filter, with the principal's excluded columns as its column wildcard's exclusions, + instead of on the table. The masked view of a filtered principal applies the same + predicate, since the view reads the table as its definer. +* **A build keeps them.** BigQuery removes every row access policy of a table that a + `WRITE_TRUNCATE` load writes, so a full-refresh build loads with + `WRITE_TRUNCATE_DATA`, which replaces the rows and keeps the table's policies, tags + and partitioning (measured on BigQuery, 5 October 2026, on an unpartitioned and on a + DAY-partitioned table with a partition expiration). It is the disposition dlt's + BigQuery destination uses for the same reason. ## Dataset grants are no longer authoritative @@ -208,6 +284,10 @@ keep data longer, set a longer `retention` instead. (create key rings and keys, set their IAM), `roles/datacatalog.categoryAdmin` (taxonomies, tags and their IAM), and `bigquery.datasets.update` on the datasets (dataset IAM members). +* For masking and row filters on GCP: the BigQuery Data Policy API + (`bigquerydatapolicy.googleapis.com`) enabled, and `bigquery.dataPolicies.*` and + `bigquery.rowAccessPolicies.*` for the identity running `fluid apply`. + `roles/bigquery.dataOwner` on the project carries both. * A key ring and a crypto key cannot be deleted on GCP. `tofu destroy` removes them from state and schedules the key's versions for destruction (30 days by default); the next apply adopts the same names (`discover_imports`). @@ -252,6 +332,22 @@ keep data longer, set a longer `retention` instead. "User has neither fine-grained reader nor masked get permission to get data protected by policy tag … on column …", not the documented "does not have permission to access policy tag". +* **Masking and row filters, measured by hand**, 5 October 2026, on the same lab's + customer profile table, with the resources this release emits made by hand: + * BigQuery: a `LAST_FOUR_CHARACTERS` data policy on a policy tag, the analyst a + masked reader, read `XXXXX` and the last four digits, and the steward, a + fine-grained reader, read the number. + * AWS: a protected Glue view with the Athena expression above returned every row + masked to the analyst, who was still refused the column on the table. + * AWS: a data cells filter with the NULL-aware consent predicate returned 752 of + 10,172 rows to the analyst. + * BigQuery: a `WRITE_TRUNCATE` load removed a table's row access policy and the + table then showed every row; a `WRITE_TRUNCATE_DATA` load kept it. +* `tofu plan` with terraform-provider-aws 6.67 against the live state of the lab's + eleven AWS products planned no change for any of them, with only the provider pin + moved from `~> 5.0`. +* **Not proven yet**: an apply of the masking, row filter and label resources exactly + as forge-cli emits them, on either cloud. * **Not proven**: the Lake Formation half against a real account as 0.17.0 derives it from `columnRestrictions`. The grant shape it emits (excluded columns beside `wildcard`) was applied and enforced on a real account from 0.16.6, written by hand diff --git a/fluid_build/build_runners/_bigquery_load.py b/fluid_build/build_runners/_bigquery_load.py index af0d0cf2..4a8956d9 100644 --- a/fluid_build/build_runners/_bigquery_load.py +++ b/fluid_build/build_runners/_bigquery_load.py @@ -40,6 +40,21 @@ ``DATETIME`` the naive one). The wall-clock value is read as UTC, which is how BigQuery reads a zone-less timestamp literal too. +**Row access policies.** A full refresh is ``WRITE_TRUNCATE_DATA``, not +``WRITE_TRUNCATE``. BigQuery documents that "operations that use the +``WRITE_TRUNCATE`` write disposition remove all existing row access policies" +(row-level security, "Limitations"), so every build would have dropped the +row filters ``tofu apply`` had just made, and each filtered principal would +have read every row until the next apply. ``WRITE_TRUNCATE_DATA`` replaces +the rows and keeps the table: its schema, policy tags and row access +policies. It is the disposition dlt's BigQuery destination loads with for +the same reason (``truncate-and-insert`` with ``enable_atomic_replace``). +Measured on BigQuery, 5 October 2026: after a ``WRITE_TRUNCATE`` load the +table had no row access policy and showed every row; after a +``WRITE_TRUNCATE_DATA`` load it kept its policy, and on a DAY-partitioned +table with a partition expiration it replaced the rows of every partition +and kept the partitioning and the expiration. + **Emulators.** With ``BIGQUERY_EMULATOR_HOST`` set, python-bigquery already sends every request to that host, but it still resolves Application Default Credentials first: with none it raises, and with some it sends a real OAuth @@ -47,7 +62,11 @@ ``AnonymousCredentials`` instead, the way goccy/bigquery-emulator documents its clients. The goccy emulator also reports no ``outputRows`` for a load job, so a load whose job reports no count is checked by counting the table -afterwards; it is never taken as zero rows loaded, nor as success. +afterwards; it is never taken as zero rows loaded, nor as success. Its load +handler knows ``WRITE_TRUNCATE`` and ``WRITE_EMPTY`` and appends for any +other disposition (goccy/bigquery-emulator ``server/handler.go``), so a full +refresh on an emulator is ``WRITE_TRUNCATE``: it has no row access policy to +keep. """ from __future__ import annotations @@ -57,8 +76,9 @@ from typing import Any, Dict, Mapping, Optional # Write disposition per source mode. The other modes (merge, dedup, CDC) need a -# MERGE this path does not do, and are refused rather than approximated. -_WRITE_DISPOSITION = {"full_refresh": "WRITE_TRUNCATE", "incremental_append": "WRITE_APPEND"} +# MERGE this path does not do, and are refused rather than approximated. A full +# refresh keeps the table's row access policies (see "Row access policies"). +_WRITE_DISPOSITION = {"full_refresh": "WRITE_TRUNCATE_DATA", "incremental_append": "WRITE_APPEND"} _SOURCE_FORMAT = {"parquet": "PARQUET"} # The project order ``terraform-provider-google`` reads, so the load goes to the @@ -87,6 +107,19 @@ def emulator_host() -> Optional[str]: return host or None +def write_disposition(mode: str) -> str: + """The load job's write disposition for ``mode``. + + ``WRITE_TRUNCATE_DATA`` for a full refresh, which keeps the table's row + access policies, except on an emulator: goccy/bigquery-emulator appends for + a disposition it does not know, and it has no row access policy to keep. + """ + disposition = _WRITE_DISPOSITION[mode] + if disposition == "WRITE_TRUNCATE_DATA" and emulator_host() is not None: + return "WRITE_TRUNCATE" + return disposition + + def bigquery_client(bigquery: Any, project: Optional[str]) -> Any: """A ``bigquery.Client`` for ``project``: ADC, or anonymous on an emulator. @@ -272,15 +305,16 @@ def load_file( The table's own schema is passed to the job. Without it, a Parquet file whose columns are all optional cannot load into REQUIRED columns - (googleapis/python-bigquery#2373), and ``WRITE_TRUNCATE`` would replace + (googleapis/python-bigquery#2373), and a truncating load would replace the declared schema with the file's. A ``TIMESTAMP`` column the file holds as a naive timestamp is loaded from a UTC-adjusted copy - (:func:`utc_adjusted_copy`). + (:func:`utc_adjusted_copy`). A full refresh is ``WRITE_TRUNCATE_DATA`` + (:func:`write_disposition`), so the table keeps its row access policies. The count is the job's ``outputRows``. A job that reports none (the goccy - emulator never does) is checked by counting the table: for - ``WRITE_TRUNCATE`` the table must then hold exactly the file's rows, and - for ``WRITE_APPEND`` it must have grown by them, which needs the count + emulator never does) is checked by counting the table: after a full + refresh the table must hold exactly the file's rows, and after + ``WRITE_APPEND`` it must have grown by them, which needs the count before the load; that is taken on an emulator only, and a real job with no count and no earlier count is refused rather than assumed. """ @@ -302,7 +336,7 @@ def load_file( before = _count_table(client, table_id, location) job_config = bigquery.LoadJobConfig( source_format=_SOURCE_FORMAT[sink_format], - write_disposition=_WRITE_DISPOSITION[mode], + write_disposition=write_disposition(mode), create_disposition="CREATE_NEVER", schema=table.schema, ) diff --git a/fluid_build/build_runners/_embedded_sql_io.py b/fluid_build/build_runners/_embedded_sql_io.py index 0e2dd032..6db965aa 100644 --- a/fluid_build/build_runners/_embedded_sql_io.py +++ b/fluid_build/build_runners/_embedded_sql_io.py @@ -1266,8 +1266,9 @@ def remove_staged(paths: Sequence[Path]) -> None: def load_bigquery_landing(landing: BigQueryLanding, *, logger: logging.Logger) -> Dict[str, Any]: """Load the staged result into its table: the acquisition runner's own load. - ``WRITE_TRUNCATE``: the embedded-SQL result replaces the table, as it - replaces the local file or the S3 object on the other targets. The count + ``WRITE_TRUNCATE_DATA``: the embedded-SQL result replaces the table's rows, + as it replaces the local file or the S3 object on the other targets, and + the table keeps its schema, policy tags and row access policies. The count the load is held to is read back from the staged file. """ from ._bigquery_load import load_file @@ -1312,7 +1313,7 @@ def write_bigquery_run_record( landed, as bronze is: ``records_total`` is the load's count, which :func:`load_bigquery_landing` has already held to the staged file's rows (``rows_from: write``), and the mode is ``full_refresh`` (the load is - ``WRITE_TRUNCATE``). It is dbt's pattern too: ``run_results.json`` keeps + ``WRITE_TRUNCATE_DATA``). It is dbt's pattern too: ``run_results.json`` keeps each node's ``relation_name`` and ``adapter_response.rows_affected``. A failed run is recorded without ``bigquery_load``: whether it changed @@ -1496,8 +1497,8 @@ def refuse_landing_into_input(io: EmbeddedSqlIO) -> None: resolve to rather than on product ids: :func:`_refuse_self_consume` already refuses a contract consuming its own id, but an overlay naming an upstream's table (a typo, or two products left to the ``default`` - dataset with the same expose id) passed it, and the build's - ``WRITE_TRUNCATE`` replaced another product's rows with its query result. + dataset with the same expose id) passed it, and the build's full-refresh + load replaced another product's rows with its query result. That is a data write, not a planned delete, so ``--allow-data-loss`` is never asked. An S3 landing inside the prefix an input reads is refused for the same reason: the object would become rows of that product's table. @@ -1518,7 +1519,7 @@ def _refuse_loading_an_input(bq: BigQueryLanding, inputs: Sequence[ResolvedInput f"{bq.table_id}, which {_entry(r.product_id, r.expose_id)} reads" ), why=( - f"The load replaces the table (WRITE_TRUNCATE), so {r.product_id}'s rows " + f"The load replaces the table's rows (WRITE_TRUNCATE_DATA), so {r.product_id}'s rows " "would be overwritten with this build's query result." ), fix=( diff --git a/fluid_build/build_runners/base.py b/fluid_build/build_runners/base.py index 65097ac5..e91f80bc 100644 --- a/fluid_build/build_runners/base.py +++ b/fluid_build/build_runners/base.py @@ -364,7 +364,7 @@ def _print_embedded_sql_io(io: Any) -> None: cprint( f" ➡ lands BigQuery table {io.bigquery_landing.table_id} " f"({io.bigquery_landing.location}): staged as Parquet, then one load job " - "(WRITE_TRUNCATE), recorded in the build's run record", + "(WRITE_TRUNCATE_DATA), recorded in the build's run record", markup=False, ) for warning in getattr(io, "warnings", None) or []: diff --git a/fluid_build/cli/_verify_lf_columns.py b/fluid_build/cli/_verify_lf_columns.py index 3f32cadf..7f1f262b 100644 --- a/fluid_build/cli/_verify_lf_columns.py +++ b/fluid_build/cli/_verify_lf_columns.py @@ -24,6 +24,13 @@ them: a table-level grant (every column), a column list naming it, or a column wildcard that does not exclude it. So a grant made outside the contract is caught too. A mismatch is CRITICAL; a check that could not run is an error. + +A grant on a data cells filter (how a contract's ``policy.authz.rowFilters`` reach +Lake Formation) counts too: ``ListPermissions`` with ``ResourceType: TABLE`` does not +return those, so they are listed apart, and each filter's own columns +(``GetDataCellsFilter``: its column list, or its column wildcard and exclusions) +are what the grant reaches. A filter that cannot be read reaches every column, so a +check that cannot see it fails rather than passes. """ from __future__ import annotations @@ -36,12 +43,65 @@ ClientFactory = Callable[[str, str], Any] -def _readable(resource: Mapping[str, Any], database: str, table: str) -> Optional[Set[str]]: +def _column_spec(spec: Mapping[str, Any]) -> Set[str]: + """A grant's or filter's ``ColumnNames``, or its ``ColumnWildcard`` as ``{"*", "-excluded"}``. + + Neither is every column (``{"*"}``). + """ + if spec.get("ColumnNames"): + return set(spec["ColumnNames"]) + wildcard = spec.get("ColumnWildcard") or {} + return {"*", *(f"-{c}" for c in (wildcard.get("ExcludedColumnNames") or ()))} + + +def _filter_columns( + lf: Any, ref: Mapping[str, Any], cache: Dict[tuple, Optional[Set[str]]] +) -> Set[str]: + """The columns a data cells filter reaches, read once per filter; every column if unreadable.""" + key = ( + ref.get("TableCatalogId"), + ref.get("DatabaseName"), + ref.get("TableName"), + ref.get("Name"), + ) + if key not in cache: + try: + got = ( + lf.get_data_cells_filter( + TableCatalogId=ref.get("TableCatalogId"), + DatabaseName=ref.get("DatabaseName"), + TableName=ref.get("TableName"), + Name=ref.get("Name"), + ).get("DataCellsFilter") + or {} + ) + except Exception as exc: # noqa: BLE001 — fail closed below + LOG.warning("verify_lf_columns could not read data cells filter %s: %s", key, exc) + got = None + cache[key] = None if got is None else _column_spec(got) + found = cache[key] + return {"*"} if found is None else found + + +def _readable( + resource: Mapping[str, Any], + database: str, + table: str, + lf: Any = None, + filters: Optional[Dict[tuple, Optional[Set[str]]]] = None, +) -> Optional[Set[str]]: """Columns of ``database.table`` this permission's resource reaches; ``None`` for none. The empty set with the ``"*"`` member means every column. """ whole = {"*"} + if "DataCellsFilter" in resource: + ref = resource.get("DataCellsFilter") or {} + if ref.get("DatabaseName") != database or ref.get("TableName") != table: + return None + if lf is None: + return whole + return _filter_columns(lf, ref, filters if filters is not None else {}) if "Table" in resource: ref = resource.get("Table") or {} if ref.get("DatabaseName") != database: @@ -53,12 +113,7 @@ def _readable(resource: Mapping[str, Any], database: str, table: str) -> Optiona ref = resource.get("TableWithColumns") or {} if ref.get("DatabaseName") != database or ref.get("Name") != table: return None - if ref.get("ColumnNames"): - return set(ref["ColumnNames"]) - wildcard = ref.get("ColumnWildcard") - if wildcard is not None: - return {"*", *(f"-{c}" for c in (wildcard.get("ExcludedColumnNames") or ()))} - return whole + return _column_spec(ref) return None @@ -85,8 +140,26 @@ def _table_permissions(lf: Any) -> List[Mapping[str, Any]]: entries.extend(page.get("PrincipalResourcePermissions") or []) token = page.get("NextToken") if not token: - return entries + break kwargs["NextToken"] = token + # Grants on data cells filters are not TABLE permissions: an unfiltered listing + # returns them, as ``Resource.DataCellsFilter``. + kwargs = {} + try: + while True: + page = lf.list_permissions(**kwargs) + entries.extend( + e + for e in page.get("PrincipalResourcePermissions") or [] + if "DataCellsFilter" in (e.get("Resource") or {}) + ) + token = page.get("NextToken") + if not token: + break + kwargs["NextToken"] = token + except Exception as exc: # noqa: BLE001 — the TABLE listing still stands + LOG.warning("verify_lf_columns could not list data cells filter grants: %s", exc) + return entries def column_restrictions_dimension( @@ -142,10 +215,11 @@ def column_restrictions_dimension( for g in ((binding.get("governance") or {}).get("lakeFormation") or {}).get("grants") or () if isinstance(g, Mapping) and g.get("principal") } + filters: Dict[tuple, Optional[Set[str]]] = {} visible = { (entry.get("Principal") or {}).get("DataLakePrincipalIdentifier") for entry in permissions - if _readable(entry.get("Resource") or {}, database, table) is not None + if _readable(entry.get("Resource") or {}, database, table, lf, filters) is not None } unseen = sorted(grantees - visible) if unseen: @@ -169,7 +243,7 @@ def column_restrictions_dimension( continue if not LF_READ_PERMISSIONS & set(entry.get("Permissions") or ()): continue - readable = _readable(entry.get("Resource") or {}, database, table) + readable = _readable(entry.get("Resource") or {}, database, table, lf, filters) if readable is None: continue leaked.extend(c for c in _leaks(readable, list(forbidden)) if c not in leaked) diff --git a/fluid_build/iac/column_access.py b/fluid_build/iac/column_access.py index 3b8503f5..b74fecc2 100644 --- a/fluid_build/iac/column_access.py +++ b/fluid_build/iac/column_access.py @@ -33,6 +33,13 @@ * ``allow``: the columns are readable only by the principals an ``allow`` names (an allow list per column); a column with no ``allow`` is readable by every reader of the expose. +* ``mask`` (with ``mask: last_four | first_four | nullify``): the principal reads the + columns masked by the platform, never their raw values. For the raw read it is + denied, so it is in no column's reader set; :func:`column_masks` says which + columns it reads masked, and by which rule (GCP: a data policy on the columns' + policy tag and ``roles/bigquerydatapolicy.maskedReader``; AWS: Lake Formation cannot + mask, so a protected Glue Data Catalog view returns them masked). A deny beats a + mask, and a mask beats an allow for the principal it names; one column has one rule. * A deny beats an allow. A restriction never grants access: the readers are intersected with the expose's readers (on GCP the ``accessPolicy`` read grants and the expose's own ``policy.authz.readers``; on AWS the Lake Formation ``SELECT`` @@ -59,6 +66,14 @@ ALLOW = "allow" DENY = "deny" +MASK = "mask" + +#: The masking rules ``access: mask`` takes, cloud-neutral: ``last_four`` returns +#: ``XXXXX`` and the value's last four characters, ``first_four`` its first four and +#: ``XXXXX``, and ``nullify`` NULL (BigQuery's LAST_FOUR_CHARACTERS, +#: FIRST_FOUR_CHARACTERS and ALWAYS_NULL; a value of four characters or fewer is +#: hashed by the first two, on each platform in its own encoding). +MASK_RULES = ("last_four", "first_four", "nullify") #: Lake Formation permissions that let a principal read a table's rows. LF_READ_PERMISSIONS = frozenset({"SELECT", "ALL"}) @@ -75,11 +90,18 @@ class Restriction: #: that holds the restriction where it has room for them (GCP's policy tag). tags: Tuple[str, ...] = () labels: Tuple[Tuple[str, str], ...] = () + #: The masking rule of an ``access: mask`` restriction (one of :data:`MASK_RULES`). + mask: Optional[str] = None -def _where(exposure: Mapping[str, Any], index: int) -> str: +def policy_where(exposure: Mapping[str, Any], index: int, field: str) -> str: + """``exposes[].policy.authz.``, where a refusal points in the contract.""" expose_id = exposure.get("exposeId") or exposure.get("id") or index - return f"exposes[{expose_id}].policy.authz.columnRestrictions" + return f"exposes[{expose_id}].policy.authz.{field}" + + +def _where(exposure: Mapping[str, Any], index: int) -> str: + return policy_where(exposure, index, "columnRestrictions") def restrictions_for(exposure: Mapping[str, Any], index: int = 0) -> Tuple[Restriction, ...]: @@ -124,16 +146,18 @@ def restrictions_for(exposure: Mapping[str, Any], index: int = 0) -> Tuple[Restr f"{at} must list the columns it restricts.", ("Set columns to one or more column names of this expose's schema.",), ) - if access not in (ALLOW, DENY): + if access not in (ALLOW, DENY, MASK): raise UnsupportedBindingError( "column-restriction", - f"{at}.access is {entry.get('access')!r}; it must be 'allow' or 'deny', so " - "the cloud can be told which way the restriction goes.", + f"{at}.access is {entry.get('access')!r}; it must be 'allow', 'deny' or " + "'mask', so the cloud can be told which way the restriction goes.", ( - "Set access: deny to hide the columns from the principal, or allow to make " - "the principal one of the columns' only readers.", + "Set access: deny to hide the columns from the principal, allow to make " + "the principal one of the columns' only readers, or mask (with a mask " + "rule) to let it read them masked.", ), ) + mask = _mask_rule(entry, access, at) unknown = [c for c in columns if c not in declared] if unknown: raise UnsupportedBindingError( @@ -155,11 +179,54 @@ def restrictions_for(exposure: Mapping[str, Any], index: int = 0) -> Tuple[Restr if isinstance(labels, Mapping) else () ), + mask=mask, ) ) + _refuse_two_rules_for_a_column(out, where) return tuple(out) +def _mask_rule(entry: Mapping[str, Any], access: str, at: str) -> Optional[str]: + """An ``access: mask`` restriction's rule; ``None`` for any other access. + + A mask needs its rule, and a rule on a restriction that does not mask is refused, + since it would mask nothing. + """ + mask = entry.get("mask") + if access == MASK and mask not in MASK_RULES: + raise UnsupportedBindingError( + "column-restriction", + f"{at} masks its columns with {mask!r}; access: mask needs a mask rule, one " + f"of {', '.join(MASK_RULES)}, so the platform knows what to return.", + ("Set mask: last_four (or first_four, or nullify).",), + ) + if access != MASK and mask is not None: + raise UnsupportedBindingError( + "column-restriction", + f"{at} sets mask: {mask!r} on an access: {access} restriction, which masks " + "nothing: only access: mask reads columns masked.", + ("Set access: mask, or remove mask.",), + ) + return mask if access == MASK else None + + +def _refuse_two_rules_for_a_column(restrictions: Sequence[Restriction], where: str) -> None: + """A column has one masking rule: the one its policy tag (GCP) or view (AWS) applies.""" + rules: Dict[str, str] = {} + for restriction in restrictions: + if restriction.mask is None: + continue + for column in restriction.columns: + if rules.setdefault(column, restriction.mask) != restriction.mask: + raise UnsupportedBindingError( + "column-restriction", + f"{where} masks {column} by {rules[column]!r} and by " + f"{restriction.mask!r}; a column has one masking rule, the one its " + "policy tag (GCP) or view (AWS) applies.", + ("Mask the column by one rule for every principal that reads it masked.",), + ) + + def authz_readers(exposure: Mapping[str, Any]) -> Tuple[str, ...]: """``exposes[].policy.authz.readers``: the expose's own readers, logical principals. @@ -203,6 +270,7 @@ def column_readers( denied: Dict[str, set[str]] = {} for restriction in restrictions: identities = set(resolve(restriction.principal)) + # A masked principal never reads the raw column: for this set it is denied. target = allowed if restriction.access == ALLOW else denied for column in restriction.columns: target.setdefault(column, set()).update(identities) @@ -226,6 +294,52 @@ def column_readers( return out +def column_masks( + exposure: Mapping[str, Any], + restrictions: Sequence[Restriction], + resolve: Callable[[str], Tuple[str, ...]], + readers: FrozenSet[str], + *, + where: str, +) -> Dict[str, Tuple[str, FrozenSet[str]]]: + """``{masked column: (rule, identities that read it masked)}``, in schema order. + + The identities are the readers an ``access: mask`` names, less any a ``deny`` on the + same column names (a deny beats a mask). A masked principal that is not a reader + is reported, not added: a restriction never grants access. A column whose masked + identities are all denied, or none of them readers, is left out. + """ + masked: Dict[str, set[str]] = {} + rule: Dict[str, str] = {} + denied: Dict[str, set[str]] = {} + for restriction in restrictions: + identities = set(resolve(restriction.principal)) + for column in restriction.columns: + if restriction.access == MASK and restriction.mask: + masked.setdefault(column, set()).update(identities) + rule[column] = restriction.mask + elif restriction.access == DENY: + denied.setdefault(column, set()).update(identities) + out: Dict[str, Tuple[str, FrozenSet[str]]] = {} + for column in restricted_columns(exposure, restrictions): + if column not in masked: + continue + not_readers = sorted(masked[column] - readers) + if not_readers: + LOG.warning( + "column_mask_not_a_reader %s column=%s identities=%s: a restriction never " + "grants access, so these identities, which have no read grant on the expose, " + "still cannot read the column, masked or not", + where, + column, + not_readers, + ) + identities = frozenset((masked[column] & readers) - denied.get(column, set())) + if identities: + out[column] = (rule[column], identities) + return out + + # ── AWS: Lake Formation excluded columns ──────────────────────────────── @@ -335,7 +449,8 @@ def lf_expected_exclusions( for i, columns in exclusions.items(): expected.setdefault(str(grants[i].get("principal")), set()).update(columns) for restriction in restrictions: - if restriction.access != DENY: + # A masked principal must not read the raw column either: it reads the view. + if restriction.access not in (DENY, MASK): continue for identity in resolve_principal( restriction.principal, mapping, platform=AWS, where=where @@ -356,8 +471,11 @@ def lf_expected_exclusions( "ALLOW", "DENY", "LF_READ_PERMISSIONS", + "MASK", + "MASK_RULES", "Restriction", "authz_readers", + "column_masks", "column_readers", "lf_exclusions", "lf_expected_exclusions", diff --git a/fluid_build/iac/governance_labels.py b/fluid_build/iac/governance_labels.py new file mode 100644 index 00000000..5777bc2b --- /dev/null +++ b/fluid_build/iac/governance_labels.py @@ -0,0 +1,78 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""A contract's governance as resource labels, the same on every cloud. + +The contract's and the expose's ``labels``, and its classification, jurisdiction, +regulatory framework and residency, written onto the resources a product owns: GCP +labels on the dataset, table and key, AWS tags on the bucket and key and parameters on +the Glue table. So the regulation a product falls under is on the resource and can be +found there, not only in the contract. The values are coerced to GCP's label alphabet +(the stricter of the two), so both clouds carry the same keys and values. +""" + +from __future__ import annotations + +import re +from typing import Any, Dict, Mapping + +_LABEL_KEY_RE = re.compile(r"[^a-z0-9_-]") + + +def _label(value: Any) -> str: + """A string as a GCP label value: lowercase letters, digits, ``_`` and ``-``, 63 at most.""" + return _LABEL_KEY_RE.sub("-", str(value).strip().lower())[:63] + + +def governance_labels(contract: Mapping[str, Any], exposure: Mapping[str, Any]) -> Dict[str, str]: + """The contract's governance as GCP labels: its labels, classification and regulation. + + The contract's and the expose's ``labels`` (each key and value coerced to GCP's + label alphabet), plus ``fluid_classification``, ``fluid_jurisdiction``, + ``fluid_regulation`` and ``fluid_residency`` from the expose's classification and + the contract's ``sovereignty``. A key that does not start with a letter is + prefixed ``l_``; empty values are dropped; at most 60 are kept, in key order. + """ + out: Dict[str, str] = {} + for source in (contract.get("labels"), exposure.get("labels")): + if isinstance(source, Mapping): + for key, value in source.items(): + k = _label(key) + if k and not k[0].isalpha(): + k = f"l_{k}"[:63] + v = _label(value) + if k and v: + out[k] = v + policy = exposure.get("policy") if isinstance(exposure.get("policy"), Mapping) else {} + classification = policy.get("classification") or (contract.get("metadata") or {}).get( + "classification" + ) + if classification: + out["fluid_classification"] = _label(classification) + sovereignty = ( + contract.get("sovereignty") if isinstance(contract.get("sovereignty"), Mapping) else {} + ) + if sovereignty.get("jurisdiction"): + out["fluid_jurisdiction"] = _label(sovereignty["jurisdiction"]) + frameworks = sovereignty.get("regulatoryFramework") or sovereignty.get("regulations") + if isinstance(frameworks, str): + frameworks = [frameworks] + if isinstance(frameworks, list) and frameworks: + out["fluid_regulation"] = _label("_".join(str(f) for f in frameworks)) + if sovereignty.get("dataResidency") is not None: + out["fluid_residency"] = "true" if sovereignty.get("dataResidency") else "false" + return {k: out[k] for k in sorted(out)[:60]} + + +__all__ = ["governance_labels"] diff --git a/fluid_build/iac/providers/aws.py b/fluid_build/iac/providers/aws.py index 7a26cf1e..df544c91 100644 --- a/fluid_build/iac/providers/aws.py +++ b/fluid_build/iac/providers/aws.py @@ -49,7 +49,7 @@ import os import re from dataclasses import dataclass, field -from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple +from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple import yaml @@ -57,6 +57,7 @@ from ...providers.aws.util import warehouse as _warehouse from .. import column_access from ..base import UnsupportedBindingError +from ..governance_labels import governance_labels from ..importer import ImportBlock from ..naming import TofuExpr, safe_ident, tofu_ref from ..packaging import ( @@ -65,7 +66,9 @@ PackagingResolution, resolve_packaging, ) +from ..principals import AWS, principal_map, resolve_principal from ..provider_match import is_cloud +from ..row_access import RowFilter, row_filters_for from ..versions import required_providers from . import aws_storage from .aws_storage import Retention @@ -120,6 +123,27 @@ def _has_custom_aws_endpoint() -> bool: ) +def _emulator_catalog_id() -> Dict[str, Any]: + """``{"catalog_id": }`` for a Glue database or table on an emulator. + + terraform-provider-aws 6 makes a Glue database's id ``:`` + (a table's ``::``), the catalog id being the + resource's ``catalog_id`` or else the provider's account id, and refuses + to read back an id with an empty catalog ("unexpected format for ID + (:sales), expected catalog-id:database-name"; + hashicorp/terraform-provider-aws#31626). On an emulator the provider + skips requesting the account id (:meth:`AwsIacPlugin.provider_block`), + which it must, because given one, provider 6.23 and later read a bucket's + tags through S3 Control at ``.``, which neither moto + nor LocalStack routes (hashicorp/terraform-provider-aws#45292). So the + catalog id comes from ``data.aws_caller_identity``, which the emulator + answers. On real AWS this is ``{}``: the module is unchanged. + """ + if not _has_custom_aws_endpoint(): + return {} + return {"catalog_id": tofu_ref("data.aws_caller_identity.fluid_lf_caller.account_id")} + + def _resolve_catalog_id() -> str: """Resolve the AWS account id for Glue catalog import ids. @@ -382,10 +406,26 @@ def emit( # excluded columns (``iac/column_access.py``); refused when nothing # on this binding could enforce them. exclusions = lf_column_exclusions(exposure, binding, index) + # The contract's row filters, each on the read grant of its principal, + # and the protected views its masked principals read (Lake Formation + # cannot mask): ``lf_row_filters`` / ``lf_masked_views``. + row_filters = lf_row_filters(exposure, binding, index) + masked_views = lf_masked_views(exposure, binding, index, exclusions or {}, row_filters) placement = _placement(packaging, exposure) - tags = _tags_for(base_tags, placement) + # The contract's governance labels (classification, jurisdiction, + # regulation, its own labels) under fluid's own two, which win. + gov_labels = governance_labels(contract, exposure) + tags = _tags_for({**gov_labels, **base_tags}, placement) _emit_glue( - resources, loc, fmt, schema, cid, tags, contract=contract, placement=placement + resources, + loc, + fmt, + schema, + cid, + tags, + contract=contract, + placement=placement, + labels=gov_labels, ) _emit_s3(resources, loc, cid, tags, placement=placement) _emit_kinesis(resources, loc, cid, tags) @@ -405,6 +445,8 @@ def emit( tags, placement=placement, exclusions=exclusions or {}, + row_filters=row_filters, + masked_views=masked_views, ) # Retention (exposes[].lifecycle) and encryption at rest # (binding.encryption), per bucket this product owns. Nothing is @@ -460,10 +502,13 @@ def emit_data( # ``{account}-fluid-data`` bucket, whose token references this source. # ...and whenever a product KMS key's policy names the account. storage = _storage_by_bucket(contract, cid) + # ...and on an emulator, where it is every Glue resource's catalog id + # (see :func:`_emulator_catalog_id`). if ( _contract_uses_lakeformation(contract) or _references_caller_account(contract) or any(s.product_key for s in storage.values()) + or _has_custom_aws_endpoint() ): data.setdefault("aws_caller_identity", {})["fluid_lf_caller"] = {} # A ``cross-account`` LF bucket policy filters its grantees at plan @@ -622,7 +667,9 @@ def provider_block(self) -> Dict[str, Any]: Gated on :func:`_has_custom_aws_endpoint`, so real-AWS applies are byte-for-byte unchanged (no provider block emitted). Mirrors - LocalStack's documented Terraform provider setup. + LocalStack's documented Terraform provider setup. Without the account + id, provider 6 builds Glue ids with an empty catalog, so each Glue + database and table names its catalog (:func:`_emulator_catalog_id`). """ if not _has_custom_aws_endpoint(): return {} @@ -745,6 +792,7 @@ def _emit_glue( *, contract: Optional[Mapping[str, Any]] = None, placement: _Placement = _LEGACY_PLACEMENT, + labels: Optional[Mapping[str, str]] = None, ) -> None: database = loc.get("database") if not database: @@ -764,6 +812,7 @@ def _emit_glue( db_name, { "name": database, + **_emulator_catalog_id(), "lifecycle": {"ignore_changes": ["parameters"]}, }, ) @@ -831,6 +880,11 @@ def _emit_glue( # block, which is what keeps the byte-parity pin green. if placement.pool: parameters["fluid_pool"] = str(placement.pool) + # The contract's governance labels (classification, jurisdiction, regulation + # and its own labels; ``iac/governance_labels.py``), beside fluid's own + # parameters, which they never overwrite: ``classification`` is the format. + for key, value in (labels or {}).items(): + parameters.setdefault(key, value) # Column-level tags from the contract's ``schema[].tags`` field # (already in v0.7.3 — ``$defs.column.properties.tags``). Emitted # as the legacy ``forge.pii.`` Glue parameter the retired @@ -882,6 +936,7 @@ def _emit_glue( } if description: table_body["description"] = description + table_body.update(_emulator_catalog_id()) resources.setdefault("aws_glue_catalog_table", {})[ safe_ident(f"{cid}_{database}_{table}") @@ -1233,7 +1288,7 @@ def _emit_s3_notification(resources: Dict[str, Any], action: Mapping[str, Any], # # There is NO first-party resource in `hashicorp/aws` for # `CREATE EXTERNAL SCHEMA ... FROM DATA CATALOG`. The community pattern -# (and the only one that works with the `hashicorp/aws ~> 5.0` pin) is a +# (and the only one that works with the `hashicorp/aws ~> 6.0` pin) is a # `null_resource` + `provisioner.local-exec` calling the `redshift-data` # API. The plugin emits this bridge and orders it after the workgroup + # the upstream Glue catalog database via an explicit `depends_on` (see @@ -2033,6 +2088,212 @@ def lf_column_exclusions( return exclusions +def lf_row_filters( + exposure: Mapping[str, Any], binding: Mapping[str, Any], index: int = 0 +) -> Dict[int, RowFilter]: + """``{grant index: row filter}``: the contract's row filters on their principals' grants. + + A filter's principal resolves through ``binding.principals`` to its IAM identities; + each read grant held by one of them carries the filter (:func:`_emit_lakeformation` + grants it on a data cells filter instead of the table). A filter with no read grant + to carry it is refused: Lake Formation would let nobody read less. + """ + filters = row_filters_for(exposure, index) + if not filters: + return {} + where = f"exposes[{exposure.get('exposeId') or index}].policy.authz.rowFilters" + gov = (binding.get("governance") or {}).get("lakeFormation") or {} + grants = gov.get("grants") if isinstance(gov, Mapping) else None + loc = binding.get("location") or {} + if not grants or not loc.get("table"): + raise UnsupportedBindingError( + "row-filter-unenforceable", + f"{where} filters rows, but this aws binding has no governance.lakeFormation " + "grants on a Glue table to carry the filter, so nothing would enforce it.", + ("Grant the filtered principal SELECT in governance.lakeFormation.grants.",), + ) + mapping = principal_map(binding) + out: Dict[int, RowFilter] = {} + for flt in filters: + identities = set(resolve_principal(flt.principal, mapping, platform=AWS, where=where)) + held = [ + i + for i, g in enumerate(grants) + if isinstance(g, Mapping) + and str(g.get("principal") or "") in identities + and column_access.LF_READ_PERMISSIONS & set(g.get("permissions") or ()) + ] + if not held: + raise UnsupportedBindingError( + "row-filter-unenforceable", + f"{where} filters {flt.principal}, but no governance.lakeFormation grant reads " + "the table as it, so the filter would narrow nothing.", + ("Grant the principal SELECT, or remove the filter.",), + ) + for i in held: + out[i] = flt + return out + + +@dataclass(frozen=True) +class _LfMaskedView: + """A protected Glue Data Catalog view that returns masked columns to one principal.""" + + principal: str + name: str + sql: str + definer: str + validation_connection: str + + +def _athena_mask(column_sql: str, rule: str) -> str: + """The column masked by ``rule``, in Athena SQL, as BigQuery's rule of the same name. + + ``XXXXX`` and the last (or first) four characters, a value of four characters or + fewer hashed (SHA-256, hex) instead, and NULL kept NULL; ``nullify`` is NULL. + """ + if rule == "nullify": + return "CAST(NULL AS varchar)" + shown = ( + f"concat('XXXXX', substr({column_sql}, -4))" + if rule == "last_four" + else f"concat(substr({column_sql}, 1, 4), 'XXXXX')" + ) + return ( + f"CASE WHEN {column_sql} IS NULL THEN NULL " + f"WHEN length({column_sql}) <= 4 THEN to_hex(sha256(to_utf8({column_sql}))) " + f"ELSE {shown} END" + ) + + +_STRING_TYPES = frozenset({"string", "varchar", "text", "char"}) + + +def _grant_raw_columns( + grant: Mapping[str, Any], + schema: Sequence[Mapping[str, Any]], + derived_exclusions: Optional[Sequence[str]], +) -> set: + """The columns a Lake Formation grant lets its principal read raw from the table. + + A protected view reads the table as its definer, so whatever it selects its reader + reads, whatever the reader's own grant allows: the view must select no column the + grant does not. A ``columns:`` list is the grant's allow list; otherwise every + column less its exclusions, derived or written as ``excludedColumns`` (the two + agree, or the emit is refused), as :func:`_emit_lakeformation` grants them. + """ + declared = {str(c.get("name")) for c in schema if c.get("name")} + listed = grant.get("columns") + if listed: + return declared & {str(c) for c in listed} + excluded = derived_exclusions or grant.get("excludedColumns") or () + return declared - {str(c) for c in excluded} + + +def lf_masked_views( + exposure: Mapping[str, Any], + binding: Mapping[str, Any], + index: int, + exclusions: Mapping[int, Tuple[str, ...]], + row_filters: Mapping[int, RowFilter], +) -> List[_LfMaskedView]: + """The protected views the expose's masked principals read (Lake Formation cannot mask). + + One view per read grant whose principal ``access: mask`` names: every column the + principal may read raw, plus its masked columns through :func:`_athena_mask`, and + its row filter as the view's ``WHERE``, since the view reads the table as its + definer. The definer and the Athena validation connection are the binding's + ``governance.lakeFormation.maskedViews`` (``definer``, ``validationConnection``): + without them a mask is refused, never dropped. + """ + restrictions = column_access.restrictions_for(exposure, index) + if not restrictions or not any(r.access == column_access.MASK for r in restrictions): + return [] + where = f"exposes[{exposure.get('exposeId') or index}].policy.authz.columnRestrictions" + gov = (binding.get("governance") or {}).get("lakeFormation") or {} + grants = gov.get("grants") or [] + views_cfg = gov.get("maskedViews") if isinstance(gov, Mapping) else None + definer = (views_cfg or {}).get("definer") if isinstance(views_cfg, Mapping) else None + connection = ( + (views_cfg or {}).get("validationConnection") if isinstance(views_cfg, Mapping) else None + ) + if not definer or not connection: + raise UnsupportedBindingError( + "column-mask-unenforceable", + f"{where} masks columns, but Lake Formation cannot mask, and this aws binding " + "declares no governance.lakeFormation.maskedViews (definer, validationConnection) " + "for the protected view that would return them masked.", + ( + "Add governance.lakeFormation.maskedViews: definer (an IAM role Glue and Lake " + "Formation may assume) and validationConnection (a VIEW_VALIDATION_ATHENA Glue " + "connection).", + "Or deny the columns instead of masking them.", + ), + ) + loc = binding.get("location") or {} + database, table = loc.get("database"), loc.get("table") + mapping = principal_map(binding) + + def resolve(principal: str) -> Tuple[str, ...]: + return resolve_principal(principal, mapping, platform=AWS, where=where) + + readers = frozenset( + str(g.get("principal")) + for g in grants + if isinstance(g, Mapping) + and g.get("principal") + and column_access.LF_READ_PERMISSIONS & set(g.get("permissions") or ()) + ) + masks = column_access.column_masks(exposure, restrictions, resolve, readers, where=where) + schema = [ + c for c in ((exposure.get("contract") or {}).get("schema") or []) if isinstance(c, Mapping) + ] + types = {str(c.get("name")): str(c.get("type") or "").lower() for c in schema} + for column in masks: + if masks[column][0] != "nullify" and types.get(column) not in _STRING_TYPES: + raise UnsupportedBindingError( + "column-mask-type", + f"{where} masks {column} ({types.get(column) or 'no type'}) by " + f"{masks[column][0]}, which masks text; the protected view on AWS returns it " + "as a string.", + ("Mask a string column, or use mask: nullify.",), + ) + views: List[_LfMaskedView] = [] + for i, grant in enumerate(grants): + principal = str(grant.get("principal") or "") if isinstance(grant, Mapping) else "" + mine = [c for c, (_rule, ids) in masks.items() if principal in ids] + if not mine: + continue + raw = _grant_raw_columns(grant, schema, exclusions.get(i)) - set(mine) + select = [] + for col in schema: + name = str(col.get("name")) + if name not in mine and name not in raw: + continue + quoted = f'"{validate_ident(name)}"' + select.append( + f"{_athena_mask(quoted, masks[name][0])} AS {quoted}" if name in mine else quoted + ) + sql = ( + f"SELECT {', '.join(select)} FROM " + f'"{validate_ident(str(database))}"."{validate_ident(str(table))}"' + ) + flt = row_filters.get(i) + if flt is not None: + sql += f" WHERE {flt.where}" + suffix = "" if len(views) == 0 else f"_{len(views) + 1}" + views.append( + _LfMaskedView( + principal=principal, + name=f"{table}_masked{suffix}", + sql=sql, + definer=str(definer), + validation_connection=str(connection), + ) + ) + return views + + def _refuse_grants_left_no_column( exposure: Mapping[str, Any], binding: Mapping[str, Any], @@ -2082,10 +2343,20 @@ def _emit_lakeformation( *, placement: _Placement = _LEGACY_PLACEMENT, exclusions: Optional[Mapping[int, Tuple[str, ...]]] = None, + row_filters: Optional[Mapping[int, RowFilter]] = None, + masked_views: Sequence["_LfMaskedView"] = (), ) -> None: """Emit per-exposure LF resources. No-op when the binding has no ``governance.lakeFormation`` block. + ``row_filters`` is :func:`lf_row_filters`: a grant that carries one reads through a + data cells filter (the filter's predicate, and the grant's excluded columns as its + column wildcard's exclusions), so its ``SELECT`` is granted on the filter, not on + the table. ``masked_views`` is :func:`lf_masked_views`: each becomes a protected + Glue Data Catalog view (``VIRTUAL_VIEW``, SECURITY DEFINER), its definer is + granted ``SELECT`` with grant option on the table and ``CREATE_TABLE`` and + ``DESCRIBE`` on the database, and its principal ``SELECT`` and ``DESCRIBE`` on the view. + ``exclusions`` is ``column_access.lf_exclusions``: for each grant index, the columns the contract's ``policy.authz.columnRestrictions`` do not let that grant's principal read. They become the grant's ``excluded_column_names``; @@ -2146,6 +2417,12 @@ def _emit_lakeformation( body["permissions_with_grant_option"] = list(gp) cols = grant.get("columns") excluded = (exclusions or {}).get(idx) or grant.get("excludedColumns") + flt = (row_filters or {}).get(idx) + if flt is not None and table_key: + _emit_lf_filtered_grant( + resources, cid, table, table_key, idx, principal, flt, cols, excluded + ) + continue if (cols or excluded) and table_key: if cols and excluded: # One block cannot hold both: Lake Formation takes either a column @@ -2266,24 +2543,148 @@ def _emit_lakeformation( # because LF requires one of these and "wildcard" is the # natural row-only-filter behaviour. col_block = {"column_wildcard": [{}]} - body = { - "table_data": [ - { - "table_catalog_id": tofu_ref( - "data.aws_caller_identity.fluid_lf_caller.account_id" - ), - "database_name": tofu_ref( - f"aws_glue_catalog_table.{table_key}.database_name" - ), - "table_name": tofu_ref(f"aws_glue_catalog_table.{table_key}.name"), - "name": filter_name, - "row_filter": [{"filter_expression": row_expr}], - **col_block, - } - ] - } filter_key = safe_ident(f"{cid}_lf_filter_{table}_{filter_name}") - resources.setdefault("aws_lakeformation_data_cells_filter", {})[filter_key] = body + resources.setdefault("aws_lakeformation_data_cells_filter", {})[filter_key] = { + "table_data": [_lf_filter_table_data(table_key, filter_name, row_expr, col_block)] + } + + # 5. The protected views masked principals read: Lake Formation cannot mask. + if masked_views and table_key: + _emit_lf_masked_views(resources, cid, database, table, table_key, masked_views) + + +def _lf_filter_table_data( + table_key: str, name: str, row_expr: str, col_block: Mapping[str, Any] +) -> Dict[str, Any]: + """An ``aws_lakeformation_data_cells_filter``'s ``table_data`` on the product's table.""" + return { + "table_catalog_id": tofu_ref("data.aws_caller_identity.fluid_lf_caller.account_id"), + "database_name": tofu_ref(f"aws_glue_catalog_table.{table_key}.database_name"), + "table_name": tofu_ref(f"aws_glue_catalog_table.{table_key}.name"), + "name": name, + "row_filter": [{"filter_expression": row_expr}], + **col_block, + } + + +def _emit_lf_filtered_grant( + resources: Dict[str, Any], + cid: str, + table: str, + table_key: str, + idx: int, + principal: str, + flt: RowFilter, + cols: Any, + excluded: Any, +) -> None: + """Grant ``idx`` read through the contract's row filter: a data cells filter holding + the predicate and the grant's excluded columns, and ``SELECT`` on the filter, never + on the table.""" + if cols: + raise UnsupportedBindingError( + "lakeformation-grant-columns", + f"governance.lakeFormation.grants[{idx}] lists columns, and the contract's " + f"row filter {flt.name} filters its principal: a data cells filter takes " + "excluded columns, not a column list.", + ("Restrict its columns with policy.authz.columnRestrictions instead.",), + ) + filter_key = safe_ident(f"{cid}_lf_rows_{table}_{flt.name}") + columns_block: Dict[str, Any] = { + "column_wildcard": [{"excluded_column_names": list(excluded)} if excluded else {}] + } + resources.setdefault("aws_lakeformation_data_cells_filter", {})[filter_key] = { + "table_data": [_lf_filter_table_data(table_key, flt.name, flt.where, columns_block)] + } + # SELECT only, as on a column-limited grant: on the filter, never the table. + resources.setdefault("aws_lakeformation_permissions", {})[ + safe_ident(f"{cid}_lf_grant_{table}_{idx}") + ] = { + "principal": principal, + "permissions": ["SELECT"], + "data_cells_filter": [ + { + "database_name": tofu_ref(f"aws_glue_catalog_table.{table_key}.database_name"), + "table_name": tofu_ref(f"aws_glue_catalog_table.{table_key}.name"), + "table_catalog_id": tofu_ref("data.aws_caller_identity.fluid_lf_caller.account_id"), + "name": flt.name, + } + ], + "depends_on": [f"aws_lakeformation_data_cells_filter.{filter_key}"], + } + + +def _emit_lf_masked_views( + resources: Dict[str, Any], + cid: str, + database: str, + table: str, + table_key: str, + masked_views: Sequence["_LfMaskedView"], +) -> None: + """Each masked view as a protected Glue Data Catalog view (``VIRTUAL_VIEW``, run as its + definer), the definer's grants it is validated with, and its principal's grant.""" + db_ref = tofu_ref(f"aws_glue_catalog_table.{table_key}.database_name") + tbl_ref = tofu_ref(f"aws_glue_catalog_table.{table_key}.name") + perms_res = resources.setdefault("aws_lakeformation_permissions", {}) + for view in masked_views: + definer_table = safe_ident(f"{cid}_lf_view_definer_{table}_{view.definer}") + definer_db = safe_ident(f"{cid}_lf_view_definer_db_{database}_{view.definer}") + perms_res.setdefault( + definer_table, + { + "principal": view.definer, + "permissions": ["SELECT"], + "permissions_with_grant_option": ["SELECT"], + "table": [{"database_name": db_ref, "name": tbl_ref}], + }, + ) + # Lake Formation's prerequisites for a view: its definer holds SELECT with the + # grant option on every table it reads and CREATE_TABLE on the view's database. + perms_res.setdefault( + definer_db, + { + "principal": view.definer, + "permissions": ["CREATE_TABLE", "DESCRIBE"], + "database": [{"name": db_ref}], + }, + ) + view_key = safe_ident(f"{cid}_{database}_{view.name}") + resources.setdefault("aws_glue_catalog_table", {})[view_key] = { + "name": view.name, + "database_name": db_ref, + **_emulator_catalog_id(), + "table_type": "VIRTUAL_VIEW", + "view_definition": [ + { + "definer": view.definer, + "is_protected": True, + "representations": [ + { + "dialect": "ATHENA", + "dialect_version": "3", + "validation_connection": view.validation_connection, + "view_original_text": view.sql, + } + ], + } + ], + # Glue validates the view as its definer, so the definer's grants first. + "depends_on": [ + f"aws_lakeformation_permissions.{definer_table}", + f"aws_lakeformation_permissions.{definer_db}", + ], + } + perms_res[safe_ident(f"{cid}_lf_view_grant_{view.name}")] = { + "principal": view.principal, + "permissions": ["SELECT", "DESCRIBE"], + "table": [ + { + "database_name": db_ref, + "name": tofu_ref(f"aws_glue_catalog_table.{view_key}.name"), + } + ], + } def _check_lf_grant_columns( diff --git a/fluid_build/iac/providers/gcp.py b/fluid_build/iac/providers/gcp.py index bb545412..db09988e 100644 --- a/fluid_build/iac/providers/gcp.py +++ b/fluid_build/iac/providers/gcp.py @@ -399,7 +399,11 @@ def emit( loc = binding.get("location") or {} schema = (exposure.get("contract") or {}).get("schema") or [] placement = _placement(packaging, exposure) - labels = _labels_for(base_labels, placement) + # The contract's governance labels (its classification, jurisdiction, + # regulation and own labels) under fluid's own two, which win. + labels = _labels_for( + {**_gov.governance_labels(contract, exposure), **base_labels}, placement + ) # The contract's principals are logical: each GCP expose's binding maps # them to real IAM members (``binding.principals``), and an unmapped or # placeholder principal is refused rather than written into IAM. @@ -828,6 +832,7 @@ class _BqGovernance: retention: Optional[_gov.BqRetention] = None encryption: Optional[_gov.BqEncryption] = None tags: Tuple[_gov.TagGroup, ...] = () + row_policies: Tuple[_gov.RowPolicy, ...] = () _NO_GOVERNANCE = _BqGovernance() @@ -866,6 +871,13 @@ def _bigquery_governance( ) ) ), + row_policies=( + () + if is_view + else tuple( + _gov.row_policies(contract, exposure, cid, _bq_table_name(exposure, loc), index) + ) + ), ) @@ -972,6 +984,10 @@ def _emit_policy_tags( one ``google_data_catalog_policy_tag_iam_member`` each, to exactly the readers ``column_access`` derives; a denied principal gets an access error on the column (``SELECT * EXCEPT (...)`` still works for it). + + A masked group's tag also gets a data policy (``DATA_MASKING_POLICY`` with the rule's + predefined expression), in the taxonomy's location, and the masked reader role on + it for exactly the group's masked principals: they read the column masked. """ if not tags: return {} @@ -1006,11 +1022,69 @@ def _emit_policy_tags( "role": _gov.FINE_GRAINED_READER_ROLE, "member": member, } + if group.mask: + _emit_data_policy(resources, group, name, body["region"], loc) for column in group.columns: refs[column] = name return refs +def _emit_row_policies( + resources: Dict[str, Any], + policies: Sequence[_gov.RowPolicy], + ds_ref: Any, + tbl_name: str, + project: Optional[str], +) -> None: + """A ``google_bigquery_row_access_policy`` per row policy of the table.""" + for policy in policies: + row: Dict[str, Any] = { + "dataset_id": ds_ref, + "table_id": tofu_ref(f"google_bigquery_table.{tbl_name}.table_id"), + "policy_id": policy.policy_id, + "filter_predicate": policy.predicate, + "grantees": list(policy.grantees), + } + if project: + row["project"] = project + resources.setdefault("google_bigquery_row_access_policy", {})[policy.key] = row + + +def _emit_data_policy( + resources: Dict[str, Any], + group: _gov.TagGroup, + tag_name: Any, + region: str, + loc: Mapping[str, Any], +) -> None: + """A masked group's data policy on its tag, and its masked readers.""" + dp_id = group.data_policy_id + body: Dict[str, Any] = { + "location": region, + "data_policy_id": dp_id, + "policy_tag": tag_name, + "data_policy_type": "DATA_MASKING_POLICY", + "data_masking_policy": {"predefined_expression": _gov.MASK_EXPRESSIONS[group.mask]}, + } + if loc.get("project"): + body["project"] = loc["project"] + resources.setdefault("google_bigquery_datapolicy_data_policy", {})[dp_id] = body + for member in group.masked_readers: + iam: Dict[str, Any] = { + "location": region, + "data_policy_id": tofu_ref( + f"google_bigquery_datapolicy_data_policy.{dp_id}.data_policy_id" + ), + "role": _gov.MASKED_READER_ROLE, + "member": member, + } + if loc.get("project"): + iam["project"] = loc["project"] + resources.setdefault("google_bigquery_datapolicy_data_policy_iam_member", {})[ + _iam_key(dp_id, member, role=_gov.MASKED_READER_ROLE) + ] = iam + + def _iam_key(stem: str, member: str, *, role: str) -> str: """A resource name for one role and member that no other member can share. @@ -1129,6 +1203,7 @@ def _emit_bigquery( if gov.retention is not None: _partition_table(resources, body, tbl_name, gov.retention) resources.setdefault("google_bigquery_table", {})[tbl_name] = body + _emit_row_policies(resources, gov.row_policies, ds_ref, tbl_name, loc.get("project")) # Table-level IAM replaces the dataset grants under shared mode: # the same principals, the same intent, the narrowest scope that still diff --git a/fluid_build/iac/providers/gcp_governance.py b/fluid_build/iac/providers/gcp_governance.py index e0e0d0e2..1d6662ae 100644 --- a/fluid_build/iac/providers/gcp_governance.py +++ b/fluid_build/iac/providers/gcp_governance.py @@ -53,6 +53,25 @@ ``policy_tags``), and ``roles/datacatalog.categoryFineGrainedReader`` for exactly those readers. The resource shapes follow cloud-foundation-fabric's ``data-catalog-policy-tag`` module. + +**Masking** is a restriction with ``access: mask`` (fluid-schema 0.7.6): the masked +columns get a policy tag of their own (a tag's data policy applies to every column +the tag carries, so they never share one with a denied column), a data policy on it +(``google_bigquery_datapolicy_data_policy``, ``DATA_MASKING_POLICY`` with the rule's +predefined expression), and ``roles/bigquerydatapolicy.maskedReader`` on the data +policy for exactly the masked principals. A masked principal reads the same column, +masked; a fine-grained reader of the tag reads it raw (BigQuery's column data +masking). The grant takes a minute or two to apply. + +**Row filters** are ``exposes[].policy.authz.rowFilters`` through ``iac/row_access.py``: +a row access policy per filter (``google_bigquery_row_access_policy``), and one that +selects every row (``TRUE``) for every other reader and writer of the expose, since a +principal no row access policy names reads no row (BigQuery row-level security). + +**Labels**: the contract's and the expose's ``labels``, and its classification, +jurisdiction and regulatory framework, go onto the dataset, table and key +(:func:`governance_labels`), so the regulation is on the resource and can be found +there, not only in the contract. """ from __future__ import annotations @@ -66,10 +85,12 @@ from ..access import normalize_access_grants from ..base import UnsupportedBindingError -from ..column_access import authz_readers, column_readers, restrictions_for +from ..column_access import authz_readers, column_masks, column_readers, restrictions_for +from ..governance_labels import governance_labels from ..naming import safe_ident from ..principals import GCP, gcp_grants, principal_map, resolve_principal from ..provider_match import canonical_cloud +from ..row_access import row_filters_for LOG = logging.getLogger(__name__) @@ -99,6 +120,24 @@ #: What a principal needs on a policy tag to read the columns it is attached to. FINE_GRAINED_READER_ROLE = "roles/datacatalog.categoryFineGrainedReader" +#: Lets a principal read a policy tag's columns masked, by the tag's data policy. +MASKED_READER_ROLE = "roles/bigquerydatapolicy.maskedReader" + +#: The longest data policy id BigQuery takes. It refuses a longer one ("should only contain +#: letters, numbers and underscores while under 200 characters"; measured 5 October 2026: +#: 200 characters accepted, 300 refused). +DATA_POLICY_ID_MAX = 199 + +#: ``columnRestrictions[].mask`` rule -> BigQuery data policy ``predefined_expression``. +MASK_EXPRESSIONS = { + "last_four": "LAST_FOUR_CHARACTERS", + "first_four": "FIRST_FOUR_CHARACTERS", + "nullify": "ALWAYS_NULL", +} + +#: The row access policy that lets every reader no filter names read every row. +ALL_ROWS_POLICY_ID = "fluid_all_rows" + #: Only DAY partitions are emitted: the one granularity retention in days maps onto. PARTITION_TYPE = "DAY" @@ -153,6 +192,23 @@ class TagGroup: #: policy tag's description (a policy tag has no labels of its own). rule_tags: Tuple[str, ...] = () rule_labels: Tuple[Tuple[str, str], ...] = () + #: The masking rule of a masked group (one of ``column_access.MASK_RULES``). + mask: Optional[str] = None + #: IAM members granted the masked reader role on the group's data policy. + masked_readers: Tuple[str, ...] = () + + @property + def data_policy_id(self) -> str: + """The data policy's id: ``[A-Za-z0-9_]``, unique in the project and location. + + A longer id than BigQuery takes keeps its first characters and ends in a hash of + the whole, so two long ids stay distinct. + """ + ident = safe_ident(f"{self.key}_mask") + if len(ident) <= DATA_POLICY_ID_MAX: + return ident + digest = hashlib.sha256(ident.encode("utf-8")).hexdigest()[:12] + return f"{ident[: DATA_POLICY_ID_MAX - len(digest) - 1]}_{digest}" @property def description(self) -> str: @@ -161,6 +217,11 @@ def description(self) -> str: f"Restricted columns {', '.join(self.columns)}: readable only by the principals " "granted the fine-grained reader role on this tag." ) + if self.mask: + text += ( + f" Masked ({self.mask}) for the principals granted the masked reader role on " + f"its data policy {self.data_policy_id}." + ) if self.rule_tags: text += f" Tags: {', '.join(self.rule_tags)}." if self.rule_labels: @@ -409,6 +470,7 @@ def expose_readers( where: str, exposure: Optional[Mapping[str, Any]] = None, readers_where: str = "policy.authz.readers", + any_grant: bool = False, ) -> FrozenSet[str]: """Every IAM member that reads the expose, mapped through ``binding.principals``. @@ -416,11 +478,11 @@ def expose_readers( the expose's own ``policy.authz.readers`` (which it does not: their table access is managed elsewhere). Both are readers a restriction narrows; leaving the second out gave a restricted column no reader at all, so a deny for one group locked it - for everyone. + for everyone. With ``any_grant``, every member with a grant of any verb. """ members: set[str] = set() for grant in gcp_grants(normalize_access_grants(contract), binding, where=where): - if READ_VERBS & set(grant.permissions): + if any_grant or READ_VERBS & set(grant.permissions): members.add(f"{grant.principal_type}:{grant.principal}") if exposure is not None: mapping = principal_map(binding) @@ -474,11 +536,15 @@ def resolve(principal: str) -> Tuple[str, ...]: ), ) by_column = column_readers(exposure, restrictions, resolve, readers, where=where) - grouped: Dict[FrozenSet[str], List[str]] = {} + masks = column_masks(exposure, restrictions, resolve, readers, where=where) + # A tag's data policy masks every column the tag carries, so a masked column + # shares a tag only with columns masked alike for the same principals. + grouped: Dict[Tuple[FrozenSet[str], Optional[str], FrozenSet[str]], List[str]] = {} for column, allowed in by_column.items(): - grouped.setdefault(allowed, []).append(column) + rule, masked = masks.get(column, (None, frozenset())) + grouped.setdefault((allowed, rule, masked), []).append(column) groups: List[TagGroup] = [] - for allowed, columns in grouped.items(): + for (allowed, rule, masked), columns in grouped.items(): rules = [r for r in restrictions if set(r.columns) & set(columns)] rule_tags = tuple(dict.fromkeys(t for r in rules for t in r.tags)) rule_labels = tuple(sorted({pair for r in rules for pair in r.labels})) @@ -490,11 +556,109 @@ def resolve(principal: str) -> Tuple[str, ...]: readers=tuple(sorted(allowed)), rule_tags=rule_tags, rule_labels=rule_labels, + mask=rule, + masked_readers=tuple(sorted(masked)), ) ) return groups +@dataclass(frozen=True) +class RowPolicy: + """One BigQuery row access policy of a table.""" + + #: Resource-name stem, unique in the module. + key: str + policy_id: str + predicate: str + grantees: Tuple[str, ...] + + +def expose_members( + contract: Mapping[str, Any], exposure: Mapping[str, Any], binding: Mapping[str, Any], index: int +) -> FrozenSet[str]: + """Every IAM member with any grant on the expose, readers and writers alike. + + The members a row access policy must name for them to see rows: a writer's own + checks (``fluid verify``, the drift gate) read the table too. + """ + at = _where(exposure, index) + return expose_readers( + contract, + binding, + where=f"{at} accessPolicy", + exposure=exposure, + readers_where=f"{at}.policy.authz.readers", + any_grant=True, + ) + + +def row_policies( + contract: Mapping[str, Any], + exposure: Mapping[str, Any], + cid: str, + table: str, + index: int = 0, +) -> List[RowPolicy]: + """The row access policies one table's row filters become, filters first. + + Each filter's grantees are every identity of its principal, whatever it holds on + the expose: a row access policy never grants the table, so naming an identity + narrows only what it reads through whichever grant it has, a writer's included + (``roles/bigquery.dataEditor`` reads rows too). The last policy selects every row + for every other member with a grant on the expose, and never for an identity a + filter names: BigQuery shows a principal the union of the policies naming it, so + a filtered identity in it would read every row. + """ + filters = row_filters_for(exposure, index) + if not filters: + return [] + binding = exposure.get("binding") or {} + where = f"{_where(exposure, index)}.policy.authz.rowFilters" + mapping = principal_map(binding) + members = expose_members(contract, exposure, binding, index) + out: List[RowPolicy] = [] + filtered: set[str] = set() + for flt in filters: + identities = set(resolve_principal(flt.principal, mapping, platform=GCP, where=where)) + filtered.update(identities) + if not identities: + LOG.info( + "row_filter_no_identity %s principal=%s: binding.principals maps it to no " + "identity on GCP, so there is no one to filter", + where, + flt.principal, + ) + continue + if not identities & members: + LOG.warning( + "row_filter_principal_without_grant %s principal=%s: the contract grants it " + "nothing on this expose; the filter narrows what it reads through any other " + "grant", + where, + flt.principal, + ) + out.append( + RowPolicy( + key=safe_ident(f"{cid}_{table}_{flt.name}"), + policy_id=flt.name, + predicate=flt.where, + grantees=tuple(sorted(identities)), + ) + ) + rest = sorted(members - filtered) + if rest: + out.append( + RowPolicy( + key=safe_ident(f"{cid}_{table}_{ALL_ROWS_POLICY_ID}"), + policy_id=ALL_ROWS_POLICY_ID, + predicate="TRUE", + grantees=tuple(rest), + ) + ) + return out + + def validate_bigquery_governance( contract: Mapping[str, Any], exposure: Mapping[str, Any], index: int, *, is_view: bool ) -> None: @@ -512,6 +676,13 @@ def validate_bigquery_governance( "attach to table columns, so restrict the columns of the table the view reads.", (), ) + if row_filters_for(exposure, index) and is_view: + raise UnsupportedBindingError( + "row-filter-view", + f"{where}.policy.authz.rowFilters is set on a BigQuery view; row access policies " + "attach to tables, so filter the rows of the table the view reads.", + (), + ) tag_groups(contract, exposure, "c", "d", "t", index) @@ -578,8 +749,9 @@ def refuse_mixed_dataset_encryption(contract: Mapping[str, Any]) -> None: def refuse_unsupported_target(exposure: Mapping[str, Any], index: int, target: str) -> None: """A GCP expose that is not a BigQuery table must not carry a policy it would drop. - The GCS, Pub/Sub and Iceberg-storage emitters write no lifecycle rule, key or column - control, so a contract asking for one there is refused rather than applied without it. + The GCS, Pub/Sub and Iceberg-storage emitters write no lifecycle rule, key, column + control or row filter, so a contract asking for one there is refused rather than + applied without it. """ where = _where(exposure, index) binding = exposure.get("binding") or {} @@ -591,6 +763,8 @@ def refuse_unsupported_target(exposure: Mapping[str, Any], index: int, target: s asked.append("binding.encryption") if restrictions_for(exposure, index): asked.append("policy.authz.columnRestrictions") + if row_filters_for(exposure, index): + asked.append("policy.authz.rowFilters") if asked: raise UnsupportedBindingError( "gcp-governance-unsupported-target", @@ -603,16 +777,22 @@ def refuse_unsupported_target(exposure: Mapping[str, Any], index: int, target: s __all__ = [ "BqEncryption", "BqRetention", + "ALL_ROWS_POLICY_ID", "FINE_GRAINED_READER_ROLE", "KEY_ROTATION_PERIOD", "KMS_ENCRYPTER_ROLE", + "MASKED_READER_ROLE", + "MASK_EXPRESSIONS", "PARTITION_TYPE", "PRODUCT_KEY_NAME", + "RowPolicy", "TagGroup", "dataset_location", "encryption_for", + "expose_members", "expose_readers", "gcp_owned", + "governance_labels", "kms_location", "partition_trigger_input", "product_key_name", @@ -620,6 +800,7 @@ def refuse_unsupported_target(exposure: Mapping[str, Any], index: int, target: s "refuse_mixed_dataset_encryption", "refuse_unsupported_target", "retention_for", + "row_policies", "tag_groups", "taxonomy_display_name", "taxonomy_region", diff --git a/fluid_build/iac/row_access.py b/fluid_build/iac/row_access.py new file mode 100644 index 00000000..9d758b24 --- /dev/null +++ b/fluid_build/iac/row_access.py @@ -0,0 +1,182 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Row filters: which rows a principal may read, one derivation for every cloud. + +``exposes[].policy.authz.rowFilters[]`` (``{principal, where, name?}``, fluid-schema +0.7.6) lets its principal read only the rows its ``where`` predicate selects. A reader +no filter names reads every row, and a principal has at most one filter per expose. +Each emitter writes the filters in its platform's own terms: + +* GCP BigQuery: a row access policy per filter (``google_bigquery_row_access_policy``, + ``filter_predicate`` and ``grantees``), and one more that selects every row for + every other reader and writer of the table, because BigQuery shows a principal that + no row access policy names no row at all (its row-level security docs). +* AWS Glue: a Lake Formation data cells filter (``aws_lakeformation_data_cells_filter``, + ``row_filter.filter_expression``, the principal's excluded columns as its column + wildcard's exclusions), and the principal's ``SELECT`` is granted on the filter + instead of on the table. + +The predicate is one SQL boolean expression over the expose's columns. It is parsed +(sqlglot) rather than trusted: a statement separator, a comment or a column the schema +does not declare is refused, so the text written into the platform's policy is the +predicate and nothing else. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any, List, Mapping, Tuple + +from .base import UnsupportedBindingError +from .column_access import policy_where + +_NAME = re.compile(r"^[a-z][a-z0-9_]{0,62}$") + + +@dataclass(frozen=True) +class RowFilter: + """One ``rowFilters[]`` entry, validated.""" + + principal: str + #: The predicate, exactly as written (trimmed). + where: str + #: The filter's name on the platform: ``name``, or derived from the principal. + name: str + #: The columns the predicate reads, in the order it names them. + columns: Tuple[str, ...] = () + + +def _where(exposure: Mapping[str, Any], index: int) -> str: + return policy_where(exposure, index, "rowFilters") + + +def _derived_name(principal: str) -> str: + """``analysts_rows`` for ``group:analysts@northwind.example``.""" + local = principal.split(":", 1)[-1].split("@", 1)[0] + stem = re.sub(r"[^a-z0-9]+", "_", local.lower()).strip("_") or "principal" + if not stem[0].isalpha(): + stem = f"p_{stem}" + return f"{stem[:56]}_rows" + + +def _predicate_columns(where: str, at: str) -> Tuple[str, ...]: + """The columns ``where`` reads; a predicate that is not one boolean expression is refused.""" + if ";" in where or "--" in where or "/*" in where or "*/" in where: + raise UnsupportedBindingError( + "row-filter", + f"{at}.where holds a statement separator or a comment; a row filter is one " + "boolean expression, written into the platform's policy as it stands.", + ("Write the predicate alone, e.g. consent_data_analytics = true.",), + ) + try: + import sqlglot + from sqlglot import expressions as exp + except ImportError: # pragma: no cover - sqlglot is a core dependency + return () + try: + tree = sqlglot.parse_one(f"SELECT 1 FROM t WHERE {where}") + except sqlglot.errors.ParseError as exc: + raise UnsupportedBindingError( + "row-filter", + f"{at}.where does not parse as a SQL boolean expression: {str(exc).splitlines()[0]}", + ("Write one boolean expression over the expose's columns.",), + ) from exc + clause = tree.args.get("where") + if ( + not isinstance(tree, exp.Select) + or clause is None + or tree.args.get("group") + or tree.args.get("order") + or tree.args.get("limit") + or tree.find(exp.Subquery, exp.Union) + ): + raise UnsupportedBindingError( + "row-filter", + f"{at}.where must be one boolean expression over the expose's columns, with no " + "subquery, so the platform evaluates it on each row by itself.", + ("Write the predicate alone, e.g. consent_data_analytics = true.",), + ) + return tuple(dict.fromkeys(col.name for col in clause.find_all(exp.Column))) + + +def row_filters_for(exposure: Mapping[str, Any], index: int = 0) -> Tuple[RowFilter, ...]: + """The expose's row filters, or ``()``; a malformed one is refused.""" + policy = exposure.get("policy") if isinstance(exposure, Mapping) else None + authz = policy.get("authz") if isinstance(policy, Mapping) else None + raw = authz.get("rowFilters") if isinstance(authz, Mapping) else None + if not raw: + return () + where_at = _where(exposure, index) + if not isinstance(raw, list): + raise UnsupportedBindingError("row-filter", f"{where_at} must be a list of filters.", ()) + schema = (exposure.get("contract") or {}).get("schema") or [] + declared = {col.get("name") for col in schema if isinstance(col, Mapping)} + out: List[RowFilter] = [] + seen_principals: set[str] = set() + seen_names: set[str] = set() + for i, entry in enumerate(raw): + at = f"{where_at}[{i}]" + if not isinstance(entry, Mapping): + raise UnsupportedBindingError("row-filter", f"{at} must be a mapping.", ()) + principal = entry.get("principal") + where = entry.get("where") + if not isinstance(principal, str) or not principal.strip(): + raise UnsupportedBindingError( + "row-filter", + f"{at} names no principal, so there is no one to filter.", + ("Set principal to the logical principal the filter applies to.",), + ) + if not isinstance(where, str) or not where.strip(): + raise UnsupportedBindingError( + "row-filter", + f"{at} has no where predicate, so it selects nothing.", + ("Set where to one SQL boolean expression over the expose's columns.",), + ) + principal, where = principal.strip(), where.strip() + if principal in seen_principals: + raise UnsupportedBindingError( + "row-filter", + f"{at} filters {principal} a second time; a principal has one row filter per " + "expose, which the platform applies as it stands.", + ("Combine the two predicates with AND into one filter.",), + ) + seen_principals.add(principal) + name = entry.get("name") or _derived_name(principal) + if not isinstance(name, str) or not _NAME.match(name): + raise UnsupportedBindingError( + "row-filter", + f"{at}.name {name!r} must be lowercase letters, digits and underscores, " + "starting with a letter, at most 63 characters.", + (), + ) + if name in seen_names: + raise UnsupportedBindingError( + "row-filter", f"{at}.name {name!r} names another filter of this expose too.", () + ) + seen_names.add(name) + columns = _predicate_columns(where, at) + unknown = [c for c in columns if c not in declared] + if unknown: + raise UnsupportedBindingError( + "row-filter", + f"{at}.where reads {unknown}, which the expose's schema does not declare.", + ("Name columns of exposes[].contract.schema in the predicate.",), + ) + out.append(RowFilter(principal=principal, where=where, name=name, columns=columns)) + return tuple(out) + + +__all__ = ["RowFilter", "row_filters_for"] diff --git a/fluid_build/iac/state_migration.py b/fluid_build/iac/state_migration.py index 6ebed551..09634d08 100644 --- a/fluid_build/iac/state_migration.py +++ b/fluid_build/iac/state_migration.py @@ -37,7 +37,7 @@ provider the *state* names, not only the module's: a ``{"terraform": {}}`` module beside a state naming ``hashicorp/null`` installed the latest ``hashicorp/null`` (measured, tofu 1.12), and for an aws state that is the -latest ``hashicorp/aws``, not the pinned ``~> 5.0``. The probe runs on every +latest ``hashicorp/aws``, not the pinned ``~> 6.0``. The probe runs on every apply whose new key is still empty (every ``--dry-run`` while a move is pending, every gcp run while the old key holds the aws state), so it installs nothing: ``-plugin-dir`` names an empty directory, the init stops at its diff --git a/fluid_build/iac/versions.py b/fluid_build/iac/versions.py index 7f802159..74f9cf90 100644 --- a/fluid_build/iac/versions.py +++ b/fluid_build/iac/versions.py @@ -32,7 +32,7 @@ #: ``Snowflake-Labs`` org to the official ``snowflakedb`` org at v2. PROVIDER_PINS: Dict[str, Dict[str, str]] = { "google": {"source": "hashicorp/google", "version": "~> 6.0"}, - "aws": {"source": "hashicorp/aws", "version": "~> 5.0"}, + "aws": {"source": "hashicorp/aws", "version": "~> 6.0"}, "snowflake": {"source": "snowflakedb/snowflake", "version": "~> 2.0"}, # Confluent Cloud — Tableflow (managed Kafka→Iceberg) + Glue catalog / # provider integration. The managed control plane owns compaction + diff --git a/fluid_build/schemas/fluid-schema-0.7.6.json b/fluid_build/schemas/fluid-schema-0.7.6.json index 28ca63eb..1c52b7d9 100644 --- a/fluid_build/schemas/fluid-schema-0.7.6.json +++ b/fluid_build/schemas/fluid-schema-0.7.6.json @@ -1184,7 +1184,7 @@ }, "columnRestrictions": { "type": "array", - "description": "Column-level access control. A column named in any restriction is restricted. 'deny': the principal may not read the columns. 'allow': the columns are readable only by the principals an allow names. A deny beats an allow, and a restriction never grants access: the readers are the expose's readers. Principals are logical and resolve through binding.principals. NEW in v0.7.6 (the field itself is older): `fluid apply` enforces it. GCP BigQuery: a Data Catalog taxonomy per product and dataset with fine-grained access control, a policy tag per set of restricted columns that share their readers, attached through the table schema's policyTags, and roles/datacatalog.categoryFineGrainedReader for exactly the expose's readers allowed to read them: the accessPolicy read grantees and the expose's own policy.authz.readers (a denied principal gets an access error on the columns). A restriction on an expose with no reader on the binding is refused, as it would lock the columns for everyone. A restriction's tags and labels are written into the policy tag's description; AWS has no place for them. AWS: each governance.lakeFormation grant with SELECT excludes the restricted columns its principal may not read (table_with_columns.excluded_column_names); a grant's hand-written excludedColumns must agree, and a binding with no Lake Formation grants is refused. `fluid verify` checks the policy tags and their readers on GCP, and the principals' Lake Formation permissions on AWS.", + "description": "Column-level access control. A column named in any restriction is restricted. 'deny': the principal may not read the columns. 'allow': the columns are readable only by the principals an allow names. A deny beats an allow, and a restriction never grants access: the readers are the expose's readers. Principals are logical and resolve through binding.principals. NEW in v0.7.6 (the field itself is older): `fluid apply` enforces it. GCP BigQuery: a Data Catalog taxonomy per product and dataset with fine-grained access control, a policy tag per set of restricted columns that share their readers, attached through the table schema's policyTags, and roles/datacatalog.categoryFineGrainedReader for exactly the expose's readers allowed to read them: the accessPolicy read grantees and the expose's own policy.authz.readers (a denied principal gets an access error on the columns). A restriction on an expose with no reader on the binding is refused, as it would lock the columns for everyone. A restriction's tags and labels are written into the policy tag's description; AWS has no place for them. AWS: each governance.lakeFormation grant with SELECT excludes the restricted columns its principal may not read (table_with_columns.excluded_column_names); a grant's hand-written excludedColumns must agree, and a binding with no Lake Formation grants is refused. `fluid verify` checks the policy tags and their readers on GCP, and the principals' Lake Formation permissions on AWS. NEW in v0.7.6: access: mask lets the principal read the columns masked by the platform (see mask): a deny beats a mask, and a mask beats an allow for the principal it names.", "items": { "type": "object", "additionalProperties": false, @@ -1204,9 +1204,10 @@ "type": "string", "enum": [ "allow", - "deny" + "deny", + "mask" ], - "description": "deny: the principal may not read the columns. allow: only the principals an allow names may read them." + "description": "deny: the principal may not read the columns. allow: only the principals an allow names may read them. mask: the principal reads the columns masked by the platform, by the rule in mask, and never their raw values (NEW in v0.7.6)." }, "tags": { "$ref": "#/$defs/tags", @@ -1215,8 +1216,53 @@ "labels": { "$ref": "#/$defs/labels", "description": "Access control rule labels." + }, + "mask": { + "type": "string", + "enum": [ + "last_four", + "first_four", + "nullify" + ], + "description": "The masking rule an access: mask restriction applies, required with it and refused without it. last_four: 'XXXXX' and the value's last four characters; first_four: its first four characters and 'XXXXX' (a value of four characters or fewer is hashed instead, by each platform); nullify: NULL. GCP BigQuery: a data policy (DATA_MASKING_POLICY, LAST_FOUR_CHARACTERS, FIRST_FOUR_CHARACTERS or ALWAYS_NULL) on the columns' policy tag, and roles/bigquerydatapolicy.maskedReader for the masked principals, who read the same column masked. AWS Glue: Lake Formation cannot mask, so the masked principals' grant on the table excludes the columns and they read a protected Glue Data Catalog view (SECURITY DEFINER, binding.governance.lakeFormation.maskedViews) that returns them masked by the same rule (NEW in v0.7.6)." } - } + }, + "allOf": [ + { + "if": { + "properties": { + "access": { + "const": "mask" + } + }, + "required": [ + "access" + ] + }, + "then": { + "required": [ + "mask" + ] + } + }, + { + "if": { + "required": [ + "mask" + ] + }, + "then": { + "properties": { + "access": { + "const": "mask" + } + }, + "required": [ + "access" + ] + } + } + ] } }, "tags": { @@ -1226,6 +1272,37 @@ "labels": { "$ref": "#/$defs/labels", "description": "Authorization policy labels." + }, + "rowFilters": { + "type": "array", + "description": "Row-level access control (NEW in v0.7.6). Each filter lets its principal read only the rows its where predicate selects; a reader no filter names reads every row, and a principal may have one filter per expose. The predicate is one SQL boolean expression over this expose's columns, with no statement separator or comment. Principals are logical and resolve through binding.principals. GCP BigQuery: a row access policy (filter_predicate, grantees) per filter, plus one that selects every row for every other reader and writer, since a principal no row access policy names reads no row at all. AWS Glue: a Lake Formation data cells filter (row_filter and the principal's excluded columns), and the principal's SELECT is granted on the filter instead of the table.", + "items": { + "type": "object", + "additionalProperties": false, + "required": [ + "principal", + "where" + ], + "properties": { + "principal": { + "type": "string", + "description": "Logical principal the filter applies to, written as in accessPolicy; binding.principals maps it per cloud." + }, + "name": { + "type": "string", + "pattern": "^[a-z][a-z0-9_]{0,62}$", + "description": "The filter's name on the platform (the row access policy or data cells filter); derived from the principal when omitted." + }, + "where": { + "type": "string", + "minLength": 1, + "description": "One SQL boolean expression over the expose's columns, e.g. consent_data_analytics = true AND is_suppressed = false." + }, + "description": { + "type": "string" + } + } + } } } }, @@ -5966,6 +6043,25 @@ "name", "rowExpression" ] + }, + "maskedViews": { + "type": "object", + "additionalProperties": false, + "required": [ + "definer", + "validationConnection" + ], + "description": "Where the protected Glue Data Catalog views of the expose's masked columns come from (NEW in v0.7.6). Lake Formation cannot mask, so a principal an access: mask column restriction names reads a protected, SECURITY DEFINER multi-dialect view,
_masked, that returns the columns masked by the restriction's rule and applies its row filter; its grant on the table excludes the columns. The definer is granted SELECT with grant option on the table and DESCRIBE on the database; the principal SELECT and DESCRIBE on the view.", + "properties": { + "definer": { + "type": "string", + "description": "ARN of the IAM role the views are defined as: Glue and Lake Formation must be allowed to assume it (sts:AssumeRole, sts:SetContext), and the applying principal to pass it (iam:PassRole)." + }, + "validationConnection": { + "type": "string", + "description": "The Glue connection of type VIEW_VALIDATION_ATHENA (WORKGROUP_NAME) Glue validates each view's SQL through; the applying principal and the definer need glue:PassConnection on it." + } + } } } } diff --git a/tests/build_runners/test_duckdb_bigquery_load.py b/tests/build_runners/test_duckdb_bigquery_load.py index e52be26f..9397a5b6 100644 --- a/tests/build_runners/test_duckdb_bigquery_load.py +++ b/tests/build_runners/test_duckdb_bigquery_load.py @@ -163,7 +163,7 @@ def test_the_built_file_is_loaded_into_the_declared_table(bq, tmp_path): assert table_id == "acme-eu.bronze.orders" assert location == "europe-west1" assert cfg["source_format"] == "PARQUET" - assert cfg["write_disposition"] == "WRITE_TRUNCATE" + assert cfg["write_disposition"] == "WRITE_TRUNCATE_DATA", "keeps row access policies" assert cfg["create_disposition"] == "CREATE_NEVER", "the table is the IaC's" assert cfg["schema"] == ["the-declared-schema"], "the table's own schema, not the file's" staged = tmp_path / ".fluid" / "staging" / "ingest_orders" / "orders.parquet" @@ -310,3 +310,33 @@ def test_an_unrelated_first_expose_does_not_capture_the_build(bq, tmp_path): assert _run(contract, tmp_path) == 0 assert [c[1] for c in fake.calls if c[0] == "load"] == ["acme-eu.bronze.orders"] assert not decoy.exists() + + +# ── A full refresh keeps the table's row access policies ─────────────── +# +# Measured on BigQuery, 5 October 2026: a WRITE_TRUNCATE load removed the +# table's row access policy and the table showed every row; a +# WRITE_TRUNCATE_DATA load kept it. BigQuery documents the first ("operations +# that use the WRITE_TRUNCATE write disposition remove all existing row access +# policies"), so every build would have undone the row filters tofu apply made. + + +def test_a_full_refresh_keeps_the_table_s_row_access_policies(monkeypatch): + monkeypatch.delenv(_bigquery_load.EMULATOR_HOST_ENV, raising=False) + assert _bigquery_load.write_disposition("full_refresh") == "WRITE_TRUNCATE_DATA" + assert _bigquery_load.write_disposition("incremental_append") == "WRITE_APPEND" + + +def test_no_mode_loads_with_a_disposition_that_drops_row_access_policies(monkeypatch): + monkeypatch.delenv(_bigquery_load.EMULATOR_HOST_ENV, raising=False) + for mode in _bigquery_load._WRITE_DISPOSITION: + assert _bigquery_load.write_disposition(mode) != "WRITE_TRUNCATE", mode + + +def test_on_an_emulator_a_full_refresh_is_write_truncate(monkeypatch): + """goccy/bigquery-emulator appends for a disposition it does not know + (``server/handler.go`` handles WRITE_TRUNCATE and WRITE_EMPTY), so a full + refresh there would double the table; it has no row access policy to keep.""" + monkeypatch.setenv(_bigquery_load.EMULATOR_HOST_ENV, "localhost:9050") + assert _bigquery_load.write_disposition("full_refresh") == "WRITE_TRUNCATE" + assert _bigquery_load.write_disposition("incremental_append") == "WRITE_APPEND" diff --git a/tests/build_runners/test_embedded_sql_bigquery.py b/tests/build_runners/test_embedded_sql_bigquery.py index 4a3a1c0f..c03454ec 100644 --- a/tests/build_runners/test_embedded_sql_bigquery.py +++ b/tests/build_runners/test_embedded_sql_bigquery.py @@ -143,7 +143,14 @@ def load_table_from_file(self, fh, table_id, job_config=None, location=None): if load_error is not None: raise load_error entry = fake.tables[table_id] - if job_config.write_disposition == "WRITE_TRUNCATE" or not entry["data"].num_rows: + if ( + job_config.write_disposition + in ( + "WRITE_TRUNCATE", + "WRITE_TRUNCATE_DATA", + ) + or not entry["data"].num_rows + ): entry["data"] = data else: entry["data"] = pa.concat_tables([entry["data"], data]) @@ -401,7 +408,7 @@ def test_the_sql_reads_the_bigquery_rows_and_lands_them_in_bigquery(bq, tmp_path [load] = fake.loads assert load["table"] == SILVER_TABLE assert load["location"] == "europe-west1" - assert load["config"]["write_disposition"] == "WRITE_TRUNCATE" + assert load["config"]["write_disposition"] == "WRITE_TRUNCATE_DATA", "keeps row access policies" assert load["config"]["create_disposition"] == "CREATE_NEVER" # Nothing was written to a local file named after the binding's gs:// path. assert _gs_files(tmp_path) == [] diff --git a/tests/iac/_moto_s3.py b/tests/iac/_moto_s3.py new file mode 100644 index 00000000..101fdb30 --- /dev/null +++ b/tests/iac/_moto_s3.py @@ -0,0 +1,71 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""moto keeps the tags a CreateBucket request carries, as S3 does. + +S3's CreateBucket takes tags in its ``CreateBucketConfiguration`` (````, +added to the API in 2025), and terraform-provider-aws 6.23 and later tag a bucket +that way, skipping ``PutBucketTagging`` when the create succeeds. moto 5.2.3 reads +only ``LocationConstraint`` from that body and drops the tags, so every bucket the +moto lanes apply was left untagged, and a clean apply read back as tag drift +(``tags.fluid_contract``, ``tags.managed_by``). On real S3 the tags are kept: the +lab's eleven AWS products planned no change on provider 6.67. + +:func:`keep_create_bucket_tags` wraps moto's bucket PUT so a create also stores the +body's tags, the fix fakecloud made in its own S3 (faiscadev/fakecloud#2557). It +changes nothing when moto is not installed, and is idempotent. +""" + +from __future__ import annotations + +from typing import Any, Dict + + +def _create_tags(body: Any) -> Dict[str, str]: + """The tags of a CreateBucketConfiguration body, or ``{}``.""" + import xmltodict + + if not body: + return {} + try: + cfg = (xmltodict.parse(body) or {}).get("CreateBucketConfiguration") or {} + except Exception: # noqa: BLE001 — not a configuration body: no tags + return {} + tags = (cfg.get("Tags") or {}).get("Tag") if isinstance(cfg, dict) else None + if isinstance(tags, dict): + tags = [tags] + return {str(t["Key"]): str(t.get("Value") or "") for t in tags or () if t.get("Key")} + + +def keep_create_bucket_tags() -> bool: + """Make moto's CreateBucket keep its tags. True when moto is installed and patched.""" + try: + from moto.s3.responses import S3Response + except ImportError: + return False + original = S3Response._bucket_response_put + if getattr(original, "_fluid_keeps_create_tags", False): + return True + + def bucket_response_put(self: Any, request: Any, bucket_name: str, querystring: Any) -> Any: + result = original(self, request, bucket_name, querystring) + if not querystring: # a CreateBucket, not a sub-resource PUT + tags = _create_tags(self.body) + if tags: + self.backend.put_bucket_tagging(bucket_name, tags) + return result + + bucket_response_put._fluid_keeps_create_tags = True # type: ignore[attr-defined] + S3Response._bucket_response_put = bucket_response_put + return True diff --git a/tests/iac/conftest.py b/tests/iac/conftest.py index efac2625..8c259ec1 100644 --- a/tests/iac/conftest.py +++ b/tests/iac/conftest.py @@ -71,6 +71,14 @@ ) from fluid_build.iac.credentials import build_tofu_env +from ._moto_s3 import keep_create_bucket_tags + +# moto's CreateBucket drops the tags of its request body, which terraform-provider-aws +# 6.23+ tags a bucket with; S3 keeps them (tests/iac/_moto_s3.py). Patched once per +# process, before any moto server starts. +keep_create_bucket_tags() + + # --------------------------------------------------------------------------- # tofu init resilience — tolerate transient provider-registry outages # --------------------------------------------------------------------------- diff --git a/tests/iac/test_iac_aws.py b/tests/iac/test_iac_aws.py index 54daef6e..63d3fdf0 100644 --- a/tests/iac/test_iac_aws.py +++ b/tests/iac/test_iac_aws.py @@ -204,6 +204,27 @@ def test_localstack_endpoint_emits_path_style_and_skips(self, monkeypatch): assert aws["skip_metadata_api_check"] is True assert aws["skip_region_validation"] is True + def test_on_an_emulator_each_glue_resource_names_its_catalog(self, monkeypatch): + """Provider 6 builds a Glue id from the catalog; with the account id skipped it was + ``:``, which 6.x refuses to read back (hashicorp/terraform-provider-aws#31626).""" + self._clear_endpoint_env(monkeypatch) + monkeypatch.setenv("AWS_ENDPOINT_URL", "http://localhost:4566") + c = _contract([_glue_exposure(database="d", table="t", bucket="b")]) + doc = json.loads(build_module(_aws(), c)) + ref = "${data.aws_caller_identity.fluid_lf_caller.account_id}" + (db,) = doc["resource"]["aws_glue_catalog_database"].values() + (table,) = doc["resource"]["aws_glue_catalog_table"].values() + assert db["catalog_id"] == ref and table["catalog_id"] == ref + assert "fluid_lf_caller" in doc["data"]["aws_caller_identity"] + + def test_on_real_aws_no_glue_resource_names_its_catalog(self, monkeypatch): + self._clear_endpoint_env(monkeypatch) + c = _contract([_glue_exposure(database="d", table="t", bucket="b")]) + doc = json.loads(build_module(_aws(), c)) + (db,) = doc["resource"]["aws_glue_catalog_database"].values() + (table,) = doc["resource"]["aws_glue_catalog_table"].values() + assert "catalog_id" not in db and "catalog_id" not in table + def test_per_service_endpoint_also_triggers_compat(self, monkeypatch): self._clear_endpoint_env(monkeypatch) monkeypatch.setenv("AWS_ENDPOINT_URL_S3", "http://localhost:4566") diff --git a/tests/iac/test_iac_aws_apply_engine_moto.py b/tests/iac/test_iac_aws_apply_engine_moto.py index d8b89e9c..dfe22f8b 100644 --- a/tests/iac/test_iac_aws_apply_engine_moto.py +++ b/tests/iac/test_iac_aws_apply_engine_moto.py @@ -18,10 +18,11 @@ just the plugin — so the full pipeline is exercised: contract → ``.tf.json`` → ``tofu init/plan/apply``, including the data-loss gate and ``--dry-run``. -A moto ``ThreadedMotoServer`` provides the AWS API surface; a sidecar -``provider.tf.json`` aiming the AWS provider at moto is dropped into the +A moto ``ThreadedMotoServer`` provides the AWS API surface, named by +``AWS_ENDPOINT_URL`` so the plugin emits its module for an emulator; a sidecar +``provider_override.tf.json`` aiming each service at moto is dropped into the engine's own workdir (``/.fluid/iac/aws//``) before -apply, so ``tofu init`` merges it with the plugin's credential-free +apply, so ``tofu init`` merges it into the plugin's credential-free ``main.tf.json``. Skipped unless ``tofu`` is on PATH and moto's ``server`` extra is installed. """ @@ -68,9 +69,14 @@ def _have_moto_server() -> bool: @pytest.fixture -def moto_endpoint() -> Iterator[str]: +def moto_endpoint(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]: """Start a fresh moto AWS server and clear its process-wide state. + ``AWS_ENDPOINT_URL`` names it, as an emulator user sets it, so the engine + compiles the module for an emulator: provider 6 needs each Glue resource's + catalog id there (``aws._emulator_catalog_id``), since the provider skips + requesting the account id. + moto's ``ThreadedMotoServer`` shares state with every other moto process-level mock — stopping and starting a new server does NOT clear the backends. The ``/moto-api/reset`` admin endpoint clears all state, @@ -84,6 +90,7 @@ def moto_endpoint() -> Iterator[str]: try: _, port = server.get_host_and_port() endpoint = f"http://127.0.0.1:{port}" + monkeypatch.setenv("AWS_ENDPOINT_URL", endpoint) # Reset before yielding — also covers state leaked in by a previous # test running in the same process. with contextlib.suppress(Exception): @@ -178,13 +185,14 @@ def _apply_args(contract_path: Path, workspace_dir: Path, **overrides) -> argpar def _drop_moto_provider_sidecar(workspace_dir: Path, contract_id: str, endpoint: str) -> Path: - """Drop ``provider.tf.json`` into the apply engine's workdir BEFORE the - engine runs — ``tofu init`` merges every ``*.tf.json`` in the directory, - so this overlays the moto endpoint + dummy creds onto the plugin's - credential-free ``main.tf.json``.""" + """Drop ``provider_override.tf.json`` into the apply engine's workdir BEFORE + the engine runs. An override file merges into the provider block the plugin + emits for an emulator (a second plain ``provider "aws"`` block would be a + duplicate configuration), overlaying the moto endpoint per service and the + dummy creds onto the credential-free ``main.tf.json``.""" workdir = workspace_dir / ".fluid" / "iac" / "aws" / safe_ident(contract_id) workdir.mkdir(parents=True, exist_ok=True) - sidecar = workdir / "provider.tf.json" + sidecar = workdir / "provider_override.tf.json" sidecar.write_text(json.dumps(_provider_override(endpoint)), encoding="utf-8") return workdir diff --git a/tests/iac/test_iac_aws_retention_encryption_moto.py b/tests/iac/test_iac_aws_retention_encryption_moto.py index 03b494cd..86288397 100644 --- a/tests/iac/test_iac_aws_retention_encryption_moto.py +++ b/tests/iac/test_iac_aws_retention_encryption_moto.py @@ -44,10 +44,12 @@ import datetime as dt import json +import os import subprocess import time from pathlib import Path from typing import Any, Dict, Iterator, List, Optional +from unittest import mock import pytest @@ -187,9 +189,19 @@ def _tofu(workdir: Path, env: Dict[str, str], *args: str) -> "subprocess.Complet def _write(contract: Dict[str, Any], workdir: Path, endpoint: str) -> None: + """The module as forge-cli emits it for an emulator, with moto's routing as its provider. + + Emitted with ``AWS_ENDPOINT_URL`` set, as an emulator user runs it: the provider skips + requesting the account id, so each Glue resource names its catalog + (``aws._emulator_catalog_id``); provider 6 refuses to read back a Glue id with an empty + one. The emitted provider block is replaced by this file's, which routes each service to + moto, so there is one provider configuration. + """ workdir.mkdir(parents=True, exist_ok=True) - (workdir / "main.tf.json").write_text(build_module(get_iac_plugin("aws"), contract)) - (workdir / "provider.tf.json").write_text(json.dumps(_provider_override(endpoint))) + with mock.patch.dict(os.environ, {"AWS_ENDPOINT_URL": endpoint}): + module = json.loads(build_module(get_iac_plugin("aws"), contract)) + module["provider"] = _provider_override(endpoint)["provider"] + (workdir / "main.tf.json").write_text(json.dumps(module, indent=2)) def _init(workdir: Path, env: Dict[str, str]) -> None: diff --git a/tests/iac/test_iac_aws_validate.py b/tests/iac/test_iac_aws_validate.py index 76a12604..d8e06221 100644 --- a/tests/iac/test_iac_aws_validate.py +++ b/tests/iac/test_iac_aws_validate.py @@ -17,7 +17,7 @@ The contract-shape gate for the AWS / Iceberg / Redshift mesh path. Each representative contract — Iceberg-on-Glue, Redshift Serverless, the external-schema bridge, and the full data-mesh dual-port scenario — must -compile to a module the real provider schemas (``hashicorp/aws ~> 5.0`` +compile to a module the real provider schemas (``hashicorp/aws ~> 6.0`` + ``hashicorp/null ~> 3.0``) accept. Needs ``tofu`` on PATH and registry network access (``tofu init`` downloads diff --git a/tests/iac/test_iac_gcp_governance.py b/tests/iac/test_iac_gcp_governance.py index 90954ed7..ffda35d0 100644 --- a/tests/iac/test_iac_gcp_governance.py +++ b/tests/iac/test_iac_gcp_governance.py @@ -528,7 +528,7 @@ def test_an_allow_never_grants_access_to_a_non_reader(self, caplog): "restriction, fragment", [ ({"principal": ANALYSTS, "columns": ["nope"], "access": "deny"}, "does not declare"), - ({"principal": ANALYSTS, "columns": ["msisdn"]}, "must be 'allow' or 'deny'"), + ({"principal": ANALYSTS, "columns": ["msisdn"]}, "must be 'allow', 'deny' or 'mask'"), ({"columns": ["msisdn"], "access": "deny"}, "names no principal"), ({"principal": ANALYSTS, "columns": [], "access": "deny"}, "must list the columns"), ], diff --git a/tests/iac/test_iac_mask_rowfilters_labels.py b/tests/iac/test_iac_mask_rowfilters_labels.py new file mode 100644 index 00000000..25b59229 --- /dev/null +++ b/tests/iac/test_iac_mask_rowfilters_labels.py @@ -0,0 +1,557 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Masked columns, row filters and governance labels, on GCP and on AWS. + +``policy.authz.columnRestrictions[]`` with ``access: mask`` lets a principal read a +column masked by the platform; ``policy.authz.rowFilters[]`` lets it read only the +rows a predicate selects; the contract's labels, classification and sovereignty land +on the resources as labels (GCP) and tags or parameters (AWS). These pin what each +emitter writes, and what is refused rather than dropped. +""" + +from __future__ import annotations + +import copy +import json +from pathlib import Path +from typing import Any, Dict, List, Optional + +import pytest + +from fluid_build.iac import build_module, get_iac_plugin +from fluid_build.iac.base import UnsupportedBindingError +from fluid_build.iac.column_access import column_masks, restrictions_for +from fluid_build.iac.governance_labels import governance_labels +from fluid_build.iac.row_access import row_filters_for + +ANALYSTS = "group:analysts@northwind.example" +STEWARDS = "group:data-stewards@northwind.example" +PIPELINE = "serviceAccount:fluid-pipeline@northwind.example" +GCP_MAP = { + ANALYSTS: "serviceAccount:analyst@northwind-demo.iam.gserviceaccount.com", + STEWARDS: "serviceAccount:steward@northwind-demo.iam.gserviceaccount.com", + PIPELINE: "serviceAccount:deploy@northwind-demo.iam.gserviceaccount.com", +} +ACCOUNT = "123456789012" +ANALYST_ARN = f"arn:aws:iam::{ACCOUNT}:role/analyst" +STEWARD_ARN = f"arn:aws:iam::{ACCOUNT}:role/steward" +DEFINER_ARN = f"arn:aws:iam::{ACCOUNT}:role/view-definer" +AWS_MAP = { + ANALYSTS: ANALYST_ARN, + STEWARDS: STEWARD_ARN, + PIPELINE: f"arn:aws:iam::{ACCOUNT}:role/box", +} + +DENY_ID = {"principal": ANALYSTS, "columns": ["customer_id"], "access": "deny"} +MASK_MSISDN = {"principal": ANALYSTS, "columns": ["msisdn"], "access": "mask", "mask": "last_four"} +CONSENT = { + "principal": ANALYSTS, + "name": "analysts_consented", + "where": "consent = true AND (suppressed IS NULL OR suppressed = false)", +} +SCHEMA = [ + {"name": "customer_id", "type": "string"}, + {"name": "msisdn", "type": "string"}, + {"name": "msisdn_hash", "type": "string"}, + {"name": "consent", "type": "boolean"}, + {"name": "suppressed", "type": "boolean"}, + {"name": "score", "type": "integer"}, +] + + +def _expose( + binding: Dict[str, Any], + restrictions: Optional[List[Dict[str, Any]]] = None, + row_filters: Optional[List[Dict[str, Any]]] = None, +) -> Dict[str, Any]: + authz: Dict[str, Any] = {} + if restrictions is not None: + authz["columnRestrictions"] = restrictions + if row_filters is not None: + authz["rowFilters"] = row_filters + expose: Dict[str, Any] = { + "exposeId": "profile", + "binding": binding, + "labels": {"layer": "silver"}, + "contract": {"schema": copy.deepcopy(SCHEMA)}, + "policy": {"classification": "Confidential", "authz": authz}, + } + return expose + + +def _contract(expose: Dict[str, Any]) -> Dict[str, Any]: + return { + "fluidVersion": "0.7.6", + "id": "silver.profile", + "labels": {"gdpr-article": "art6-1b-contractual", "Cost Center": "NW CX 001"}, + "sovereignty": { + "jurisdiction": "EU", + "regulatoryFramework": ["GDPR"], + "dataResidency": True, + }, + "accessPolicy": { + "grants": [ + {"principal": STEWARDS, "permissions": ["read", "select", "query"]}, + {"principal": ANALYSTS, "permissions": ["read", "select", "query"]}, + {"principal": PIPELINE, "permissions": ["read", "write", "insert"]}, + ] + }, + "exposes": [expose], + } + + +def _gcp(restrictions=None, row_filters=None) -> Dict[str, Any]: + binding = { + "platform": "gcp", + "format": "bigquery_table", + "location": { + "project": "northwind-demo", + "dataset": "silver", + "table": "profile", + "region": "europe-west3", + }, + "principals": GCP_MAP, + } + return _contract(_expose(binding, restrictions, row_filters)) + + +def _aws(restrictions=None, row_filters=None, masked_views=True) -> Dict[str, Any]: + lake_formation: Dict[str, Any] = { + "registerLocation": True, + "grants": [ + {"principal": STEWARD_ARN, "permissions": ["SELECT", "DESCRIBE"]}, + {"principal": ANALYST_ARN, "permissions": ["SELECT", "DESCRIBE"]}, + ], + } + if masked_views: + lake_formation["maskedViews"] = { + "definer": DEFINER_ARN, + "validationConnection": "views-conn", + } + binding = { + "platform": "aws", + "format": "parquet", + "location": { + "database": "silver", + "table": "profile", + "bucket": "nw-silver", + "path": "silver/profile/", + }, + "principals": AWS_MAP, + "governance": {"lakeFormation": lake_formation}, + } + return _contract(_expose(binding, restrictions, row_filters)) + + +def _gcp_resources(contract: Dict[str, Any]) -> Dict[str, Any]: + return json.loads(build_module(get_iac_plugin("gcp"), contract)).get("resource") or {} + + +def _aws_resources(contract: Dict[str, Any]) -> Dict[str, Any]: + return get_iac_plugin("aws").emit(contract) + + +def _refusal(cloud: str, contract: Dict[str, Any]) -> UnsupportedBindingError: + with pytest.raises(UnsupportedBindingError) as raised: + if cloud == "gcp": + build_module(get_iac_plugin("gcp"), contract) + else: + get_iac_plugin("aws").emit(contract) + return raised.value + + +# ── the shared derivations ─────────────────────────────────────────────── + + +class TestTheSharedDerivations: + def test_a_mask_needs_a_rule_and_only_a_mask_takes_one(self): + for bad, fragment in [ + ({"principal": ANALYSTS, "columns": ["msisdn"], "access": "mask"}, "needs a mask rule"), + ({**MASK_MSISDN, "mask": "rot13"}, "needs a mask rule"), + ({**DENY_ID, "mask": "last_four"}, "masks nothing"), + ]: + with pytest.raises(UnsupportedBindingError) as raised: + restrictions_for(_expose({}, [bad])) + assert fragment in str(raised.value) + + def test_a_column_has_one_masking_rule(self): + two = [MASK_MSISDN, {**MASK_MSISDN, "principal": STEWARDS, "mask": "first_four"}] + with pytest.raises(UnsupportedBindingError) as raised: + restrictions_for(_expose({}, two)) + assert "one masking rule" in str(raised.value) + + def test_a_deny_beats_a_mask_and_a_mask_never_grants(self): + expose = _expose({}, [MASK_MSISDN, {**DENY_ID, "columns": ["msisdn"]}]) + restrictions = restrictions_for(expose) + masks = column_masks( + expose, restrictions, lambda p: (p,), frozenset({ANALYSTS, STEWARDS}), where="t" + ) + assert masks == {} + masks = column_masks( + _expose({}, [MASK_MSISDN]), + restrictions_for(_expose({}, [MASK_MSISDN])), + lambda p: (p,), + frozenset({STEWARDS}), + where="t", + ) + assert masks == {}, "a masked principal that is not a reader reads nothing" + + @pytest.mark.parametrize( + "where, fragment", + [ + ("consent = true; DROP TABLE x", "statement separator"), + ("consent = true -- x", "statement separator"), + ("nope = 1", "does not declare"), + ("consent IN (SELECT 1)", "one boolean expression"), + ("consent =", "does not parse"), + ], + ) + def test_a_row_filter_is_one_boolean_expression_over_the_schema(self, where, fragment): + with pytest.raises(UnsupportedBindingError) as raised: + row_filters_for(_expose({}, row_filters=[{"principal": ANALYSTS, "where": where}])) + assert fragment in str(raised.value) + + def test_a_principal_has_one_row_filter(self): + with pytest.raises(UnsupportedBindingError) as raised: + row_filters_for(_expose({}, row_filters=[CONSENT, {**CONSENT, "name": "other"}])) + assert "a second time" in str(raised.value) + + def test_governance_labels_are_the_contracts_in_the_gcp_label_alphabet(self): + contract = _gcp() + labels = governance_labels(contract, contract["exposes"][0]) + assert labels["gdpr-article"] == "art6-1b-contractual" + assert labels["cost-center"] == "nw-cx-001" + assert labels["fluid_classification"] == "confidential" + assert labels["fluid_jurisdiction"] == "eu" + assert labels["fluid_regulation"] == "gdpr" + assert labels["fluid_residency"] == "true" + assert labels["layer"] == "silver" + + +# ── GCP ────────────────────────────────────────────────────────────────── + + +class TestGcp: + def test_a_masked_column_gets_its_own_tag_a_data_policy_and_masked_readers(self): + res = _gcp_resources(_gcp([DENY_ID, MASK_MSISDN])) + tags = res["google_data_catalog_policy_tag"] + assert len(tags) == 2, "the masked column never shares a tag with a denied one" + (policy,) = res["google_bigquery_datapolicy_data_policy"].values() + assert policy["data_policy_type"] == "DATA_MASKING_POLICY" + assert policy["data_masking_policy"] == {"predefined_expression": "LAST_FOUR_CHARACTERS"} + assert policy["location"] == "europe-west3" + masked_tag = next(k for k, t in tags.items() if t["display_name"].endswith("msisdn")) + assert masked_tag in policy["policy_tag"] + members = { + (m["role"], m["member"]) + for m in res["google_bigquery_datapolicy_data_policy_iam_member"].values() + } + assert members == {("roles/bigquerydatapolicy.maskedReader", GCP_MAP[ANALYSTS])} + raw = { + m["member"] + for m in res["google_data_catalog_policy_tag_iam_member"].values() + if masked_tag in m["policy_tag"] + } + assert GCP_MAP[ANALYSTS] not in raw, "a masked principal never reads the raw column" + assert GCP_MAP[STEWARDS] in raw + + def test_a_data_policy_id_stays_under_bigquery_s_limit(self): + """BigQuery refuses a data policy id of 200 characters or more (measured: 300 refused).""" + import dataclasses + + from fluid_build.iac.providers import gcp_governance as gov + + short = gov.TagGroup( + key="silver_customer_customer_profile_360_tga_silver_customer_profile_360_x", + display_name="msisdn", + columns=("msisdn",), + readers=(), + mask="last_four", + ) + assert short.data_policy_id == f"{short.key}_mask", "a short id is unchanged" + long_a = dataclasses.replace(short, key="p" * 250 + "_a") + long_b = dataclasses.replace(short, key="p" * 250 + "_b") + assert len(long_a.data_policy_id) <= gov.DATA_POLICY_ID_MAX + assert long_a.data_policy_id != long_b.data_policy_id, "two long ids stay distinct" + assert long_a.data_policy_id.replace("_", "").isalnum() + + def test_the_hashed_column_stays_readable(self): + res = _gcp_resources(_gcp([DENY_ID, MASK_MSISDN])) + table = next(iter(res["google_bigquery_table"].values())) + schema = json.loads(table["schema"]) + tagged = {f["name"] for f in schema if f.get("policyTags")} + assert tagged == {"customer_id", "msisdn"} + + def test_each_row_filter_is_a_row_access_policy_and_everyone_else_reads_every_row(self): + res = _gcp_resources(_gcp([DENY_ID], [CONSENT])) + policies = {p["policy_id"]: p for p in res["google_bigquery_row_access_policy"].values()} + assert set(policies) == {"analysts_consented", "fluid_all_rows"} + assert policies["analysts_consented"]["filter_predicate"] == CONSENT["where"] + assert policies["analysts_consented"]["grantees"] == [GCP_MAP[ANALYSTS]] + assert policies["fluid_all_rows"]["filter_predicate"] == "TRUE" + assert set(policies["fluid_all_rows"]["grantees"]) == {GCP_MAP[STEWARDS], GCP_MAP[PIPELINE]} + + def test_a_write_only_filtered_principal_reads_only_its_rows(self): + """A writer reads rows too (roles/bigquery.dataEditor): left out of its own filter it + landed in fluid_all_rows, and BigQuery would have shown it every row.""" + contract = _gcp(row_filters=[{"principal": PIPELINE, "where": "consent = true"}]) + contract["accessPolicy"]["grants"][2]["permissions"] = ["write", "insert"] + res = _gcp_resources(contract) + policies = {p["policy_id"]: p for p in res["google_bigquery_row_access_policy"].values()} + assert policies["fluid_pipeline_rows"]["grantees"] == [GCP_MAP[PIPELINE]] + assert GCP_MAP[PIPELINE] not in policies["fluid_all_rows"]["grantees"] + + def test_a_filtered_principal_with_no_grant_is_still_filtered(self): + """Granted elsewhere (a project role), it reads only the rows its filter selects.""" + contract = _gcp(row_filters=[CONSENT]) + grants = contract["accessPolicy"]["grants"] + contract["accessPolicy"]["grants"] = [g for g in grants if g["principal"] != ANALYSTS] + res = _gcp_resources(contract) + policies = {p["policy_id"]: p for p in res["google_bigquery_row_access_policy"].values()} + assert policies["analysts_consented"]["grantees"] == [GCP_MAP[ANALYSTS]] + assert GCP_MAP[ANALYSTS] not in policies["fluid_all_rows"]["grantees"] + + def test_a_row_filter_on_a_view_is_refused_not_dropped(self): + contract = _gcp(row_filters=[CONSENT]) + contract["exposes"][0]["binding"]["format"] = "bigquery_view" + with pytest.raises(UnsupportedBindingError) as raised: + _gcp_resources(contract) + assert raised.value.kind == "row-filter-view" + + def test_a_row_filter_on_a_gcs_expose_is_refused_not_dropped(self): + contract = _gcp(row_filters=[CONSENT]) + binding = contract["exposes"][0]["binding"] + binding.update(format="gcs_bucket", location={"bucket": "lake"}) + with pytest.raises(UnsupportedBindingError) as raised: + _gcp_resources(contract) + assert raised.value.kind == "gcp-governance-unsupported-target" + assert "policy.authz.rowFilters" in str(raised.value) + + def test_the_contracts_governance_labels_reach_dataset_table_and_key(self): + res = _gcp_resources(_gcp()) + for kind in ("google_bigquery_dataset", "google_bigquery_table"): + labels = next(iter(res[kind].values()))["labels"] + assert labels["fluid_regulation"] == "gdpr" + assert labels["managed_by"] == "fluid", "fluid's own labels win" + + def test_no_mask_no_filter_emits_none_of_the_new_resources(self): + res = _gcp_resources(_gcp([DENY_ID])) + assert "google_bigquery_datapolicy_data_policy" not in res + assert "google_bigquery_row_access_policy" not in res + + +# ── AWS ────────────────────────────────────────────────────────────────── + + +def _grant(res: Dict[str, Any], principal: str) -> Dict[str, Any]: + (grant,) = [ + g + for g in res["aws_lakeformation_permissions"].values() + if g["principal"] == principal + and ("table_with_columns" in g or "data_cells_filter" in g or "table" in g) + and not str(g.get("table", [{}])[0].get("name", "")).count("masked") + ] + return grant + + +class TestAws: + def test_a_filtered_principal_reads_through_a_data_cells_filter(self): + res = _aws_resources(_aws([DENY_ID], [CONSENT])) + (flt,) = res["aws_lakeformation_data_cells_filter"].values() + data = flt["table_data"][0] + assert data["name"] == "analysts_consented" + assert data["row_filter"] == [{"filter_expression": CONSENT["where"]}] + assert data["column_wildcard"] == [{"excluded_column_names": ["customer_id"]}] + grant = _grant(res, ANALYST_ARN) + assert grant["permissions"] == ["SELECT"] + assert grant["data_cells_filter"][0]["name"] == "analysts_consented" + assert "table_with_columns" not in grant and "table" not in grant + assert "table" in _grant(res, STEWARD_ARN), "an unfiltered reader keeps its table grant" + + def test_a_masked_principal_reads_a_protected_view_and_never_the_raw_column(self): + res = _aws_resources(_aws([DENY_ID, MASK_MSISDN], [CONSENT])) + views = [ + t + for t in res["aws_glue_catalog_table"].values() + if t.get("table_type") == "VIRTUAL_VIEW" + ] + assert len(views) == 1 + view = views[0] + assert view["name"] == "profile_masked" + vd = view["view_definition"][0] + assert vd["definer"] == DEFINER_ARN and vd["is_protected"] is True + rep = vd["representations"][0] + assert rep["dialect"] == "ATHENA" and rep["validation_connection"] == "views-conn" + sql = rep["view_original_text"] + assert 'concat(\'XXXXX\', substr("msisdn", -4)) END AS "msisdn"' in sql + assert '"customer_id"' not in sql, "a denied column is not in the view" + assert '"msisdn_hash"' in sql, "the hashed column is" + assert sql.endswith( + f"WHERE {CONSENT['where']}" + ), "the view applies the principal's row filter" + grants = res["aws_lakeformation_permissions"].values() + definer = [g for g in grants if g["principal"] == DEFINER_ARN] + assert any( + g.get("permissions_with_grant_option") == ["SELECT"] and "table" in g for g in definer + ) + # Lake Formation's view prerequisites: CREATE_TABLE on the view's database. + assert any( + g["permissions"] == ["CREATE_TABLE", "DESCRIBE"] and "database" in g for g in definer + ) + view_grant = [ + g for g in grants if g["principal"] == ANALYST_ARN and "masked" in str(g.get("table")) + ] + assert view_grant and view_grant[0]["permissions"] == ["SELECT", "DESCRIBE"] + flt = next(iter(res["aws_lakeformation_data_cells_filter"].values()))["table_data"][0] + assert "msisdn" in flt["column_wildcard"][0]["excluded_column_names"] + + def test_a_masked_view_selects_only_the_columns_its_readers_grant_lists(self): + """The view reads the table as its definer: a reader whose grant lists its columns + read every other column raw through it, a denied one included.""" + deny_score = {"principal": ANALYSTS, "columns": ["score"], "access": "deny"} + contract = _aws([MASK_MSISDN, deny_score]) + grant = contract["exposes"][0]["binding"]["governance"]["lakeFormation"]["grants"][1] + grant.update(permissions=["SELECT"], columns=["consent", "suppressed"]) + res = _aws_resources(contract) + (view,) = [ + t + for t in res["aws_glue_catalog_table"].values() + if t.get("table_type") == "VIRTUAL_VIEW" + ] + sql = view["view_definition"][0]["representations"][0]["view_original_text"] + assert '"consent"' in sql and '"suppressed"' in sql + assert "concat('XXXXX', substr(\"msisdn\", -4))" in sql + for raw in ("customer_id", "msisdn_hash", "score"): + assert f'"{raw}"' not in sql, raw + + def test_a_masked_view_never_selects_a_column_the_grant_excludes(self): + res = _aws_resources(_aws([MASK_MSISDN, DENY_ID])) + (view,) = [ + t + for t in res["aws_glue_catalog_table"].values() + if t.get("table_type") == "VIRTUAL_VIEW" + ] + sql = view["view_definition"][0]["representations"][0]["view_original_text"] + assert '"customer_id"' not in sql and '"msisdn_hash"' in sql + + def test_a_mask_without_a_view_definer_is_refused(self): + error = _refusal("aws", _aws([MASK_MSISDN], masked_views=False)) + assert error.kind == "column-mask-unenforceable" + + def test_a_text_mask_on_a_non_string_column_is_refused(self): + bad = {"principal": ANALYSTS, "columns": ["score"], "access": "mask", "mask": "last_four"} + assert _refusal("aws", _aws([bad])).kind == "column-mask-type" + + def test_the_contracts_governance_labels_reach_bucket_and_table(self): + res = _aws_resources(_aws()) + bucket = next(iter(res["aws_s3_bucket"].values())) + assert bucket["tags"]["fluid_regulation"] == "gdpr" + table = next( + t + for t in res["aws_glue_catalog_table"].values() + if t.get("table_type") != "VIRTUAL_VIEW" + ) + assert table["parameters"]["gdpr-article"] == "art6-1b-contractual" + assert table["parameters"]["classification"] == "parquet", "the format parameter is kept" + + +# ── the schema ─────────────────────────────────────────────────────────── + + +class TestSchema: + def _validate(self, contract: Dict[str, Any]) -> List[str]: + import jsonschema + + import fluid_build + + path = Path(fluid_build.__file__).parent / "schemas" / "fluid-schema-0.7.6.json" + schema = json.loads(path.read_text(encoding="utf-8")) + validator = jsonschema.validators.validator_for(schema)(schema) + return [e.message for e in validator.iter_errors(contract)] + + def test_mask_row_filters_and_masked_views_are_valid_076(self): + errors = self._validate(_aws([DENY_ID, MASK_MSISDN], [CONSENT])) + assert not [e for e in errors if "mask" in e or "rowFilters" in e or "maskedViews" in e] + + def test_a_mask_without_its_rule_is_refused_by_the_schema(self): + bad = {"principal": ANALYSTS, "columns": ["msisdn"], "access": "mask"} + errors = self._validate(_aws([bad])) + assert any("mask" in e for e in errors) + + +# ── fluid verify: a grant on a data cells filter ───────────────────────── + + +class _FakeLf: + """Lake Formation as ListPermissions and GetDataCellsFilter answer it.""" + + def __init__(self, excluded: Optional[List[str]], readable: bool = True): + self.excluded, self.readable = excluded, readable + + def list_permissions(self, **kwargs): + table = {"Table": {"DatabaseName": "silver", "Name": "profile"}} + flt = { + "DataCellsFilter": { + "TableCatalogId": ACCOUNT, + "DatabaseName": "silver", + "TableName": "profile", + "Name": "analysts_consented", + } + } + steward = { + "Principal": {"DataLakePrincipalIdentifier": STEWARD_ARN}, + "Resource": table, + "Permissions": ["SELECT"], + } + analyst = { + "Principal": {"DataLakePrincipalIdentifier": ANALYST_ARN}, + "Resource": flt, + "Permissions": ["SELECT"], + } + if kwargs.get("ResourceType") == "TABLE": + return {"PrincipalResourcePermissions": [steward]} + return {"PrincipalResourcePermissions": [steward, analyst]} + + def get_data_cells_filter(self, **kwargs): + if not self.readable: + raise RuntimeError("AccessDenied") + return {"DataCellsFilter": {"ColumnWildcard": {"ExcludedColumnNames": self.excluded}}} + + +class TestVerifyDataCellsFilters: + def _check(self, lf: _FakeLf) -> Dict[str, Any]: + from fluid_build.cli._verify_lf_columns import column_restrictions_dimension + + contract = _aws([DENY_ID, MASK_MSISDN], [CONSENT]) + expose = contract["exposes"][0] + return column_restrictions_dimension( + "profile", + expose, + expose["binding"], + region="eu-north-1", + factory=lambda svc, region: lf, + ) + + def test_a_filter_that_excludes_the_restricted_columns_passes(self): + result = self._check(_FakeLf(["customer_id", "msisdn"])) + assert result["status"] == "pass", result + + def test_a_filter_that_leaves_a_masked_column_readable_fails(self): + result = self._check(_FakeLf(["customer_id"])) + assert result["status"] == "fail" and "msisdn" in result["message"] + + def test_a_filter_it_cannot_read_fails_closed(self): + result = self._check(_FakeLf(["customer_id", "msisdn"], readable=False)) + assert result["status"] == "fail" diff --git a/tests/iac/test_state_migration_safety.py b/tests/iac/test_state_migration_safety.py index 3cb907fd..9f196a28 100644 --- a/tests/iac/test_state_migration_safety.py +++ b/tests/iac/test_state_migration_safety.py @@ -260,10 +260,10 @@ def test_a_probe_that_cannot_initialise_at_all_is_an_error_not_an_empty_state(fa def test_the_move_installs_what_the_old_state_names_at_the_plugin_s_pins(fake, tmp_path): """Once per contract, the copy's init installs providers: the pinned - ``~> 5.0`` aws, not the latest, and only what the state names.""" + ``~> 6.0`` aws, not the latest, and only what the state names.""" tofu = fake([_EMPTY, _EMPTY, _state("55555555-eeee", *_OLD["resources"])]) assert _reconcile(tmp_path).outcome == mig.MIGRATED - want = {"aws": {"source": "hashicorp/aws", "version": "~> 5.0"}} + want = {"aws": {"source": "hashicorp/aws", "version": "~> 6.0"}} assert tofu.modules["move"]["terraform"]["required_providers"] == want assert tofu.modules["move"]["terraform"]["backend"] == _LEGACY assert tofu.modules["move-copy"]["terraform"]["required_providers"] == want