Recon deny-list + DB-error sanitization for the SQL guard

migrations/core/013_guardrail_audit_error_detail.sql

@@ -0,0 +1,11 @@
+-- Store the RAW driver/DB error text server-side, keyed by audit_id. The caller only ever receives a
+-- generic, value-free classified message (raw schema / column / value names must never cross the
+-- assistant boundary), so the operator needs somewhere to see the real error when debugging a failed
+-- query — that is this column. Populated ONLY on an operational failure (the executor's stderr);
+-- NULL for guardrail refusals (their reasons are already clean). Portable (SQLite + Postgres).
+--
+-- SENSITIVITY: a driver error can embed data VALUES (e.g. a Postgres unique-violation
+-- `DETAIL: Key (email)=(...)`), so this column inherits the datasource's sensitivity classification.
+-- It is strictly better than the prior behavior (which returned that text to the caller), but for a
+-- PHI/PII deployment it must fall under the audit table's access-control + retention/redaction policy.
+ALTER TABLE guardrail_audit ADD COLUMN error_detail TEXT;

packages/agami-core/src/contracts.py

@@ -217,3 +217,4 @@ class GuardrailAuditRecord(_Contract):
     execution_ms: int | None = None
     correlation_id: str | None = None
     source: str | None = None
+    error_detail: str | None = None  # RAW driver error — server-side audit only, NEVER in the Envelope

packages/agami-core/src/execute_sql.py

@@ -1347,19 +1347,25 @@ def execute_guarded(
 ) -> ExecResult:
     """The un-bypassable guarded envelope — the single execution chokepoint (REQ-002/REQ-014).
 
-    In fixed order: read-only / dangerous-SQL guard (the hard security gate — NOT bypassable via
-    ``no_safety``, which skips only the semantic-model pass, never write/RCE/DoS protection) ->
-    semantic-model safety pass (fan/chasm pre-flight + scope + PII + ``default_filters`` rewrite) ->
-    resolve the datasource -> ``executor.execute(vetted_sql, …)``. The executor only ever receives
-    SQL both guards have passed. Raises ``GuardRefused`` carrying the typed ``Refusal`` on a refusal,
-    and ``ExecutorError`` on a connect/run failure — so the subprocess ``main`` and the in-process MCP
-    handler apply the same guard and build the same envelope. The row cap rides the request-scoped
-    ``_max_rows_override`` ContextVar the caller sets."""
+    In fixed order: read-only / dangerous-SQL guard -> recon / metadata deny-list (both hard security
+    gates — NOT bypassable via ``no_safety``, which skips only the semantic-model pass, never
+    write/RCE/DoS protection) -> semantic-model safety pass (fan/chasm pre-flight + scope + PII +
+    ``default_filters`` rewrite) -> resolve the datasource -> ``executor.execute(vetted_sql, …)``. The
+    executor only ever receives SQL every guard has passed. Raises ``GuardRefused`` carrying the typed
+    ``Refusal`` on a refusal, and ``ExecutorError`` on a connect/run failure — so the subprocess
+    ``main`` and the in-process MCP handler apply the same guards and build the same envelope. The row
+    cap rides the request-scoped ``_max_rows_override`` ContextVar the caller sets."""
     import sql_guard
 
     verdict = sql_guard.check_read_only(sql)
     if verdict is not None:
         raise GuardRefused(_refusal_from_verdict("permission", verdict), code=1)
+    # Recon / metadata deny-list — refuse server-fingerprinting + system-catalog introspection
+    # (version(), current_user, information_schema, pg_* relations) as a distinct `recon` refusal.
+    # A hard gate, at the chokepoint so BOTH surfaces get it, and NOT bypassable via no_safety.
+    recon = sql_guard.check_no_recon(sql)
+    if recon is not None:
+        raise GuardRefused(_refusal_from_verdict("recon", recon), code=1)
     if not no_safety:
         sql, refusal = _model_safety(sql, profile, area)
         if refusal is not None:
@@ -1424,11 +1430,11 @@ def main() -> int:
 
     profile = args.profile or _resolve_default_profile()
 
-    # Route through the single guarded envelope with the built-in executor: guard -> model-safety ->
-    # resolve -> connect-and-run, returning native rows we then serialize to stdout as CSV (the
-    # subprocess wire). The guard is the hard security gate for EVERY caller (both MCP servers, the
-    # agami-query skill, cron), NOT bypassable via --no-safety (which skips only the semantic-model
-    # pass, never write/RCE/DoS protection).
+    # Route through the single guarded envelope with the built-in executor: guard -> recon deny-list
+    # -> model-safety -> resolve -> connect-and-run, returning native rows we then serialize to stdout
+    # as CSV (the subprocess wire). The guard is the hard security gate for EVERY caller (both MCP
+    # servers, the agami-query skill, cron), NOT bypassable via --no-safety (which skips only the
+    # semantic-model pass, never write/RCE/DoS protection).
     try:
         result = execute_guarded(
             sql, profile, args.area, executor=BUILTIN_EXECUTOR, no_safety=args.no_safety

packages/agami-core/src/guardrail.py

@@ -50,11 +50,16 @@
     # data-protection / governance — emitted by those gates
     "sensitive_columns",
     "preflight_refused",
-    # operational / execution failures — from the executor + DB driver
+    # operational / execution failures — from the executor + DB driver (classified from stderr;
+    # `permission` reuses the safety kind above). The rich column/table/network kinds mirror
+    # db_error_classifier.md.
     "timeout",
     "dsn",
+    "network",
     "driver_missing",
     "auth",
+    "column_not_found",
+    "table_not_found",
     "syntax",
     "other",
 )

packages/agami-core/src/model_store.py

@@ -350,24 +350,49 @@ def record_tool_call(self, record: Any) -> None:
         )
         self._store.commit()
 
+    _AUDIT_COLS = (
+        "audit_id, ts, datasource, status, refusal_kind, sql, row_count, execution_ms, "
+        "correlation_id, source"
+    )
+
     def record_guardrail_audit(self, record: Any) -> None:
-        self._store.execute(
-            "INSERT INTO guardrail_audit (audit_id, ts, datasource, status, refusal_kind, sql, "
-            "row_count, execution_ms, correlation_id, source) "
-            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
-            (
-                record.audit_id,
-                record.ts,
-                record.datasource,
-                record.status,
-                record.refusal_kind,
-                record.sql,
-                record.row_count,
-                record.execution_ms,
-                record.correlation_id,
-                record.source,
-            ),
+        base = (
+            record.audit_id,
+            record.ts,
+            record.datasource,
+            record.status,
+            record.refusal_kind,
+            record.sql,
+            record.row_count,
+            record.execution_ms,
+            record.correlation_id,
+            record.source,
         )
+        try:
+            self._store.execute(
+                f"INSERT INTO guardrail_audit ({self._AUDIT_COLS}, error_detail) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (*base, record.error_detail),
+            )
+        except Exception as exc:
+            # ONLY the pre-013 schema (the error_detail column isn't there yet — new code against an
+            # un-migrated DB) is retried WITHOUT the raw detail. Any OTHER failure must propagate: a
+            # blanket retry would both mask a real DB error and silently drop error_detail (Copilot
+            # review). Match the missing-column signature across drivers (sqlite "has no column named",
+            # Postgres "does not exist" / UndefinedColumn).
+            msg = str(exc).lower()
+            missing_error_detail = "error_detail" in msg and (
+                "no column" in msg or "does not exist" in msg or "undefined" in msg
+            )
+            if not missing_error_detail:
+                raise  # a genuine DB failure — surface it to the best-effort caller, don't mask it
+            # Roll back first so a Postgres aborted-transaction doesn't fail the retry.
+            self._store.rollback()
+            self._store.execute(
+                f"INSERT INTO guardrail_audit ({self._AUDIT_COLS}) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                base,
+            )
         self._store.commit()
 
 

packages/agami-core/src/sql_guard.py

@@ -15,6 +15,11 @@
 `check_read_only(sql)` returns `None` when the SQL is a single safe read-only
 statement, else a safety `Verdict` the shared executor maps to a refusal
 (`kind="permission"`). The rejection ladder itself lives in `_read_only_reason`.
+
+`check_no_recon(sql)` is the companion metadata/recon gate — it returns a `Verdict`
+(`kind="recon"`) for a query that fingerprints the server or reads the system catalog
+(`version()`, `current_user`, `information_schema`, `pg_*` relations). Both run at the
+same chokepoint over the same neutralized SQL.
 """
 
 from __future__ import annotations
@@ -233,6 +238,101 @@ def _neutralize(sql: str) -> str:
 )
 
 
+# ---------------------------------------------------------------------------
+# Recon / metadata deny-list. A query can pass read-only + object-scope and STILL fingerprint the
+# server (`version()`, `current_user`, `inet_server_addr()`), enumerate or dump the system catalog
+# (`information_schema`, `pg_catalog`, any `pg_*` relation — including `pg_stats` sampled VALUES and
+# `pg_authid` password hashes), read catalog DDL (`pg_get_viewdef`/`pg_get_functiondef`), or probe
+# object existence (`has_table_privilege`, `to_regclass`) — all leaking DB internals across the LLM
+# boundary. The read-only role can't revoke catalog access on most engines (Q1), so this is the
+# INTENDED backstop and denies it app-side, as a DISTINCT `recon` refusal. Matched over the SAME
+# neutralized SQL as `check_read_only` (no second parser); the sets/prefixes below are the source of
+# truth — adding an engine's recon builtin is a one-line edit.
+#
+# FP discipline (regex, not AST — so exact-name matching is the residual): the paren set requires a
+# trailing `(`, so bare `version`/`database`/`user` COLUMNS pass; the niladic + relation sets use a
+# `.`-lookbehind so a qualified column (`t.current_user`, `t.pg_class`) passes; schema names require a
+# trailing `.`. The `pg_`/`stl_`/… prefixes match RESERVED namespaces (Postgres reserves the `pg_`
+# object-name prefix; Redshift reserves `stl_/stv_/svl_/svv_`), so the FP surface is a bare, unqualified
+# column literally prefixed `pg_`/`stl_` — rare, documented, pinned in the FP corpus. Bare
+# `user`/`schema`/`database` (Postgres niladic synonyms) are DELIBERATELY omitted from the niladic set —
+# far too common as intended column names; only their `()`-call form is denied (a known, minor
+# username-fingerprint residual).
+_RECON_PAREN_FNS = frozenset(
+    {
+        "version",  # server version string (pg / snowflake)
+        "current_database",
+        "current_schemas",  # pg (plural, takes a bool) — the niladic current_schema is below
+        "database",
+        "schema",
+        "user",  # MySQL user() / system_user() — the CALL form only
+        "connection_id",
+        "system_user",
+        "current_account",  # snowflake
+        "current_region",
+        "current_version",
+        "current_warehouse",
+        "inet_server_addr",  # server/client network fingerprint (pg)
+        "inet_server_port",
+        "inet_client_addr",
+        "inet_client_port",
+    }
+)
+# Function FAMILIES matched by prefix/pattern (each still requires the trailing `(`): every `pg_*`
+# call (introspection + DDL-dump like pg_get_viewdef / size/topology like pg_relation_size — the
+# dangerous subset is already denied earlier by check_read_only as `permission`), the
+# has_*_privilege family (object-existence probing), and to_reg* (name→OID resolution / enumeration).
+_RECON_CALL_FAMILIES = (r"pg_\w+", r"has_\w+_privilege", r"to_reg\w+")
+_RECON_NILADIC = frozenset(
+    {"current_user", "session_user", "current_catalog", "current_schema", "current_role"}
+)
+_RECON_SCHEMAS = frozenset(
+    {"information_schema", "pg_catalog", "account_usage", "mysql", "performance_schema", "sys"}
+)
+# High-value catalog relations named explicitly for readability; the `pg_` prefix below is the actual
+# catch-all (it also covers pg_stats / pg_authid / pg_statistic / pg_stat_statements / pg_type / … ).
+_RECON_BARE_RELATIONS = frozenset(
+    {
+        "pg_tables",
+        "pg_class",
+        "pg_stat_activity",
+        "pg_roles",
+        "pg_user",
+        "pg_shadow",
+        "pg_authid",
+        "pg_stats",
+    }
+)
+# Reserved relation-name namespaces (unshadowable-by-convention): Postgres `pg_`, Redshift system tables.
+_RECON_RELATION_PREFIXES = ("pg_", "stl_", "stv_", "svl_", "svv_")
+
+
+def _recon_group(names: frozenset[str]) -> str:
+    # Longest-first so an alternation prefers the more specific name (regex alternation is greedy per
+    # position but this also keeps `current_schemas` from being shadowed by `current_schema`).
+    return "|".join(sorted(names, key=len, reverse=True))
+
+
+_RECON_PAREN_RE = re.compile(
+    rf"\b({_recon_group(_RECON_PAREN_FNS)}|{'|'.join(_RECON_CALL_FAMILIES)})\s*\(", re.IGNORECASE
+)
+# `(?<!\.)` — an unqualified, unquoted niladic keyword IS the special function; a real column of that
+# name must be qualified (`t.current_user`) or quoted, so the lookbehind lets those through.
+_RECON_NILADIC_RE = re.compile(rf"(?<!\.)\b({_recon_group(_RECON_NILADIC)})\b", re.IGNORECASE)
+_RECON_SCHEMA_RE = re.compile(rf"\b({_recon_group(_RECON_SCHEMAS)})\s*\.", re.IGNORECASE)
+# `(?<!\.)` so a qualified column matching a reserved relation name (`t.pg_class`) passes — a catalog
+# relation is referenced bare (search-path-resolved) or schema-qualified (caught by _RECON_SCHEMA_RE).
+_RECON_RELATION_RE = re.compile(
+    r"(?<!\.)\b("
+    + _recon_group(_RECON_BARE_RELATIONS)
+    + "|"
+    + "|".join(p + r"\w+" for p in _RECON_RELATION_PREFIXES)
+    + r")\b",
+    re.IGNORECASE,
+)
+_RECON_SYSVAR_RE = re.compile(r"@@\w+")  # MySQL system variables — all config/recon
+
+
 def _read_only_reason(sql: str | None) -> str | None:
     """Return None if `sql` is a single safe read-only statement, else a reason string.
 
@@ -314,3 +414,46 @@ def check_read_only(sql: str | None) -> Verdict | None:
         "Send a single read-only SELECT / WITH...SELECT — no DML, DDL, transaction/session "
         "control, or multiple statements.",
     )
+
+
+def check_no_recon(sql: str | None) -> Verdict | None:
+    """Return ``None`` if ``sql`` calls no metadata/recon function and reads no system catalog, else a
+    safety ``Verdict`` the shared executor maps to a refusal (``kind=recon``).
+
+    Runs AFTER :func:`check_read_only` at the shared executor chokepoint (and as a fail-fast pre-check
+    in the tool layer). Detection is regex over the SAME neutralized SQL — comments + literals blanked,
+    quoted identifiers unwrapped — so a recon token hidden in a string/comment can't smuggle past, and
+    (symmetrically) a legitimate mention inside a literal/comment doesn't false-trip. See the
+    ``_RECON_*`` sets for the deny-list and its false-positive discipline.
+    """
+    if not sql or not sql.strip():
+        return None  # empty — check_read_only owns that rejection
+    try:
+        stripped = _neutralize(sql)
+    except _GuardReject:
+        # Dialect-ambiguous form we can't neutralize → fail CLOSED. In the normal flow check_read_only
+        # (same neutralizer) already refused this input; failing closed here keeps the recon gate safe
+        # even for a hypothetical caller that invoked it standalone or reordered the two checks.
+        return safety_verdict(
+            "recon",
+            "the query uses a form that can't be safely analyzed for metadata/recon access",
+            "Send a single, unambiguous read-only SELECT over your declared tables and columns.",
+        )
+
+    m = (
+        _RECON_PAREN_RE.search(stripped)
+        or _RECON_NILADIC_RE.search(stripped)
+        or _RECON_SCHEMA_RE.search(stripped)
+        or _RECON_RELATION_RE.search(stripped)
+        or _RECON_SYSVAR_RE.search(stripped)
+    )
+    if m is None:
+        return None
+    hit = m.group(0).strip(" .(")  # the matched token, without the trailing `(` / `.` anchor
+    return safety_verdict(
+        "recon",
+        f"metadata/recon access is not allowed (`{hit}`) — server/version fingerprinting and "
+        "system-catalog / information_schema introspection are blocked",
+        "Query only your declared tables and columns; drop any server-metadata function or "
+        "system-catalog / information_schema reference.",
+    )

packages/agami-core/src/store.py

@@ -124,6 +124,9 @@ def query(self, sql: str, params: tuple = ()) -> list[dict[str, Any]]:
     def commit(self) -> None:
         self.conn.commit()
 
+    def rollback(self) -> None:
+        self.conn.rollback()
+
     def close(self) -> None:
         self.conn.close()