diff --git a/.secrets.baseline b/.secrets.baseline index 5275b5d0..e7822c65 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -395,7 +395,7 @@ "filename": "fluid_build/observability/secret_redactor.py", "hashed_secret": "1348b145fa1a555461c1b790a2f66614781091e9", "is_verified": false, - "line_number": 167 + "line_number": 190 } ], "fluid_build/providers/_sql_safety.py": [ @@ -447,14 +447,14 @@ "filename": "fluid_build/providers/snowflake/util/logging.py", "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8", "is_verified": false, - "line_number": 31 + "line_number": 32 }, { "type": "Private Key", "filename": "fluid_build/providers/snowflake/util/logging.py", "hashed_secret": "1348b145fa1a555461c1b790a2f66614781091e9", "is_verified": false, - "line_number": 163 + "line_number": 154 } ], "fluid_build/pypi/experimental/build-metadata/0.7.1.dev2.json": [ @@ -1304,28 +1304,28 @@ "filename": "tests/providers/test_snowflake_util_logging.py", "hashed_secret": "132d7d659faa6773d3bc2c3b0da698e60fc1a32e", "is_verified": false, - "line_number": 50 + "line_number": 52 }, { "type": "JSON Web Token", "filename": "tests/providers/test_snowflake_util_logging.py", "hashed_secret": "f1fb6becdc61282ab53f4f18b7183bb6d9fbc3c5", "is_verified": false, - "line_number": 52 + "line_number": 54 }, { "type": "GitHub Token", "filename": "tests/providers/test_snowflake_util_logging.py", "hashed_secret": "e175c6f5f2a92e8623bd9a4820edb4e8c1b0fd10", "is_verified": false, - "line_number": 54 + "line_number": 56 }, { "type": "Secret Keyword", "filename": "tests/providers/test_snowflake_util_logging.py", "hashed_secret": "f3bbbd66a63d4bf1747940578ec3d0103530e21d", "is_verified": false, - "line_number": 66 + "line_number": 68 } ], "tests/scripts/test_check_pinned_actions.py": [ @@ -2147,112 +2147,112 @@ "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "675ebcc24442cf0e52f570519d222ee4293e3aba", "is_verified": false, - "line_number": 275 + "line_number": 281 }, { "type": "Secret Keyword", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "9f0eebb9374c9b37602b8a2d8533283ec4d7165b", "is_verified": false, - "line_number": 275 + "line_number": 281 }, { "type": "Base64 High Entropy String", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "a94bff2a72682dda478d2c763bbb719dde8deda4", "is_verified": false, - "line_number": 281 + "line_number": 287 }, { "type": "Base64 High Entropy String", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "43f1674fef522318d813e0d5b061ee50568819b2", "is_verified": false, - "line_number": 296 + "line_number": 302 }, { "type": "Secret Keyword", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "fc64b21ba679848f055d95f3286770cfa248bf96", "is_verified": false, - "line_number": 296 + "line_number": 302 }, { "type": "Base64 High Entropy String", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "fb653f6836c4d3abf00b8dd0d976c416aab7f425", "is_verified": false, - "line_number": 337 + "line_number": 343 }, { "type": "Base64 High Entropy String", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "93e2332e8fa09c3c418d5573a1bd1bd5d8413cba", "is_verified": false, - "line_number": 338 + "line_number": 344 }, { "type": "Secret Keyword", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "6014c06eb0704c94e4bbbd0ec64eb7ea2dda6600", "is_verified": false, - "line_number": 342 + "line_number": 348 }, { "type": "Secret Keyword", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "c106f340ec6c5f66bc159a4d7975f7c26b92f0f1", "is_verified": false, - "line_number": 366 + "line_number": 372 }, { "type": "Private Key", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "be4fc4886bd949b369d5e092eb87494f12e57e5b", "is_verified": false, - "line_number": 383 + "line_number": 389 }, { "type": "Base64 High Entropy String", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "002a4ce517c7c70c1c4633aa496ebdfd1e7c20d0", "is_verified": false, - "line_number": 385 + "line_number": 391 }, { "type": "Private Key", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "f0778f3e140a61d5bbbed5430773e52af2f5fba4", "is_verified": false, - "line_number": 408 + "line_number": 414 }, { "type": "Base64 High Entropy String", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "8b8e2eac91abca31fcf910894b57dd40997b6daf", "is_verified": false, - "line_number": 409 + "line_number": 415 }, { "type": "JSON Web Token", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "24c3be78884b668a97ebb0a0c2c8b16339af32f7", "is_verified": false, - "line_number": 433 + "line_number": 439 }, { "type": "Private Key", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "27c6929aef41ae2bcadac15ca6abcaff72cda9cd", "is_verified": false, - "line_number": 514 + "line_number": 520 }, { "type": "Base64 High Entropy String", "filename": "tests/test_observability_secret_redactor.py", "hashed_secret": "bd8ba9d36f95e4444c27222e664300608b17cb67", "is_verified": false, - "line_number": 515 + "line_number": 521 } ], "tests/test_secrets_branches.py": [ diff --git a/CHANGELOG.md b/CHANGELOG.md index ac26f76e..0404872d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -167,6 +167,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 path exits 1 where it used to write a template instead. A `fluid ai setup` choice and a single keyring key still come first, and `GEMINI_API_KEY` with `GOOGLE_API_KEY` is one provider. Pick one with `--llm-provider` or `FLUID_LLM_PROVIDER`. +- **`fluid plan` refuses a contract its own sovereignty policy blocks, by default.** It checked + the policy only with `--check-sovereignty`, so a contract `fluid validate` refuses + (`enforcementMode: strict`, or a region in `deniedRegions`, which blocks in every mode) still + got a plan. Now plan runs the same policy engine before it plans: a blocking verdict exits 1 + (`ERR_SOVEREIGNTY_VIOLATION`) before any plan is written. The refusal names each expose and + carries the findings as the error's `errors`, the way `local_plan_validation_failed` does, so + an engine-API caller such as Command Center's plan endpoint shows it. The default check builds + no cloud provider. `--check-sovereignty` runs the same check, then asks the provider's own + hook and prints the verdict, also before the plan is written: a hook that reports nothing no + longer passes a contract `fluid validate` refuses. The new `--no-check-sovereignty` writes the + plan anyway, says so on the console and logs a WARNING (event `sovereignty_check_skipped`); it + changes nothing in `fluid apply`, which keeps its own checks (on Databricks and GCP it refuses + such a placement itself). The generated CI pipelines already pass `--check-sovereignty`; a + failing plan stage now writes no plan file (one left at `--out` by an earlier run stays as it + was). +- **Databricks: Unity Catalog tag values are written as the contract states them.** The + governance labels became tags with GCP's label slugs: lowercased, spaces and punctuation + turned into `-`, cut at 63 characters (`business_value = + 'a-single-unrenewed-ground-lease-can-strand-a-multi-tenant-tower'`, `fluid_jurisdiction = + 'us'`). A tag value takes up to 256 characters with case, spaces and punctuation, so the value + is now the contract's text less surrounding whitespace, and `fluid_regulation` joins the + frameworks with `, ` (`GDPR, CCPA`). A longer value is refused as + `databricks-tag-value-too-long` rather than cut. Tag keys are unchanged (each `-` still + becomes `_`), and GCP labels and AWS tags keep the label alphabet. + **On an existing deployment** the next `fluid apply` updates each changed + `databricks_entity_tag_assignment` in place (`~`; `tag_value` has no replace modifier in the + pinned provider, 1.140.0): no table, schema or data is replaced or destroyed, and no + `--allow-data-loss` is needed. Until that apply, `fluid verify` reports the old values as tag + drift. On Free Edition, a product carrying the old values planned `+0 ~17 -0`. - **`fluid apply` on a Databricks contract with `exposes[].semantics` needs the `[databricks]` extra** (databricks-sdk), which creates the metric view, and refuses before `tofu apply` when it is missing, as it does for column masks. A semantics block the metric view cannot hold (a @@ -290,6 +319,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 land such an expose (`payload.landed_by`) and asks no table for it. An expose that no build writes, or that another build may write, still reports its missing source, and the drift verdict and exit code are unchanged. +- **Log redaction no longer masks fully qualified names.** `fluid apply` printed + `databricks.load table=***REDACTED***` for `northwind_towers.bronze_smart_meters.smart_meters` + and the name of the next table, whose last part had 11 characters: the generic JWT rule masked + any three dotted runs of 12 or more characters. Such a run is now masked only when its first + part is a JOSE header (base64url of a JSON object, the check detect-secrets makes), so + `catalog.schema.table`, `project.dataset.table` and `DB.SCHEMA.TABLE` names stay readable in + logs and errors. Tokens are still masked, in both redaction layers (the Snowflake one now uses + the global rule), and more of them than before: a JWE (one with direct encryption has an empty + encrypted key), a JWS with a short or empty payload or signature, a header written with spaces + or newlines, and a token glued to a word (`token_eyJ…`). The JWT rules no longer scan + quadratically on input such as `a-a-a-…` or `eyJ-eyJ-…` (200,000 characters took 30 s or more; + now under a second). `redact_secret_text()` loses its `keep_dotted_names` keyword (0.28.0): + logs and printed errors now apply the same rule. ## [0.29.0] — 2026-11-12 diff --git a/fluid_build/_console.py b/fluid_build/_console.py index 33ff816b..247987d8 100644 --- a/fluid_build/_console.py +++ b/fluid_build/_console.py @@ -87,7 +87,7 @@ def _redact_error_str(value: Any) -> str: """ from fluid_build.observability.secret_redactor import redact_secret_text - return redact_secret_text(_redact_str(str(value)), keep_dotted_names=True) + return redact_secret_text(_redact_str(str(value))) def literal_error_text(text: Any, style: str = "") -> RichText: diff --git a/fluid_build/_error_catalog.py b/fluid_build/_error_catalog.py index deefc5a5..df9e2735 100644 --- a/fluid_build/_error_catalog.py +++ b/fluid_build/_error_catalog.py @@ -550,7 +550,15 @@ def slug_for(event: str) -> str: "Move the offending binding into an allowed region, or widen " "'sovereignty.allowedRegions' in the contract", "Run 'fluid validate ' to see the same findings with full detail", - "Set 'sovereignty.enforcementMode: advisory' to report without blocking", + ( + "Set 'sovereignty.enforcementMode: advisory' to report a region outside " + + "'allowedRegions' without blocking; a region in 'deniedRegions' blocks in " + + "every mode" + ), + ( + "Pass --no-check-sovereignty to write the plan anyway (logged at WARNING; " + + "fluid apply keeps its own checks)" + ), ], "sovereignty", ), diff --git a/fluid_build/agent_skills/skills/troubleshooting-fluid-errors/references/error-codes.md b/fluid_build/agent_skills/skills/troubleshooting-fluid-errors/references/error-codes.md index 219bf921..3f32b7ea 100644 --- a/fluid_build/agent_skills/skills/troubleshooting-fluid-errors/references/error-codes.md +++ b/fluid_build/agent_skills/skills/troubleshooting-fluid-errors/references/error-codes.md @@ -75,7 +75,7 @@ Docs pages are relative to https://agenticstiger.github.io/forge_docs/ | `ERR_POLICY_COMPILE_FAILED` | Check the agent-policy block in the contract; run 'fluid policy check '. | concepts/sovereignty.html | | `ERR_POLICY_COMPILER_CRASHED` | Run 'fluid validate ': policy compile reads accessPolicy and exposes without validating them against the contract schema. If the contract validates, re-run with 'fluid --log-level DEBUG policy compile ' to see the compiler's traceback. | cli/policy-compile.html#errors | | `ERR_POLICY_APPLY_FAILED` | Run 'fluid policy check ' to surface the offending rule before apply. | concepts/sovereignty.html | -| `ERR_SOVEREIGNTY_VIOLATION` | Move the offending binding into an allowed region, or widen 'sovereignty.allowedRegions' in the contract. Run 'fluid validate ' to see the same findings with full detail. Set 'sovereignty.enforcementMode: advisory' to report without blocking. | concepts/sovereignty.html | +| `ERR_SOVEREIGNTY_VIOLATION` | Move the offending binding into an allowed region, or widen 'sovereignty.allowedRegions' in the contract. Run 'fluid validate ' to see the same findings with full detail. Set 'sovereignty.enforcementMode: advisory' to report a region outside 'allowedRegions' without blocking; a region in 'deniedRegions' blocks in every mode. Pass --no-check-sovereignty to write the plan anyway (logged at WARNING; fluid apply keeps its own checks). | concepts/sovereignty.html | | `ERR_PRODUCT_NEW_FAILED` | Check the target directory is writable and the productType is valid (SDP/ADP/CDP). Pick an empty/new target dir and pass --data-product-type SDP\|ADP\|CDP explicitly. | advanced/production-troubleshooting.html | | `ERR_PRODUCT_ADD_FAILED` | Run 'fluid validate' on the contract first; --type must be source\|exposure\|dq. | advanced/production-troubleshooting.html | | `ERR_PRODUCT_ADD_EXPOSE_NOT_FOUND` | The --expose target does not exist in the contract — list exposes with 'fluid status'. Add the expose first, or target an existing exposeId. | advanced/production-troubleshooting.html | diff --git a/fluid_build/build_runners/_databricks_job_run.py b/fluid_build/build_runners/_databricks_job_run.py index 5ccfd49e..49616b99 100644 --- a/fluid_build/build_runners/_databricks_job_run.py +++ b/fluid_build/build_runners/_databricks_job_run.py @@ -668,9 +668,7 @@ def clean(text: Any) -> str: from fluid_build.observability.secret_redactor import redact_secret_text plain = _CONTROL_RE.sub("", _ANSI_RE.sub("", str(text or ""))) - return redact_secret_text( - plain, extra_secrets=credential_values(os.environ), keep_dotted_names=True - ) + return redact_secret_text(plain, extra_secrets=credential_values(os.environ)) def _failure(exc: BaseException) -> str: diff --git a/fluid_build/cli/plan.py b/fluid_build/cli/plan.py index bb47be27..3cc5041a 100644 --- a/fluid_build/cli/plan.py +++ b/fluid_build/cli/plan.py @@ -17,7 +17,7 @@ import re import time from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from fluid_build.cli.console import cprint, cprint_literal, warning @@ -206,13 +206,19 @@ def register(subparsers: argparse._SubParsersAction): action="store_true", help="ask provider to estimate cost of planned actions", ) + # Three states: unset (the default check), --check-sovereignty (the provider hook + # too, and the verdict printed) and --no-check-sovereignty (skipped, logged). p.add_argument( "--check-sovereignty", - action="store_true", + action=argparse.BooleanOptionalAction, + default=None, help=( - "check data sovereignty constraints (provider hook, else the " - "built-in policy engine); exits 1 when the contract's " - "enforcementMode is 'strict' and the check fails" + "the contract's sovereignty policy is checked before the plan is " + "written (the check fluid validate runs), and a blocking verdict " + "(enforcementMode 'strict', or an explicitly denied region) exits 1 " + "and writes no plan. --check-sovereignty also asks the provider's " + "own hook and prints the verdict; --no-check-sovereignty skips the " + "check, logged at WARNING (apply keeps its own checks)" ), ) # ``default=SUPPRESS`` so a global ``fluid --provider X plan ...`` is not @@ -306,6 +312,36 @@ def run(args, logger: logging.Logger) -> int: # and the same ``provider_unknown`` error apply raises. _gate_provider_flag(getattr(args, "provider", None), logger) + # --- Sovereignty gate ------------------------------------------------ + # ``fluid validate`` refuses a contract whose own sovereignty policy blocks + # where it puts data, and ``fluid plan`` wrote a plan for it anyway unless + # ``--check-sovereignty`` was passed. Now the built-in policy engine (the + # checker validate runs) decides by default, before any plan is generated + # or written: a blocking verdict raises and no plan.json is written. It + # builds no provider, so a plain plan stays off the cloud SDKs. + # ``--check-sovereignty`` runs it too and then asks the provider's hook, + # before the write below: a hook that reports nothing must not make the + # explicit check weaker than the default. ``--no-check-sovereignty`` skips + # the check, said on the console and logged at WARNING for audit, as + # ``fluid apply --no-verify-plan-binding`` is. + check_sovereignty = getattr(args, "check_sovereignty", None) + if check_sovereignty is not False: + _gate_sovereignty(contract, args.contract, logger) + else: + warning( + "Sovereignty check skipped (--no-check-sovereignty): the plan is written " + "whatever the contract's sovereignty policy says." + ) + warn( + logger, + "--no-check-sovereignty: the sovereignty check was SKIPPED. The plan " + "is written whatever the contract's sovereignty policy says; fluid " + "apply keeps its own checks. Make sure this is recorded in the change " + "log.", + event="sovereignty_check_skipped", + contract=args.contract, + ) + info( logger, "plan_start", @@ -427,6 +463,24 @@ def run(args, logger: logging.Logger) -> int: context={"bundle": str(bundle_path), "error": str(exc)}, ) + # --check-sovereignty: the provider's hook, else the built-in engine, with + # the verdict printed. Before the write, like the default gate, so a + # blocking verdict leaves no plan behind. The *contract* decides what + # blocks: ``_report_sovereignty`` reports True only when its + # ``sovereignty.enforcementMode`` is strict, a finding is an error (an + # explicitly denied region), or a provider hook reported violations. + hook_provider = None + provider_flag: Optional[str] = None + if check_sovereignty or getattr(args, "estimate_cost", False): + provider_flag, hook_provider = _hook_provider(args, contract, logger) + # NB: deliberately *not* gated on ``hook_provider`` — a provider that + # failed to build must not silently swallow an explicitly requested + # governance check. ``_report_sovereignty`` handles ``None``. + if check_sovereignty and _report_sovereignty( + contract, hook_provider, provider_flag, logger + ): + raise CLIError(1, "sovereignty_violation", {"contract": args.contract}) + # Write plan to file (idempotent - only if changed) write_json_idempotent(args.out, plan) @@ -440,58 +494,32 @@ def run(args, logger: logging.Logger) -> int: else: _display_plan_simple(plan, logger, output_path=args.out) - # --- Advanced hooks: cost estimation & sovereignty checking --- - sovereignty_blocked = False - if getattr(args, "estimate_cost", False) or getattr(args, "check_sovereignty", False): + # --- Advanced hook: cost estimation --- + if getattr(args, "estimate_cost", False) and hook_provider: from fluid_build.cli.hooks import run_estimate_cost - # Build provider for hook invocation - provider_flag = getattr(args, "provider", None) - project_flag = getattr(args, "project", None) - region_flag = getattr(args, "region", None) - if not provider_flag: - provider_flag, loc = resolve_provider_from_contract(contract) - if not project_flag: - project_flag = loc.get("project") - if not region_flag: - region_flag = loc.get("region") - try: - hook_provider = build_provider(provider_flag, project_flag, region_flag, logger) - except Exception: - hook_provider = None - actions_list = plan.get("actions", []) - - if getattr(args, "estimate_cost", False) and hook_provider: - estimate = run_estimate_cost(hook_provider, actions_list, logger) - if estimate is not None: - cprint( - f"\nEstimated cost: ${estimate.monthly:.2f}/month" - f" + ${estimate.one_time:.2f} one-time" - f" ({estimate.currency})" - ) - if estimate.notes: - cprint(f" Note: {estimate.notes}") - plan["cost_estimate"] = estimate.to_dict() - # Re-inject digests — cost_estimate was added AFTER the - # first inject, so planDigest no longer covers the plan - # body. Recompute to keep the stage-7 tamper gate honest. - # ``force=True``: the plan is already bound from the - # injection above; this is a deliberate, in-process - # re-bind over a mutation we just made ourselves, so we - # opt past inject_digests' overwrite guard (J7). - plan = inject_digests(plan, bundle_path=bundle_path, force=True) - write_json_idempotent(args.out, plan) - else: - cprint("\nCost estimation: not supported by this provider") - - # NB: deliberately *not* gated on ``hook_provider`` — a provider that - # failed to build must not silently swallow an explicitly requested - # governance check. ``_report_sovereignty`` handles ``None``. - if getattr(args, "check_sovereignty", False): - sovereignty_blocked = _report_sovereignty( - contract, hook_provider, provider_flag, logger + estimate = run_estimate_cost(hook_provider, actions_list, logger) + if estimate is not None: + cprint( + f"\nEstimated cost: ${estimate.monthly:.2f}/month" + f" + ${estimate.one_time:.2f} one-time" + f" ({estimate.currency})" ) + if estimate.notes: + cprint(f" Note: {estimate.notes}") + plan["cost_estimate"] = estimate.to_dict() + # Re-inject digests — cost_estimate was added AFTER the + # first inject, so planDigest no longer covers the plan + # body. Recompute to keep the stage-7 tamper gate honest. + # ``force=True``: the plan is already bound from the + # injection above; this is a deliberate, in-process + # re-bind over a mutation we just made ourselves, so we + # opt past inject_digests' overwrite guard (J7). + plan = inject_digests(plan, bundle_path=bundle_path, force=True) + write_json_idempotent(args.out, plan) + else: + cprint("\nCost estimation: not supported by this provider") # --- HTML visualization (--html flag, absorbs preview/viz-plan) --- html_path = getattr(args, "html_output", None) @@ -504,15 +532,6 @@ def run(args, logger: logging.Logger) -> int: except Exception: warn(logger, "plan visualizer not available", event="html_render_skipped") - # Fail closed on a blocking sovereignty verdict. The *contract* decides - # what blocks — ``_report_sovereignty`` only reports True when the - # contract's own ``sovereignty.enforcementMode`` is strict (or a - # provider hook reported violations outright). The plan file is still - # written; the non-zero exit is what makes ``--check-sovereignty`` - # usable as a CI gate rather than advisory output. - if sovereignty_blocked: - raise CLIError(1, "sovereignty_violation", {"contract": args.contract}) - info( logger, "plan_success", @@ -601,7 +620,7 @@ def _report_sovereignty( f" — source: {pname} provider hook" ) for v in violations: - cprint(f" - {v}") + cprint_literal(f" - {v}") warn( logger, "sovereignty violations reported by provider hook", @@ -640,46 +659,125 @@ def _report_sovereignty( ) return False - from ..policy.sovereignty import DEFAULT_ENFORCEMENT_MODE - from ..policy.sovereignty import validate_sovereignty as _policy_check - - # Read the default from the engine, never re-hardcode it — the mode shown - # to the operator must be the mode the verdict was computed under. - mode = sovereignty.get("enforcementMode", DEFAULT_ENFORCEMENT_MODE) - is_valid, messages = _policy_check(contract) + blocked, mode, messages = _policy_engine_verdict(contract) source = f"source: built-in policy engine, enforcementMode={mode}; {reason}" if messages: cprint(f"\nSovereignty check: {len(messages)} finding(s) — {source}") for m in messages: - cprint(f" - {m}") + cprint_literal(f" - {m}") else: cprint(f"\nSovereignty check: PASS — {source}") - # Block on the engine's own verdict OR on any error-severity finding. - # ``is_valid`` alone is not enough: it returns True under advisory/audit - # even for findings the engine itself classified as errors (an explicitly - # denied region is always an error, whatever the mode). ``fluid validate`` - # blocks on those — see cli/validate.py, which routes every "❌" message to - # ``add_error`` — so plan must too, or the same contract would pass one - # stage and fail the next. enforcementMode still does real work: under - # advisory it downgrades an allowedRegions mismatch to a warning, and - # warnings do not block here. - has_error_finding = any("❌" in m for m in messages) - if not is_valid or has_error_finding: - cprint(f"\n❌ Sovereignty check FAILED — enforcementMode is '{mode}'.") - warn( - logger, - "sovereignty check failed under strict enforcement", - event="sovereignty_violations", - source="policy_engine", - enforcement_mode=mode, - count=len(messages), - ) + if blocked: + _report_blocked(mode, len(messages), logger) return True return False +def _policy_engine_verdict(contract: Dict[str, Any]) -> Tuple[bool, str, List[str]]: + """The built-in policy engine's verdict: ``(blocks, enforcementMode, findings)``. + + It blocks on the engine's own verdict OR on any error-severity finding. + ``is_valid`` alone is not enough: it returns True under advisory/audit even + for findings the engine itself classified as errors (an explicitly denied + region is always an error, whatever the mode). ``fluid validate`` blocks on + those — see cli/validate.py, which routes every "❌" message to + ``add_error`` — so plan must too, or the same contract would pass one stage + and fail the next. enforcementMode still does real work: under advisory it + downgrades an allowedRegions mismatch to a warning, and warnings do not block. + """ + from ..policy.sovereignty import DEFAULT_ENFORCEMENT_MODE + from ..policy.sovereignty import validate_sovereignty as _policy_check + + sovereignty = contract.get("sovereignty") or {} + # Read the default from the engine, never re-hardcode it — the mode shown + # to the operator must be the mode the verdict was computed under. + mode = sovereignty.get("enforcementMode", DEFAULT_ENFORCEMENT_MODE) + is_valid, messages = _policy_check(contract) + has_error_finding = any("❌" in m for m in messages) + return (not is_valid or has_error_finding), mode, messages + + +def _report_blocked(mode: str, count: int, logger: logging.Logger) -> None: + cprint(f"\n❌ Sovereignty check FAILED — enforcementMode is '{mode}'.") + warn( + logger, + "sovereignty check failed", + event="sovereignty_violations", + source="policy_engine", + enforcement_mode=mode, + count=count, + ) + + +def _gate_sovereignty(contract: Dict[str, Any], where: str, logger: logging.Logger) -> None: + """``fluid plan``'s default sovereignty check: refuse what ``fluid validate`` refuses. + + The built-in policy engine decides (:func:`_policy_engine_verdict`); no provider + is built. Silent when nothing blocks, and when the contract declares no policy, + so a plan prints what it always printed. On a blocking verdict it prints the + findings and raises, before any plan is generated or written. + + Raises: + CLIError: ``sovereignty_violation`` (exit 1). + """ + sovereignty = contract.get("sovereignty") + if not isinstance(sovereignty, dict) or not sovereignty: + return + blocked, mode, messages = _policy_engine_verdict(contract) + if not blocked: + return + cprint( + f"\nSovereignty check: {len(messages)} finding(s) — source: built-in policy " + f"engine, enforcementMode={mode}" + ) + for m in messages: + # Literal: a finding names its expose as ``[id]``, which Rich would take for markup. + cprint_literal(f" - {m}") + _report_blocked(mode, len(messages), logger) + # ``errors`` and ``message``, as ``local_plan_validation_failed`` carries them: a caller + # of the engine API (Command Center's plan endpoint) shows a refusal that has them + # instead of falling back to a plan this gate never saw. + errors = [m.split("\n", 1)[0].split(" ", 1)[-1].strip() for m in messages if "❌" in m] + raise CLIError( + 1, + "sovereignty_violation", + { + "command": "plan", + "contract": where, + "errors": errors or [m.split("\n", 1)[0] for m in messages], + "message": ( + f"The contract's sovereignty policy (enforcementMode {mode}) blocks where it " + "puts data, so fluid plan wrote no plan." + ), + }, + ) + + +def _hook_provider( + args: argparse.Namespace, contract: Dict[str, Any], logger: logging.Logger +) -> Tuple[Optional[str], Any]: + """The provider the cost and sovereignty hooks ask, with its name. + + The provider is ``None`` when it cannot be built; the name is what the + contract or ``--provider`` says, for the report. + """ + provider_flag = getattr(args, "provider", None) + project_flag = getattr(args, "project", None) + region_flag = getattr(args, "region", None) + if not provider_flag: + provider_flag, loc = resolve_provider_from_contract(contract) + if not project_flag: + project_flag = loc.get("project") + if not region_flag: + region_flag = loc.get("region") + try: + return provider_flag, build_provider(provider_flag, project_flag, region_flag, logger) + except Exception: + return provider_flag, None + + def _gate_provider_flag(requested: Optional[str], logger: logging.Logger) -> None: """Reject an unknown ``--provider`` value before any plan is generated. diff --git a/fluid_build/iac/governance_labels.py b/fluid_build/iac/governance_labels.py index 04c4fd6f..40a6c658 100644 --- a/fluid_build/iac/governance_labels.py +++ b/fluid_build/iac/governance_labels.py @@ -6,9 +6,12 @@ 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. +the Glue table, Unity Catalog tags on the Databricks schema and table. So the +regulation a product falls under is on the resource and can be found there, not only +in the contract. The keys are coerced to GCP's label alphabet (the strictest), so every +cloud carries the same keys, and so are the values unless a caller asks for them +``verbatim_values``: a Unity Catalog tag value keeps case, spaces and punctuation, so +the Databricks module writes the contract's words, not a slug of them. """ from __future__ import annotations @@ -24,8 +27,26 @@ def _label(value: Any) -> str: return _LABEL_KEY_RE.sub("-", str(value).strip().lower())[:63] +def _verbatim(value: Any) -> str: + """A string as written, less the surrounding whitespace a Unity Catalog tag refuses. + + The schema allows only strings; a YAML ``true`` or ``null`` that reaches an emitter + unvalidated (``fluid generate iac`` has no schema gate) is spelled as YAML spells it, + and ``null`` makes no tag, rather than Python's ``True`` and ``None``. + """ + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + return str(value).strip() + + def governance_labels( - contract: Mapping[str, Any], exposure: Mapping[str, Any], *, limit: Optional[int] = 60 + contract: Mapping[str, Any], + exposure: Mapping[str, Any], + *, + limit: Optional[int] = 60, + verbatim_values: bool = False, ) -> Dict[str, str]: """The contract's governance as GCP labels: its labels, classification and regulation. @@ -36,7 +57,13 @@ def governance_labels( prefixed ``l_``; empty values are dropped; at most ``limit`` (60, GCP's cap) are kept, in key order. ``limit=None`` keeps them all, for a caller that must refuse a count its target cannot hold rather than drop labels silently. + + ``verbatim_values`` keeps each value as the contract writes it, less surrounding + whitespace, and joins the regulatory frameworks with ``", "``: for a target whose + tag values take any text (Unity Catalog), which refuses a value too long for it + rather than cut it. The keys are coerced either way. """ + value_of, separator = (_verbatim, ", ") if verbatim_values else (_label, "_") out: Dict[str, str] = {} for source in (contract.get("labels"), exposure.get("labels")): if isinstance(source, Mapping): @@ -44,7 +71,7 @@ def governance_labels( k = _label(key) if k and not k[0].isalpha(): k = f"l_{k}"[:63] - v = _label(value) + v = value_of(value) if k and v: out[k] = v policy = exposure.get("policy") if isinstance(exposure.get("policy"), Mapping) else {} @@ -52,17 +79,17 @@ def governance_labels( "classification" ) if classification: - out["fluid_classification"] = _label(classification) + out["fluid_classification"] = value_of(classification) sovereignty = ( contract.get("sovereignty") if isinstance(contract.get("sovereignty"), Mapping) else {} ) if sovereignty.get("jurisdiction"): - out["fluid_jurisdiction"] = _label(sovereignty["jurisdiction"]) + out["fluid_jurisdiction"] = value_of(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)) + out["fluid_regulation"] = value_of(separator.join(str(f) for f in frameworks)) if sovereignty.get("dataResidency") is not None: out["fluid_residency"] = "true" if sovereignty.get("dataResidency") else "false" keys = sorted(out) diff --git a/fluid_build/iac/providers/databricks.py b/fluid_build/iac/providers/databricks.py index 9f559c06..ab3d202b 100644 --- a/fluid_build/iac/providers/databricks.py +++ b/fluid_build/iac/providers/databricks.py @@ -38,9 +38,12 @@ residency), are Unity Catalog tags: one ``databricks_entity_tag_assignment`` per tag on each owned schema (the contract's own) and each table (the expose's too). A tag key may not contain ``-``, so each ``-`` becomes ``_`` - (``data-classification`` is ``data_classification``); two labels that meet at one - key are ``databricks-tag-collision`` and more than the 50 tags a securable holds is - ``databricks-tags-too-many``, refused rather than overwritten or dropped. An + (``data-classification`` is ``data_classification``). A value is written as the + contract states it, less surrounding whitespace: a tag value keeps case, spaces and + punctuation, so it is not the lowercase slug the GCP label alphabet makes of it. Two + labels that meet at one key are ``databricks-tag-collision``, a value longer than 256 + characters is ``databricks-tag-value-too-long`` and more than the 50 tags a securable + holds is ``databricks-tags-too-many``, refused rather than overwritten, cut or dropped. An existing assignment is imported best-effort before the first apply, and removing a label destroys only its assignment, which holds no data. Whether a tag survives a dbt ``CREATE OR REPLACE`` is undocumented, so ``fluid verify`` is the gate. @@ -1849,12 +1852,14 @@ def _uc_tags( tags (the product's identity, :func:`_product_tags`), or ``{}`` and a refusal. ``governance_labels`` keys are ``[a-z0-9_-]``, and a tag key may not contain - ``-``, so each ``-`` becomes ``_``; the values go as they are. Two labels that - meet at one key, a label that meets one of the module's own keys, or more tags - than a securable holds, are refused: the module would otherwise overwrite one tag - with another, or drop some without a word. + ``-``, so each ``-`` becomes ``_``. The values go as the contract writes them, less + surrounding whitespace (``verbatim_values``): a tag value keeps case, spaces and + punctuation, so it is not cut to GCP's label alphabet. Two labels that meet at one + key, a label that meets one of the module's own keys, a value longer than a tag + value holds, or more tags than a securable holds, are refused: the module would + otherwise overwrite one tag with another, cut one, or drop some without a word. """ - labels = governance_labels(contract, exposure, limit=None) + labels = governance_labels(contract, exposure, limit=None, verbatim_values=True) # The certification label is Unity Catalog's certification tag, not a tag of its own. labels.pop(CERTIFICATION_LABEL, None) tags: Dict[str, str] = dict(own or {}) @@ -1877,6 +1882,9 @@ def _uc_tags( ) ) return {} + if len(value) > _MAX_TAG_VALUE: + refusals.append(_too_long(f"{where}: the governance label {label!r}", value, key)) + continue tags[key] = value source[key] = label if len(tags) > _MAX_TAGS: @@ -1896,6 +1904,16 @@ def _uc_tags( return tags +def _too_long(what: str, value: str, key: str) -> UnsupportedBindingError: + """``databricks-tag-value-too-long``: ``value`` (``what``) would be cut as tag ``key``.""" + return _refusal( + "tag-value-too-long", + f"{what} is {len(value)} characters, and a Unity Catalog tag value holds at most " + f"{_MAX_TAG_VALUE}, so tag {key} would be cut.", + f"Shorten it to {_MAX_TAG_VALUE} characters or fewer.", + ) + + def _lookup(document: Mapping[str, Any], path: str) -> Any: value: Any = document for part in path.split("."): @@ -1955,14 +1973,7 @@ def _product_tags( if not value: continue if len(value) > _MAX_TAG_VALUE: - refusals.append( - _refusal( - "tag-value-too-long", - f"{where} is {len(value)} characters, and a Unity Catalog tag value holds " - f"at most {_MAX_TAG_VALUE}, so tag {key} would be cut.", - f"Shorten it to {_MAX_TAG_VALUE} characters or fewer.", - ) - ) + refusals.append(_too_long(where, value, key)) continue tags[key] = value certification = _certification(contract, table, refusals) diff --git a/fluid_build/observability/secret_redactor.py b/fluid_build/observability/secret_redactor.py index b300a9af..0a52b974 100644 --- a/fluid_build/observability/secret_redactor.py +++ b/fluid_build/observability/secret_redactor.py @@ -86,18 +86,41 @@ "session_token", "token", ) -_JWT_RE = re.compile(r"\b[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\b") -# Three-segment JWT anchored on the ``eyJ`` base64url prefix of the JSON ``{"`` -# header (gitleaks' ``jwt`` rule and detect-secrets' ``JwtTokenDetector`` anchor -# the same way). Used instead of the prefix-agnostic ``_JWT_RE`` for text a person -# reads (``redact_secret_text(..., keep_dotted_names=True)``), where ``_JWT_RE`` -# would hide a fully qualified name such as -# ``my-gcp-project.customer_analytics.daily_revenue_summary``. Logs keep ``_JWT_RE``. -_JWT_ANCHORED_RE = re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+") +# A JWT, JWS or JWE in compact form (RFC 7519, RFC 7515 §7.1, RFC 7516 §7.1): +# base64url segments joined by ``.``, the first of them the protected header, a JSON +# object. Every JOSE library writes that object as ``{"``, which base64url-encodes to +# ``eyJ``, so this rule anchors there, as gitleaks' ``jwt`` rule (``ey``) and +# detect-secrets' ``JwtTokenDetector`` (``eyJ``) do. A JWS has a payload and a +# signature after the header, a JWE an encrypted key, IV, ciphertext and tag. The +# second segment may be empty: a JWE with direct encryption has no encrypted key +# (``header..iv.ciphertext.tag``), and a detached JWS has no payload. The rule starts +# where no base64url character precedes ``eyJ``; ``\b`` also matched after every ``-``, +# which made the scan quadratic on ``eyJ-eyJ-…``. A header glued to a word +# (``x-eyJ…``, ``token_eyJ…``) is :data:`_JWT_RE`'s. +_JWT_ANCHORED_RE = re.compile( + r"(? str: return f"{match.group('key')}{match.group('sep')}{quote}{_REDACTED}{quote}" -def redact_secret_text( - text: str, *, extra_secrets: Iterable[str] = (), keep_dotted_names: bool = False -) -> str: +def _is_jose_header(segment: str) -> bool: + """Whether ``segment`` is a JOSE protected header: base64url of a JSON object with + the ``alg`` member that RFC 7515 and RFC 7516 (section 4.1.1 each) require. + + detect-secrets' ``JwtTokenDetector.is_formally_valid`` checks a JWT the same way, + by decoding its parts and parsing them as JSON. A dotted name's part decodes to + bytes that are not JSON, and ``alg`` keeps a short one such as ``e30`` (``{}``) from + counting. The imports are deferred: ``base64`` stays off the ``fluid --help`` path, + which loads this module for its logging filter. + """ + import base64 + import json + + try: + decoded = json.loads(base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4))) + return isinstance(decoded, dict) and "alg" in decoded + except (ValueError, RecursionError): + # binascii.Error, UnicodeDecodeError and JSONDecodeError are ValueErrors. + return False + + +def _holds_jose_header(segment: str) -> bool: + """Whether ``segment`` is a JOSE header, or ends in one glued to a word. + + A header is base64url of ``{``, so it starts with ``e``: a glued one starts at an + ``e`` after ``-`` or ``_`` (``x-ewog…``, ``token_eyAi…``) or at an ``eyJ`` + (``tokeneyJ…``). At most :data:`_GLUED_HEADER_TRIES` of those places are tried. + """ + if _is_jose_header(segment): + return True + tries = 0 + at = segment.find("e", 1) + while at > 0 and tries < _GLUED_HEADER_TRIES: + if segment[at - 1] in "-_" or segment.startswith("eyJ", at): + tries += 1 + if _is_jose_header(segment[at:]): + return True + at = segment.find("e", at + 1) + return False + + +def _mask_if_jose(match: "re.Match[str]", placeholder: str) -> str: + """The ``_JWT_RE`` run ``match`` with the JOSE token in it masked, else unchanged. + + The token may start in any segment: one glued after a dotted name + (``northwind_towers.x-
.``) is masked, the name kept. + """ + segments = match.group(0).split(".") + for index, segment in enumerate(segments): + if _holds_jose_header(segment): + return ".".join(segments[:index] + [placeholder]) + return match.group(0) + + +def redact_jose_tokens(text: str, *, placeholder: str = _REDACTED) -> str: + """``text`` with every JWT, JWS and JWE masked, and dotted names left as written. + + One implementation for both redaction layers: the Snowflake twin + (``providers/snowflake/util/logging.py``) calls it with its own placeholder, as it + imports ``_URL_USERINFO_RE``, so the two cannot drift. + """ + if not isinstance(text, str) or not text: + return text + text = _JWT_ANCHORED_RE.sub(placeholder, text) + # Before the two-segment rule, which would mask an ``eyJ`` payload and its + # signature and leave a header this rule no longer sees as one token. + text = _JWT_RE.sub(lambda match: _mask_if_jose(match, placeholder), text) + return _JWT_TWO_SEGMENT_RE.sub(placeholder, text) + + +def redact_secret_text(text: str, *, extra_secrets: Iterable[str] = ()) -> str: """Redact secret-like substrings in plain text. ``extra_secrets`` are literals to mask for THIS call only — for callers that hold a credential but must not register it process-wide (e.g. the Snowflake IaC table COMMENT, which carries whatever one contract spelled inline and should not affect redaction of anything else). - - ``keep_dotted_names`` masks a three-segment JWT only when it has the ``eyJ`` - header, so a fully qualified table name in an error a person reads survives. - Log redaction leaves it off and keeps the prefix-agnostic rule. """ if not isinstance(text, str) or not text: return text @@ -602,8 +689,7 @@ def redact_secret_text( # PEM blocks first: a multiline key block can itself contain ``.``/``=`` # runs that would otherwise be partially mangled by the token regexes. redacted = _PEM_PRIVATE_KEY_RE.sub(_REDACTED, redacted) - redacted = (_JWT_ANCHORED_RE if keep_dotted_names else _JWT_RE).sub(_REDACTED, redacted) - redacted = _JWT_TWO_SEGMENT_RE.sub(_REDACTED, redacted) + redacted = redact_jose_tokens(redacted) # S-010: provider-token shapes run before the assignment regex so # ``api_key=sk_live_...`` hits the Stripe pattern in addition to the # assignment pattern, which also works. diff --git a/fluid_build/providers/snowflake/util/logging.py b/fluid_build/providers/snowflake/util/logging.py index baeb7d65..1e4e73e5 100644 --- a/fluid_build/providers/snowflake/util/logging.py +++ b/fluid_build/providers/snowflake/util/logging.py @@ -17,6 +17,7 @@ from fluid_build.observability.secret_redactor import ( _URL_USERINFO_RE, is_sensitive_key_name, + redact_jose_tokens, redact_known_values, ) @@ -117,19 +118,9 @@ # Azure keys (for external stages) re.compile(r'"azure_sas_token":\s*"[^"]*"', re.IGNORECASE), re.compile(r"AZURE_STORAGE_SAS_TOKEN=[^\s;&]+", re.IGNORECASE), - # Third-party / provider token shapes (S-010) - # JWT (three base64url parts separated by dots) - re.compile(r"\beyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+"), - # JWT (two-segment header.payload, no signature) - re.compile(r"\beyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\b"), - # Generic 3-segment JWT — prefix-agnostic (does NOT require the ``eyJ`` - # header). Mirrors the global ``secret_redactor._JWT_RE`` so a JWT signed - # with a non-JSON / non-``eyJ`` header (or any three high-entropy base64url - # runs) is masked by this layer too (CLAUDE.md symmetry). Placed AFTER the - # two ``eyJ``-anchored patterns so the tighter, header-anchored spans win - # first. Each segment is bounded-min {12,} (not unbounded) to keep it from - # matching short dotted identifiers / version strings. - re.compile(r"\b[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\b"), + # Third-party / provider token shapes (S-010). JWTs are not in this list: + # redact_string() masks them with the global layer's redact_jose_tokens, the one + # implementation both layers share (CLAUDE.md symmetry). # Stripe restricted / live / test secret keys re.compile(r"\bsk_(?:live|test)_[A-Za-z0-9]{16,}"), # GitHub personal access tokens, OAuth tokens, app installation tokens @@ -264,6 +255,9 @@ def redact_string(text: str) -> str: # password containing ``;`` ``,`` ``}`` ``]`` ``"`` ``&`` or a space is # masked whole, which no assignment pattern can guarantee. redacted = redact_known_values(text, placeholder="[REDACTED]") + # JWTs, JWSs and JWEs: the global layer's rule, so a dotted table name survives + # here exactly as it does there. + redacted = redact_jose_tokens(redacted, placeholder="[REDACTED]") for pattern in SENSITIVE_PATTERNS: # Three substitution shapes, dispatched on the pattern's groups: diff --git a/tests/cli/test_plan_sovereignty_gate.py b/tests/cli/test_plan_sovereignty_gate.py index ab8a58ea..42a8667d 100644 --- a/tests/cli/test_plan_sovereignty_gate.py +++ b/tests/cli/test_plan_sovereignty_gate.py @@ -289,7 +289,7 @@ def _write(tmp_path, contract): path.write_text(yaml.safe_dump(contract), encoding="utf-8") return path - def test_violating_contract_exits_non_zero(self, tmp_path): + def test_violating_contract_exits_non_zero_and_writes_no_plan(self, tmp_path, capsys): from fluid_build._contract_loader import CLIError from fluid_build.cli.plan import run @@ -299,6 +299,8 @@ def test_violating_contract_exits_non_zero(self, tmp_path): assert excinfo.value.exit_code == 1 assert excinfo.value.event == "sovereignty_violation" + assert not (tmp_path / "plan.json").exists() + assert "Plan saved" not in capsys.readouterr().out def test_compliant_contract_exits_zero(self, tmp_path): from fluid_build.cli.plan import run @@ -306,9 +308,102 @@ def test_compliant_contract_exits_zero(self, tmp_path): path = self._write(tmp_path, _contract(region="eu-central-1")) assert run(self._args(tmp_path, path), LOG) == 0 - def test_flag_off_does_not_gate(self, tmp_path): - """Without the flag the check does not run, so nothing blocks.""" + def test_no_check_sovereignty_skips_the_check_and_says_so(self, tmp_path, caplog, capsys): + """``--no-check-sovereignty`` writes the plan anyway: on the console and at WARNING.""" from fluid_build.cli.plan import run path = self._write(tmp_path, _contract()) - assert run(self._args(tmp_path, path, check_sovereignty=False), LOG) == 0 + with caplog.at_level(logging.WARNING): + assert run(self._args(tmp_path, path, check_sovereignty=False), LOG) == 0 + assert (tmp_path / "plan.json").exists() + assert "sovereignty check was SKIPPED" in caplog.text + assert "Sovereignty check skipped (--no-check-sovereignty)" in capsys.readouterr().out + + def test_a_clean_provider_hook_does_not_weaken_check_sovereignty(self, tmp_path, monkeypatch): + """The explicit flag runs the default check first, so a hook that reports + nothing cannot pass a contract ``fluid validate`` refuses.""" + from fluid_build._contract_loader import CLIError + from fluid_build.cli import plan as plan_mod + + monkeypatch.setattr( + plan_mod, "_hook_provider", lambda *_a, **_k: ("aws", _CleanHookProvider()) + ) + path = self._write(tmp_path, _contract()) + with pytest.raises(CLIError) as excinfo: + plan_mod.run(self._args(tmp_path, path, check_sovereignty=True), LOG) + assert excinfo.value.event == "sovereignty_violation" + assert not (tmp_path / "plan.json").exists() + + +class TestPlanChecksSovereigntyByDefault: + """``fluid plan`` with no flag refuses what ``fluid validate`` refuses.""" + + pytestmark = pytest.mark.unit + + def _run(self, tmp_path, contract): + from fluid_build.cli.plan import run + + path = TestPlanExitCode._write(tmp_path, contract) + return run(TestPlanExitCode._args(tmp_path, path, check_sovereignty=None), LOG) + + @pytest.mark.parametrize( + "kwargs", + [ + {}, # strict, outside allowedRegions and in deniedRegions + {"denied": False}, # strict, outside allowedRegions + {"mode": "advisory"}, # an explicit deny blocks in every mode + ], + ids=["strict-denied", "strict-not-allowed", "advisory-denied"], + ) + def test_a_blocking_contract_is_refused_and_no_plan_is_written(self, tmp_path, kwargs, capsys): + from fluid_build._contract_loader import CLIError + + with pytest.raises(CLIError) as excinfo: + self._run(tmp_path, _contract(**kwargs)) + assert (excinfo.value.exit_code, excinfo.value.event) == (1, "sovereignty_violation") + assert not (tmp_path / "plan.json").exists() + out = capsys.readouterr().out + assert "us-east-1" in out and "Sovereignty check FAILED" in out + # The expose is named, printed literally (Rich would take "[id]" for markup). + assert "[customer_events]" in out + # The findings ride on the error, as local_plan_validation_failed's do, so an + # engine-API caller (Command Center's plan endpoint) can show the refusal. + context = excinfo.value.context + assert context["errors"] and all("[customer_events]" in e for e in context["errors"]) + assert "us-east-1" in " ".join(context["errors"]) and context["message"] + + @pytest.mark.parametrize( + "kwargs", + [ + {"region": "eu-central-1"}, # compliant + {"mode": "advisory", "denied": False}, # advisory: a warning, not a block + {"sovereignty": False}, # no policy to check + ], + ids=["compliant", "advisory-not-allowed", "no-policy"], + ) + def test_a_contract_that_does_not_block_plans_as_before(self, tmp_path, kwargs, capsys): + assert self._run(tmp_path, _contract(**kwargs)) == 0 + assert (tmp_path / "plan.json").exists() + assert "Sovereignty check" not in capsys.readouterr().out + + def test_the_default_check_builds_no_provider(self, tmp_path, monkeypatch): + """A plain plan stays off the cloud SDKs (Command Center's plan relies on it).""" + from fluid_build._contract_loader import CLIError + from fluid_build.cli import plan as plan_mod + + def no_provider(*_args, **_kwargs): + raise AssertionError("the default sovereignty check built a provider") + + monkeypatch.setattr(plan_mod, "build_provider", no_provider) + with pytest.raises(CLIError): + self._run(tmp_path, _contract()) + + @pytest.mark.parametrize( + "argv, expected", + [([], None), (["--check-sovereignty"], True), (["--no-check-sovereignty"], False)], + ) + def test_the_flag_has_three_states(self, argv, expected): + from fluid_build.cli import build_parser + + args = build_parser().parse_args(["plan", "contract.fluid.yaml", *argv]) + assert args.check_sovereignty is expected diff --git a/tests/iac/test_iac_databricks.py b/tests/iac/test_iac_databricks.py index f7b2708d..d7c61354 100644 --- a/tests/iac/test_iac_databricks.py +++ b/tests/iac/test_iac_databricks.py @@ -488,8 +488,8 @@ def test_validate_reports_the_policy_once_not_from_the_binding_check(self): SCHEMA_TAGS = { "cost_center": "cx-001", "data_classification": "confidential", - "fluid_jurisdiction": "eu", - "fluid_regulation": "gdpr", + "fluid_jurisdiction": "EU", + "fluid_regulation": "GDPR", "fluid_residency": "true", "gdpr_article": "art6-1b-contractual", } @@ -530,6 +530,21 @@ def _tags_on(doc: Dict[str, Any], entity_type: str) -> Dict[str, str]: } +def _refused_on_both_securables(contract: Dict[str, Any], kind: str, text: str) -> None: + """``kind`` from emit and from the planner, and one validate error per securable: the + schema and the table both carry the contract's labels.""" + for call in ( + lambda: IAC_PLUGINS["databricks"].emit(contract), + lambda: dbx.planned_objects(contract), + ): + with pytest.raises(UnsupportedBindingError) as raised: + call() + assert raised.value.kind == kind + assert text in str(raised.value) + errors, _ = dbx.validate_databricks_binding(contract) + assert len(errors) == 2 and all(text in e for e in errors) + + class TestGovernanceTags: def test_the_schema_and_the_table_carry_the_contracts_governance_as_tags(self): doc = _module(_governed()) @@ -620,17 +635,7 @@ def test_the_planner_names_each_assignment_by_its_import_id(self): def test_two_labels_that_meet_at_one_tag_key_are_refused_never_overwritten(self): contract = _governed() contract["labels"] = {"cost-center": "a", "cost_center": "b"} - for call in ( - lambda: IAC_PLUGINS["databricks"].emit(contract), - lambda: dbx.planned_objects(contract), - ): - with pytest.raises(UnsupportedBindingError) as raised: - call() - assert raised.value.kind == "databricks-tag-collision" - assert "cost_center" in str(raised.value) - errors, _ = dbx.validate_databricks_binding(contract) - # the schema and the table both carry the contract's labels - assert len(errors) == 2 and all("cost_center" in e for e in errors) + _refused_on_both_securables(contract, "databricks-tag-collision", "cost_center") def test_more_tags_than_a_securable_holds_is_refused_never_dropped(self): contract = _contract() @@ -665,6 +670,75 @@ def test_managed_objects_carry_the_expected_tags_for_verify(self): assert dict(table.schema_tags) == SCHEMA_TAGS +class TestTagValuesAsWritten: + """A Unity Catalog tag value is the contract's text, not the GCP label slug of it.""" + + def test_a_value_is_written_as_the_contract_states_it_not_as_a_label_slug(self): + # The 0.29.0 shape: lowercased, punctuation and spaces as '-', cut at 63 + # ('a-single-unrenewed-ground-lease-can-strand-a-multi-tenant-tower'). + sentence = ( + "A single unrenewed ground lease can strand a multi-tenant tower and its entire " + "revenue stream; mastered terms and expiry dates are what the renewal radar runs on." + ) + contract = _governed() + contract["labels"] = { + "business-value": f" {sentence} ", + "Standards": "IFRS 16 leases; ISO 4217 currency codes", + "consumers": "real-estate-managers, finance-analysts, legal-counsel", + "office": "Zürich: Bahnhofstrasse 1/3 (key=value)", + } + contract["sovereignty"].update(jurisdiction="US", regulatoryFramework=["GDPR", "CCPA"]) + contract["exposes"][0].setdefault("policy", {})["classification"] = "Restricted" + tags = _tags_on(_module(contract), "tables") + assert tags["business_value"] == sentence + assert tags["standards"] == "IFRS 16 leases; ISO 4217 currency codes" + assert tags["consumers"] == "real-estate-managers, finance-analysts, legal-counsel" + assert tags["office"] == "Zürich: Bahnhofstrasse 1/3 (key=value)" + assert tags["fluid_jurisdiction"] == "US" + assert tags["fluid_regulation"] == "GDPR, CCPA" + assert tags["fluid_classification"] == "Restricted" + # The keys are as before: GCP's label alphabet, each '-' an '_'. + assert {"business_value", "standards", "consumers", "office"} <= set(tags) + + def test_a_changed_value_keeps_the_address_and_import_id_so_it_updates_in_place(self): + before, after = _governed(), _governed() + after["labels"]["cost-center"] = "CX 001 (Finance)" + assert set(_module(before)["resource"][TAG]) == set(_module(after)["resource"][TAG]) + ids = [ + sorted(b.id for b in IAC_PLUGINS["databricks"].discover_imports(c) if TAG in b.to) + for c in (before, after) + ] + assert ids[0] == ids[1] + + def test_a_value_longer_than_a_tag_value_holds_is_refused_never_cut(self): + contract = _governed() + contract["labels"]["grain"] = "x" * 256 + assert _tags_on(_module(contract), "tables")["grain"] == "x" * 256 + contract["labels"]["grain"] = "x" * 257 + _refused_on_both_securables( + contract, "databricks-tag-value-too-long", "'grain' is 257 characters" + ) + + def test_a_value_that_is_not_a_string_is_spelled_as_yaml_spells_it(self): + # The schema allows only strings, but ``fluid generate iac`` emits without it. + from fluid_build.iac.governance_labels import governance_labels + + contract = _contract() + contract["labels"] = {"pii": True, "archived": False, "gone": None, "ratio": 1.5} + labels = governance_labels(contract, contract["exposes"][0], verbatim_values=True) + assert labels == {"pii": "true", "archived": "false", "ratio": "1.5"} + + def test_gcp_and_aws_labels_keep_the_label_alphabet(self): + from fluid_build.iac.governance_labels import governance_labels + + contract = _governed() + contract["labels"] = {"business-value": "A Tower; Its Revenue"} + labels = governance_labels(contract, contract["exposes"][0]) + assert labels["business-value"] == "a-tower--its-revenue" + assert labels["fluid_jurisdiction"] == "eu" + assert labels["fluid_regulation"] == "gdpr" + + def test_a_build_that_names_its_own_host_is_refused(): # The dbt runner reads runtime.resources.host before DATABRICKS_HOST and # would send the workspace token there. diff --git a/tests/iac/test_iac_databricks_fake_e2e.py b/tests/iac/test_iac_databricks_fake_e2e.py index 750dfe03..e614dc4a 100644 --- a/tests/iac/test_iac_databricks_fake_e2e.py +++ b/tests/iac/test_iac_databricks_fake_e2e.py @@ -214,6 +214,35 @@ def test_a_changed_label_value_updates_the_tag_in_place(workspace, fake, capfd): ) in (fake.requests) +def test_slug_values_a_deployment_carries_are_rewritten_in_place( + workspace, fake, capfd, monkeypatch +): + """Up to 0.29.0 a tag value was its GCP label slug (``cx-001--finance-``). The next + apply rewrites each changed tag in place, one ``~`` each, and replaces nothing.""" + from fluid_build.iac.providers import databricks as dbx + + real = dbx.governance_labels + + def as_slugs(contract, exposure, **kwargs): + return real(contract, exposure, **{**kwargs, "verbatim_values": False}) + + path = write_contract(workspace, _governed(**{**LABELS, "cost-center": "CX 001 (Finance)"})) + monkeypatch.setattr(dbx, "governance_labels", as_slugs) + assert apply_contract(capfd, path)[0] == 0 + assert fake.tags_on("tables", TABLE)["cost_center"] == "cx-001--finance-" + marker = fake.tables[TABLE]["marker"] + + monkeypatch.setattr(dbx, "governance_labels", real) + rc, out = apply_contract(capfd, path, "--mode", "dry-run") + assert rc == 0 and "tofu plan: +0 ~2 -0" in out, out # the schema's tag and the table's + rc, out = apply_contract(capfd, path) + assert rc == 0, out + assert fake.tags_on("tables", TABLE)["cost_center"] == "CX 001 (Finance)" + assert fake.tags_on("schemas", "main.sales")["cost_center"] == "CX 001 (Finance)" + assert fake.tables[TABLE]["marker"] == marker and fake.replaced == [] + assert "tofu plan: +0 ~0 -0" in apply_contract(capfd, path, "--mode", "dry-run")[1] + + def test_removing_a_label_destroys_only_its_assignment_without_allow_data_loss( workspace, fake, capfd ): diff --git a/tests/providers/test_snowflake_util_logging.py b/tests/providers/test_snowflake_util_logging.py index c01b99f6..884b9d03 100644 --- a/tests/providers/test_snowflake_util_logging.py +++ b/tests/providers/test_snowflake_util_logging.py @@ -16,6 +16,8 @@ import re +import pytest + from fluid_build.providers.snowflake.util.logging import ( SENSITIVE_PATTERNS, redact_dict, @@ -99,19 +101,47 @@ def test_sensitive_patterns_group_counts_are_handled_by_redact_string() -> None: # --------------------------------------------------------------------------- +def _spaced_header_jwt() -> str: + """A JWT whose header does NOT base64url-encode to ``eyJ``: one written with spaces + (``{ "alg": ... }`` encodes to ``eyA``). Built at runtime so no literal token sits in + the file.""" + import base64 + + parts = (b'{ "alg": "HS256" }', b'{"sub":"user1234567"}', bytes(32)) + return ".".join(base64.urlsafe_b64encode(part).rstrip(b"=").decode() for part in parts) + + class TestSnowflakeRedactionSymmetryGaps: """Gaps ``redact_string`` previously leaked while the global ``redact_secret_text`` twin caught them.""" + @pytest.fixture(autouse=True) + def _no_registered_secrets(self): + # The exact-match registry both layers share is process-wide; a literal another + # test in this worker registered would mask part of a token here first. + import sys + + from fluid_build.observability.secret_redactor import ( + forget_known_secrets, + redact_secret_text, + register_secrets, + ) + + held = sys.modules[redact_secret_text.__module__]._known_secrets + forget_known_secrets() + yield + forget_known_secrets() + register_secrets(held) + # -- Gap 5: generic non-``eyJ`` 3-segment JWT -------------------------- def test_generic_non_eyj_three_segment_jwt_masked(self) -> None: - # A JWT whose header does NOT base64url-encode to ``eyJ`` (or any three - # high-entropy base64url runs) — previously only the ``eyJ``-anchored - # patterns existed here, so this leaked. - jwt = "aGVhZGVyMTIzNDU2.cGF5bG9hZDEyMzQ1Ng.c2lnbmF0dXJlMTIz" - out = redact_string(f"stage token {jwt} accepted") - assert jwt not in out - assert "[REDACTED]" in out + jwt = _spaced_header_jwt() + assert not jwt.startswith("eyJ") + assert redact_string(f"stage token {jwt} accepted") == "stage token [REDACTED] accepted" + # A fully qualified table name has the same shape and is no token. The rule this + # layer shared with the global one masked it ("table=[REDACTED]"). + name = "loaded northwind_towers.bronze_smart_meters.smart_meters rows=340" + assert redact_string(name) == name def test_eyj_anchored_jwt_still_masked(self) -> None: # The tighter eyJ-anchored patterns must still win (positioned first). @@ -165,9 +195,13 @@ def test_private_key_prose_not_over_redacted(self) -> None: def test_gaps_symmetric_with_global_twin(self) -> None: from fluid_build.observability.secret_redactor import redact_secret_text - jwt = "aGVhZGVyMTIzNDU2.cGF5bG9hZDEyMzQ1Ng.c2lnbmF0dXJlMTIz" - assert jwt not in redact_string(f"t {jwt} x") - assert jwt not in redact_secret_text(f"t {jwt} x") + # A non-``eyJ`` JWT, masked by both layers... + jwt = _spaced_header_jwt() + assert jwt.split(".")[0] not in redact_string(f"t {jwt} x") + assert jwt.split(".")[0] not in redact_secret_text(f"t {jwt} x") + # ...and a fully qualified name of the same shape, kept by both. + name = "northwind_towers.bronze_smart_meters.smart_meters" + assert redact_string(f"t {name} x") == redact_secret_text(f"t {name} x") == f"t {name} x" value = "MIIEvQIBADANBgkqhkiNONPEMVALUE0123456789" # pragma: allowlist secret line = f"private_key={value}" diff --git a/tests/test_observability_secret_redactor.py b/tests/test_observability_secret_redactor.py index 2e7cea6f..504cf8da 100644 --- a/tests/test_observability_secret_redactor.py +++ b/tests/test_observability_secret_redactor.py @@ -12,7 +12,13 @@ import pytest from fluid_build.cli import _setup_enhanced_logging -from fluid_build.observability.secret_redactor import SecretRedactingFilter, redact_secret_text +from fluid_build.observability.secret_redactor import ( + SecretRedactingFilter, + forget_known_secrets, + redact_jose_tokens, + redact_secret_text, + register_secrets, +) @pytest.fixture @@ -689,33 +695,170 @@ def test_key_names_masked(self, name: str) -> None: assert "v4lue-xyz" not in redact_secret_text('{"%s": "v4lue-xyz"}' % name) -class TestJwtAnchoring: - """Error text a person reads keeps a dotted table name; logs keep the #389 rule.""" +def _b64url(raw: bytes) -> str: + import base64 + + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode() + + +def _jose(header: bytes, *rest: bytes) -> str: + """A compact JOSE token built at runtime, so no literal token sits in the file.""" + return ".".join([_b64url(header), *(_b64url(part) for part in rest)]) + + +_SIGNATURE = bytes(range(32)) +_PAYLOAD = b'{"sub":"1234567890","name":"x"}' + +#: Each shape the JOSE rule must mask, in both layers and through the logging filter. +_JOSE_TOKENS = { + "compact JWS": _jose(b'{"alg":"HS256","typ":"JWT"}', _PAYLOAD, _SIGNATURE), + "header with spaces": _jose(b'{ "alg": "HS256" }', _PAYLOAD, _SIGNATURE), + "header with newlines": _jose(b'{\n "alg": "HS256"\n}', _PAYLOAD, _SIGNATURE), + "JWE, direct encryption (no encrypted key)": _jose( + b'{"alg":"dir","enc":"A256GCM"}', b"", b"0123456789ab", b"ciphertext here", bytes(16) + ), + "JWE with an encrypted key": _jose( + b'{"alg":"RSA-OAEP","enc":"A256GCM"}', bytes(64), b"0123456789ab", b"ciphertext", bytes(16) + ), + "unsecured JWS (empty signature)": _jose(b'{"alg":"none"}', _PAYLOAD, b""), + "short payload": _jose(b'{"alg":"HS256"}', b"{}", _SIGNATURE), + "glued after _": "token_" + _jose(b'{"alg":"HS256","typ":"JWT"}', _PAYLOAD, _SIGNATURE), + "glued after -": "x-" + _jose(b'{"alg":"HS256","typ":"JWT"}', _PAYLOAD, _SIGNATURE), + "two segments (no signature)": _jose(b'{"alg":"none"}', _PAYLOAD), + "JWE, direct encryption, header with spaces": _jose( + b'{ "alg": "dir", "enc": "A256GCM" }', b"", b"0123456789ab", b"ciphertext here", bytes(16) + ), + "JWE with a key, header with newlines": _jose( + b'{\n "alg": "RSA-OAEP", "enc": "A256GCM"\n}', bytes(64), b"0123456789ab", b"ct", bytes(16) + ), + "two segments glued after -": "x-" + _jose(b'{"alg":"none"}', _PAYLOAD), +} + +#: Fully qualified names the old prefix-agnostic rule masked (three runs of 12+), and +#: the 47-character one it never did (its last part has 11). +_DOTTED_NAMES = [ + "northwind_towers.bronze_smart_meters.smart_meters", + "northwind_towers.bronze_site_master.site_master", + "northwind_towers.bronze_smart_meters.smart_meters.reading_timestamp", + "ANALYTICS_DB.PUBLIC_SCHEMA.CUSTOMER_ORDERS", + "my-gcp-project.customer_analytics.daily_revenue_summary", + "my-gcp-project.CustomerAnalytics.DailyRevenueSummary", + "module.databricks_tables_main.databricks_sql_table.northwind_meters", + "monkeyJar_inventory.bronze_inventory_x.inventory_items", +] + + +def _leaked(token: str, out: str) -> list: + return [part for part in token.split(".") if len(part) >= 8 and part in out] + + +class TestJoseTokensAndDottedNames: + """JOSE tokens are masked by their header; a dotted name is never a token.""" + + @pytest.fixture(autouse=True) + def _no_registered_secrets(self): + # The exact-match registry is process-wide: a literal another test in this + # worker registered (a base64 key of bytes(range(32)), say) would mask part + # of these tokens first. The expectations here are the JOSE rule's alone. + held = sys.modules[redact_secret_text.__module__]._known_secrets + forget_known_secrets() + yield + forget_known_secrets() + register_secrets(held) + + @pytest.mark.parametrize("name", _DOTTED_NAMES) + def test_a_fully_qualified_name_survives_every_path(self, name): + from fluid_build._console import literal_error_text + from fluid_build.providers.snowflake.util.logging import redact_string + + text = f"databricks.load table={name} rows=340" + assert redact_secret_text(text) == text + assert redact_string(text) == text + assert literal_error_text(text).plain == text + + @pytest.mark.parametrize("name", _DOTTED_NAMES) + def test_a_fully_qualified_name_survives_the_logging_filter(self, name): + logger, stream = _exc_logger("test.secret_redactor.dotted_name") + logger.info("databricks.load table=%s rows=%d deleted=true", name, 340) + assert stream.getvalue().strip() == f"databricks.load table={name} rows=340 deleted=true" + + @pytest.mark.parametrize("shape", sorted(_JOSE_TOKENS)) + def test_a_jose_token_is_masked_in_both_layers(self, shape): + from fluid_build.providers.snowflake.util.logging import redact_string + + token = _JOSE_TOKENS[shape] + line = f"token {token} end" + assert _leaked(token, redact_secret_text(line)) == [] + assert _leaked(token, redact_string(line)) == [] + + @pytest.mark.parametrize("shape", sorted(_JOSE_TOKENS)) + def test_a_jose_token_is_masked_by_the_logging_filter(self, shape): + token = _JOSE_TOKENS[shape] + logger, stream = _exc_logger("test.secret_redactor.jose") + logger.info("refresh returned %s", token) + assert _leaked(token, stream.getvalue()) == [] + assert "***REDACTED***" in stream.getvalue() + + def test_three_runs_that_are_not_a_token_are_left_alone(self): + # The old rule's own example of "any three high-entropy runs": base64 of + # "header123456", "payload123456" and "signature123", whose first part is no + # JSON object, so it is no JOSE header. + runs = "aGVhZGVyMTIzNDU2.cGF5bG9hZDEyMzQ1Ng.c2lnbmF0dXJlMTIz" + assert redact_secret_text(f"id {runs} end") == f"id {runs} end" @pytest.mark.parametrize( - "name", + "glued", [ - "my-gcp-project.customer_analytics.daily_revenue_summary", - "ANALYTICS_DB.PUBLIC_SCHEMA.CUSTOMER_ORDERS", + "northwind_towers.x-" + _jose(b'{"alg":"none"}', _PAYLOAD), + "northwind_towers.bronze_smart_meters.smart_meters.reading.x-" + + _jose(b'{"alg":"none"}', _PAYLOAD), + "token_" + + _jose(b'{ "alg": "dir", "enc": "A256GCM" }', b"", b"0123456789ab", b"ct", b"tag"), + "x-" + _jose(b'{\n "alg": "HS256"\n}', _PAYLOAD, _SIGNATURE), ], ) - def test_fully_qualified_table_names_survive_in_error_text(self, name): - from fluid_build._console import literal_error_text + def test_a_header_glued_anywhere_in_a_run_is_found(self, glued): + from fluid_build.providers.snowflake.util.logging import redact_string - text = f"Table {name} not found" - assert redact_secret_text(text, keep_dotted_names=True) == text - assert literal_error_text(text).plain == text - # Log redaction keeps the prefix-agnostic three-segment rule. - assert name not in redact_secret_text(text) - - @pytest.mark.parametrize("keep_dotted_names", [False, True]) - def test_three_and_two_segment_jwts_are_still_masked(self, keep_dotted_names): - header = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9" - three = f"{header}.eyJzdWIiOiIxMjM0NTY3ODkwIn0.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV" - two = f"{header}.eyJzdWIiOiIxMjM0NTY3ODkwIn0" - for token in (three, two): - out = redact_secret_text(f"token {token} end", keep_dotted_names=keep_dotted_names) - assert "eyJ" not in out + for out in (redact_secret_text(f"got {glued} end"), redact_string(f"got {glued} end")): + assert _leaked(glued.split("x-")[-1].split("token_")[-1], out) == [] + assert redact_secret_text(f"got {glued} end").startswith( + "got northwind_towers." if glued.startswith("northwind") else "got " + ) + + def test_a_short_json_part_of_a_name_is_no_header(self): + # e30 is base64url of {}: JSON, but no JOSE header has no alg. + name = "northwind_towers.e30.daily_revenue_summary" + assert redact_secret_text(f"loaded {name}") == f"loaded {name}" + + def test_the_words_after_a_token_are_kept(self): + token = _JOSE_TOKENS["compact JWS"] + assert redact_secret_text(f"got {token}.Next step") == "got ***REDACTED***.Next step" + + def test_the_snowflake_twin_shares_the_rule(self): + from fluid_build.providers.snowflake.util.logging import redact_jose_tokens as twin + + assert twin is redact_jose_tokens + + @pytest.mark.parametrize( + "evil", + [ + "a-" * 100_000, + "eyJ-" * 50_000, + "eyJ_" * 50_000, + "eyJ." * 50_000, + ("abcdefghijklmn." * 3 + " ") * 5_000, + ], + ) + def test_the_scan_stays_linear(self, evil): + import time + + start = time.perf_counter() + redact_jose_tokens(evil) + elapsed = time.perf_counter() - start + # Linear: well under 0.2 s here. The ``\b``-anchored rules re-scanned the run + # from every ``-`` (``a-`` * 100,000 took about 30 s, ``eyJ-`` * 50,000 about 40 s). + assert elapsed < 2.0, f"possible ReDoS regression: {elapsed:.2f}s" class TestUrlUserinfoKeepsAMaskedPassword: