fix(pii): require a whole-output proof before masking instead of refusing

CHANGELOG.md

@@ -14,6 +14,28 @@ below corresponds to one such version.
 
 ### Security
 
+- **Gated PII masking on a whole-output proof; an output that cannot be proven clean is
+  now refused.** The mask decision quantified over the projections the detector *found*
+  (`all_maskable`), which proves "every offending projection we saw is maskable" but never
+  "no sensitive value reaches the output". While any detected projection refused the whole
+  query that gap was harmless; once a maskable projection let the query **run**, one seen
+  column unblocked the entire result set and every co-projected value the name-based
+  detector missed came back raw, beside a `***` implying the row had been protected.
+  `SensitiveCheckResult.output_provable` is the missing half: it is False when an
+  output-bearing arm draws through a source whose body is not walked (a derived table, a CTE
+  the arm selects from, a `VALUES`/`LATERAL` source), projects a scalar subquery, or projects
+  a qualifier naming none of that arm's own sources. `certainty="provable"` now requires it,
+  so an unprovable output falls back to a refusal. Two further binding bugs made the proof
+  itself unsound and are closed with it: a subquery alias could **shadow** an outer alias of
+  the same name (`FROM customers t` outside, `FROM orders t` inside an `EXISTS` body), silently
+  rebinding the outer `t.ssn` to a table declaring no `ssn` so it was never detected while the
+  qualifier still "resolved"; and two aliases folding to the same name (`FROM x "AB", customers
+  ab` with `AB.ssn`) resolved by exact-match-first to the quoted one, where the engine folds
+  onto the other. Alias binding is now shadow-aware and fails closed on a genuinely ambiguous
+  fold. The proof is scoped to what can actually put a value in the output, so a subquery in
+  `WHERE`/`HAVING`/`ORDER BY`, a JOIN's `ON` condition, or a CTE no arm selects from no longer
+  downgrades an ordinary maskable query to a refusal.
+
 - **Closed a read-only-guard bypass via a welded quoted identifier.** A double-quoted
   identifier is self-delimiting in SQL on **both** ends, so `SELECT*FROM"pg_read_file"(…)`
   and `SELECT "x"INTO evil FROM t` are valid statements with no whitespace either side of

packages/agami-core/src/execute_sql.py

@@ -1296,14 +1296,27 @@ def _model_safety(
     mask_plan: MaskPlan | None = None
     sens = RT.check_sensitive_projection(sql, org, ctx=ctx)
     if sens.action == "refuse":
+        # Two conditions, and BOTH are required. `all_maskable` quantifies over the projections
+        # the detector FOUND, so on its own it proves only "every offending projection we saw is
+        # maskable" — never "no sensitive value reaches the output". Under the earlier refuse-only
+        # behaviour that gap leaked only when nothing was detected (anything detected blocked the
+        # whole query); once a maskable projection let the query RUN, one seen column unblocked the
+        # entire result set and every co-projected value the name-based detector missed came back
+        # raw — beside a `***` implying the row had been protected. `output_provable` is the
+        # missing half: the detector could see the whole output, not just part of it.
         all_maskable = bool(sens.projections) and all(
             p.maskable and p.output_index is not None for p in sens.projections
         )
+        # `getattr`, not attribute access: this file is VENDORED into the plugin while
+        # semantic_model/runtime.py resolves from the separately-versioned installed package, so a
+        # newer plugin can meet an older `SensitiveCheckResult`. A missing field then means "this
+        # build cannot prove the output", which is exactly the fail-closed answer.
+        provable = all_maskable and getattr(sens, "output_provable", False)
         verdict = Verdict(
             cls="data_protection",
             rule="sensitive_projection",
             severity="high",
-            certainty="provable" if all_maskable else "uncertain",
+            certainty="provable" if provable else "uncertain",
             detail=sens.reason,
             remediation=sens.suggestion or "",
         )

packages/agami-core/src/semantic_model/runtime.py

@@ -31,6 +31,7 @@
 from __future__ import annotations
 
 import re
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from difflib import SequenceMatcher
 from typing import Any, Callable, Optional
@@ -457,11 +458,23 @@ class SensitiveCheckResult:
     reason: str = ""
     suggestion: Optional[str] = None
     # Per-offending-projection traceability (ACE-041): the maskable-vs-must-refuse decision and, for
-    # maskable ones, the output-column index. Latent in THIS slice (the caller still refuses on any
-    # sensitive projection); a later masking slice reads it. Deterministic order: arm order, then
-    # projection index; duplicates and multi-arm collisions on `output_index` are preserved (unlike
-    # `columns`, which is the de-duplicated sorted set the refuse message is built from).
+    # maskable ones, the output-column index. LOAD-BEARING, not informational — `_model_safety`
+    # reads `maskable` / `output_index` off this list to build `MaskPlan.indices`, i.e. the exact
+    # output columns that get redacted, so order and duplicates are significant and must not be
+    # re-sorted or de-duplicated. Deterministic order: arm order, then projection index; duplicates
+    # and multi-arm collisions on `output_index` are preserved (unlike `columns`, which is the
+    # de-duplicated sorted set the refuse message is built from).
     projections: list["SensitiveProjection"] = field(default_factory=list)
+    # Whether the WHOLE output was resolvable, not just whether each FOUND projection was
+    # maskable (ACE-041 review). `projections` can only describe what the detector saw; this
+    # says whether it could have seen everything. False when an output-bearing arm draws through
+    # a source whose body is not walked (a derived table, a CTE the arm selects from, or a
+    # non-table source such as VALUES / LATERAL), or projects a scalar subquery, or projects a
+    # column whose qualifier names none of that arm's own sources — i.e. exactly where the
+    # name-based match is blind. See `_output_lineage_provable` for why a bare column needs no
+    # such test. Masking a partially-understood output is worse than refusing it: the caller gets
+    # a `***` that certifies the other columns.
+    output_provable: bool = False
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -470,6 +483,7 @@ def as_dict(self) -> dict[str, Any]:
             "reason": self.reason,
             "suggestion": self.suggestion,
             "projections": [p.as_dict() for p in self.projections],
+            "output_provable": self.output_provable,
         }
 
 
@@ -521,6 +535,108 @@ def _direct_from_tables(tree: "exp.Select") -> set[str]:
     return names
 
 
+def _arm_sources(sel: "exp.Select") -> list["exp.Expression"]:
+    """The FROM / JOIN clauses belonging to THIS select, never a nested one.
+
+    Read off `sel`'s own args by NODE TYPE rather than by arg key: sqlglot renamed the FROM key
+    (`from` -> `from_`) inside the `sqlglot>=20` range this package accepts, and a missing key
+    would silently degrade to "no sources", which reads as provable. Matching on `exp.From` /
+    `exp.Join` is stable across that range."""
+    out: list["exp.Expression"] = []
+    for value in sel.args.values():
+        for item in value if isinstance(value, list) else [value]:
+            if isinstance(item, (exp.From, exp.Join)):
+                out.append(item)
+    return out
+
+
+def _direct_scope(sel: "exp.Select") -> dict[str, str]:
+    """alias (or table name) -> table, for THIS arm's own FROM/JOIN sources only.
+
+    Used by the whole-output proof to decide whether a projected qualifier names one of the
+    sources the arm actually draws from. Deliberately narrower than `_sensitive_scope`."""
+    scope: dict[str, str] = {}
+    for src in _arm_sources(sel):
+        node = src.this
+        if isinstance(node, exp.Table):
+            scope[node.alias_or_name] = node.name
+    return scope
+
+
+def _sensitive_scope(sel: "exp.Select") -> dict[str, Optional[str]]:
+    """alias -> table for the sensitive gate, with a SHADOWED alias resolved to None.
+
+    Same whole-subtree walk as `_tables_in_scope`, so a bare column still resolves through a
+    derived-table or CTE body exactly as it did before. The one difference is what happens when a
+    single alias is bound to two different tables — `FROM customers t` outside and `FROM orders t`
+    inside an EXISTS body:
+
+        SELECT c.ssn, t.ssn FROM customers c, customers t WHERE EXISTS (SELECT 1 FROM orders t)
+
+    A last-one-wins map silently rebinds the OUTER `t.ssn` to `orders`, which declares no `ssn`,
+    so the column reads as un-sensitive and is never detected — while the qualifier still
+    "resolves", so the output reads as provable and the query masks one column and hands back its
+    twin raw. A subquery alias is scoped to its own body and must not capture an outer qualifier.
+    Mapping the conflict to None makes the binding UNRESOLVED, which `_sensitive_col_ref` already
+    handles by falling back to the conservative folded bare name (offending, and maskable)."""
+    bound: dict[str, set[str]] = {}
+    for tbl in sel.find_all(exp.Table):
+        bound.setdefault(tbl.alias_or_name, set()).add(tbl.name)
+    return {alias: (next(iter(t)) if len(t) == 1 else None) for alias, t in bound.items()}
+
+
+def _output_lineage_provable(tree: "exp.Expression") -> bool:
+    """Could the sensitive-projection detector have seen EVERY value reaching the output?
+
+    The detector matches sensitive columns by name against the tables in scope. That is sound
+    only while every output value traces back to a declared table it can name. Two things break
+    the trace:
+
+      * an OUTPUT-BEARING source whose body is not walked — a derived table, a CTE the arm
+        selects from, or a non-table source (VALUES / LATERAL / table function). `(SELECT ssn
+        FROM customers) t` presents an opaque `t` whose columns cannot be attributed, and a
+        rename inside makes even the name useless;
+      * a qualifier that names nothing among the arm's own sources, so the column cannot be
+        bound to a table at all.
+
+    Both tests are deliberately scoped to what can actually PUT A VALUE IN THE OUTPUT: an arm's
+    own FROM/JOIN sources, and its projection list. A subquery in WHERE / HAVING / ORDER BY, or a
+    JOIN's ON condition, yields a filter rather than a column, and a CTE no arm selects from
+    contributes nothing — withholding the proof for those refuses ordinary analytics SQL and buys
+    no safety. (This mirrors the distinction `check_column_scope` already draws.)
+
+    A BARE column needs no test here. The sensitive match is by NAME against the whole model, so
+    it does not depend on which table the column binds to: either the name is declared sensitive
+    somewhere and `_sensitive_col_ref` flags it (falling back to the folded bare name when the
+    table is ambiguous), or it is declared sensitive nowhere and no binding could make it
+    sensitive. Refusing on ambiguity alone would reject every unqualified column in a join.
+
+    Returning False here does NOT refuse the query. It only withholds the *proof* that the output
+    is fully understood, which downgrades a mask decision to a refusal. This runs only after an
+    offending projection has already been found, so a query that projects nothing sensitive never
+    reaches it — see the caller."""
+    cte_names = {cte.alias_or_name.lower() for cte in tree.find_all(exp.CTE)}
+    for sel in _output_selects(tree):
+        for src in _arm_sources(sel):
+            node = src.this
+            if not isinstance(node, exp.Table):
+                return False  # derived table / VALUES / LATERAL / table function: body not walked
+            if node.name.lower() in cte_names:
+                return False  # a CTE body is not walked either, and a rename inside hides the name
+        scope = _direct_scope(sel)
+        # Case-fold the scope keys: SQL identifiers are case-insensitive unless quoted, so
+        # `FROM customers C` + `c.ssn` is one table, and a case-sensitive miss would silently
+        # read as "not a sensitive column" rather than "could not resolve".
+        lowered = {alias.lower() for alias in scope}
+        for proj in sel.expressions:
+            if proj.find(exp.Subquery) is not None:
+                return False  # a scalar subquery in the PROJECTION does reach the output
+            for col in proj.find_all(exp.Column):
+                if col.table and col.table.lower() not in lowered:
+                    return False  # qualifier names nothing among this arm's own sources
+    return True
+
+
 def _output_selects(node: "exp.Expression") -> list["exp.Select"]:
     """The SELECTs whose projection reaches the query OUTPUT: the top-level SELECT, or
     — for a set operation (UNION / INTERSECT / EXCEPT) — every arm.
@@ -553,7 +669,7 @@ def _output_selects(node: "exp.Expression") -> list["exp.Select"]:
 
 def _sensitive_col_ref(
     col: "exp.Column",
-    scope: dict[str, str],
+    scope: Mapping[str, Optional[str]],
     by_table: dict[str, dict[str, str]],
     allnames: set[str],
 ) -> Optional[str]:
@@ -579,7 +695,7 @@ def _sensitive_col_ref(
 def _classify_projection(
     proj: "exp.Expression",
     index: int,
-    scope: dict[str, str],
+    scope: Mapping[str, Optional[str]],
     direct: set[str],
     by_table: dict[str, dict[str, str]],
     allnames: set[str],
@@ -664,7 +780,10 @@ def check_sensitive_projection(
     offending: set[str] = set()
     projections: list[SensitiveProjection] = []
     for sel in _output_selects(tree):
-        scope = _tables_in_scope(sel)
+        # `_sensitive_scope`, NOT `_tables_in_scope`: a last-one-wins alias map lets an alias
+        # declared inside a subquery overwrite the outer alias of the same name, which binds an
+        # outer qualified column to the wrong table and hides it from the match entirely.
+        scope = _sensitive_scope(sel)
         direct = _direct_from_tables(sel)
         for index, proj in enumerate(sel.expressions):
             _classify_projection(
@@ -683,6 +802,7 @@ def check_sensitive_projection(
         suggestion="Aggregate it (e.g. COUNT(DISTINCT <col>)) for a count, or omit it and "
         "select the entity's non-sensitive key (e.g. id) instead.",
         projections=projections,
+        output_provable=_output_lineage_provable(tree),
     )
 
 
@@ -1689,9 +1809,32 @@ def collect(node):
     return referenced
 
 
-def _resolve_col_table(col: "exp.Column", scope: dict[str, str]) -> Optional[str]:
+def _resolve_col_table(col: "exp.Column", scope: Mapping[str, Optional[str]]) -> Optional[str]:
+    """The table `col` binds to, or None when that cannot be decided.
+
+    `scope` values are Optional because the sensitive gate passes `_sensitive_scope`, which maps a
+    SHADOWED alias (one bound to two different tables) to None on purpose. `Mapping` rather than
+    `dict` so the fan/chasm and aggregation callers can keep passing a plain `dict[str, str]` — a
+    Mapping's value type is covariant, a dict's is not."""
     if col.table:
-        return scope.get(col.table, col.table)
+        # SQL identifiers are case-insensitive unless quoted, so `FROM customers C` and `c.ssn`
+        # name the same table. A case-sensitive miss does not read as "unresolved" downstream —
+        # it falls through to the raw qualifier, which the sensitive gate then fails to match
+        # against its table index and treats as NOT sensitive. ACE-041 slice 3 (`4bd39bb`) folded
+        # the COLUMN name for exactly this reason; the alias lookup is the other half of that fix.
+        #
+        # Resolve by FOLDED match rather than exact-match-first. When two aliases in scope fold
+        # together — `FROM x "AB", customers ab` with `AB.ssn` — an exact hit on the quoted `"AB"`
+        # would bind to `x`, while the engine folds the unquoted `AB` onto `ab` and reads
+        # `customers`. The binding is genuinely ambiguous, so return None (unresolved) and let the
+        # caller fail closed; picking either one is a coin flip that can read as un-sensitive.
+        folded = col.table.lower()
+        matches = {name for alias, name in scope.items() if alias.lower() == folded}
+        if len(matches) == 1:
+            return next(iter(matches))
+        if matches:
+            return None  # ambiguous binding -> unresolved, never a guess
+        return col.table
     # unqualified column: ambiguous; only safe to attribute if single table
     if len(scope) == 1:
         return next(iter(scope.values()))

plugins/agami/lib/execute_sql.py

@@ -1296,14 +1296,27 @@ def _model_safety(
     mask_plan: MaskPlan | None = None
     sens = RT.check_sensitive_projection(sql, org, ctx=ctx)
     if sens.action == "refuse":
+        # Two conditions, and BOTH are required. `all_maskable` quantifies over the projections
+        # the detector FOUND, so on its own it proves only "every offending projection we saw is
+        # maskable" — never "no sensitive value reaches the output". Under the earlier refuse-only
+        # behaviour that gap leaked only when nothing was detected (anything detected blocked the
+        # whole query); once a maskable projection let the query RUN, one seen column unblocked the
+        # entire result set and every co-projected value the name-based detector missed came back
+        # raw — beside a `***` implying the row had been protected. `output_provable` is the
+        # missing half: the detector could see the whole output, not just part of it.
         all_maskable = bool(sens.projections) and all(
             p.maskable and p.output_index is not None for p in sens.projections
         )
+        # `getattr`, not attribute access: this file is VENDORED into the plugin while
+        # semantic_model/runtime.py resolves from the separately-versioned installed package, so a
+        # newer plugin can meet an older `SensitiveCheckResult`. A missing field then means "this
+        # build cannot prove the output", which is exactly the fail-closed answer.
+        provable = all_maskable and getattr(sens, "output_provable", False)
         verdict = Verdict(
             cls="data_protection",
             rule="sensitive_projection",
             severity="high",
-            certainty="provable" if all_maskable else "uncertain",
+            certainty="provable" if provable else "uncertain",
             detail=sens.reason,
             remediation=sens.suggestion or "",
         )

tests/e2e/harness.py

@@ -120,7 +120,7 @@ def seed_db_model(url: str, ds: str = "acme") -> None:
 
     s = Store.connect(url)
     s.run_migrations()
-    model_store.write_organization(s, ds, build_org())
+    model_store.write_datasource(s, ds, build_org())
     s.close()