From e57605bec7ff6c18624d75c1edec1e635cd3976b Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Tue, 3 Nov 2026 21:43:32 +0100
Subject: [PATCH 1/8] security: refuse a trailing newline in whole-string
 validators

Python's `$` also matches just before a final newline, so
`re.match(r"^[a-z]+$", "orders\n")` succeeds and the caller goes on to
use "orders\n". #788 closed this in providers/_sql_safety.py. The same
`^...$` plus `.match` shape guarded 36 more validator patterns across
fluid_build/, fed by contract YAML (a `|` block scalar ends in a
newline), .env values, CLI flags, LLM tool calls and percent-encoded
URL queries.

31 of those patterns now end in `\Z`. Two call sites use `fullmatch`
instead: the catalog quoting chokepoint, which also covers the BigQuery
patterns callers pass in, and the federation workspace id, whose error
message prints the pattern. Seen through the CLI before the fix,
`fluid generate iac` put `region = "eu-west-1\n"` on the AWS provider,
and `branch = "main\n"` on a Databricks job from `?branch=main%0A`.

81 other `$`-ending patterns stay as they are. Their input is stripped
first, or they are deny lists and detectors where `\Z` would loosen
the check, or they are parsers whose captured groups are what is used.

tests/security/test_trailing_newline_anchors.py pins each fixed site
with a value and the same value plus "\n". All 39 cases fail on the
parent commit.

The fix follows Django's username validators (ticket #30257, `$` to
`\Z`) and the MCP Python SDK's tool-name check (#3076, `fullmatch`).
---
 fluid_build/build_runners/_ids.py             |   2 +-
 fluid_build/build_runners/meltano/runner.py   |   4 +-
 fluid_build/cli/_forge_data_model_jdbc.py     |   2 +-
 fluid_build/cli/discover/_jdbc_introspect.py  |   2 +-
 fluid_build/cli/forge_copilot_prompts.py      |   2 +-
 fluid_build/cli/forge_prompt_overlays.py      |   2 +-
 fluid_build/cli/forge_validation.py           |   4 +-
 fluid_build/cli/import_workflow/dbt.py        |   2 +-
 fluid_build/cli/security.py                   |   2 +-
 fluid_build/copilot/catalog/_patterns.py      |   2 +-
 fluid_build/copilot/missions/spec.py          |   2 +-
 fluid_build/forge/core/bundle.py              |   2 +-
 fluid_build/forge/federation.py               |   2 +-
 fluid_build/iac/providers/aws.py              |   2 +-
 fluid_build/iac/providers/databricks_jobs.py  |   6 +-
 fluid_build/iac/row_access.py                 |   2 +-
 fluid_build/providers/aws/util/validation.py  |   8 +-
 fluid_build/providers/bigquery_validation.py  |   2 +-
 fluid_build/providers/gcp/plan/planner.py     |   2 +-
 fluid_build/providers/gcp/util/names.py       |   8 +-
 fluid_build/providers/quality_engine.py       |   2 +-
 .../providers/snowflake/util/config.py        |   4 +-
 fluid_build/schedulers/airflow/fluid_apply.py |   6 +-
 .../security/test_trailing_newline_anchors.py | 372 ++++++++++++++++++
 24 files changed, 408 insertions(+), 36 deletions(-)
 create mode 100644 tests/security/test_trailing_newline_anchors.py

diff --git a/fluid_build/build_runners/_ids.py b/fluid_build/build_runners/_ids.py
index b2d3abde5..831263a63 100644
--- a/fluid_build/build_runners/_ids.py
+++ b/fluid_build/build_runners/_ids.py
@@ -44,7 +44,7 @@
 # DOS-via-filename attacks. Crucially this excludes ``/``, ``\``, ``..``
 # (a leading dot is rejected) and any path separator, so a validated id
 # can never traverse out of the directory it is joined into.
-_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.\-]{0,127}$")
+_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.\-]{0,127}\Z")
 
 
 class IdentifierViolation(ValueError):
diff --git a/fluid_build/build_runners/meltano/runner.py b/fluid_build/build_runners/meltano/runner.py
index 25dc10565..25b2769f6 100644
--- a/fluid_build/build_runners/meltano/runner.py
+++ b/fluid_build/build_runners/meltano/runner.py
@@ -112,8 +112,8 @@ def collect_singer_output(
 # ── Tap invocation ──────────────────────────────────────────────────────
 
 
-_TAP_NAME_RE = re.compile(r"^tap-[a-z0-9](?:[a-z0-9._-]*[a-z0-9])?$")
-_TARGET_NAME_RE = re.compile(r"^target-[a-z0-9](?:[a-z0-9._-]*[a-z0-9])?$")
+_TAP_NAME_RE = re.compile(r"^tap-[a-z0-9](?:[a-z0-9._-]*[a-z0-9])?\Z")
+_TARGET_NAME_RE = re.compile(r"^target-[a-z0-9](?:[a-z0-9._-]*[a-z0-9])?\Z")
 
 
 def _resolve_tap_binary(tap_name: str, *, project_dir: Optional[Path] = None) -> Optional[str]:
diff --git a/fluid_build/cli/_forge_data_model_jdbc.py b/fluid_build/cli/_forge_data_model_jdbc.py
index 6daddcfac..0be635a48 100644
--- a/fluid_build/cli/_forge_data_model_jdbc.py
+++ b/fluid_build/cli/_forge_data_model_jdbc.py
@@ -36,7 +36,7 @@
 
 # Canonical FLUID contract-id pattern (lowercased form — ``product_id`` is
 # lowercased before the check). Mirrors the JSON-schema id constraint.
-_FLUID_ID_RE = re.compile(r"^[a-z0-9_][a-z0-9_.-]*[a-z0-9_]$|^[a-z0-9_]$")
+_FLUID_ID_RE = re.compile(r"^[a-z0-9_][a-z0-9_.-]*[a-z0-9_]\Z|^[a-z0-9_]\Z")
 
 
 def _run_from_jdbc_source(
diff --git a/fluid_build/cli/discover/_jdbc_introspect.py b/fluid_build/cli/discover/_jdbc_introspect.py
index 39d04be79..146739408 100644
--- a/fluid_build/cli/discover/_jdbc_introspect.py
+++ b/fluid_build/cli/discover/_jdbc_introspect.py
@@ -541,7 +541,7 @@ def _attach_string_sqlite(args: Dict[str, str], alias: str) -> str:
     return f"ATTACH {quote_ansi_string_literal(path)} AS {alias} (TYPE sqlite)"
 
 
-_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*\Z")
 
 
 def _validate_alias(alias: str) -> str:
diff --git a/fluid_build/cli/forge_copilot_prompts.py b/fluid_build/cli/forge_copilot_prompts.py
index cb5dade65..319c3acd2 100644
--- a/fluid_build/cli/forge_copilot_prompts.py
+++ b/fluid_build/cli/forge_copilot_prompts.py
@@ -253,7 +253,7 @@ def get_active_domain() -> Optional[str]:
 # leading dot/underscore. This is the first line of defence against
 # ``--prompt-profile ../../etc``; ``_resolve_profile_dir`` adds a resolved-path
 # containment check as the second.
-_PROFILE_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+_PROFILE_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*\Z")
 
 
 class PromptProfileError(ValueError):
diff --git a/fluid_build/cli/forge_prompt_overlays.py b/fluid_build/cli/forge_prompt_overlays.py
index 5842f2ac7..6accf0781 100644
--- a/fluid_build/cli/forge_prompt_overlays.py
+++ b/fluid_build/cli/forge_prompt_overlays.py
@@ -96,7 +96,7 @@
 # leading dot/underscore. First line of defence against
 # ``--prompt-overlay ../../etc``; :func:`_resolve_overlay_path` adds a
 # resolved-realpath containment check as the second.
-_OVERLAY_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+_OVERLAY_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*\Z")
 
 _ALLOWED_MODES = ("replace", "append", "prepend")
 
diff --git a/fluid_build/cli/forge_validation.py b/fluid_build/cli/forge_validation.py
index c6cb37867..966ee0058 100644
--- a/fluid_build/cli/forge_validation.py
+++ b/fluid_build/cli/forge_validation.py
@@ -46,7 +46,7 @@ def validate_project_name(name: str) -> Tuple[bool, Optional[str]]:
     if len(name) > 63:
         return False, "Project name must be 63 characters or less"
 
-    if not re.match(r"^[a-z][a-z0-9-]*[a-z0-9]$", name):
+    if not re.match(r"^[a-z][a-z0-9-]*[a-z0-9]\Z", name):
         return (
             False,
             "Project name must start with a letter, end with letter/number, and contain only lowercase letters, numbers, and hyphens",
@@ -147,7 +147,7 @@ def validate_template_name(template: str) -> Tuple[bool, Optional[str]]:
     if len(template) < 2:
         return False, "Template name must be at least 2 characters"
 
-    if not re.match(r"^[a-z][a-z0-9-]*$", template):
+    if not re.match(r"^[a-z][a-z0-9-]*\Z", template):
         return (
             False,
             "Template name must start with a letter and contain only lowercase letters, numbers, and hyphens",
diff --git a/fluid_build/cli/import_workflow/dbt.py b/fluid_build/cli/import_workflow/dbt.py
index 34edaad3e..422d5f7a2 100644
--- a/fluid_build/cli/import_workflow/dbt.py
+++ b/fluid_build/cli/import_workflow/dbt.py
@@ -64,7 +64,7 @@
 _REF_RE = re.compile(r"ref\(\s*(?:['\"][^'\"]+['\"]\s*,\s*)?['\"]([^'\"]+)['\"]\s*\)")
 
 # FLUID identifier pattern (schema `$defs.identifier`).
-_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.-]*[A-Za-z0-9_]$|^[A-Za-z0-9_]$")
+_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.-]*[A-Za-z0-9_]\Z|^[A-Za-z0-9_]\Z")
 
 # dbt folder layer → medallion layer (mirrors engines/dbt/models.py).
 _LAYER_TO_MEDALLION = {"staging": "Bronze", "intermediate": "Silver", "marts": "Gold"}
diff --git a/fluid_build/cli/security.py b/fluid_build/cli/security.py
index 3e2f82f9d..d2add1e77 100644
--- a/fluid_build/cli/security.py
+++ b/fluid_build/cli/security.py
@@ -662,7 +662,7 @@ def validate_project_name(name: str) -> bool:
             return False
 
         # Check format: alphanumeric, hyphens, underscores only
-        if not re.match(r"^[a-zA-Z0-9_-]+$", name):
+        if not re.match(r"^[a-zA-Z0-9_-]+\Z", name):
             return False
 
         # Check length
diff --git a/fluid_build/copilot/catalog/_patterns.py b/fluid_build/copilot/catalog/_patterns.py
index 19edce90e..a38d47f81 100644
--- a/fluid_build/copilot/catalog/_patterns.py
+++ b/fluid_build/copilot/catalog/_patterns.py
@@ -143,7 +143,7 @@ def validate_and_quote_identifier(
         non-SQL-shaped identifiers.
     """
     rule = pattern or _DEFAULT_IDENTIFIER
-    if not rule.match(value or ""):
+    if not rule.fullmatch(value or ""):
         raise CatalogConfigError(
             message=f"Invalid {kind} identifier: {value!r}",
             suggestions=[
diff --git a/fluid_build/copilot/missions/spec.py b/fluid_build/copilot/missions/spec.py
index 879c019fc..efe81f744 100644
--- a/fluid_build/copilot/missions/spec.py
+++ b/fluid_build/copilot/missions/spec.py
@@ -70,7 +70,7 @@
 _BUDGET_KEYS = frozenset({"max_usd", "max_iterations", "max_wall_seconds"})
 
 #: One dotted-path segment: a key, optionally fanned out over an array.
-_SEGMENT_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_-]*)(\[\*\])?$")
+_SEGMENT_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_-]*)(\[\*\])?\Z")
 
 _MISSION_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]*$")
 
diff --git a/fluid_build/forge/core/bundle.py b/fluid_build/forge/core/bundle.py
index 3758ed067..ca4e757af 100644
--- a/fluid_build/forge/core/bundle.py
+++ b/fluid_build/forge/core/bundle.py
@@ -492,7 +492,7 @@ def bundle_overlay_warnings(source: Optional[Mapping[str, Any]]) -> List[str]:
 #: path. It selects the overlay file whose ``id`` the recorded source
 #: contract is checked with, so a value like ``../../x`` must not be able to
 #: pick an arbitrary file for that check.
-_ENV_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
+_ENV_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*\Z")
 
 
 def bundle_source_contract(
diff --git a/fluid_build/forge/federation.py b/fluid_build/forge/federation.py
index 76df5e109..62452fa5d 100644
--- a/fluid_build/forge/federation.py
+++ b/fluid_build/forge/federation.py
@@ -589,7 +589,7 @@ def from_dict(cls, d: Mapping[str, Any]) -> "FederatedWorkspace":
         # anything outside a conservative slug so a tampered manifest
         # row cannot smuggle ``../`` traversal or absolute segments into
         # those path sinks.
-        if not _WORKSPACE_ID_RE.match(ws_id):
+        if not _WORKSPACE_ID_RE.fullmatch(ws_id):
             raise ValueError(
                 f"federation workspace id {ws_id!r} is invalid: must match "
                 f"{_WORKSPACE_ID_RE.pattern} (alphanumerics, dot, dash, "
diff --git a/fluid_build/iac/providers/aws.py b/fluid_build/iac/providers/aws.py
index da0f890cc..a7cac3287 100644
--- a/fluid_build/iac/providers/aws.py
+++ b/fluid_build/iac/providers/aws.py
@@ -103,7 +103,7 @@
 # binding carries (a jurisdiction such as "EU", a Google region, an unresolved
 # "{{ env.AWS_REGION }}") must not become the provider's region; it used to be
 # ignored, and still is.
-_AWS_REGION_RE = re.compile(r"^[a-z]{2}(-[a-z]+)+-\d+$")
+_AWS_REGION_RE = re.compile(r"^[a-z]{2}(-[a-z]+)+-\d+\Z")
 
 
 def recorded_regions(resources: Iterable[Mapping[str, Any]]) -> set:
diff --git a/fluid_build/iac/providers/databricks_jobs.py b/fluid_build/iac/providers/databricks_jobs.py
index d78abac85..3efa289a0 100644
--- a/fluid_build/iac/providers/databricks_jobs.py
+++ b/fluid_build/iac/providers/databricks_jobs.py
@@ -123,9 +123,9 @@ def _under_home(prefix: str, suffix: str) -> TofuExpr:
     "dev.azure.com": "azureDevOpsServices",
 }
 _GIT_REFS = ("branch", "tag", "commit")
-_GIT_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$")
-_COMMIT_RE = re.compile(r"^[0-9a-fA-F]{7,40}$")
-_REPO_PATH_RE = re.compile(r"^[A-Za-z0-9._/-]+$")
+_GIT_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*\Z")
+_COMMIT_RE = re.compile(r"^[0-9a-fA-F]{7,40}\Z")
+_REPO_PATH_RE = re.compile(r"^[A-Za-z0-9._/-]+\Z")
 
 #: ``1h``, ``30m``, ``45s``, ``2d``, a bare number of seconds, or ISO 8601 ``PT1H30M``.
 _DURATION_RE = re.compile(r"^(\d+)\s*([smhd]?)$")
diff --git a/fluid_build/iac/row_access.py b/fluid_build/iac/row_access.py
index 9d758b24f..e234ac5b0 100644
--- a/fluid_build/iac/row_access.py
+++ b/fluid_build/iac/row_access.py
@@ -43,7 +43,7 @@
 from .base import UnsupportedBindingError
 from .column_access import policy_where
 
-_NAME = re.compile(r"^[a-z][a-z0-9_]{0,62}$")
+_NAME = re.compile(r"^[a-z][a-z0-9_]{0,62}\Z")
 
 
 @dataclass(frozen=True)
diff --git a/fluid_build/providers/aws/util/validation.py b/fluid_build/providers/aws/util/validation.py
index 2eb820e40..bfb436b8d 100644
--- a/fluid_build/providers/aws/util/validation.py
+++ b/fluid_build/providers/aws/util/validation.py
@@ -37,10 +37,10 @@ class ResourceValidator:
     """Validates AWS resource configurations before execution."""
 
     # AWS naming rules
-    S3_BUCKET_PATTERN = re.compile(r"^[a-z0-9][a-z0-9\-]{1,61}[a-z0-9]$")
-    GLUE_DATABASE_PATTERN = re.compile(r"^[a-z0-9_]+$")
-    GLUE_TABLE_PATTERN = re.compile(r"^[a-z0-9_]+$")
-    LAMBDA_FUNCTION_PATTERN = re.compile(r"^[a-zA-Z0-9\-_]+$")
+    S3_BUCKET_PATTERN = re.compile(r"^[a-z0-9][a-z0-9\-]{1,61}[a-z0-9]\Z")
+    GLUE_DATABASE_PATTERN = re.compile(r"^[a-z0-9_]+\Z")
+    GLUE_TABLE_PATTERN = re.compile(r"^[a-z0-9_]+\Z")
+    LAMBDA_FUNCTION_PATTERN = re.compile(r"^[a-zA-Z0-9\-_]+\Z")
 
     # AWS service quotas (defaults - can be increased)
     DEFAULT_QUOTAS = {
diff --git a/fluid_build/providers/bigquery_validation.py b/fluid_build/providers/bigquery_validation.py
index 4d6864e90..6482cb21c 100644
--- a/fluid_build/providers/bigquery_validation.py
+++ b/fluid_build/providers/bigquery_validation.py
@@ -43,7 +43,7 @@
 # need a dedicated allowlist: 6–30 chars, lowercase-or-mixed letter start,
 # letters/digits/hyphens in the middle, and a letter/digit terminator.
 # See https://cloud.google.com/resource-manager/docs/creating-managing-projects.
-_BQ_PROJECT_ID = re.compile(r"^[A-Za-z][A-Za-z0-9-]{4,28}[A-Za-z0-9]$")
+_BQ_PROJECT_ID = re.compile(r"^[A-Za-z][A-Za-z0-9-]{4,28}[A-Za-z0-9]\Z")
 
 
 def _quote_bq_project(project: str) -> str:
diff --git a/fluid_build/providers/gcp/plan/planner.py b/fluid_build/providers/gcp/plan/planner.py
index a4d4ce62b..91883c8aa 100644
--- a/fluid_build/providers/gcp/plan/planner.py
+++ b/fluid_build/providers/gcp/plan/planner.py
@@ -37,7 +37,7 @@
 # contain hyphens, so ``_sql_safety.validate_ident`` (which rejects ``-``)
 # cannot vet them — this dedicated check guards the project component of a
 # BigQuery fully-qualified name before it is interpolated into DDL/DML.
-_GCP_PROJECT_RE = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$")
+_GCP_PROJECT_RE = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]\Z")
 
 
 def _validate_bq_project(project: str) -> str:
diff --git a/fluid_build/providers/gcp/util/names.py b/fluid_build/providers/gcp/util/names.py
index 6a3f80fe5..c25db891b 100644
--- a/fluid_build/providers/gcp/util/names.py
+++ b/fluid_build/providers/gcp/util/names.py
@@ -324,7 +324,7 @@ def validate_name(name: str, resource_type: str) -> bool:
     try:
         if resource_type in ["dataset", "table"]:
             # BigQuery naming rules
-            if not re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", name):
+            if not re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*\Z", name):
                 return False
             return len(name) <= 1024
 
@@ -332,7 +332,7 @@ def validate_name(name: str, resource_type: str) -> bool:
             # Cloud Storage naming rules
             if len(name) < 3 or len(name) > 63:
                 return False
-            if not re.match(r"^[a-z0-9][a-z0-9\-]*[a-z0-9]$", name):
+            if not re.match(r"^[a-z0-9][a-z0-9\-]*[a-z0-9]\Z", name):
                 return False
             if re.match(r"^\d+\.\d+\.\d+\.\d+$", name):
                 return False
@@ -344,7 +344,7 @@ def validate_name(name: str, resource_type: str) -> bool:
                 return False
             if not name[0].isalpha():
                 return False
-            return re.match(r"^[a-zA-Z][a-zA-Z0-9\-\._~%+]*$", name) is not None
+            return re.match(r"^[a-zA-Z][a-zA-Z0-9\-\._~%+]*\Z", name) is not None
 
         elif resource_type == "job":
             # Job naming rules
@@ -354,7 +354,7 @@ def validate_name(name: str, resource_type: str) -> bool:
                 return False
             if name[-1] == "-":
                 return False
-            return re.match(r"^[a-z][a-z0-9\-]*[a-z0-9]$", name) is not None
+            return re.match(r"^[a-z][a-z0-9\-]*[a-z0-9]\Z", name) is not None
 
         else:
             # Unknown resource type
diff --git a/fluid_build/providers/quality_engine.py b/fluid_build/providers/quality_engine.py
index d4ba6191c..a3f8741d7 100644
--- a/fluid_build/providers/quality_engine.py
+++ b/fluid_build/providers/quality_engine.py
@@ -41,7 +41,7 @@
 LOG = logging.getLogger("fluid.providers.quality_engine")
 
 # Safe SQL identifier regex (letters, digits, underscores only)
-_SAFE_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+_SAFE_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*\Z")
 
 
 def _validate_ident(name: str) -> str:
diff --git a/fluid_build/providers/snowflake/util/config.py b/fluid_build/providers/snowflake/util/config.py
index 6a44ade67..e2fc67f9f 100644
--- a/fluid_build/providers/snowflake/util/config.py
+++ b/fluid_build/providers/snowflake/util/config.py
@@ -36,8 +36,8 @@
 # share the canonical pattern without redefining or importing a private name.
 ENV_TEMPLATE_RE = _ENV_TEMPLATE_RE
 _ACCOUNT_HOST_RE = re.compile(r"\.snowflakecomputing\.com$", re.IGNORECASE)
-_ACCOUNT_SEGMENT_RE = re.compile(r"^[A-Za-z0-9_-]+$")
-_ACCOUNT_REGION_SEGMENT_RE = re.compile(r"^[A-Za-z0-9]+(?:-[A-Za-z0-9]+)+$")
+_ACCOUNT_SEGMENT_RE = re.compile(r"^[A-Za-z0-9_-]+\Z")
+_ACCOUNT_REGION_SEGMENT_RE = re.compile(r"^[A-Za-z0-9]+(?:-[A-Za-z0-9]+)+\Z")
 _ACCOUNT_CLOUD_SUFFIXES = frozenset({"aws", "azure", "gcp"})
 
 # Keys that carry secret material. These MUST never appear in the `_sources`
diff --git a/fluid_build/schedulers/airflow/fluid_apply.py b/fluid_build/schedulers/airflow/fluid_apply.py
index 7a38496fe..685d547de 100644
--- a/fluid_build/schedulers/airflow/fluid_apply.py
+++ b/fluid_build/schedulers/airflow/fluid_apply.py
@@ -489,9 +489,9 @@ def validate_contract_path(raw: Any) -> str:
 def validate_id(raw: Any, *, kind: str) -> str:
     """:func:`validate_identifier`, refusing a trailing newline as well.
 
-    The shared grammar ends in ``$``, which also matches before a final
-    ``\\n``; an id or env that carries one names nothing fluid can find
-    (``--env 'aws\\n'`` silently applies no overlay).
+    An id or env that carries one names nothing fluid can find (``--env
+    'aws\\n'`` silently applies no overlay). The shared grammar ends in
+    ``\\Z`` and refuses one already; the check below still holds if it changes.
     """
     try:
         value = validate_identifier(raw, kind=kind)
diff --git a/tests/security/test_trailing_newline_anchors.py b/tests/security/test_trailing_newline_anchors.py
new file mode 100644
index 000000000..8505dfe67
--- /dev/null
+++ b/tests/security/test_trailing_newline_anchors.py
@@ -0,0 +1,372 @@
+# 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.
+
+"""Whole-string validators refuse a value that ends in a newline.
+
+Python's ``$`` matches at the end of the string and also just before a final
+``\\n``, so ``re.match(r"^[a-z]+$", "orders\\n")`` succeeds and the caller goes on
+to use ``"orders\\n"``: as a path segment, a SQL identifier or literal, a
+Terraform attribute, a cache directory. Contract YAML yields such values easily
+(a ``|`` block scalar ends in one), and so do ``.env`` files and percent-encoded
+URL queries. The validators below end in ``\\Z`` (or call ``fullmatch``), which
+matches only at the end of the string.
+
+Each test hands one validator a value with a trailing newline, next to the same
+value without it. It fails while that validator's pattern still ends in ``$``.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sqlite3
+from types import SimpleNamespace
+
+import pytest
+
+pytestmark = [pytest.mark.unit]
+
+
+# --- SQL identifiers and literals --------------------------------------------
+
+
+def test_quality_engine_refuses_a_selector_ending_in_a_newline():
+    from fluid_build.providers import quality_engine
+
+    assert '"amount"' in quality_engine._completeness_sql("t", "amount")
+    with pytest.raises(ValueError, match="Invalid SQL identifier"):
+        quality_engine._completeness_sql("t", "amount\n")
+
+
+def test_catalog_identifier_quoting_refuses_a_trailing_newline():
+    from fluid_build.copilot.catalog._patterns import validate_and_quote_identifier
+    from fluid_build.copilot.catalog.base import CatalogConfigError
+
+    assert validate_and_quote_identifier("SALES", kind="database") == '"SALES"'
+    with pytest.raises(CatalogConfigError):
+        validate_and_quote_identifier("SALES\n", kind="database")
+
+
+@pytest.mark.parametrize(
+    ("kind", "value"), [("project", "acme-proj"), ("dataset", "sales"), ("table", "orders")]
+)
+def test_bigquery_catalog_identifiers_refuse_a_trailing_newline(kind, value):
+    from fluid_build.copilot.catalog.base import CatalogConfigError
+    from fluid_build.copilot.catalog.bigquery import _quote_ident
+
+    assert _quote_ident(value, kind=kind) == f"`{value}`"
+    with pytest.raises(CatalogConfigError):
+        _quote_ident(f"{value}\n", kind=kind)
+
+
+def test_jdbc_schema_filter_refuses_a_trailing_newline(monkeypatch, tmp_path):
+    pytest.importorskip("duckdb")
+    from fluid_build.cli.discover import _jdbc_introspect
+    from fluid_build.providers import _duckdb_sandbox
+
+    def _no_connect(*_args, **_kwargs):
+        raise AssertionError("introspection connected with an invalid schema filter")
+
+    monkeypatch.setattr(_duckdb_sandbox, "secure_duckdb_connect", _no_connect)
+    db_path = tmp_path / "source.sqlite"
+    sqlite3.connect(str(db_path)).close()
+    with pytest.raises(ValueError, match="Invalid schema filter"):
+        _jdbc_introspect.introspect_jdbc(
+            source="sqlite", uri=f"sqlite:///{db_path}", schema_filter="main\n"
+        )
+
+
+def test_bigquery_project_quoting_refuses_a_trailing_newline():
+    from fluid_build.providers.bigquery_validation import _quote_bq_project
+
+    assert _quote_bq_project("acme-analytics") == "`acme-analytics`"
+    with pytest.raises(ValueError, match="Invalid BigQuery project id"):
+        _quote_bq_project("acme-analytics\n")
+
+
+def test_gcp_planner_project_refuses_a_trailing_newline():
+    from fluid_build.providers.gcp.plan.planner import _validate_bq_project
+
+    assert _validate_bq_project("acme-analytics") == "acme-analytics"
+    with pytest.raises(ValueError, match="Invalid GCP project ID"):
+        _validate_bq_project("acme-analytics\n")
+
+
+def test_snowflake_account_segment_refuses_a_trailing_newline():
+    from fluid_build.providers.snowflake.util import config
+
+    assert config._normalize_account_identifier("xy12345.eu-west-1") == "xy12345.eu-west-1"
+    with pytest.raises(ValueError, match="Invalid Snowflake account identifier"):
+        config._normalize_account_identifier("xy12345\n.eu-west-1")
+
+
+def test_snowflake_account_region_pattern_refuses_a_trailing_newline():
+    # The segment rule runs first and now refuses this input on its own; the
+    # region rule must hold without it.
+    from fluid_build.providers.snowflake.util import config
+
+    assert config._ACCOUNT_REGION_SEGMENT_RE.match("eu-west-1")
+    assert config._ACCOUNT_REGION_SEGMENT_RE.match("eu-west-1\n") is None
+
+
+# --- Names that become paths, files and cache directories ---------------------
+
+
+def test_build_identifier_refuses_a_trailing_newline():
+    from fluid_build.build_runners._ids import IdentifierViolation, validate_identifier
+
+    assert validate_identifier("orders", kind="contract.id") == "orders"
+    with pytest.raises(IdentifierViolation):
+        validate_identifier("orders\n", kind="contract.id")
+
+
+def test_federation_workspace_id_refuses_a_trailing_newline():
+    from fluid_build.forge.federation import FederatedWorkspace
+
+    row = {"id": "telco", "kind": "git_registry", "endpoint": "https://github.com/acme/telco"}
+    assert FederatedWorkspace.from_dict(row).id == "telco"
+    with pytest.raises(ValueError, match="federation workspace id"):
+        FederatedWorkspace.from_dict({**row, "id": "telco\n"})
+
+
+def test_bundle_recorded_env_refuses_a_trailing_newline(tmp_path):
+    from fluid_build.forge.core.bundle import resolve_bundle_source_contract
+
+    tgz = tmp_path / "bundle.tgz"
+    _, why = resolve_bundle_source_contract(tgz, {"contract": "contract.fluid.yaml", "env": "aws"})
+    assert "environment name" not in why
+    path, why = resolve_bundle_source_contract(
+        tgz, {"contract": "contract.fluid.yaml", "env": "aws\n"}
+    )
+    assert path is None
+    assert "is not an environment name" in why
+
+
+def test_consumes_overlay_env_refuses_a_trailing_newline():
+    from fluid_build.build_runners._embedded_sql_io import ConsumesResolutionError, _validated_env
+
+    assert _validated_env("aws") == "aws"
+    with pytest.raises(ConsumesResolutionError):
+        _validated_env("aws\n")
+
+
+@pytest.mark.skipif(os.name == "nt", reason="a Windows file name cannot hold a newline")
+@pytest.mark.parametrize(
+    ("resolver", "name"),
+    [("_resolve_tap_binary", "tap-postgres"), ("_resolve_target_binary", "target-jsonl")],
+)
+def test_singer_binary_lookup_refuses_a_name_ending_in_a_newline(
+    monkeypatch, tmp_path, resolver, name
+):
+    from fluid_build.build_runners.meltano import runner
+
+    for filename in (name, f"{name}\n"):
+        binary = tmp_path / filename
+        binary.write_text("#!/bin/sh\n", encoding="utf-8")
+        binary.chmod(0o755)
+    monkeypatch.setenv("PATH", str(tmp_path))
+    resolve = getattr(runner, resolver)
+
+    assert resolve(name) == str(tmp_path / name)
+    assert resolve(f"{name}\n") is None
+
+
+def test_prompt_profile_name_refuses_a_trailing_newline():
+    from fluid_build.cli.forge_copilot_prompts import PromptProfileError, _resolve_profile_dir
+
+    assert _resolve_profile_dir("eu-gdpr-strict").name == "eu-gdpr-strict"
+    with pytest.raises(PromptProfileError, match="invalid prompt profile name"):
+        _resolve_profile_dir("eu-gdpr-strict\n")
+
+
+def test_prompt_overlay_name_refuses_a_trailing_newline():
+    from fluid_build.cli.forge_prompt_overlays import PromptOverlayError, _resolve_overlay_path
+
+    assert _resolve_overlay_path("pii-lockdown").name == "pii-lockdown.yaml"
+    with pytest.raises(PromptOverlayError, match="invalid prompt overlay name"):
+        _resolve_overlay_path("pii-lockdown\n")
+
+
+# --- Identifiers written into a contract -------------------------------------
+
+
+def test_jdbc_forged_product_id_drops_a_trailing_newline(monkeypatch, tmp_path):
+    import yaml
+
+    from fluid_build.cli import _forge_data_model_jdbc
+    from fluid_build.cli.discover import _jdbc_introspect
+
+    table = _jdbc_introspect.IntrospectedTable(
+        schema="public",
+        name="orders",
+        columns=[_jdbc_introspect.IntrospectedColumn(name="id", type_name="integer")],
+    )
+    db = _jdbc_introspect.IntrospectedDatabase(
+        source_kind="postgres", database="orders", tables=[table]
+    )
+    monkeypatch.setattr(_jdbc_introspect, "introspect_jdbc", lambda **_kwargs: db)
+
+    def _forged_id(name):
+        out = tmp_path / "out.fluid.yaml"
+        args = SimpleNamespace(
+            source="postgres",
+            uri="postgresql://forge@db.example/orders",
+            schema_name=None,
+            tables=None,
+            name=name,
+            output=str(out),
+        )
+        assert _forge_data_model_jdbc._run_from_jdbc_source(args, logging.getLogger("test")) == 0
+        return yaml.safe_load(out.read_text(encoding="utf-8"))["id"]
+
+    assert _forged_id("orders") == "orders"
+    assert _forged_id("orders\n") == "orders"
+
+
+def test_dbt_importer_slugifies_an_identifier_ending_in_a_newline():
+    from fluid_build.cli.import_workflow.dbt import _safe_identifier
+
+    assert _safe_identifier("orders") == "orders"
+    assert _safe_identifier("orders\n") == "orders"
+
+
+def test_mission_predicate_path_refuses_a_trailing_newline():
+    from fluid_build.copilot.missions.spec import MissionSpecError, validate_predicate_path
+
+    validate_predicate_path("exposes[*].contract", label="checks[0]")
+    with pytest.raises(MissionSpecError, match="invalid path segment"):
+        validate_predicate_path("exposes[*].contract\n", label="checks[0]")
+
+
+# --- Values emitted into OpenTofu ---------------------------------------------
+
+
+def test_row_filter_name_refuses_a_trailing_newline():
+    from fluid_build.iac.base import UnsupportedBindingError
+    from fluid_build.iac.row_access import row_filters_for
+
+    def _expose(name):
+        row_filter = {
+            "principal": "group:analysts@northwind.example",
+            "name": name,
+            "where": "consent = true",
+        }
+        return {
+            "exposeId": "profile",
+            "contract": {"schema": [{"name": "consent", "type": "boolean"}]},
+            "policy": {"authz": {"rowFilters": [row_filter]}},
+        }
+
+    assert row_filters_for(_expose("analysts_consented"))[0].name == "analysts_consented"
+    with pytest.raises(UnsupportedBindingError, match="must be lowercase letters"):
+        row_filters_for(_expose("analysts_consented\n"))
+
+
+def _aws_contract(region):
+    location = {
+        "bucket": "acme-lake",
+        "path": "bronze/orders/",
+        "database": "bronze",
+        "table": "orders",
+        "region": region,
+    }
+    expose = {
+        "exposeId": "orders",
+        "binding": {"platform": "aws", "format": "parquet", "location": location},
+        "contract": {"schema": [{"name": "id", "type": "string"}]},
+    }
+    return {"id": "bronze.orders", "exposes": [expose]}
+
+
+def test_aws_binding_region_ending_in_a_newline_is_not_the_provider_region(monkeypatch):
+    from fluid_build.iac import get_iac_plugin, provider_config
+
+    monkeypatch.delenv("AWS_ENDPOINT_URL", raising=False)
+    plugin = get_iac_plugin("aws")
+
+    assert provider_config(plugin, _aws_contract("eu-west-1")) == {"region": "eu-west-1"}
+    assert provider_config(plugin, _aws_contract("eu-west-1\n")) == {}
+
+
+def test_aws_recorded_region_ending_in_a_newline_is_not_a_region():
+    from fluid_build.iac.providers.aws import recorded_regions
+
+    assert recorded_regions([{"values": {"region": "eu-west-1"}}]) == {"eu-west-1"}
+    assert recorded_regions([{"values": {"region": "eu-west-1\n"}}]) == set()
+
+
+@pytest.mark.parametrize(
+    "query", ["branch=main%0A", "tag=v1.2%0A", "commit=0a1b2c3%0A", "branch=main&path=dbt%0A"]
+)
+def test_databricks_git_source_refuses_a_ref_or_path_ending_in_a_newline(query):
+    from fluid_build.iac.providers.databricks_jobs import _git_source
+
+    clean, refused = [], []
+    repository = f"https://github.com/acme/jobs?{query}"
+    assert _git_source(repository.replace("%0A", ""), "builds[0]", clean) is not None
+    assert not clean
+    assert _git_source(repository, "builds[0]", refused) is None
+    assert refused
+
+
+# --- Name validators other code calls ---------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("method", "value"),
+    [
+        ("validate_s3_bucket_name", "acme-lake"),
+        ("validate_glue_database_name", "bronze"),
+        ("validate_glue_table_name", "orders"),
+        ("validate_lambda_function_name", "ingest-orders"),
+    ],
+)
+def test_aws_resource_names_refuse_a_trailing_newline(method, value):
+    from fluid_build.providers.aws.util.validation import ResourceValidator
+
+    check = getattr(ResourceValidator("123456789012", "eu-west-1"), method)
+    assert check(value) == (True, None)
+    assert check(f"{value}\n")[0] is False
+
+
+@pytest.mark.parametrize(
+    ("resource_type", "name"),
+    [("dataset", "sales"), ("bucket", "acme-lake"), ("topic", "orders-events"), ("job", "nightly")],
+)
+def test_gcp_resource_names_refuse_a_trailing_newline(resource_type, name):
+    from fluid_build.providers.gcp.util.names import validate_name
+
+    assert validate_name(name, resource_type) is True
+    assert validate_name(f"{name}\n", resource_type) is False
+
+
+def test_project_name_check_refuses_a_trailing_newline():
+    from fluid_build.cli.security import InputSanitizer
+
+    assert InputSanitizer.validate_project_name("my-project") is True
+    assert InputSanitizer.validate_project_name("my-project\n") is False
+
+
+def test_forge_project_name_refuses_a_trailing_newline():
+    from fluid_build.cli.forge_validation import validate_project_name
+
+    assert validate_project_name("orders-product") == (True, None)
+    assert validate_project_name("orders-product\n")[0] is False
+
+
+def test_forge_template_name_refuses_a_trailing_newline():
+    from fluid_build.cli.forge_validation import validate_template_name
+
+    assert validate_template_name("starter") == (True, None)
+    assert validate_template_name("starter\n")[0] is False

From 58f3719b621d713658b175d0918718930dda7328 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Tue, 3 Nov 2026 21:44:36 +0100
Subject: [PATCH 2/8] docs(changelog): 36 more validators refuse a value with a
 trailing newline

---
 CHANGELOG.md | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5afdd08dd..3e8636faa 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -207,6 +207,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `Invalid SQL identifier: 'spectrum_events\n'`. That includes a name that reaches it with a
   trailing newline from an environment variable or a file; strip the value where it is set.
 
+- **36 more validators let a value with a trailing newline through.** The same `$` ended other
+  whole-string patterns, so a value written as a YAML `|` block, read from a `.env` file or decoded
+  from `%0A` in a URL kept its newline: `fluid generate iac` put `region = "eu-west-1\n"` on the
+  AWS provider, and a Databricks job got `branch = "main\n"` from `?branch=main%0A`. They now end
+  in `\Z` (two call sites use `fullmatch`), so such a value is refused with the validator's usual
+  message, or takes the path the code already has for a value that is not a plain name: an AWS
+  binding region is left to the environment, and a product id forged from a JDBC source or
+  imported from dbt is slugified. Covered: quality-check selectors, catalog and JDBC identifiers,
+  BigQuery and GCP project ids, contract, build, federation-workspace and overlay-env ids,
+  row-filter names, Singer tap and target names, prompt profile and overlay names, mission
+  predicate paths, AWS and GCP resource names, Snowflake account segments and Databricks git refs.
+
 ## [0.26.0] — 2026-11-01
 
 ### Added

From ce7956f339065cc3718fe8dd8b7c618a92d78a60 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Tue, 3 Nov 2026 21:51:00 +0100
Subject: [PATCH 3/8] fix(iac): refuse an AWS binding region with whitespace
 around it

With `\Z`, provider_block_for dropped a binding region of
"eu-west-1\n" as no region code. That left the provider on the
environment's region and skipped _guard_region_move. Pinning the
stripped value is no better. Both sovereignty checkers
(policy/sovereignty.py, providers/aws/util/sovereignty.py) compare
the region as written, so "us-east-1\n" passes
`deniedRegions: [us-east-1]` while the provider would be pinned to
us-east-1.

provider_block_for now raises UnsupportedBindingError
(aws-region-whitespace) for a region code with whitespace around it.
generate iac and apply render it as a typed CLI error before anything
is applied, and fluid diff reports the inspector error. On main the
value went onto the provider with its newline.
---
 CHANGELOG.md                                  | 14 ++++++-----
 fluid_build/iac/providers/aws.py              | 23 +++++++++++++++++++
 .../security/test_trailing_newline_anchors.py | 14 +++++++----
 3 files changed, 41 insertions(+), 10 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3e8636faa..c9467813f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -212,12 +212,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   from `%0A` in a URL kept its newline: `fluid generate iac` put `region = "eu-west-1\n"` on the
   AWS provider, and a Databricks job got `branch = "main\n"` from `?branch=main%0A`. They now end
   in `\Z` (two call sites use `fullmatch`), so such a value is refused with the validator's usual
-  message, or takes the path the code already has for a value that is not a plain name: an AWS
-  binding region is left to the environment, and a product id forged from a JDBC source or
-  imported from dbt is slugified. Covered: quality-check selectors, catalog and JDBC identifiers,
-  BigQuery and GCP project ids, contract, build, federation-workspace and overlay-env ids,
-  row-filter names, Singer tap and target names, prompt profile and overlay names, mission
-  predicate paths, AWS and GCP resource names, Snowflake account segments and Databricks git refs.
+  message, or takes the path the code already has for a value that is not a plain name: a product
+  id forged from a JDBC source or imported from dbt is slugified. An AWS binding region is read
+  without the newline, as `fluid verify` already reads it, so it is still the provider's region
+  rather than being dropped in favour of the environment's. Covered: quality-check selectors,
+  catalog and JDBC identifiers, BigQuery and GCP project ids, contract, build,
+  federation-workspace and overlay-env ids, row-filter names, Singer tap and target names, prompt
+  profile and overlay names, mission predicate paths, AWS and GCP resource names, Snowflake
+  account segments and Databricks git refs.
 
 ## [0.26.0] — 2026-11-01
 
diff --git a/fluid_build/iac/providers/aws.py b/fluid_build/iac/providers/aws.py
index a7cac3287..d9c04dfc1 100644
--- a/fluid_build/iac/providers/aws.py
+++ b/fluid_build/iac/providers/aws.py
@@ -735,6 +735,29 @@ def provider_block_for(self, contract: Mapping[str, Any]) -> Dict[str, Any]:
         provider v6).
         """
         cfg = dict(self.provider_block())
+        for index, expose in enumerate(contract.get("exposes") or []):
+            binding = expose.get("binding") or {}
+            raw = (binding.get("location") or {}).get("region")
+            # A region code with whitespace around it (a YAML ``|`` block ends in a
+            # newline) is refused. Dropping it leaves the provider on the
+            # environment's region; pinning it stripped applies a region the
+            # sovereignty policy, which reads the value as written, never checked.
+            if (
+                is_cloud(binding, "aws")
+                and isinstance(raw, str)
+                and raw != raw.strip()
+                and _AWS_REGION_RE.match(raw.strip())
+            ):
+                raise UnsupportedBindingError(
+                    "aws-region-whitespace",
+                    f"exposes[{expose.get('exposeId') or index}] binding.location.region is "
+                    f"{raw!r}: a region code with whitespace around it. The sovereignty policy "
+                    "checks the region as written, so it is refused rather than pinned stripped.",
+                    (
+                        f"Write the region alone: region: {raw.strip()} (a YAML '|' block ends "
+                        "the value in a newline).",
+                    ),
+                )
         named = {
             str(((e.get("binding") or {}).get("location") or {}).get("region"))
             for e in contract.get("exposes") or []
diff --git a/tests/security/test_trailing_newline_anchors.py b/tests/security/test_trailing_newline_anchors.py
index 8505dfe67..637fd3a1c 100644
--- a/tests/security/test_trailing_newline_anchors.py
+++ b/tests/security/test_trailing_newline_anchors.py
@@ -289,14 +289,20 @@ def _aws_contract(region):
     return {"id": "bronze.orders", "exposes": [expose]}
 
 
-def test_aws_binding_region_ending_in_a_newline_is_not_the_provider_region(monkeypatch):
+def test_aws_binding_region_ending_in_a_newline_is_refused(monkeypatch):
+    # Neither dropped (the provider would take the environment's region) nor
+    # stripped (sovereignty checks the value as written, so a stripped pin
+    # could apply a region its deniedRegions names).
     from fluid_build.iac import get_iac_plugin, provider_config
+    from fluid_build.iac.base import UnsupportedBindingError
 
-    monkeypatch.delenv("AWS_ENDPOINT_URL", raising=False)
+    for name in [k for k in os.environ if k.startswith("AWS_ENDPOINT_URL")]:
+        monkeypatch.delenv(name)
     plugin = get_iac_plugin("aws")
 
-    assert provider_config(plugin, _aws_contract("eu-west-1")) == {"region": "eu-west-1"}
-    assert provider_config(plugin, _aws_contract("eu-west-1\n")) == {}
+    assert provider_config(plugin, _aws_contract("eu-west-1"))["region"] == "eu-west-1"
+    with pytest.raises(UnsupportedBindingError, match="whitespace around it"):
+        provider_config(plugin, _aws_contract("eu-west-1\n"))
 
 
 def test_aws_recorded_region_ending_in_a_newline_is_not_a_region():

From 50a522f0db8e0b4f643c15417fe622c58fa765e9 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Tue, 3 Nov 2026 22:03:00 +0100
Subject: [PATCH 4/8] security: match the BigQuery catalog and imported-schema
 patterns whole

copilot/catalog/bigquery.py::_validate_ident checked the same `$`
patterns with `.match` that the quoting chokepoint now checks with
`fullmatch`. So `scope.tables` values and FQN parts such as
"orders\n" reached `table_name IN ('...')`. The ODCS and ODPS
importers compile the bundled schema's identifier and semver
patterns, which end in `$`, and also used `.match`: an imported id
"orders\n" was kept verbatim, and "1.2.3\n" counted as semver. All
three now call `fullmatch`.

The two patterns that keep `$` because their call sites use
`fullmatch` (`_DEFAULT_IDENTIFIER`, `_WORKSPACE_ID_RE`) now say so
in a comment beside them.
---
 fluid_build/copilot/catalog/_patterns.py      |  3 +-
 fluid_build/copilot/catalog/bigquery.py       |  4 ++-
 fluid_build/forge/federation.py               |  1 +
 .../providers/odcs/mappers/normalize.py       |  2 +-
 .../providers/odps_standard/provider.py       |  2 +-
 .../security/test_trailing_newline_anchors.py | 28 +++++++++++++++++++
 6 files changed, 36 insertions(+), 4 deletions(-)

diff --git a/fluid_build/copilot/catalog/_patterns.py b/fluid_build/copilot/catalog/_patterns.py
index a38d47f81..7c443ce30 100644
--- a/fluid_build/copilot/catalog/_patterns.py
+++ b/fluid_build/copilot/catalog/_patterns.py
@@ -99,7 +99,8 @@
 
 # Conservative default identifier shape — covers Snowflake / BigQuery /
 # Unity / Glue without quoting weirdness. Catalogs with looser rules
-# (e.g., DataHub URN strings) override the regex per-adapter.
+# (e.g., DataHub URN strings) override the regex per-adapter. Matched
+# with ``fullmatch``: ``$`` also matches before a trailing newline.
 _DEFAULT_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
 
 
diff --git a/fluid_build/copilot/catalog/bigquery.py b/fluid_build/copilot/catalog/bigquery.py
index bfaba7e61..aa9de5af3 100644
--- a/fluid_build/copilot/catalog/bigquery.py
+++ b/fluid_build/copilot/catalog/bigquery.py
@@ -86,6 +86,8 @@
 #
 # The pattern table here keeps both rules visible — adding a new
 # identifier kind (e.g., a connection name) just adds one entry.
+# Match these with ``fullmatch``: their ``$`` also matches before a
+# trailing newline, so ``.match`` lets ``"orders\n"`` through.
 _BQ_IDENT_PATTERNS = {
     # Allow hyphens for project IDs.
     "project": re.compile(r"^[A-Za-z][A-Za-z0-9-]*$"),
@@ -113,7 +115,7 @@ def _quote_ident(value: str, *, kind: str) -> str:
 def _validate_ident(value: str, *, kind: str) -> str:
     """Return a BigQuery identifier after strict kind-specific validation."""
     pattern = _BQ_IDENT_PATTERNS.get(kind)
-    if pattern is None or not pattern.match(value or ""):
+    if pattern is None or not pattern.fullmatch(value or ""):
         raise CatalogConfigError(
             message=f"Invalid BigQuery {kind} identifier: {value!r}",
             suggestions=[
diff --git a/fluid_build/forge/federation.py b/fluid_build/forge/federation.py
index 62452fa5d..4a1c7086f 100644
--- a/fluid_build/forge/federation.py
+++ b/fluid_build/forge/federation.py
@@ -236,6 +236,7 @@ def _federation_git_timeout() -> float:
 # (``~/.cache/fluid/federation-git/<id>`` and the digest-cache file).
 # Constrain it to a conservative slug so a tampered manifest row can't
 # smuggle ``../`` traversal or absolute-path segments into those sinks.
+# Matched with ``fullmatch``: ``$`` also matches before a trailing newline.
 _WORKSPACE_ID_RE = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
 
 
diff --git a/fluid_build/providers/odcs/mappers/normalize.py b/fluid_build/providers/odcs/mappers/normalize.py
index 94ff451dc..b7e028deb 100644
--- a/fluid_build/providers/odcs/mappers/normalize.py
+++ b/fluid_build/providers/odcs/mappers/normalize.py
@@ -462,7 +462,7 @@ def _root_identifier(raw: Any, version: str) -> tuple[str, Optional[str]]:
         return _IMPORTED_ID_FALLBACK, None
     text = str(raw)
     pattern = _identifier_pattern(version)
-    if pattern is None or pattern.match(text):
+    if pattern is None or pattern.fullmatch(text):
         return text, None
     slug = _slugged_root_id(text)
     # %r: the id is document text and may carry control characters.
diff --git a/fluid_build/providers/odps_standard/provider.py b/fluid_build/providers/odps_standard/provider.py
index e03e17edd..c03a07c50 100644
--- a/fluid_build/providers/odps_standard/provider.py
+++ b/fluid_build/providers/odps_standard/provider.py
@@ -1141,7 +1141,7 @@ def _schema_defs(version: str) -> Mapping[str, Any]:
 
 def _is_semver(value: Any, version: str) -> bool:
     pattern = _semver_pattern(version)
-    return pattern is None or bool(pattern.match(str(value)))
+    return pattern is None or bool(pattern.fullmatch(str(value)))
 
 
 def _rehydrate_document(fluid: Mapping[str, Any]) -> Mapping[str, Any]:
diff --git a/tests/security/test_trailing_newline_anchors.py b/tests/security/test_trailing_newline_anchors.py
index 637fd3a1c..dc3814c57 100644
--- a/tests/security/test_trailing_newline_anchors.py
+++ b/tests/security/test_trailing_newline_anchors.py
@@ -70,6 +70,19 @@ def test_bigquery_catalog_identifiers_refuse_a_trailing_newline(kind, value):
         _quote_ident(f"{value}\n", kind=kind)
 
 
+@pytest.mark.parametrize(
+    ("kind", "value"), [("project", "acme-proj"), ("dataset", "sales"), ("table", "orders")]
+)
+def test_bigquery_catalog_name_check_refuses_a_trailing_newline(kind, value):
+    # The unquoted check behind `table_name IN ('...')` and FQN parsing.
+    from fluid_build.copilot.catalog.base import CatalogConfigError
+    from fluid_build.copilot.catalog.bigquery import _validate_ident
+
+    assert _validate_ident(value, kind=kind) == value
+    with pytest.raises(CatalogConfigError):
+        _validate_ident(f"{value}\n", kind=kind)
+
+
 def test_jdbc_schema_filter_refuses_a_trailing_newline(monkeypatch, tmp_path):
     pytest.importorskip("duckdb")
     from fluid_build.cli.discover import _jdbc_introspect
@@ -241,6 +254,21 @@ def test_dbt_importer_slugifies_an_identifier_ending_in_a_newline():
     assert _safe_identifier("orders\n") == "orders"
 
 
+def test_odcs_import_slugifies_an_id_ending_in_a_newline():
+    # The bundled schema's identifier pattern ends in `$`; it is matched whole.
+    from fluid_build.providers.odcs.mappers.normalize import _root_identifier
+
+    assert _root_identifier("orders", "0.7.5") == ("orders", None)
+    assert _root_identifier("orders\n", "0.7.5")[0] == "orders"
+
+
+def test_odps_import_semver_check_refuses_a_trailing_newline():
+    from fluid_build.providers.odps_standard.provider import _is_semver
+
+    assert _is_semver("1.2.3", "0.7.5") is True
+    assert _is_semver("1.2.3\n", "0.7.5") is False
+
+
 def test_mission_predicate_path_refuses_a_trailing_newline():
     from fluid_build.copilot.missions.spec import MissionSpecError, validate_predicate_path
 

From 27c051830b009f9a68973f9d8fea739ef2fc6c42 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Tue, 3 Nov 2026 22:03:00 +0100
Subject: [PATCH 5/8] fix: show the refused newline in AWS name and mission
 path messages

These messages quoted the value as '{value}', so a name refused for a
trailing newline printed as a valid-looking name split over two
lines. They now use repr, which prints any other value exactly as
before.
---
 fluid_build/copilot/missions/spec.py            |  2 +-
 fluid_build/providers/aws/util/validation.py    |  8 ++++----
 tests/security/test_trailing_newline_anchors.py | 11 +++++++----
 3 files changed, 12 insertions(+), 9 deletions(-)

diff --git a/fluid_build/copilot/missions/spec.py b/fluid_build/copilot/missions/spec.py
index efe81f744..64c435977 100644
--- a/fluid_build/copilot/missions/spec.py
+++ b/fluid_build/copilot/missions/spec.py
@@ -172,7 +172,7 @@ def validate_predicate_path(path: str, *, label: str) -> None:
     for segment in path.split("."):
         if not _SEGMENT_RE.match(segment):
             raise MissionSpecError(
-                f"{label}: invalid path segment '{segment}' in '{path}'. The predicate "
+                f"{label}: invalid path segment {segment!r} in {path!r}. The predicate "
                 "grammar is frozen: dotted keys with optional [*] array fan-out "
                 "(e.g. exposes[*].contract.dq.rules)."
             )
diff --git a/fluid_build/providers/aws/util/validation.py b/fluid_build/providers/aws/util/validation.py
index bfb436b8d..38a7bb770 100644
--- a/fluid_build/providers/aws/util/validation.py
+++ b/fluid_build/providers/aws/util/validation.py
@@ -91,7 +91,7 @@ def validate_s3_bucket_name(self, bucket: str) -> Tuple[bool, Optional[str]]:
 
         if not self.S3_BUCKET_PATTERN.match(bucket):
             return False, (
-                f"Bucket name '{bucket}' is invalid. "
+                f"Bucket name {bucket!r} is invalid. "
                 "Must contain only lowercase letters, numbers, and hyphens. "
                 "Must start and end with letter or number."
             )
@@ -135,7 +135,7 @@ def validate_glue_database_name(self, database: str) -> Tuple[bool, Optional[str
 
         if not self.GLUE_DATABASE_PATTERN.match(database):
             return False, (
-                f"Database name '{database}' is invalid. "
+                f"Database name {database!r} is invalid. "
                 "Must contain only lowercase letters, numbers, and underscores."
             )
 
@@ -164,7 +164,7 @@ def validate_glue_table_name(self, table: str) -> Tuple[bool, Optional[str]]:
 
         if not self.GLUE_TABLE_PATTERN.match(table):
             return False, (
-                f"Table name '{table}' is invalid. "
+                f"Table name {table!r} is invalid. "
                 "Must contain only lowercase letters, numbers, and underscores."
             )
 
@@ -193,7 +193,7 @@ def validate_lambda_function_name(self, function: str) -> Tuple[bool, Optional[s
 
         if not self.LAMBDA_FUNCTION_PATTERN.match(function):
             return False, (
-                f"Function name '{function}' is invalid. "
+                f"Function name {function!r} is invalid. "
                 "Must contain only letters, numbers, hyphens, and underscores."
             )
 
diff --git a/tests/security/test_trailing_newline_anchors.py b/tests/security/test_trailing_newline_anchors.py
index dc3814c57..ae60fc3f6 100644
--- a/tests/security/test_trailing_newline_anchors.py
+++ b/tests/security/test_trailing_newline_anchors.py
@@ -272,9 +272,10 @@ def test_odps_import_semver_check_refuses_a_trailing_newline():
 def test_mission_predicate_path_refuses_a_trailing_newline():
     from fluid_build.copilot.missions.spec import MissionSpecError, validate_predicate_path
 
-    validate_predicate_path("exposes[*].contract", label="checks[0]")
-    with pytest.raises(MissionSpecError, match="invalid path segment"):
-        validate_predicate_path("exposes[*].contract\n", label="checks[0]")
+    # A stripped path can still carry one at the end of an inner segment.
+    validate_predicate_path("exposes[*].contract.dq", label="checks[0]")
+    with pytest.raises(MissionSpecError, match=r"invalid path segment 'contract\\n'"):
+        validate_predicate_path("exposes[*].contract\n.dq", label="checks[0]")
 
 
 # --- Values emitted into OpenTofu ---------------------------------------------
@@ -371,7 +372,9 @@ def test_aws_resource_names_refuse_a_trailing_newline(method, value):
 
     check = getattr(ResourceValidator("123456789012", "eu-west-1"), method)
     assert check(value) == (True, None)
-    assert check(f"{value}\n")[0] is False
+    valid, message = check(f"{value}\n")
+    assert valid is False
+    assert f"{value}\\n" in message  # the refused newline is shown, not printed
 
 
 @pytest.mark.parametrize(

From f9997075a83f7c9858828f35f2e445cb38b1ea56 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Tue, 3 Nov 2026 22:03:00 +0100
Subject: [PATCH 6/8] chore: keep the domain slug patterns in step with the
 profile-name guard

The comment on forge_domain_enrichment._SAFE_DOMAIN_RE says it
matches forge_copilot_prompts._PROFILE_NAME_RE, and the comment on
_ai_setup_storage._DOMAIN_SLUG_RE says it mirrors _SAFE_DOMAIN_RE.
The profile-name guard now ends in `\Z`, so both do too. Neither is
reachable with a trailing newline: one runs after a disk lookup, the
other on stripped input.
---
 fluid_build/cli/_ai_setup_storage.py            |  2 +-
 fluid_build/cli/forge_domain_enrichment.py      |  2 +-
 tests/security/test_trailing_newline_anchors.py | 17 +++++++++++++++++
 3 files changed, 19 insertions(+), 2 deletions(-)

diff --git a/fluid_build/cli/_ai_setup_storage.py b/fluid_build/cli/_ai_setup_storage.py
index ecef764be..7d74ff9dd 100644
--- a/fluid_build/cli/_ai_setup_storage.py
+++ b/fluid_build/cli/_ai_setup_storage.py
@@ -115,7 +115,7 @@ def _plaintext_env_var() -> str:
 # Privacy guard: only a bare domain slug is ever persisted. Mirrors
 # ``forge_domain_enrichment._SAFE_DOMAIN_RE``, lower-cased and length-capped —
 # the regex is what keeps prompt text / paths / PII out of the history file.
-_DOMAIN_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
+_DOMAIN_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}\Z")
 
 
 def _allow_plaintext_ai_secrets() -> bool:
diff --git a/fluid_build/cli/forge_domain_enrichment.py b/fluid_build/cli/forge_domain_enrichment.py
index ba581b4cc..04d9620e3 100644
--- a/fluid_build/cli/forge_domain_enrichment.py
+++ b/fluid_build/cli/forge_domain_enrichment.py
@@ -47,7 +47,7 @@
 # override — defence in depth against a ``../``-style traversal reaching a
 # fragment block in an out-of-tree YAML. Matches the profile-name guard in
 # ``forge_copilot_prompts._PROFILE_NAME_RE``.
-_SAFE_DOMAIN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+_SAFE_DOMAIN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*\Z")
 
 
 @lru_cache(maxsize=1)
diff --git a/tests/security/test_trailing_newline_anchors.py b/tests/security/test_trailing_newline_anchors.py
index ae60fc3f6..a697e8e92 100644
--- a/tests/security/test_trailing_newline_anchors.py
+++ b/tests/security/test_trailing_newline_anchors.py
@@ -203,6 +203,23 @@ def test_prompt_profile_name_refuses_a_trailing_newline():
         _resolve_profile_dir("eu-gdpr-strict\n")
 
 
+@pytest.mark.parametrize(
+    ("module", "name"),
+    [
+        ("fluid_build.cli.forge_domain_enrichment", "_SAFE_DOMAIN_RE"),
+        ("fluid_build.cli._ai_setup_storage", "_DOMAIN_SLUG_RE"),
+    ],
+)
+def test_domain_slug_patterns_refuse_a_trailing_newline(module, name):
+    # Fed a stripped or on-disk value, so pinned on the pattern: their comments
+    # say they mirror the profile-name guard, and they now do.
+    import importlib
+
+    pattern = getattr(importlib.import_module(module), name)
+    assert pattern.match("finance")
+    assert pattern.match("finance\n") is None
+
+
 def test_prompt_overlay_name_refuses_a_trailing_newline():
     from fluid_build.cli.forge_prompt_overlays import PromptOverlayError, _resolve_overlay_path
 

From 8b7a6601ebfab9957af2651f6c06ce431843854a Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Tue, 3 Nov 2026 22:03:00 +0100
Subject: [PATCH 7/8] docs: drop the validator count and reword validate_id's
 summary

The count in the changelog entry depended on how patterns were
grouped. The entry now lists what is covered instead, including the
AWS region refusal and the imports. validate_id's summary said it
refused a trailing newline "as well", which the shared grammar now
does itself.
---
 CHANGELOG.md                                  | 18 +++++++++---------
 fluid_build/schedulers/airflow/fluid_apply.py |  4 ++--
 2 files changed, 11 insertions(+), 11 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index c9467813f..ee003020d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -207,19 +207,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `Invalid SQL identifier: 'spectrum_events\n'`. That includes a name that reaches it with a
   trailing newline from an environment variable or a file; strip the value where it is set.
 
-- **36 more validators let a value with a trailing newline through.** The same `$` ended other
+- **Other validators let a value with a trailing newline through, too.** The same `$` ended other
   whole-string patterns, so a value written as a YAML `|` block, read from a `.env` file or decoded
   from `%0A` in a URL kept its newline: `fluid generate iac` put `region = "eu-west-1\n"` on the
   AWS provider, and a Databricks job got `branch = "main\n"` from `?branch=main%0A`. They now end
-  in `\Z` (two call sites use `fullmatch`), so such a value is refused with the validator's usual
+  in `\Z` or are matched with `fullmatch`, so such a value is refused with the validator's usual
   message, or takes the path the code already has for a value that is not a plain name: a product
-  id forged from a JDBC source or imported from dbt is slugified. An AWS binding region is read
-  without the newline, as `fluid verify` already reads it, so it is still the provider's region
-  rather than being dropped in favour of the environment's. Covered: quality-check selectors,
-  catalog and JDBC identifiers, BigQuery and GCP project ids, contract, build,
-  federation-workspace and overlay-env ids, row-filter names, Singer tap and target names, prompt
-  profile and overlay names, mission predicate paths, AWS and GCP resource names, Snowflake
-  account segments and Databricks git refs.
+  id forged from a JDBC source or imported from dbt or ODCS is slugified. An AWS binding region
+  with whitespace around it is refused (`aws-region-whitespace`), because the sovereignty policy
+  checks the region as written. Covered: quality-check selectors, catalog and JDBC identifiers,
+  BigQuery and GCP project ids, contract, build, federation-workspace and overlay-env ids,
+  row-filter names, Singer tap and target names, prompt profile and overlay names, mission
+  predicate paths, AWS and GCP resource names, Snowflake account segments, Databricks git refs,
+  and the identifier and version checks of ODCS and ODPS imports.
 
 ## [0.26.0] — 2026-11-01
 
diff --git a/fluid_build/schedulers/airflow/fluid_apply.py b/fluid_build/schedulers/airflow/fluid_apply.py
index 685d547de..5c38c18bd 100644
--- a/fluid_build/schedulers/airflow/fluid_apply.py
+++ b/fluid_build/schedulers/airflow/fluid_apply.py
@@ -487,9 +487,9 @@ def validate_contract_path(raw: Any) -> str:
 
 
 def validate_id(raw: Any, *, kind: str) -> str:
-    """:func:`validate_identifier`, refusing a trailing newline as well.
+    """:func:`validate_identifier` as a :class:`ScheduleRenderError`, with its own newline check.
 
-    An id or env that carries one names nothing fluid can find (``--env
+    An id or env that ends in a newline names nothing fluid can find (``--env
     'aws\\n'`` silently applies no overlay). The shared grammar ends in
     ``\\Z`` and refuses one already; the check below still holds if it changes.
     """

From f4f5b74fe6e2d018f12f685b2aaf95e45751d078 Mon Sep 17 00:00:00 2001
From: Speculator55005 <50082482+fas89@users.noreply.github.com>
Date: Tue, 3 Nov 2026 22:21:28 +0100
Subject: [PATCH 8/8] test: skip the BigQuery validation pin without the GCP
 SDK

bigquery_validation imports google.cloud.bigquery at module load. The
iac Stage 1 unit lane installs no GCP extra, so the new pin failed
collection there with ModuleNotFoundError. It now skips via
pytest.importorskip, the guard tests/providers/test_validation_sql_injection.py
already applies to the same module.
---
 tests/security/test_trailing_newline_anchors.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/tests/security/test_trailing_newline_anchors.py b/tests/security/test_trailing_newline_anchors.py
index a697e8e92..445209b87 100644
--- a/tests/security/test_trailing_newline_anchors.py
+++ b/tests/security/test_trailing_newline_anchors.py
@@ -101,6 +101,9 @@ def _no_connect(*_args, **_kwargs):
 
 
 def test_bigquery_project_quoting_refuses_a_trailing_newline():
+    # bigquery_validation imports google.cloud.bigquery at module load, and the
+    # GCP SDK is a provider extra the minimal unit lanes do not install.
+    pytest.importorskip("google.cloud.bigquery")
     from fluid_build.providers.bigquery_validation import _quote_bq_project
 
     assert _quote_bq_project("acme-analytics") == "`acme-analytics`"