Add shared credential-safe SQL connection diagnostics

application/single_app/config.py

@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.350"
+VERSION = "0.261.353"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform

application/single_app/functions_appinsights.py

@@ -12,6 +12,9 @@
 from azure.monitor.opentelemetry import configure_azure_monitor
 import app_settings_cache
 from functions_assist_diagnostics import ASSIST_DIAGNOSTIC_TEXT, safe_assist_diagnostics
+from semantic_kernel_plugins.sql_connection_diagnostics import (
+    SQL_ENGINES, valid_native_error, valid_sqlstate,
+)
 
 # Singleton for the logger and Azure Monitor configuration
 _appinsights_logger = None
@@ -307,6 +310,15 @@
             if _is_sensitive_log_key(key):
                 logger_extra[f"{normalized_key}_present"] = value is not None
                 continue
+            diagnostic_key = _normalize_log_key(key)
+            if diagnostic_key in {"sqlstate", "sqlengine", "sqlnativeerror"}:
+                if (
+                    (diagnostic_key == "sqlstate" and valid_sqlstate(value))
+                    or (diagnostic_key == "sqlengine" and type(value) is str and value in SQL_ENGINES)
+                    or (diagnostic_key == "sqlnativeerror" and valid_native_error(value))
+                ):
+                    logger_extra[normalized_key] = value
+                continue
             if isinstance(value, dict):
                 logger_extra[f"{normalized_key}_count"] = len(value)
             elif isinstance(value, (list, tuple, set)):

application/single_app/route_backend_plugins.py

@@ -17,6 +17,7 @@
     build_sql_server_odbc_connection_string,
     connect_with_sql_server_odbc_fallback,
 )
+from semantic_kernel_plugins.sql_connection_diagnostics import classify_sql_connection_error
 from semantic_kernel_plugins.rocksdb_plugin import (
     AUTH_SCHEME_API_KEY,
     AUTH_SCHEME_BEARER,
@@ -3417,30 +3418,16 @@
         else:
             return jsonify({'success': False, 'error': f'Unsupported database type: {database_type}'}), 400
 
-    except ImportError as e:
-        if editor_call:
-            return jsonify({'success': False, 'error': 'The required database driver is not available on the server.'}), 400
-        if database_type == 'sqlserver' and 'libodbc' in str(e):
-            return jsonify({
-                'success': False,
-                'error': 'Database driver not installed: the container image is missing the unixODBC runtime required for SQL Server connections.'
-            }), 400
-        return jsonify({'success': False, 'error': f'Database driver not installed: {str(e)}'}), 400
     except Exception as e:
-        if editor_call:
-            log_event(
-                '[PLUGINS] SQL editor connection test failed.',
-                level=logging.WARNING,
-                extra={'database_type': database_type, 'error_type': type(e).__name__},
-            )
-            return jsonify({'success': False, 'error': 'SQL connection failed. Verify the connection and authentication settings.'}), 400
-        error_msg = str(e)
-        if database_type == 'sqlserver' and "Can't open lib 'ODBC Driver 17 for SQL Server'" in error_msg:
-            error_msg = 'The selected ODBC Driver 17 is not installed in this container image. Select ODBC Driver 18 for SQL Server or rebuild the image with Driver 17.'
-        # Sanitize error message to avoid leaking sensitive details
-        if 'password' in error_msg.lower() or 'pwd' in error_msg.lower():
-            error_msg = 'Authentication failed. Please check your credentials.'
-        return jsonify({'success': False, 'error': f'Connection failed: {error_msg}'}), 400
+        diagnostic = classify_sql_connection_error(e, database_type)
+        log_event(
+            '[SQL_CONNECTION] SQL connection check failed.',
+            level=logging.WARNING,
+            extra={**diagnostic.log_properties(), 'operation': 'sql_connection_test'},
+        )
+        return jsonify({
+            'success': False, 'error': diagnostic.message, 'error_code': diagnostic.code,
+        }), 400
 
 
 @bpap.route('/api/plugins/test-cosmos-connection', methods=['POST'])

application/single_app/semantic_kernel_plugins/sql_connection_diagnostics.py

@@ -0,0 +1,117 @@
+# sql_connection_diagnostics.py
+"""Credential-free SQL connection diagnostics, with no application bootstrap or I/O."""
+
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+
+SQL_ENGINES = frozenset({"sqlserver", "postgresql", "mysql", "sqlite", "unknown"})
+SQLSTATE_PATTERN = re.compile(r"(?:[0-9]{2}|HY|IM|S[0-9]|HV|P0|XX|F0)[0-9A-Z]{3}", re.ASCII)
+MAX_NATIVE_ERROR = 2**31 - 1
+_ODBC_NATIVE_ERROR = re.compile(r"\((-?[0-9]{1,10})\) \((?:SQLDriverConnect|SQLConnect)\)$", re.ASCII)
+_ERROR_TYPES = frozenset({
+    "Error", "DatabaseError", "InterfaceError", "OperationalError", "InternalError",
+    "ProgrammingError", "NotSupportedError", "IntegrityError", "DataError",
+    "ImportError", "ModuleNotFoundError", "TimeoutError", "ConnectionError", "OSError",
+})
+_MESSAGES = {
+    "sql_connection_failed": "SQL connection failed. Verify the connection and authentication settings.",
+    "sql_driver_unavailable": "The required database driver is not available on the server. Contact your administrator.",
+    "sql_connection_timeout": "SQL connection timed out. Check database availability and network access.",
+    "sql_authentication_failed": "SQL authentication was rejected. Verify the selected authentication settings and database access.",
+}
+
+
+def valid_sqlstate(value: object) -> bool:
+    """Accept only bounded SQLSTATE values, not provider messages or arbitrary strings."""
+    return type(value) is str and len(value) == 5 and SQLSTATE_PATTERN.fullmatch(value) is not None
+
+
+def valid_native_error(value: object) -> bool:
+    return type(value) is int and -MAX_NATIVE_ERROR <= value <= MAX_NATIVE_ERROR
+
+
+@dataclass(frozen=True)
+class SqlConnectionDiagnostic:
+    code: str
+    engine: str
+    error_type: str
+    sqlstate: Optional[str] = None
+    native_error: Optional[int] = None
+
+    @property
+    def message(self) -> str:
+        return _MESSAGES[self.code]
+
+    def log_properties(self) -> dict:
+        properties = {
+            "failure_code": self.code,
+            "sql_engine": self.engine,
+            "error_type": self.error_type,
+        }
+        if self.sqlstate is not None:
+            properties["sqlstate"] = self.sqlstate
+        if self.native_error is not None:
+            properties["sql_native_error"] = self.native_error
+        return properties
+
+
+class SqlConnectionError(RuntimeError):
+    """A connection failure safe for existing plugin result and logging handlers."""
+
+    def __init__(self, diagnostic: SqlConnectionDiagnostic):
+        self.diagnostic = diagnostic
+        super().__init__(diagnostic.message)
+
+
+def classify_sql_connection_error(error: Exception, database_type: str) -> SqlConnectionDiagnostic:
+    """Classify structured driver evidence without retaining raw exception messages."""
+    engine = database_type if type(database_type) is str else "unknown"
+    if engine in {"azure_sql", "azuresql"}:
+        engine = "sqlserver"
+    if engine not in SQL_ENGINES:
+        engine = "unknown"
+
+    sqlstate = None
+    native_error = None
+    args = error.args
+    if engine in {"sqlserver", "postgresql"}:
+        for candidate in (getattr(error, "sqlstate", None), getattr(error, "pgcode", None), args[0] if args else None):
+            if valid_sqlstate(candidate):
+                sqlstate = candidate
+                break
+    if engine == "sqlserver" and sqlstate and len(args) > 1:
+        detail = args[1]
+        # pyodbc puts the native code at the end of its SQLDriverConnect diagnostic.
+        # Never collect arbitrary numbers (ports, IDs, or credentials) from provider text.
+        if type(detail) is str and len(detail) <= 8192 and detail.startswith(f"[{sqlstate}]"):
+            match = _ODBC_NATIVE_ERROR.search(detail)
+            if match:
+                candidate = int(match.group(1))
+                if valid_native_error(candidate):
+                    native_error = candidate
+    elif engine == "mysql" and args and valid_native_error(args[0]):
+        native_error = args[0]
+    elif engine == "sqlite":
+        candidate = getattr(error, "sqlite_errorcode", None)
+        if valid_native_error(candidate):
+            native_error = candidate
+
+    code = "sql_connection_failed"
+    if isinstance(error, ImportError) or sqlstate in {"IM002", "IM003"}:
+        code = "sql_driver_unavailable"
+    elif isinstance(error, TimeoutError) or sqlstate in {"HYT00", "HYT01"}:
+        code = "sql_connection_timeout"
+    elif (
+        (sqlstate is not None and sqlstate.startswith("28"))
+        or (engine == "sqlserver" and native_error in {18452, 18456})
+        or (engine == "mysql" and native_error in {1045, 1698})
+    ):
+        code = "sql_authentication_failed"
+
+    error_type = type(error).__name__
+    return SqlConnectionDiagnostic(
+        code, engine, error_type if error_type in _ERROR_TYPES else "Exception",
+        sqlstate, native_error,
+    )

application/single_app/semantic_kernel_plugins/sql_query_plugin.py

@@ -1,3 +1,4 @@
+# sql_query_plugin.py
 """
 SQL Query Plugin for Semantic Kernel
 - Executes SQL queries against various databases (SQL Server, PostgreSQL, MySQL, SQLite)
@@ -19,6 +20,9 @@
     build_sql_server_odbc_connection_string,
     connect_with_sql_server_odbc_fallback,
 )
+from semantic_kernel_plugins.sql_connection_diagnostics import (
+    SqlConnectionError, classify_sql_connection_error,
+)
 
 # Helper class to wrap results with metadata
 class ResultWithMetadata:
@@ -189,16 +193,14 @@ def _create_connection(self):
                 conn.row_factory = sqlite3.Row
                 return conn
 
-        except ImportError as e:
-            debug_print(f"[SQL_QUERY_PLUGIN] Database driver import failed database_type={self.database_type} error={e}")
-            raise ImportError(f"Required database driver not installed for {self.database_type}: {e}")
         except Exception as e:
-            debug_print(
-                f"[SQL_QUERY_PLUGIN] Connection failed database_type={self.database_type} server={self.server} "
-                f"database={self.database} exception_type={type(e).__name__} message={e}"
+            diagnostic = classify_sql_connection_error(e, self.database_type)
+            log_event(
+                "[SQL_CONNECTION] SQL connection failed.",
+                level=logging.WARNING,
+                extra={**diagnostic.log_properties(), "operation": "sql_query_connect"},
             )
-            log_event(f"[SQL_QUERY_PLUGIN] Connection failed: {e}", extra={"database_type": self.database_type})
-            raise
+            raise SqlConnectionError(diagnostic) from None
 
     @property
     def display_name(self) -> str:
@@ -356,6 +358,8 @@ def execute_query(
                 "data": [],
                 "row_count": 0
             }
+            if isinstance(e, SqlConnectionError):
+                error_result["error_code"] = e.diagnostic.code
             return ResultWithMetadata(error_result, self.metadata)
 
     @kernel_function(description="Execute a query that returns a single scalar value (e.g., COUNT, SUM, MAX, MIN). If the database schema is provided in your instructions, use it directly to construct the query. Otherwise, call get_database_schema from the SQL Schema plugin first to discover table and column names.")
@@ -424,6 +428,8 @@ def execute_scalar(
                 "query": query,
                 "value": None
             }
+            if isinstance(e, SqlConnectionError):
+                error_result["error_code"] = e.diagnostic.code
             return ResultWithMetadata(error_result, self.metadata)
 
     @kernel_function(description="Validate a SQL query for syntax correctness and safety without executing it. Use this to pre-check complex queries before execution, especially when constructing multi-table JOINs or complex WHERE clauses.")
@@ -530,6 +536,8 @@ def query_database(
                 "data": [],
                 "row_count": 0
             }
+            if isinstance(e, SqlConnectionError):
+                error_result["error_code"] = e.diagnostic.code
             return ResultWithMetadata(error_result, self.metadata)
 
     def _clean_query(self, query: str) -> str:

application/single_app/semantic_kernel_plugins/sql_schema_plugin.py

@@ -1,3 +1,4 @@
+# sql_schema_plugin.py
 """
 SQL Schema Plugin for Semantic Kernel
 - Connects to various SQL databases (SQL Server, PostgreSQL, MySQL, SQLite)
@@ -17,6 +18,9 @@
     build_sql_server_odbc_connection_string,
     connect_with_sql_server_odbc_fallback,
 )
+from semantic_kernel_plugins.sql_connection_diagnostics import (
+    SqlConnectionError, classify_sql_connection_error,
+)
 
 # Helper class to wrap results with metadata
 class ResultWithMetadata:
@@ -175,16 +179,14 @@ def _create_connection(self):
                 database_path = self.connection_string or self.database
                 return sqlite3.connect(database_path)
 
-        except ImportError as e:
-            debug_print(f"[SQL_SCHEMA_PLUGIN] Database driver import failed database_type={self.database_type} error={e}")
-            raise ImportError(f"Required database driver not installed for {self.database_type}: {e}")
         except Exception as e:
-            debug_print(
-                f"[SQL_SCHEMA_PLUGIN] Connection failed database_type={self.database_type} server={self.server} "
-                f"database={self.database} exception_type={type(e).__name__} message={e}"
+            diagnostic = classify_sql_connection_error(e, self.database_type)
+            log_event(
+                "[SQL_CONNECTION] SQL connection failed.",
+                level=logging.WARNING,
+                extra={**diagnostic.log_properties(), "operation": "sql_schema_connect"},
             )
-            log_event(f"[SQL_SCHEMA_PLUGIN] Connection failed: {e}", extra={"database_type": self.database_type})
-            raise
+            raise SqlConnectionError(diagnostic) from None
 
     @property
     def display_name(self) -> str:
@@ -338,8 +340,11 @@ def get_database_schema(
                 "database_type": self.database_type,
                 "database": self.database
             })
+            error_result = {"error": error_msg}
+            if isinstance(e, SqlConnectionError):
+                error_result["error_code"] = e.diagnostic.code
             return ResultWithMetadata(
-                {"error": error_msg},
+                error_result,
                 {"source": "sql_schema_plugin", "success": False}
             )