From 1faedd8cc5cc69662681288828cb3cdbc4ce7662 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?FabioLeit=C3=A3o?= Date: Thu, 2 Jul 2026 22:13:08 -0300 Subject: [PATCH 1/3] fix(connector): record scan_failures on SQL column sampling errors (#1140) Sampling query failures now persist sampling_error rows, raise ColumnSampleError, and skip detection instead of returning empty samples that masquerade as clean scans. --- connectors/sql_connector.py | 120 ++++++++++++++++++++++++------------ connectors/sql_sampling.py | 5 +- tests/test_sql_connector.py | 61 ++++++++++++++++++ 3 files changed, 143 insertions(+), 43 deletions(-) diff --git a/connectors/sql_connector.py b/connectors/sql_connector.py index 006251144..f95238ab3 100644 --- a/connectors/sql_connector.py +++ b/connectors/sql_connector.py @@ -98,6 +98,26 @@ } ) +# scan_failures reason for column sampling errors (#1140; taxonomy shared with #828). +SCAN_FAILURE_REASON_SAMPLING_ERROR = "sampling_error" + + +class ColumnSampleError(RuntimeError): + """Column sampling failed; the connector persisted a scan_failures row.""" + + +def format_column_sample_failure_details( + *, + schema: str, + table: str, + column_name: str, + dialect: str, + exc: BaseException, +) -> str: + """Stable detail string for scan_failures when SQL column sampling fails.""" + loc = f"{schema}.{table}.{column_name}" if schema else f"{table}.{column_name}" + return f"{loc} dialect={dialect}: {type(exc).__name__}: {exc}" + def _get_skip_schemas(dialect: str) -> frozenset[str]: """Return schema names to skip when discovering (dialect-specific blacklist).""" @@ -324,7 +344,10 @@ def _process_one_finding( """Sample column, run detection, optionally full-scan for minor; save finding and log.""" if self._inter_query_delay_s > 0: time.sleep(self._inter_query_delay_s) - sample = self.sample(schema, table, cname) + try: + sample = self.sample(schema, table, cname) + except ColumnSampleError: + return res = self.scanner.scan_column(cname, sample, connector_data_type=ctype) res = augment_low_id_like_for_persist(res, cname, self.detection_config) if ( @@ -337,7 +360,10 @@ def _process_one_finding( res.get("pattern_detected") or "" ) and self.detection_config.get("minor_full_scan"): full_scan_limit = self.detection_config.get("minor_full_scan_limit", 100) - full_sample = self.sample(schema, table, cname, limit=full_scan_limit) + try: + full_sample = self.sample(schema, table, cname, limit=full_scan_limit) + except ColumnSampleError: + return full_res = self.scanner.scan_column( cname, full_sample, connector_data_type=ctype ) @@ -397,59 +423,71 @@ def sample( safe_col = column_name.replace('"', '""') safe_table = table.replace('"', '""') safe_schema = (schema or "").replace('"', '""') - try: - tkey = (schema or "", table) - if tkey not in self._table_row_cache: + tkey = (schema or "", table) + if tkey not in self._table_row_cache: + try: from connectors.sql_table_row_estimate import estimate_table_rows self._table_row_cache[tkey] = estimate_table_rows( self._connection, dialect, schema or "", table ) - est = self._table_row_cache[tkey] - table_meta = TableSamplingMetadata(estimated_row_count=est) - to = self._sample_statement_timeout_ms - plan = SamplingManager.build_column_sample( - dialect, - safe_col=safe_col, - safe_table=safe_table, - safe_schema=safe_schema, - schema=schema, - limit=use_limit, - table_metadata=table_meta, - statement_timeout_ms=to, - ) - audit_key = f"{schema or ''}.{table}" - if self._sql_sampling_audit_key != audit_key: - self._sql_sampling_audit_key = audit_key - try: - from utils.logger import get_logger - - human = plan.human_strategy or plan.strategy_label - msg = "Sampling %s using %s strategy (label=%s dialect=%s)" % ( - audit_key, - human, - plan.strategy_label, - dialect, - ) - if plan.audit_notes: - msg += " notes=%s" % (plan.audit_notes,) - get_logger().info(msg) - except Exception: - pass + except Exception: + self._table_row_cache[tkey] = None + est = self._table_row_cache[tkey] + table_meta = TableSamplingMetadata(estimated_row_count=est) + to = self._sample_statement_timeout_ms + plan = SamplingManager.build_column_sample( + dialect, + safe_col=safe_col, + safe_table=safe_table, + safe_schema=safe_schema, + schema=schema, + limit=use_limit, + table_metadata=table_meta, + statement_timeout_ms=to, + ) + audit_key = f"{schema or ''}.{table}" + if self._sql_sampling_audit_key != audit_key: + self._sql_sampling_audit_key = audit_key + try: + from utils.logger import get_logger + + human = plan.human_strategy or plan.strategy_label + msg = "Sampling %s using %s strategy (label=%s dialect=%s)" % ( + audit_key, + human, + plan.strategy_label, + dialect, + ) + if plan.audit_notes: + msg += " notes=%s" % (plan.audit_notes,) + get_logger().info(msg) + except Exception: + pass + try: with self._connection.begin(): if dialect in ("postgresql", "postgres") and to: self._connection.execute( text("SET LOCAL statement_timeout = :mt").bindparams(mt=int(to)) ) rows = self._connection.execute(plan.query).fetchall() - parts = [str(r[0])[:200] for r in rows if r[0] is not None] - return " ".join(parts) except Exception as e: + target_name = str(self.config.get("name") or "database") + details = format_column_sample_failure_details( + schema=schema or "", + table=table, + column_name=column_name, + dialect=dialect, + exc=e, + ) + self.db_manager.save_failure( + target_name, SCAN_FAILURE_REASON_SAMPLING_ERROR, details + ) try: from utils.logger import get_logger get_logger().warning( - "sample_skipped schema=%s table=%s col=%s dialect=%s err=%s", + "sample_failed schema=%s table=%s col=%s dialect=%s err=%s", schema or "", table, column_name, @@ -458,7 +496,9 @@ def sample( ) except Exception: pass - return "" + raise ColumnSampleError from e + parts = [str(r[0])[:200] for r in rows if r[0] is not None] + return " ".join(parts) def run(self) -> None: """Connect, discover, sample each column, detect, save_finding; on error save_failure.""" diff --git a/connectors/sql_sampling.py b/connectors/sql_sampling.py index 5b0628c10..45464b649 100644 --- a/connectors/sql_sampling.py +++ b/connectors/sql_sampling.py @@ -357,9 +357,8 @@ def _plan_mssql_column_sample( # SQL Server has **no** query-level execution-time hint equivalent to # MySQL's ``/*+ MAX_EXECUTION_TIME(N) */``; ``OPTION (MAX_EXECUTION_TIME = …)`` # is *not* valid T-SQL and emitting it makes every sample fail with a - # syntax error -- which the connector's broad ``except Exception`` then - # swallows, returning empty samples and producing **silent zero-finding - # scans** on every SQL Server target. Statement-time bounds for MSSQL + # syntax error -- the connector records ``scan_failures`` (reason + # ``sampling_error``) and skips detection for that column (#1140). Statement-time bounds for MSSQL # come from the SQLAlchemy connection-level ``connect_timeout`` (set via # ``_connect_args_from_target``); for finer per-statement budgets a # dedicated ADR + driver-level ``LOCK_TIMEOUT`` or session option is the diff --git a/tests/test_sql_connector.py b/tests/test_sql_connector.py index 3df945bce..6a830655d 100644 --- a/tests/test_sql_connector.py +++ b/tests/test_sql_connector.py @@ -7,17 +7,22 @@ import sqlite3 from unittest.mock import MagicMock +import pytest from connectors.sql_connector import ( DRIVER_MAP, + SCAN_FAILURE_REASON_SAMPLING_ERROR, + ColumnSampleError, SQLConnector, _connect_args_from_target, _discover_fallback_no_schemas, _get_skip_schemas, _resolve_sample_statement_timeout_ms, _should_skip_schema, + format_column_sample_failure_details, ) from sqlalchemy import create_engine, inspect from sqlalchemy.engine.url import make_url +from sqlalchemy.exc import ProgrammingError def test_driver_map_uses_registered_sqlalchemy_dialects(): @@ -254,3 +259,59 @@ def test_sql_connector_sample_sparse_column_prefers_non_null(tmp_path): connector.close() assert "marker_value" in sample + + +def test_format_column_sample_failure_details_includes_location_and_dialect(): + err = ProgrammingError("stmt", {}, Exception("Incorrect syntax near 'OPTION'")) + detail = format_column_sample_failure_details( + schema="dbo", + table="users", + column_name="email", + dialect="mssql", + exc=err, + ) + assert "dbo.users.email" in detail + assert "dialect=mssql" in detail + assert "ProgrammingError" in detail + + +def test_sql_connector_sample_syntax_error_records_scan_failure(): + """Sampling SQL errors must persist scan_failures, not return empty clean samples (#1140).""" + target = {"type": "database", "driver": "mssql", "name": "ProdDB"} + scanner = MagicMock() + db_manager = MagicMock() + connector = SQLConnector(target, scanner, db_manager, sample_limit=5) + connector.engine = MagicMock() + connector.engine.dialect.name = "mssql" + connector._connection = MagicMock() + ctx = MagicMock() + connector._connection.begin.return_value = ctx + ctx.__enter__ = MagicMock(return_value=None) + ctx.__exit__ = MagicMock(return_value=False) + connector._connection.execute.side_effect = ProgrammingError( + "SELECT", + {}, + Exception("Incorrect syntax near 'OPTION'"), + ) + + with pytest.raises(ColumnSampleError): + connector.sample("dbo", "users", "email") + + db_manager.save_failure.assert_called_once() + args = db_manager.save_failure.call_args[0] + assert args[0] == "ProdDB" + assert args[1] == SCAN_FAILURE_REASON_SAMPLING_ERROR + assert "users" in args[2] + + +def test_process_one_finding_sampling_error_skips_scan_column(): + """Failed sampling must not run detection on an empty pseudo-clean sample.""" + target = {"type": "database", "driver": "sqlite", "name": "T"} + scanner = MagicMock() + db_manager = MagicMock() + connector = SQLConnector(target, scanner, db_manager) + connector.sample = MagicMock(side_effect=ColumnSampleError()) + + connector._process_one_finding("T", "localhost", "sqlite", "", "t1", "c1", "TEXT") + + scanner.scan_column.assert_not_called() From edd052edfa03b088ac48998e29b85f46f1766b55 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?FabioLeit=C3=A3o?= Date: Thu, 2 Jul 2026 22:33:25 -0300 Subject: [PATCH 2/3] fix(connector): keep minor finding when full SQL sample fails (#1140) Bugbot: full-scan ColumnSampleError must not drop the first-pass DOB_POSSIBLE_MINOR result. Add failure_hint for sampling_error in reports. --- connectors/sql_connector.py | 21 ++++++++++++--------- core/database.py | 8 +++++++- tests/test_database.py | 1 + 3 files changed, 20 insertions(+), 10 deletions(-) diff --git a/connectors/sql_connector.py b/connectors/sql_connector.py index f95238ab3..2a01a1ab3 100644 --- a/connectors/sql_connector.py +++ b/connectors/sql_connector.py @@ -363,16 +363,19 @@ def _process_one_finding( try: full_sample = self.sample(schema, table, cname, limit=full_scan_limit) except ColumnSampleError: - return - full_res = self.scanner.scan_column( - cname, full_sample, connector_data_type=ctype - ) - if "DOB_POSSIBLE_MINOR" in (full_res.get("pattern_detected") or ""): - res = full_res - suffix = " (full-scan confirmed)" - norm_tag = ( - (norm_tag or "").rstrip() + suffix if norm_tag else suffix.lstrip() + pass + else: + full_res = self.scanner.scan_column( + cname, full_sample, connector_data_type=ctype ) + if "DOB_POSSIBLE_MINOR" in (full_res.get("pattern_detected") or ""): + res = full_res + suffix = " (full-scan confirmed)" + norm_tag = ( + (norm_tag or "").rstrip() + suffix + if norm_tag + else suffix.lstrip() + ) self.db_manager.save_finding( source_type="database", target_name=target_name, diff --git a/core/database.py b/core/database.py index 3e3704355..21b1463e2 100644 --- a/core/database.py +++ b/core/database.py @@ -131,7 +131,7 @@ class ScanFailure(Base): target_name = Column(String(100)) reason = Column( String(50) - ) # unreachable, auth_failed, permission_denied, timeout, error + ) # unreachable, auth_failed, permission_denied, timeout, sampling_error, error details = Column(Text, nullable=True) created_at = Column(DateTime, default=_utc_now) @@ -163,6 +163,12 @@ def failure_hint(reason: str) -> str: "Consider increasing timeout values and re-running during off-peak hours. " "You can set timeouts in config (timeouts.connect_seconds, timeouts.read_seconds) or per target; see USAGE.md." ) + if r == "sampling_error": + return ( + "Column sampling failed (SQL syntax, permissions, or connectivity). The column was not " + "evaluated — this is not a clean result. Check dialect-specific sampling notes, credentials, " + "and the detailed message; fix the target or sampling plan before re-running." + ) return ( "Unexpected error. Review the detailed message and audit log, verify the target configuration " "(host, port, path, credentials) and test connectivity manually before re-running." diff --git a/tests/test_database.py b/tests/test_database.py index 78d09d298..dbd362fbc 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -219,6 +219,7 @@ def test_failure_hint_other_reasons(): ) assert "auth" in failure_hint("auth_failed").lower() assert "permission" in failure_hint("permission_denied").lower() + assert "sampling" in failure_hint("sampling_error").lower() assert ( "unexpected" in failure_hint("unknown").lower() or "error" in failure_hint("unknown").lower() From 99c1f756a86e8e7f29fc6add448fec3c7e65ef1f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?FabioLeit=C3=A3o?= Date: Thu, 2 Jul 2026 22:51:08 -0300 Subject: [PATCH 3/3] test(connector): minor full-scan regression + CodeQL logging note (#1140) - Document best-effort audit logging except at sample() strategy log (CodeQL :468). - Assert DOB_POSSIBLE_MINOR first-pass finding survives full-scan ColumnSampleError and sampling_error scan_failure is still recorded. --- connectors/sql_connector.py | 2 +- tests/test_sql_connector.py | 45 +++++++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/connectors/sql_connector.py b/connectors/sql_connector.py index 2a01a1ab3..cdbba705b 100644 --- a/connectors/sql_connector.py +++ b/connectors/sql_connector.py @@ -466,7 +466,7 @@ def sample( msg += " notes=%s" % (plan.audit_notes,) get_logger().info(msg) except Exception: - pass + pass # best-effort audit logging — never break sampling for a log line try: with self._connection.begin(): if dialect in ("postgresql", "postgres") and to: diff --git a/tests/test_sql_connector.py b/tests/test_sql_connector.py index 6a830655d..dd5a2764a 100644 --- a/tests/test_sql_connector.py +++ b/tests/test_sql_connector.py @@ -315,3 +315,48 @@ def test_process_one_finding_sampling_error_skips_scan_column(): connector._process_one_finding("T", "localhost", "sqlite", "", "t1", "c1", "TEXT") scanner.scan_column.assert_not_called() + + +def test_minor_full_scan_sample_error_keeps_first_pass_dob_and_records_failure(): + """Full-scan ColumnSampleError must not discard DOB_POSSIBLE_MINOR from the first pass (#1140).""" + target = {"type": "database", "driver": "sqlite", "name": "MinorDB"} + scanner = MagicMock() + scanner.scan_column.return_value = { + "sensitivity_level": "MEDIUM", + "pattern_detected": "DOB_POSSIBLE_MINOR", + "norm_tag": "", + "ml_confidence": 50, + } + db_manager = MagicMock() + connector = SQLConnector( + target, + scanner, + db_manager, + detection_config={"minor_full_scan": True, "minor_full_scan_limit": 100}, + ) + + def sample_side_effect(schema, table, cname, limit=None): + if limit is not None: + db_manager.save_failure( + "MinorDB", + SCAN_FAILURE_REASON_SAMPLING_ERROR, + "dbo.minors.dob dialect=sqlite: ProgrammingError: full scan failed", + ) + raise ColumnSampleError() + return "2005-01-01" + + connector.sample = MagicMock(side_effect=sample_side_effect) + + connector._process_one_finding( + "MinorDB", "localhost", "sqlite", "", "minors", "dob", "DATE" + ) + + db_manager.save_failure.assert_called_once_with( + "MinorDB", + SCAN_FAILURE_REASON_SAMPLING_ERROR, + "dbo.minors.dob dialect=sqlite: ProgrammingError: full scan failed", + ) + db_manager.save_finding.assert_called_once() + finding_kwargs = db_manager.save_finding.call_args.kwargs + assert "DOB_POSSIBLE_MINOR" in finding_kwargs["pattern_detected"] + assert "(full-scan confirmed)" not in (finding_kwargs.get("norm_tag") or "")