From b6575e1853a893cf69850a2cf6ef4e07df7855f0 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 6 Oct 2026 14:43:19 +0200 Subject: [PATCH] fix(aws): refuse a row filter on a binding that names no Glue-catalog table What: lf_row_filters now refuses a filter (row-filter-unenforceable) on an aws binding whose format is not a Glue-catalog format (redshift_serverless, redshift_table, redshift_external_schema, kinesis_stream, glue_table, ...), or that has no location.database or location.table: the predicate lf_column_exclusions already refuses column restrictions with. The message names the formats that can carry a filter. A new lf_governance runs lf_column_exclusions, lf_row_filters and lf_masked_views in the emitter's order. AwsIacPlugin.emit and validate_governance both call it, so fluid validate reports every refusal those derivations raise, with the emitter's own message. Why: on AWS a row filter is a Lake Formation data cells filter on a Glue table, and _emit_lakeformation returns early for any other format and for a binding with no database. lf_row_filters checked only for grants and a table. So a filter on a Redshift or Kinesis binding passed fluid validate (exit 0), and fluid generate iac and fluid apply wrote the Redshift namespace and workgroup with no filter (exit 0). fluid validate also ran only lf_column_exclusions for an aws binding, so the no-grant, no-read-grant, unmapped-principal, no-maskedViews and text-mask refusals surfaced first at generate iac or apply. A mask or a deny on a non-Glue binding was already refused by lf_column_exclusions (column-restriction-unenforceable); that kind is kept, and the mask case is now pinned. lf_masked_views quotes names through validate_ident, which raises a plain ValueError. Run at validate, that would have escaped validate_governance, and cli/validate.py skips the whole governance check on any exception: a contract with a Redshift row filter and a masked expose on a "silver-zone" database printed "Valid FLUID contract". The aws branch now goes through _aws_refusal, which reports that ValueError as the expose's error. docs/governance-parity.md and the governance_validation docstring said these refusals happen at fluid plan, which never runs the emitter; corrected to fluid validate, fluid generate iac, fluid apply and fluid diff's state check. Prior art: the refusal is the in-repo triple of lf_column_exclusions and gcp_governance.refuse_unsupported_target (UnsupportedBindingError with a stable kind, a message ending "It would not be enforced.", one remediation sentence); no new kind is minted. Reporting at validate as well as at emit follows the Terraform plugin framework's ValidateConfig (diagnostics at validate, plan and apply, before any change) and Kubernetes KEP-2885 (silently dropped fields became strict validation errors). dbt's warn-and-drop for unsupported constraints was considered and rejected for an access-control policy. Validated: tests/iac/test_iac_aws_nonglue_governance.py, 55 tests: the research's 22 failed 11 on main; all 55 green with the fix. Seeded defects: disabling the guard turns 20 red, validate running only lf_column_exclusions turns 18 red, dropping the ValueError clause turns 1 red; each restored byte-identical, 55 green. tests/iac plus the startup budget and validate suites: 1661 passed, 243 skipped (tofu hidden). The tofu-gated tests/iac files with tofu on PATH, against moto and the in-process BigQuery fake only: 255 passed, 5 skipped; the three that reach validate_governance rerun on the final code: 175 passed. Full suite, tofu hidden: 20097 passed; the 6 failures are the dbt tests that need the dbt-duckdb adapter, absent from the local venv, and fail identically on main. Real CLI: fluid validate and fluid generate iac exit 1 on the Redshift repro (no main.tf.json written); the parquet Glue control validates and emits its data cells filter. fluid validate over the 53 example and template contracts: same exit codes and governance lines as main; validate_governance over 76 YAML contracts in the repo: no change. ruff, black 24.10.0 --check, the product guardrail and detect-secrets pass. --- CHANGELOG.md | 26 + docs/governance-parity.md | 20 +- fluid_build/iac/governance_validation.py | 43 +- fluid_build/iac/providers/aws.py | 49 +- tests/iac/test_iac_aws_nonglue_governance.py | 507 +++++++++++++++++++ 5 files changed, 615 insertions(+), 30 deletions(-) create mode 100644 tests/iac/test_iac_aws_nonglue_governance.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7425c398..66f09741 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 now builds the tag's commit, also when a manual run starts from `main`, and the quality gate checks `ci.yml` on that commit. +### Fixed + +- **A row filter on an AWS binding that is not a Glue-catalog table is refused, not + dropped.** On AWS a row filter is a Lake Formation data cells filter on a Glue table, + and the emitter writes no Lake Formation resource for any other format. So + `rowFilters` on a `redshift_serverless`, `redshift_table`, `redshift_external_schema` + or Kinesis binding, or on a Glue format with no `location.database`, passed + `fluid validate`, and `fluid generate iac` and `fluid apply` wrote the Redshift + namespace and workgroup with no filter and reported success. It is now refused + (`row-filter-unenforceable`), and the message names the formats that can carry a + filter. A mask or a deny on such a binding was already refused + (`column-restriction-unenforceable`), and still is. +- **`fluid validate` reports every row-filter and mask refusal the AWS emitter makes.** + For an aws binding it ran only the column-restriction check, so a row filter with no + Lake Formation read grant or with an unmapped principal, a mask with no + `maskedViews`, a text mask on a column that is not a string, and a masked view over + a name its SQL cannot quote all passed stage 2 and failed first at + `fluid generate iac` or `fluid apply`. Both now run the same derivation + (`lf_governance`) and refuse with the same message. + +### Documentation + +- `docs/governance-parity.md` said these refusals happen at `fluid plan`. `fluid plan` + never runs the emitter: they happen at `fluid validate`, `fluid generate iac`, + `fluid apply` and the state check of `fluid diff`. + ## [0.19.0] — 2026-10-06 A contract can now say that a principal reads a column masked rather than not at all, diff --git a/docs/governance-parity.md b/docs/governance-parity.md index 375efcd3..267e674b 100644 --- a/docs/governance-parity.md +++ b/docs/governance-parity.md @@ -17,17 +17,21 @@ contract needs `fluidVersion: "0.7.6"` to validate with them. | **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. | +| **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. Only on a Glue-catalog table (format `iceberg`, `parquet`, `csv`, `json`, `avro`, `orc` or `delta`, with `location.database` and `location.table`): on a Redshift or Kinesis binding the filter is refused. | 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, 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 +A policy that a binding cannot apply is refused, never dropped: at `fluid validate` +(stage 2), and again by the emitter at `fluid generate iac`, `fluid apply` and the +state check of `fluid diff`, before any module is written. `fluid plan` does not run +the emitter, so it does not refuse them. Refused: 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 (a deny or a mask) on an AWS binding with no Lake +Formation grants or on a non-Glue format (`column-restriction-unenforceable`), 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 on an AWS binding that names no Glue-catalog +table, such as a Redshift or Kinesis format, or one 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. diff --git a/fluid_build/iac/governance_validation.py b/fluid_build/iac/governance_validation.py index 4a68db08..52542f89 100644 --- a/fluid_build/iac/governance_validation.py +++ b/fluid_build/iac/governance_validation.py @@ -14,13 +14,15 @@ """Validate-time gate for the governance a cloud binding cannot apply. -``fluid generate iac``, ``fluid plan`` and ``fluid apply`` refuse a principal that -is unmapped or a placeholder, a column restriction nothing on the binding enforces, -a Cloud KMS key on AWS or an AWS key on GCP, and retention or encryption on a GCP -target that is not a BigQuery table. This runs the SAME derivations -(``iac/providers/gcp_governance.py``, ``iac/column_access.py``, -``iac/principals.py``) so ``fluid validate`` reports each refusal at stage 2, with -the same message, instead of at apply. The same shape as ``validate_gcp_binding``. +``fluid generate iac`` and ``fluid apply`` (and the state check of ``fluid diff``, +which emits the same module) refuse a principal that is unmapped or a placeholder, a +column restriction, mask or row filter nothing on the binding enforces, a Cloud KMS +key on AWS or an AWS key on GCP, and retention or encryption on a GCP target that is +not a BigQuery table; ``fluid plan`` does not run the emitter. This runs the SAME +derivations (``iac/providers/gcp_governance.py``, ``lf_governance`` in +``iac/providers/aws.py``, ``iac/column_access.py``, ``iac/principals.py``) so +``fluid validate`` reports each refusal at stage 2, with the same message, instead of +at apply. The same shape as ``validate_gcp_binding``. """ from __future__ import annotations @@ -45,6 +47,25 @@ def _lf_grants(binding: Mapping[str, Any]) -> List[Any]: return list(grants) if isinstance(grants, list) else [] +def _aws_refusal(exposure: Mapping[str, Any], binding: Mapping[str, Any], index: int) -> str: + """What the AWS emitter's ``lf_governance`` refuses this expose with, or ``""``. + + A name the masked view's SQL cannot quote raises a plain ``ValueError`` + (``validate_ident``), on which the emitter stops too. It is reported here as the + expose's error: escaping, it would end :func:`validate_governance`, and + ``fluid validate`` would report no governance finding at all. + """ + from .providers.aws import lf_governance + + try: + lf_governance(exposure, binding, index) + except UnsupportedBindingError as exc: + return _message(exc) + except ValueError as exc: + return f"exposes[{exposure.get('exposeId') or index}]: {exc}" + return "" + + def validate_governance(contract: Mapping[str, Any]) -> Tuple[List[str], List[str]]: """``(errors, warnings)`` for the contract's cloud bindings. @@ -73,10 +94,10 @@ def validate_governance(contract: Mapping[str, Any]) -> Tuple[List[str], List[st try: principal_map(binding) if is_cloud(binding, "aws"): - from .providers.aws import lf_column_exclusions - - lf_column_exclusions(exposure, binding, index) - if not _lf_grants(binding): + refusal = _aws_refusal(exposure, binding, index) + if refusal: + errors.append(refusal) + elif not _lf_grants(binding): unenforced.append(str(exposure.get("exposeId") or index)) continue if not _gov.gcp_owned(binding): diff --git a/fluid_build/iac/providers/aws.py b/fluid_build/iac/providers/aws.py index df544c91..c6f8ecd5 100644 --- a/fluid_build/iac/providers/aws.py +++ b/fluid_build/iac/providers/aws.py @@ -403,14 +403,12 @@ def emit( fmt = binding.get("format") or "parquet" schema = (exposure.get("contract") or {}).get("schema") or [] # The contract's column restrictions, as each Lake Formation grant's - # 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) + # excluded columns (``iac/column_access.py``), its row filters, each on + # the read grant of its principal, and the protected views its masked + # principals read (Lake Formation cannot mask); each refused when + # nothing on this binding could enforce it. ``fluid validate`` runs the + # same :func:`lf_governance`, so it refuses what this would. + exclusions, row_filters, masked_views = lf_governance(exposure, binding, index) placement = _placement(packaging, exposure) # The contract's governance labels (classification, jurisdiction, # regulation, its own labels) under fluid's own two, which win. @@ -2096,16 +2094,29 @@ def lf_row_filters( 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. + to carry it is refused: Lake Formation would let nobody read less. So is a filter on + a binding that names no Glue-catalog table (a Redshift or Kinesis format, or no + ``location.database``), where :func:`_emit_lakeformation` writes nothing at all. """ filters = row_filters_for(exposure, index) if not filters: return {} where = f"exposes[{exposure.get('exposeId') or index}].policy.authz.rowFilters" + loc = binding.get("location") or {} + fmt = str(binding.get("format") or "parquet") + if fmt.lower() not in _GLUE_CATALOG_FORMATS or not loc.get("database") or not loc.get("table"): + formats = sorted(_GLUE_CATALOG_FORMATS) + raise UnsupportedBindingError( + "row-filter-unenforceable", + f"{where} filters rows, but its {fmt} binding names no Glue-catalog table; the " + "AWS emitter enforces row filters only as Lake Formation data cells filters on " + f"one, a binding of format {', '.join(formats[:-1])} or {formats[-1]} with a " + "location.database and location.table. It would not be enforced.", + ("Move the filter to such an expose, or remove it from this one.",), + ) 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"): + if not grants: raise UnsupportedBindingError( "row-filter-unenforceable", f"{where} filters rows, but this aws binding has no governance.lakeFormation " @@ -2294,6 +2305,22 @@ def resolve(principal: str) -> Tuple[str, ...]: return views +def lf_governance( + exposure: Mapping[str, Any], binding: Mapping[str, Any], index: int = 0 +) -> Tuple[Optional[Dict[int, Tuple[str, ...]]], Dict[int, RowFilter], List[_LfMaskedView]]: + """``(exclusions, row_filters, masked_views)``: one aws expose's Lake Formation policy. + + :func:`lf_column_exclusions`, :func:`lf_row_filters` and :func:`lf_masked_views`, in + that order, as :meth:`AwsIacPlugin.emit` writes them, raising whatever it would. + ``fluid validate`` runs this too (``iac/governance_validation.py``), so each refusal + is reported at stage 2 with the emitter's own message, never first at apply. + """ + exclusions = lf_column_exclusions(exposure, binding, index) + row_filters = lf_row_filters(exposure, binding, index) + masked_views = lf_masked_views(exposure, binding, index, exclusions or {}, row_filters) + return exclusions, row_filters, masked_views + + def _refuse_grants_left_no_column( exposure: Mapping[str, Any], binding: Mapping[str, Any], diff --git a/tests/iac/test_iac_aws_nonglue_governance.py b/tests/iac/test_iac_aws_nonglue_governance.py new file mode 100644 index 00000000..9c191c1e --- /dev/null +++ b/tests/iac/test_iac_aws_nonglue_governance.py @@ -0,0 +1,507 @@ +# 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 row filter or a column mask on a non-Glue AWS binding is refused, never dropped. + +The AWS emitter enforces ``policy.authz.rowFilters`` and ``columnRestrictions`` (a deny +or a mask) only through Lake Formation: a data cells filter, an excluded column or a +protected view, all on a Glue-catalog table. ``_emit_lakeformation`` returns early for +any other format (Redshift, Kinesis, ...) and for a binding with no +``location.database``, so a filter that nothing refused was written nowhere: the +module held the Redshift namespace and workgroup and no filter, and ``fluid apply`` +reported success. + +Column restrictions (deny and mask) were already refused on such a binding +(``column-restriction-unenforceable``); row filters were not. These pin both, on each +command that can surface the refusal: the emitter (``fluid generate iac`` and +``fluid apply``) and ``fluid validate``, which must report it at stage 2 with the same +message. +""" + +from __future__ import annotations + +import argparse +import copy +import logging +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Callable, Dict, List, Optional + +import pytest +import yaml + +from fluid_build.cli import _apply_opentofu_engine, generate_iac +from fluid_build.cli._common import CLIError +from fluid_build.iac import get_iac_plugin +from fluid_build.iac.base import UnsupportedBindingError +from fluid_build.iac.governance_validation import validate_governance +from fluid_build.iac.providers.aws import ( + _GLUE_CATALOG_FORMATS, + lf_column_exclusions, + lf_governance, + lf_masked_views, + lf_row_filters, +) + +ACCOUNT = "123456789012" +ANALYSTS = "group:analysts@northwind.example" +STEWARDS = "group:data-stewards@northwind.example" +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" +PRINCIPALS = {ANALYSTS: ANALYST_ARN, STEWARDS: STEWARD_ARN} + +CONSENT = {"principal": ANALYSTS, "name": "analysts_consented", "where": "consent = true"} +MASK_MSISDN = {"principal": ANALYSTS, "columns": ["msisdn"], "access": "mask", "mask": "last_four"} +DENY_ID = {"principal": ANALYSTS, "columns": ["customer_id"], "access": "deny"} +SCHEMA = [ + {"name": "customer_id", "type": "string"}, + {"name": "msisdn", "type": "string"}, + {"name": "consent", "type": "boolean"}, +] + +# The location of each format as the existing emitter tests write it, with the +# ``database`` and ``table`` a binding needs for the old guard (``not loc.get("table")``) +# to let it through to the silent drop. +GLUE = { + "database": "silver", + "table": "profile", + "bucket": "nw-silver", + "path": "silver/profile/", +} +NON_GLUE_BINDINGS = { + # The reported repro: a Redshift Serverless namespace + workgroup. + "redshift_serverless": { + "namespace": "nw_ns", + "workgroup": "nw_wg", + "database": "dev", + "table": "profile", + }, + "redshift_table": { + "namespace": "nw_ns", + "workgroup": "nw_wg", + "database": "dev", + "table": "profile", + }, + "redshift_external_schema": { + "workgroup": "nw_wg", + "database": "dev", + "table": "profile", + "external_schema": "ext_silver", + "glue_database": "silver", + "iam_role_arn": f"arn:aws:iam::{ACCOUNT}:role/spectrum", + }, + "kinesis_stream": {"stream": "profile", "database": "silver", "table": "profile"}, +} + + +def _contract( + fmt: Optional[str], + location: Dict[str, Any], + *, + restrictions: Optional[List[Dict[str, Any]]] = None, + row_filters: Optional[List[Dict[str, Any]]] = None, + masked_views: bool = True, + grants: bool = True, +) -> 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 + 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: Dict[str, Any] = { + "platform": "aws", + "location": copy.deepcopy(location), + "principals": dict(PRINCIPALS), + } + if fmt is not None: + binding["format"] = fmt + if grants: + binding["governance"] = {"lakeFormation": lake_formation} + return { + "fluidVersion": "0.7.6", + "id": "silver.profile", + "exposes": [ + { + "exposeId": "profile", + "binding": binding, + "contract": {"schema": copy.deepcopy(SCHEMA)}, + "policy": {"authz": authz}, + } + ], + } + + +def _refusal(contract: Dict[str, Any]) -> UnsupportedBindingError: + with pytest.raises(UnsupportedBindingError) as raised: + get_iac_plugin("aws").emit(contract) + return raised.value + + +def _without(location: Dict[str, Any], key: str) -> Dict[str, Any]: + return {k: v for k, v in location.items() if k != key} + + +def _analyst_grant_describe_only(contract: Dict[str, Any]) -> Dict[str, Any]: + grants = contract["exposes"][0]["binding"]["governance"]["lakeFormation"]["grants"] + grants[1]["permissions"] = ["DESCRIBE"] + return contract + + +# ── the emitter: `fluid generate iac`, `fluid apply` ───────────────────── + + +class TestTheEmitterRefusesWhatItWouldDrop: + @pytest.mark.parametrize("fmt", sorted(NON_GLUE_BINDINGS)) + def test_a_row_filter_on_a_non_glue_binding_is_refused_not_dropped(self, fmt): + """The repro. Before the fix this emitted the Redshift pair and no filter.""" + contract = _contract(fmt, NON_GLUE_BINDINGS[fmt], row_filters=[CONSENT]) + with pytest.raises(UnsupportedBindingError) as raised: + resources = get_iac_plugin("aws").emit(contract) + pytest.fail( + f"a rowFilter on a {fmt} binding was dropped: the module holds " + f"{sorted(resources)} and no data cells filter, no refusal" + ) + assert raised.value.kind == "row-filter-unenforceable" + message = str(raised.value) + assert "policy.authz.rowFilters" in message + assert fmt in message, "the refusal names the format that cannot carry the filter" + assert raised.value.remediation, "a refusal says what to do instead" + + def test_the_refusal_lists_the_formats_that_can_carry_a_filter(self): + """A ``glue_table`` author is told which formats work, not to "use a Glue table".""" + error = _refusal( + _contract("glue_table", NON_GLUE_BINDINGS["kinesis_stream"], row_filters=[CONSENT]) + ) + assert error.kind == "row-filter-unenforceable" + message = str(error) + for fmt in _GLUE_CATALOG_FORMATS: + assert fmt in message, f"{fmt} is a format the filter could be moved to" + assert "It would not be enforced." in message + (remedy,) = error.remediation + assert remedy.endswith("."), "one sentence, as the other refusals give" + + @pytest.mark.parametrize("missing", ["database", "table"]) + def test_a_row_filter_on_a_glue_format_with_no_glue_table_is_refused(self, missing): + """``_emit_lakeformation`` returns early with no database too: the same drop.""" + contract = _contract("parquet", _without(GLUE, missing), row_filters=[CONSENT]) + error = _refusal(contract) + assert error.kind == "row-filter-unenforceable" + assert "names no Glue-catalog table" in str(error) + + def test_the_same_filter_on_a_glue_table_is_still_enforced(self): + """The control: the refusal is about the format, not about the filter.""" + res = get_iac_plugin("aws").emit(_contract("parquet", GLUE, row_filters=[CONSENT])) + (flt,) = res["aws_lakeformation_data_cells_filter"].values() + assert flt["table_data"][0]["row_filter"] == [{"filter_expression": "consent = true"}] + + @pytest.mark.parametrize("fmt", sorted(_GLUE_CATALOG_FORMATS) + ["PARQUET", None], ids=str) + def test_every_glue_catalog_format_still_enforces_the_filter(self, fmt): + """No format the emitter writes a filter for is refused: case-blind, parquet default.""" + res = get_iac_plugin("aws").emit(_contract(fmt, GLUE, row_filters=[CONSENT])) + assert len(res.get("aws_lakeformation_data_cells_filter") or {}) == 1, sorted(res) + + @pytest.mark.parametrize("fmt", sorted(NON_GLUE_BINDINGS)) + @pytest.mark.parametrize("restriction", [MASK_MSISDN, DENY_ID], ids=["mask", "deny"]) + def test_a_column_restriction_on_a_non_glue_binding_is_refused_not_dropped( + self, fmt, restriction + ): + """Pinned, not new: a mask is a column restriction, and these are refused already. + + ``lf_column_exclusions`` runs first and refuses any restriction on a non-Glue + binding, so ``column-mask-unenforceable`` cannot fire there and the kind stays + ``column-restriction-unenforceable`` (a stable tag log parsers key on). + """ + contract = _contract(fmt, NON_GLUE_BINDINGS[fmt], restrictions=[restriction]) + error = _refusal(contract) + assert error.kind == "column-restriction-unenforceable" + assert fmt in str(error) + + def test_a_filter_and_a_mask_on_a_non_glue_binding_is_refused(self): + """Column restrictions are derived first, so their refusal is the one raised.""" + contract = _contract( + "redshift_serverless", + NON_GLUE_BINDINGS["redshift_serverless"], + restrictions=[MASK_MSISDN], + row_filters=[CONSENT], + ) + assert _refusal(contract).kind == "column-restriction-unenforceable" + + def test_a_non_glue_binding_with_no_filter_and_no_restriction_still_emits(self): + """Redshift compute with Lake Formation grants and no policy to drop is unchanged.""" + res = get_iac_plugin("aws").emit( + _contract("redshift_serverless", NON_GLUE_BINDINGS["redshift_serverless"]) + ) + assert "aws_redshiftserverless_namespace" in res + assert "aws_redshiftserverless_workgroup" in res + + +class TestOneDerivationForBoth: + def test_lf_governance_is_the_three_derivations_in_the_emitters_order(self): + """What ``fluid validate`` runs is what the emitter writes from.""" + contract = _contract("parquet", GLUE, restrictions=[MASK_MSISDN], row_filters=[CONSENT]) + exposure = contract["exposes"][0] + binding = exposure["binding"] + exclusions = lf_column_exclusions(exposure, binding, 0) + row_filters = lf_row_filters(exposure, binding, 0) + views = lf_masked_views(exposure, binding, 0, exclusions or {}, row_filters) + assert lf_governance(exposure, binding, 0) == (exclusions, row_filters, views) + # The filter reaches the masked principal's protected view as its WHERE. + (view,) = views + assert view.sql.endswith(" WHERE consent = true") + + +# ── the commands: `fluid generate iac` and `fluid apply` exit 1, write nothing ─ + + +class TestTheCommandSurfacesTheRefusal: + def test_generate_iac_exits_one_with_an_unsupported_binding_and_writes_nothing( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setattr(generate_iac, "native_actions", lambda contract, logger, **_: []) + contract = _contract( + "redshift_serverless", + NON_GLUE_BINDINGS["redshift_serverless"], + row_filters=[CONSENT], + ) + path = tmp_path / "contract.fluid.yaml" + path.write_text(yaml.safe_dump(contract), encoding="utf-8") + out = tmp_path / "iac" + args = argparse.Namespace(contract=str(path), provider="auto", out=str(out), env=None) + with pytest.raises(CLIError) as exc: + generate_iac.run(args, logging.getLogger("test")) + assert exc.value.exit_code == 1 + assert exc.value.event == "unsupported_binding" + assert exc.value.context["kind"] == "row-filter-unenforceable" + assert not (out / "main.tf.json").exists(), "nothing is written for a later apply" + + def test_apply_refuses_before_any_module_reaches_tofu(self, monkeypatch: pytest.MonkeyPatch): + """``fluid apply`` (and ``fluid diff``'s state check) emit through ``emit_module``.""" + monkeypatch.setattr( + _apply_opentofu_engine, "native_actions", lambda contract, logger, **_: [] + ) + contract = _contract( + "redshift_serverless", + NON_GLUE_BINDINGS["redshift_serverless"], + row_filters=[CONSENT], + ) + with pytest.raises(CLIError) as exc: + _apply_opentofu_engine.emit_module( + get_iac_plugin("aws"), + contract, + SimpleNamespace(backend=None), + logging.getLogger("test"), + ) + assert exc.value.exit_code == 1 + assert exc.value.event == "unsupported_binding" + assert exc.value.context["kind"] == "row-filter-unenforceable" + + +# ── `fluid validate`: stage 2 reports what the emitter refuses ─────────── + + +#: Every refusal ``lf_governance`` can raise for an aws expose, as a contract. +_REFUSED: Dict[str, Callable[[], Dict[str, Any]]] = { + **{ + f"row-filter-on-{fmt}": ( + lambda fmt=fmt: _contract(fmt, NON_GLUE_BINDINGS[fmt], row_filters=[CONSENT]) + ) + for fmt in NON_GLUE_BINDINGS + }, + **{ + f"mask-on-{fmt}": ( + lambda fmt=fmt: _contract(fmt, NON_GLUE_BINDINGS[fmt], restrictions=[MASK_MSISDN]) + ) + for fmt in NON_GLUE_BINDINGS + }, + "deny-on-kinesis_stream": lambda: _contract( + "kinesis_stream", NON_GLUE_BINDINGS["kinesis_stream"], restrictions=[DENY_ID] + ), + "row-filter-no-database": lambda: _contract( + "parquet", _without(GLUE, "database"), row_filters=[CONSENT] + ), + "row-filter-no-grants": lambda: _contract("parquet", GLUE, row_filters=[CONSENT], grants=False), + "row-filter-no-read-grant": lambda: _analyst_grant_describe_only( + _contract("parquet", GLUE, row_filters=[CONSENT]) + ), + "row-filter-unmapped-principal": lambda: _contract( + "parquet", + GLUE, + row_filters=[{**CONSENT, "principal": "group:auditors@northwind.example"}], + ), + "mask-no-masked-views": lambda: _contract( + "parquet", GLUE, restrictions=[MASK_MSISDN], masked_views=False + ), + "text-mask-on-a-boolean": lambda: _contract( + "parquet", GLUE, restrictions=[{**MASK_MSISDN, "columns": ["consent"]}] + ), +} + + +class TestValidateReportsTheSameRefusal: + """``validate_governance`` ran only ``lf_column_exclusions`` for an aws binding. + + Its docstring says it runs the same derivations as the emitter so each refusal is + reported at stage 2. ``lf_row_filters`` and ``lf_masked_views`` were never run, so + ``fluid validate`` passed every contract those refuse at ``generate iac``. + """ + + @pytest.mark.parametrize("fmt", sorted(NON_GLUE_BINDINGS)) + def test_a_row_filter_on_a_non_glue_binding_fails_validate(self, fmt): + contract = _contract(fmt, NON_GLUE_BINDINGS[fmt], row_filters=[CONSENT]) + errors, _warnings = validate_governance(contract) + assert any("policy.authz.rowFilters" in e for e in errors), errors + + def test_validate_and_the_emitter_refuse_a_mask_with_no_view_definer_alike(self): + """Not specific to non-Glue: the drift this item's fix must not widen.""" + contract = _contract("parquet", GLUE, restrictions=[MASK_MSISDN], masked_views=False) + assert _refusal(contract).kind == "column-mask-unenforceable" + errors, _warnings = validate_governance(contract) + assert any("maskedViews" in e for e in errors), errors + + def test_validate_and_the_emitter_refuse_a_row_filter_with_no_grant_alike(self): + contract = _contract("parquet", GLUE, row_filters=[CONSENT], grants=False) + assert _refusal(contract).kind == "row-filter-unenforceable" + errors, _warnings = validate_governance(contract) + assert any("policy.authz.rowFilters" in e for e in errors), errors + + @pytest.mark.parametrize("case", sorted(_REFUSED)) + def test_validate_reports_every_emitter_refusal_with_its_message(self, case): + """Stage 2 and the emitter cannot disagree: the same refusal, word for word.""" + contract = _REFUSED[case]() + error = _refusal(contract) + expected = f"{error} {' '.join(error.remediation)}".strip() + errors, _warnings = validate_governance(contract) + assert errors == [expected], f"{error.kind}: {errors}" + + def test_a_name_the_masked_view_cannot_quote_fails_validate_and_ends_no_check(self): + """``validate_ident`` raises a plain ``ValueError``; it must not end the whole check. + + ``fluid validate`` skips the governance check on any exception it raises, so the + other expose's refusal went unreported and the contract passed stage 2. + """ + masked = _contract( + "parquet", {**GLUE, "database": "silver-zone"}, restrictions=[MASK_MSISDN] + ) + with pytest.raises(ValueError, match="silver-zone"): + get_iac_plugin("aws").emit(masked) + contract = _contract( + "redshift_serverless", NON_GLUE_BINDINGS["redshift_serverless"], row_filters=[CONSENT] + ) + contract["exposes"].append({**masked["exposes"][0], "exposeId": "masked"}) + errors, _warnings = validate_governance(contract) + assert len(errors) == 2, errors + assert "policy.authz.rowFilters" in errors[0] + assert errors[1] == "exposes[masked]: Invalid SQL identifier: 'silver-zone'" + + def test_a_glue_contract_the_emitter_writes_validates_clean(self): + """The counterpart: a filter, a mask and a deny the emitter enforces pass stage 2.""" + contract = _contract( + "parquet", GLUE, restrictions=[MASK_MSISDN, DENY_ID], row_filters=[CONSENT] + ) + resources = get_iac_plugin("aws").emit(contract) + assert resources["aws_lakeformation_data_cells_filter"] + assert validate_governance(contract) == ([], []) + + +# ── the real `fluid validate` command ──────────────────────────────────── + + +_CLI_CONTRACT = """\ +fluidVersion: "0.7.6" +kind: DataProduct +id: silver.profile +name: Profile +description: A profile with a row filter +domain: customer +metadata: + layer: Silver + owner: + team: data-platform + email: dp@northwind.example +exposes: + - exposeId: profile + kind: table + title: Profile + version: "1.0.0" + binding: + platform: aws + format: {fmt} + location: +{location} + principals: + "{analysts}": "{analyst_arn}" + governance: + lakeFormation: + grants: + - principal: "{analyst_arn}" + permissions: [SELECT, DESCRIBE] + contract: + schema: + - {{name: customer_id, type: string}} + - {{name: consent, type: boolean}} + policy: + authz: + rowFilters: + - principal: "{analysts}" + name: analysts_consented + where: "consent = true" +""" + + +class TestTheValidateCommand: + def _validate(self, tmp_path: Path, fmt: str, location: Dict[str, Any]): + path = tmp_path / "contract.fluid.yaml" + path.write_text( + _CLI_CONTRACT.format( + fmt=fmt, + location="\n".join(f" {k}: {v}" for k, v in location.items()), + analysts=ANALYSTS, + analyst_arn=ANALYST_ARN, + ), + encoding="utf-8", + ) + return subprocess.run( + [sys.executable, "-m", "fluid_build.cli", "validate", str(path)], + capture_output=True, + text=True, + cwd=tmp_path, + ) + + def test_the_redshift_repro_fails_fluid_validate(self, tmp_path): + """It printed "Valid FLUID contract" and exited 0, then applied with no filter.""" + result = self._validate( + tmp_path, "redshift_serverless", NON_GLUE_BINDINGS["redshift_serverless"] + ) + assert result.returncode == 1, result.stdout + result.stderr + assert "filters rows" in result.stdout + assert "redshift_serverless" in result.stdout + + def test_the_same_filter_on_a_glue_table_passes_fluid_validate(self, tmp_path): + result = self._validate(tmp_path, "parquet", GLUE) + assert result.returncode == 0, result.stdout + result.stderr