669 lines
29 KiB
Python
669 lines
29 KiB
Python
#!/usr/bin/env python3
|
|
"""Derive Leo's canonical PostgreSQL read surface from its runtime SQL.
|
|
|
|
The runtime queries are the source of truth. This helper extracts canonical
|
|
``public`` and ``kb_stage`` relation aliases and their qualified column
|
|
references directly from ``cloudsql_memory_tool.py``. It deliberately rejects
|
|
unaliased canonical relations, alias collisions, direct canonical DML, and
|
|
unqualified identifiers that are not mechanically known query outputs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
CANONICAL_RELATION = re.compile(
|
|
r"\b(?:from|join)\s+(?:only\s+)?(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)",
|
|
re.IGNORECASE,
|
|
)
|
|
ALIASED_CANONICAL_RELATION = re.compile(
|
|
r"\b(?:from|join)\s+(?:only\s+)?(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)"
|
|
r"\s+as\s+([a-z_][a-z0-9_]*)",
|
|
re.IGNORECASE,
|
|
)
|
|
MUTATING_SQL_KEYWORD = re.compile(
|
|
r"\b(?:insert|update|delete|truncate|merge)\b",
|
|
re.IGNORECASE,
|
|
)
|
|
LARGE_OBJECT_MUTATOR_REFERENCE = re.compile(
|
|
r'(?<![a-z0-9_])(?:(?:"pg_catalog"|pg_catalog)\s*\.\s*)?'
|
|
r'(?:(?:"lo_creat"|lo_creat)|(?:"lo_create"|lo_create)|'
|
|
r'(?:"lo_export"|lo_export)|(?:"lo_from_bytea"|lo_from_bytea)|'
|
|
r'(?:"lo_import"|lo_import)|(?:"lo_open"|lo_open)|'
|
|
r'(?:"lo_put"|lo_put)|(?:"lo_truncate64"|lo_truncate64)|'
|
|
r'(?:"lo_truncate"|lo_truncate)|(?:"lo_unlink"|lo_unlink)|'
|
|
r'(?:"lowrite"|lowrite))\s*\(',
|
|
re.IGNORECASE,
|
|
)
|
|
PSQL_META_COMMAND = re.compile(r"\\")
|
|
CANONICAL_NAMESPACE_REFERENCE = re.compile(r"\b(?:public|kb_stage)\s*\.", re.IGNORECASE)
|
|
QUOTED_SQL_IDENTIFIER = re.compile(r'"(?:""|[^"])*"')
|
|
DYNAMIC_SQL_VALUE = "__teleo_dynamic_sql_value__"
|
|
UNREVIEWED_DYNAMIC_SQL = "__teleo_unreviewed_dynamic_sql__"
|
|
REVIEWED_SQL_INTERPOLATION_CALLS = frozenset({"sql_array", "sql_integer", "sql_json_array", "sql_literal"})
|
|
QUERY_RUNNERS = frozenset({"run_psql", "psql_json_lines"})
|
|
REVIEWED_RUNTIME_SOURCE_SHA256 = "5f98218622803e0a1bbedc1e23e538d1f152ecb516e31a387a5eb39080aaf2cd"
|
|
DYNAMIC_CALL_BUILTINS = frozenset({"eval", "exec", "globals", "locals"})
|
|
READ_CAPABILITY = "read"
|
|
STAGE_PROPOSAL_CAPABILITY = "stage-proposal"
|
|
SQL_IDENTIFIER = re.compile(r"\b[a-z_][a-z0-9_]*\b", re.IGNORECASE)
|
|
SQL_STRUCTURE_TOKEN = re.compile(r"[a-z_][a-z0-9_]*|[(),.;]", re.IGNORECASE)
|
|
FROM_CLAUSE_END_KEYWORDS = frozenset(
|
|
{"except", "fetch", "for", "group", "having", "intersect", "limit", "offset", "order", "returning", "union", "where", "window"}
|
|
)
|
|
SQL_KEYWORDS = frozenset(
|
|
"""
|
|
all and any array as asc between bigint bool boolean by case cast cross
|
|
current_database current_user desc distinct else end exists false filter
|
|
first following for from full group having ilike in inner interval is join
|
|
json jsonb last lateral left like limit not null nulls offset on or order
|
|
only outer over partition preceding range recursive right row rows select some
|
|
symmetric table text then ties time timestamp timestamptz to true union
|
|
unknown using uuid values varchar when where window with zone
|
|
""".split() # noqa: SIM905 - the compact vocabulary is easier to audit as SQL.
|
|
)
|
|
NONCANONICAL_QUERY_IDENTIFIERS = frozenset(
|
|
{
|
|
# The status query reads these PostgreSQL metadata columns; they are not
|
|
# part of the canonical application-role contract.
|
|
"extname",
|
|
"pg_extension",
|
|
"table_schema",
|
|
"table_type",
|
|
}
|
|
)
|
|
UNQUALIFIED_NONCANONICAL_RELATIONS = frozenset({"pg_extension", "pg_tables"})
|
|
|
|
|
|
class QueryContractError(ValueError):
|
|
"""The runtime SQL cannot be safely reduced to an exact read contract."""
|
|
|
|
|
|
def _assert_exact_runtime_source(source: str) -> None:
|
|
observed = hashlib.sha256(source.encode("utf-8")).hexdigest()
|
|
if observed != REVIEWED_RUNTIME_SOURCE_SHA256:
|
|
raise QueryContractError(
|
|
"runtime query surface differs from the exact reviewed runtime source: "
|
|
f"expected={REVIEWED_RUNTIME_SOURCE_SHA256}, observed={observed}"
|
|
)
|
|
|
|
|
|
def _is_reviewed_sql_interpolation(node: ast.expr) -> bool:
|
|
return (
|
|
isinstance(node, ast.Call)
|
|
and isinstance(node.func, ast.Name)
|
|
and node.func.id in REVIEWED_SQL_INTERPOLATION_CALLS
|
|
)
|
|
|
|
|
|
def _render_sql_expression(node: ast.expr) -> str | None:
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str):
|
|
return node.value
|
|
if isinstance(node, ast.JoinedStr):
|
|
parts: list[str] = []
|
|
for value in node.values:
|
|
if isinstance(value, ast.Constant) and isinstance(value.value, str):
|
|
parts.append(value.value)
|
|
elif isinstance(value, ast.FormattedValue) and _is_reviewed_sql_interpolation(value.value):
|
|
# Runtime interpolation supplies values, limits, or predicates;
|
|
# canonical relation and column identifiers remain static.
|
|
parts.append(DYNAMIC_SQL_VALUE)
|
|
else:
|
|
parts.append(UNREVIEWED_DYNAMIC_SQL)
|
|
return "".join(parts)
|
|
return None
|
|
|
|
|
|
def _statement_path(
|
|
node: ast.AST,
|
|
function: ast.FunctionDef | ast.AsyncFunctionDef,
|
|
parents: dict[ast.AST, ast.AST],
|
|
) -> tuple[tuple[ast.AST, str, int, ast.stmt], ...]:
|
|
"""Return the ordered statement blocks containing ``node``."""
|
|
|
|
path: list[tuple[ast.AST, str, int, ast.stmt]] = []
|
|
current = node
|
|
while current is not function:
|
|
parent = parents.get(current)
|
|
if parent is None:
|
|
break
|
|
if isinstance(current, ast.stmt):
|
|
for field, value in ast.iter_fields(parent):
|
|
if isinstance(value, list) and current in value:
|
|
path.append((parent, field, value.index(current), current))
|
|
break
|
|
current = parent
|
|
return tuple(reversed(path))
|
|
|
|
|
|
def _ancestor_chain(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> tuple[ast.AST, ...]:
|
|
ancestors: list[ast.AST] = []
|
|
current = node
|
|
while current in parents:
|
|
current = parents[current]
|
|
ancestors.append(current)
|
|
return tuple(ancestors)
|
|
|
|
|
|
def _assignment_dominates_call(
|
|
assignment: ast.stmt,
|
|
call: ast.Call,
|
|
function: ast.FunctionDef | ast.AsyncFunctionDef,
|
|
parents: dict[ast.AST, ast.AST],
|
|
) -> bool:
|
|
"""Return whether one unconditional earlier assignment reaches ``call``."""
|
|
|
|
assignment_path = _statement_path(assignment, function, parents)
|
|
call_path = _statement_path(call, function, parents)
|
|
for index, assignment_entry in enumerate(assignment_path):
|
|
if index >= len(call_path):
|
|
return False
|
|
assignment_owner, assignment_field, assignment_index, assignment_statement = assignment_entry
|
|
call_owner, call_field, call_index, call_statement = call_path[index]
|
|
if assignment_owner is not call_owner or assignment_field != call_field:
|
|
return False
|
|
if assignment_index < call_index:
|
|
# An assignment hidden inside an earlier conditional does not
|
|
# dominate code after that conditional. The assignment itself must
|
|
# be the direct statement that precedes the call's branch.
|
|
return index == len(assignment_path) - 1
|
|
if assignment_index > call_index or assignment_statement is not call_statement:
|
|
return False
|
|
return False
|
|
|
|
|
|
def _reject_query_runner_indirection(tree: ast.AST, parents: dict[ast.AST, ast.AST]) -> None:
|
|
"""Require every runtime query runner reference to be a direct call."""
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Call) and isinstance(node.func, (ast.Call, ast.Lambda, ast.Subscript)):
|
|
raise QueryContractError(f"line {node.lineno}: indirect call expressions are not auditable")
|
|
if isinstance(node, ast.Name) and node.id in DYNAMIC_CALL_BUILTINS and isinstance(node.ctx, ast.Load):
|
|
raise QueryContractError(f"line {node.lineno}: dynamic call indirection is not auditable")
|
|
if isinstance(node, ast.Attribute) and node.attr in DYNAMIC_CALL_BUILTINS:
|
|
raise QueryContractError(f"line {node.lineno}: dynamic call indirection is not auditable")
|
|
if isinstance(node, ast.Name) and node.id in QUERY_RUNNERS and isinstance(node.ctx, ast.Load):
|
|
parent = parents.get(node)
|
|
if not (isinstance(parent, ast.Call) and parent.func is node):
|
|
raise QueryContractError(
|
|
f"line {node.lineno}: query runner {node.id!r} must be called directly"
|
|
)
|
|
elif isinstance(node, ast.Attribute) and node.attr in QUERY_RUNNERS:
|
|
raise QueryContractError(
|
|
f"line {node.lineno}: query runner {node.attr!r} must be called directly"
|
|
)
|
|
elif isinstance(node, ast.Constant) and node.value in QUERY_RUNNERS:
|
|
raise QueryContractError(
|
|
f"line {node.lineno}: query runner {node.value!r} cannot be resolved through indirection"
|
|
)
|
|
|
|
|
|
def _is_reviewed_psql_json_lines_forwarder(
|
|
call: ast.Call,
|
|
function: ast.FunctionDef | ast.AsyncFunctionDef,
|
|
) -> bool:
|
|
"""Recognize the one query-transparent ``psql_json_lines`` forwarding call."""
|
|
|
|
if function.name != "psql_json_lines":
|
|
return False
|
|
if not (isinstance(call.func, ast.Name) and call.func.id == "run_psql"):
|
|
return False
|
|
if len(call.args) != 2 or not all(isinstance(argument, ast.Name) for argument in call.args):
|
|
return False
|
|
if [argument.id for argument in call.args] != ["args", "sql"]:
|
|
return False
|
|
keywords = {keyword.arg: keyword.value for keyword in call.keywords if keyword.arg is not None}
|
|
if len(keywords) != len(call.keywords) or set(keywords) != {"db", "capability"}:
|
|
return False
|
|
return all(
|
|
isinstance(keywords[name], ast.Name) and keywords[name].id == name
|
|
for name in ("db", "capability")
|
|
)
|
|
|
|
|
|
def extract_runtime_query_literals(
|
|
source: str,
|
|
*,
|
|
enforce_source_pin: bool = True,
|
|
) -> tuple[tuple[int, str, str], ...]:
|
|
"""Return every statically rendered SQL statement and its capability."""
|
|
|
|
tree = ast.parse(source)
|
|
parents = {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)}
|
|
_reject_query_runner_indirection(tree, parents)
|
|
queries: set[tuple[int, str, str]] = set()
|
|
for function in (node for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))):
|
|
function_calls = [node for node in ast.walk(function) if isinstance(node, ast.Call)]
|
|
reviewed_forwarders = [
|
|
call for call in function_calls if _is_reviewed_psql_json_lines_forwarder(call, function)
|
|
]
|
|
if function.name == "psql_json_lines" and len(reviewed_forwarders) != 1:
|
|
raise QueryContractError(
|
|
f"line {function.lineno}: psql_json_lines must contain exactly one reviewed run_psql forwarder"
|
|
)
|
|
assignments: dict[str, list[tuple[ast.stmt, str | None]]] = defaultdict(list)
|
|
for node in ast.walk(function):
|
|
if isinstance(node, ast.Assign):
|
|
rendered = _render_sql_expression(node.value)
|
|
for target in node.targets:
|
|
if isinstance(target, ast.Name):
|
|
assignments[target.id].append((node, rendered))
|
|
elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
|
|
assignments[node.target.id].append((node, _render_sql_expression(node.value)))
|
|
elif isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name):
|
|
assignments[node.target.id].append((node, None))
|
|
elif isinstance(node, ast.NamedExpr) and isinstance(node.target, ast.Name):
|
|
# Named expressions are deliberately not an accepted SQL
|
|
# construction surface; record them as an inexact write.
|
|
statement = next(
|
|
(ancestor for ancestor in _ancestor_chain(node, parents) if isinstance(ancestor, ast.stmt)),
|
|
function,
|
|
)
|
|
assignments[node.target.id].append((statement, None))
|
|
|
|
for call in sorted(
|
|
function_calls,
|
|
key=lambda node: (node.lineno, node.col_offset),
|
|
):
|
|
runner = call.func.id if isinstance(call.func, ast.Name) else None
|
|
if runner not in QUERY_RUNNERS:
|
|
continue
|
|
if _is_reviewed_psql_json_lines_forwarder(call, function):
|
|
continue
|
|
if len(call.args) < 2:
|
|
raise QueryContractError(f"line {call.lineno}: {runner} requires a statically inspectable SQL argument")
|
|
capability_keywords = [keyword for keyword in call.keywords if keyword.arg == "capability"]
|
|
if len(capability_keywords) > 1:
|
|
raise QueryContractError(f"line {call.lineno}: {runner} has ambiguous SQL capabilities")
|
|
capability = READ_CAPABILITY
|
|
if capability_keywords:
|
|
capability_expression = capability_keywords[0].value
|
|
if (
|
|
isinstance(capability_expression, ast.Name)
|
|
and capability_expression.id == "RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY"
|
|
) or (
|
|
isinstance(capability_expression, ast.Constant)
|
|
and capability_expression.value == STAGE_PROPOSAL_CAPABILITY
|
|
):
|
|
capability = STAGE_PROPOSAL_CAPABILITY
|
|
else:
|
|
raise QueryContractError(
|
|
f"line {call.lineno}: {runner} SQL capability must remain statically reviewed"
|
|
)
|
|
expression = call.args[1]
|
|
if isinstance(expression, ast.Name):
|
|
writes = assignments.get(expression.id, [])
|
|
if len(writes) != 1:
|
|
raise QueryContractError(
|
|
f"line {call.lineno}: {runner} SQL variable {expression.id!r} must have exactly one assignment"
|
|
)
|
|
assignment, sql = writes[0]
|
|
if not _assignment_dominates_call(assignment, call, function, parents):
|
|
raise QueryContractError(
|
|
f"line {call.lineno}: {runner} SQL assignment must unambiguously precede the call"
|
|
)
|
|
line = assignment.lineno
|
|
else:
|
|
line, sql = call.lineno, _render_sql_expression(expression)
|
|
if sql is None:
|
|
raise QueryContractError(f"line {call.lineno}: {runner} SQL must remain a local string literal")
|
|
scrubbed = _normalize_sql_identifier_syntax(sql, line)
|
|
_reject_unqualified_relation_references(scrubbed, line)
|
|
_reject_comma_separated_base_relations(scrubbed, line)
|
|
if UNREVIEWED_DYNAMIC_SQL in sql:
|
|
raise QueryContractError(
|
|
f"line {line}: dynamic SQL interpolation is not part of the reviewed runtime query surface"
|
|
)
|
|
if PSQL_META_COMMAND.search(scrubbed):
|
|
raise QueryContractError(
|
|
f"line {line}: psql meta-commands are not part of the reviewed runtime query surface"
|
|
)
|
|
if MUTATING_SQL_KEYWORD.search(scrubbed) and CANONICAL_NAMESPACE_REFERENCE.search(scrubbed):
|
|
raise QueryContractError(f"line {line}: runtime query performs direct canonical DML")
|
|
if LARGE_OBJECT_MUTATOR_REFERENCE.search(scrubbed):
|
|
raise QueryContractError(
|
|
f"line {line}: runtime query invokes a provider-owned large-object mutator"
|
|
)
|
|
queries.add((line, sql, capability))
|
|
|
|
if not queries:
|
|
raise QueryContractError("runtime source contains no SQL query literals")
|
|
if enforce_source_pin:
|
|
_assert_exact_runtime_source(source)
|
|
return tuple(sorted(queries, key=lambda item: item[0]))
|
|
|
|
|
|
def extract_canonical_query_literals(
|
|
source: str,
|
|
*,
|
|
enforce_source_pin: bool = True,
|
|
) -> tuple[tuple[int, str], ...]:
|
|
"""Return canonical SQL literals actually passed to the runtime query runners."""
|
|
|
|
queries: list[tuple[int, str]] = []
|
|
for line, sql, _capability in extract_runtime_query_literals(source, enforce_source_pin=enforce_source_pin):
|
|
normalized = _normalize_sql_identifier_syntax(sql, line)
|
|
if CANONICAL_RELATION.search(normalized):
|
|
queries.append((line, normalized))
|
|
if not queries:
|
|
raise QueryContractError("runtime source contains no canonical SQL query literals")
|
|
return tuple(queries)
|
|
|
|
|
|
def _strip_sql_strings_and_comments(sql: str) -> str:
|
|
without_strings = re.sub(r"(?:[eE])?'(?:''|[^'])*'", " ", sql)
|
|
without_block_comments = re.sub(r"/\*.*?\*/", " ", without_strings, flags=re.DOTALL)
|
|
return re.sub(r"--[^\n]*", " ", without_block_comments)
|
|
|
|
|
|
def _normalize_sql_identifier_syntax(sql: str, line: int) -> str:
|
|
"""Return structural SQL with exact, lowercase quoted identifiers normalized."""
|
|
|
|
scrubbed = _strip_sql_strings_and_comments(sql)
|
|
|
|
def normalize_quoted_identifier(match: re.Match[str]) -> str:
|
|
identifier = match.group(0)[1:-1].replace('""', '"')
|
|
if (
|
|
not SQL_IDENTIFIER.fullmatch(identifier)
|
|
or identifier != identifier.lower()
|
|
or identifier in SQL_KEYWORDS
|
|
):
|
|
raise QueryContractError(
|
|
f"line {line}: quoted SQL identifier {identifier!r} cannot be reduced exactly"
|
|
)
|
|
return identifier
|
|
|
|
normalized = QUOTED_SQL_IDENTIFIER.sub(normalize_quoted_identifier, scrubbed)
|
|
if '"' in normalized:
|
|
raise QueryContractError(f"line {line}: unterminated quoted SQL identifier")
|
|
return normalized
|
|
|
|
|
|
def _reject_unqualified_relation_references(sql: str, line: int) -> None:
|
|
"""Reject search-path-dependent base relations while allowing local CTEs/functions."""
|
|
|
|
cte_names = {
|
|
match.group(1).lower()
|
|
for match in re.finditer(
|
|
r"(?:\bwith\s+(?:recursive\s+)?|,)\s*([a-z_][a-z0-9_]*)"
|
|
r"\s*(?:\([^)]*\))?\s+as\s+(?:(?:not\s+)?materialized\s+)?\(",
|
|
sql,
|
|
re.IGNORECASE,
|
|
)
|
|
}
|
|
for match in re.finditer(
|
|
r"\b(?:from|join)\s+(?:only\s+)?([a-z_][a-z0-9_]*)",
|
|
sql,
|
|
re.IGNORECASE,
|
|
):
|
|
relation = match.group(1).lower()
|
|
remainder = sql[match.end() :].lstrip()
|
|
if remainder.startswith((".", "(")):
|
|
continue
|
|
if relation in cte_names or relation in UNQUALIFIED_NONCANONICAL_RELATIONS:
|
|
continue
|
|
raise QueryContractError(
|
|
f"line {line}: unqualified relation {relation!r} is search-path dependent"
|
|
)
|
|
|
|
|
|
def _reject_comma_separated_base_relations(sql: str, line: int) -> None:
|
|
"""Require explicit JOIN syntax for base relations after a FROM-list comma."""
|
|
|
|
cte_names = {
|
|
match.group(1).lower()
|
|
for match in re.finditer(
|
|
r"(?:\bwith\s+(?:recursive\s+)?|,)\s*([a-z_][a-z0-9_]*)"
|
|
r"\s*(?:\([^)]*\))?\s+as\s+(?:(?:not\s+)?materialized\s+)?\(",
|
|
sql,
|
|
re.IGNORECASE,
|
|
)
|
|
}
|
|
tokens = [(match.group(0).lower(), match.start()) for match in SQL_STRUCTURE_TOKEN.finditer(sql)]
|
|
depth = 0
|
|
from_depths: set[int] = set()
|
|
for index, (token, _position) in enumerate(tokens):
|
|
if token == "(":
|
|
depth += 1
|
|
continue
|
|
if token == ")":
|
|
from_depths.discard(depth)
|
|
depth = max(0, depth - 1)
|
|
continue
|
|
if token == "from":
|
|
from_depths.add(depth)
|
|
continue
|
|
if token == ";" or token in FROM_CLAUSE_END_KEYWORDS:
|
|
from_depths.discard(depth)
|
|
continue
|
|
if token != "," or depth not in from_depths:
|
|
continue
|
|
|
|
item_index = index + 1
|
|
while item_index < len(tokens) and tokens[item_index][0] in {"lateral", "only"}:
|
|
item_index += 1
|
|
if item_index >= len(tokens):
|
|
raise QueryContractError(f"line {line}: incomplete comma-separated FROM item")
|
|
relation = tokens[item_index][0]
|
|
if relation == "(":
|
|
continue
|
|
if not SQL_IDENTIFIER.fullmatch(relation):
|
|
raise QueryContractError(f"line {line}: comma-separated FROM item cannot be reduced exactly")
|
|
|
|
next_token = tokens[item_index + 1][0] if item_index + 1 < len(tokens) else ""
|
|
if next_token == "(":
|
|
continue
|
|
if next_token == ".":
|
|
function_token = tokens[item_index + 3][0] if item_index + 3 < len(tokens) else ""
|
|
if function_token == "(":
|
|
continue
|
|
qualified = relation
|
|
if item_index + 2 < len(tokens):
|
|
qualified = f"{relation}.{tokens[item_index + 2][0]}"
|
|
raise QueryContractError(
|
|
f"line {line}: comma-separated base relation {qualified!r} must use explicit JOIN"
|
|
)
|
|
if relation in cte_names:
|
|
continue
|
|
raise QueryContractError(
|
|
f"line {line}: comma-separated base relation {relation!r} must use explicit JOIN"
|
|
)
|
|
|
|
|
|
def _unqualified_identifier_candidates(sql: str) -> set[str]:
|
|
"""Find identifiers that are neither qualified columns nor query structure."""
|
|
|
|
scrubbed = _strip_sql_strings_and_comments(sql)
|
|
allowed = {*NONCANONICAL_QUERY_IDENTIFIERS, DYNAMIC_SQL_VALUE}
|
|
|
|
# Output aliases and relation aliases may be consumed unqualified by an
|
|
# enclosing query. CTE names/header columns and subquery aliases are also
|
|
# query structure rather than base-relation column reads.
|
|
allowed.update(match.group(1).lower() for match in re.finditer(r"\bas\s+([a-z_][a-z0-9_]*)", scrubbed, re.I))
|
|
allowed.update(match.group(3).lower() for match in ALIASED_CANONICAL_RELATION.finditer(scrubbed))
|
|
allowed.update(
|
|
match.group(1).lower()
|
|
for match in re.finditer(
|
|
r"(?:\bwith|,)\s*([a-z_][a-z0-9_]*)\s*(?:\([^)]*\))?\s+as\s*\(",
|
|
scrubbed,
|
|
re.I,
|
|
)
|
|
)
|
|
for match in re.finditer(
|
|
r"(?:\bwith|,)\s*[a-z_][a-z0-9_]*\s*\(([^)]*)\)\s+as\s*\(",
|
|
scrubbed,
|
|
re.I,
|
|
):
|
|
allowed.update(identifier.lower() for identifier in SQL_IDENTIFIER.findall(match.group(1)))
|
|
allowed.update(
|
|
match.group(1).lower() for match in re.finditer(r"\)\s+(?:as\s+)?([a-z_][a-z0-9_]*)", scrubbed, re.I)
|
|
)
|
|
allowed.update(
|
|
match.group(1).lower()
|
|
for match in re.finditer(
|
|
r"\b(?:from|join)\s+[a-z_][a-z0-9_]*(?:\s+as)?\s+([a-z_][a-z0-9_]*)",
|
|
scrubbed,
|
|
re.I,
|
|
)
|
|
)
|
|
# A qualified projection becomes a mechanically known output name for an
|
|
# enclosing CTE query (for example ``ce.claim_id`` -> ``claim_id``).
|
|
allowed.update(match.group(1).lower() for match in re.finditer(r"\.\s*([a-z_][a-z0-9_]*)", scrubbed, re.I))
|
|
|
|
scrubbed = re.sub(r"\b[a-z_][a-z0-9_]*\s*\.\s*[a-z_][a-z0-9_]*", " ", scrubbed, flags=re.I)
|
|
scrubbed = re.sub(r"::\s*[a-z_][a-z0-9_]*", " ", scrubbed, flags=re.I)
|
|
|
|
candidates: set[str] = set()
|
|
for match in SQL_IDENTIFIER.finditer(scrubbed):
|
|
identifier = match.group(0).lower()
|
|
if identifier in SQL_KEYWORDS or identifier in allowed:
|
|
continue
|
|
if re.match(r"\s*\(", scrubbed[match.end() :]):
|
|
continue
|
|
candidates.add(identifier)
|
|
return candidates
|
|
|
|
|
|
def _reject_inexact_canonical_alias_uses(
|
|
sql: str,
|
|
line: int,
|
|
aliases: dict[str, tuple[str, str]],
|
|
) -> None:
|
|
"""Reject canonical reads that cannot be reduced to named columns."""
|
|
|
|
scrubbed = _strip_sql_strings_and_comments(sql)
|
|
for alias in aliases:
|
|
if re.search(
|
|
rf"(?<![a-z0-9_]){re.escape(alias)}\s*\.\s*{re.escape(DYNAMIC_SQL_VALUE)}(?![a-z0-9_])",
|
|
scrubbed,
|
|
re.IGNORECASE,
|
|
):
|
|
raise QueryContractError(
|
|
f"line {line}: dynamic SQL interpolation cannot supply a canonical column identifier"
|
|
)
|
|
if re.search(
|
|
rf"(?<![a-z0-9_]){re.escape(alias)}\s*\.\s*\*",
|
|
scrubbed,
|
|
re.IGNORECASE,
|
|
):
|
|
raise QueryContractError(f"line {line}: canonical wildcard reads prevent exact derivation")
|
|
|
|
without_relations = ALIASED_CANONICAL_RELATION.sub(" ", scrubbed)
|
|
without_qualified_columns = re.sub(
|
|
rf"(?<![a-z0-9_]){re.escape(alias)}\s*\.\s*[a-z_][a-z0-9_]*",
|
|
" ",
|
|
without_relations,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
if re.search(
|
|
rf"(?<![a-z0-9_]){re.escape(alias)}(?![a-z0-9_])",
|
|
without_qualified_columns,
|
|
re.IGNORECASE,
|
|
):
|
|
raise QueryContractError(
|
|
f"line {line}: whole-row canonical alias {alias!r} prevents exact column derivation"
|
|
)
|
|
|
|
self_alias = re.search(
|
|
r"(?<![a-z0-9_.:])([a-z_][a-z0-9_]*)\s+as\s+\1(?![a-z0-9_])",
|
|
scrubbed,
|
|
re.IGNORECASE,
|
|
)
|
|
if self_alias:
|
|
raise QueryContractError(
|
|
f"line {line}: unqualified SQL identifier {self_alias.group(1).lower()!r} is masked by an output alias"
|
|
)
|
|
|
|
|
|
def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[str]]:
|
|
"""Mechanically derive canonical relation columns from runtime SQL."""
|
|
|
|
observed: dict[tuple[str, str], set[str]] = defaultdict(set)
|
|
for line, sql in extract_canonical_query_literals(source, enforce_source_pin=False):
|
|
aliases: dict[str, tuple[str, str]] = {}
|
|
explicit_starts: set[int] = set()
|
|
for match in ALIASED_CANONICAL_RELATION.finditer(sql):
|
|
relation = (match.group(1).lower(), match.group(2).lower())
|
|
alias = match.group(3).lower()
|
|
existing = aliases.get(alias)
|
|
if existing is not None and existing != relation:
|
|
raise QueryContractError(
|
|
f"line {line}: canonical alias {alias!r} maps to both {existing!r} and {relation!r}"
|
|
)
|
|
aliases[alias] = relation
|
|
explicit_starts.add(match.start())
|
|
|
|
for match in CANONICAL_RELATION.finditer(sql):
|
|
if match.start() not in explicit_starts:
|
|
relation = f"{match.group(1).lower()}.{match.group(2).lower()}"
|
|
raise QueryContractError(f"line {line}: canonical relation {relation} requires an explicit AS alias")
|
|
|
|
_reject_inexact_canonical_alias_uses(sql, line, aliases)
|
|
|
|
for alias, relation in aliases.items():
|
|
observed[relation].update(
|
|
column.lower()
|
|
for column in re.findall(
|
|
rf"(?<![a-z0-9_]){re.escape(alias)}\s*\.\s*([a-z_][a-z0-9_]*)",
|
|
sql,
|
|
re.IGNORECASE,
|
|
)
|
|
)
|
|
|
|
unqualified = _unqualified_identifier_candidates(sql)
|
|
if unqualified:
|
|
joined = ", ".join(sorted(unqualified))
|
|
raise QueryContractError(f"line {line}: unqualified SQL identifiers prevent exact derivation: {joined}")
|
|
|
|
_assert_exact_runtime_source(source)
|
|
return {relation: frozenset(columns) for relation, columns in observed.items()}
|
|
|
|
|
|
def verify_runtime_read_columns(
|
|
source: str,
|
|
expected: dict[tuple[str, str], frozenset[str]],
|
|
) -> dict[tuple[str, str], frozenset[str]]:
|
|
"""Fail when the runtime query surface and reviewed allowlist diverge."""
|
|
|
|
observed = derive_runtime_read_columns(source)
|
|
normalized_expected = {relation: frozenset(columns) for relation, columns in expected.items()}
|
|
if observed != normalized_expected:
|
|
missing_relations = sorted(normalized_expected.keys() - observed.keys())
|
|
extra_relations = sorted(observed.keys() - normalized_expected.keys())
|
|
column_drift = {
|
|
f"{schema}.{relation}": {
|
|
"missing": sorted(normalized_expected[(schema, relation)] - observed[(schema, relation)]),
|
|
"extra": sorted(observed[(schema, relation)] - normalized_expected[(schema, relation)]),
|
|
}
|
|
for schema, relation in sorted(normalized_expected.keys() & observed.keys())
|
|
if observed[(schema, relation)] != normalized_expected[(schema, relation)]
|
|
}
|
|
detail = json.dumps(
|
|
{
|
|
"missing_relations": [".".join(relation) for relation in missing_relations],
|
|
"extra_relations": [".".join(relation) for relation in extra_relations],
|
|
"column_drift": column_drift,
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
raise QueryContractError(f"runtime query surface differs from reviewed read allowlist: {detail}")
|
|
return observed
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("runtime", type=Path)
|
|
args = parser.parse_args()
|
|
contract = derive_runtime_read_columns(args.runtime.read_text(encoding="utf-8"))
|
|
payload = {f"{schema}.{relation}": sorted(columns) for (schema, relation), columns in sorted(contract.items())}
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|