[sqlserver] Use parameterized queries in AO and fragmentation metrics (#23426)

* [sqlserver] Use parameterized queries in AO and fragmentation metrics

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* [sqlserver] Fix hardcoded FreeTDS version in get_local_driver

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: jasonmp85

sqlserver/changelog.d/23426.fixed

@@ -0,0 +1 @@
+Escape single quotes in SQL string literal interpolation for robustness against unusual database and object names.

sqlserver/datadog_checks/sqlserver/database_metrics/availability_groups_metrics.py

@@ -52,7 +52,8 @@ def enabled(self):
     def queries(self):
         query = AVAILABILITY_GROUPS_METRICS_QUERY.copy()
         if self.availability_group:
-            query['query'] += f" where resource_group_id = '{self.availability_group}'"
+            query['query'] += " where resource_group_id = ?"
+            query['params'] = (self.availability_group,)
         return [query]
 
     def __repr__(self) -> str:

sqlserver/datadog_checks/sqlserver/database_metrics/availability_replicas_metrics.py

@@ -72,13 +72,17 @@ def queries(self):
         query = AVAILABILITY_REPLICAS_METRICS_QUERY.copy()
         if self.availability_group or self.only_emit_local or self.ao_database:
             where_clauses = []
+            params = []
             if self.availability_group:
-                where_clauses.append(f"resource_group_id = '{self.availability_group}'")
+                where_clauses.append("resource_group_id = ?")
+                params.append(self.availability_group)
             if self.only_emit_local:
                 where_clauses.append("is_local = 1")
             if self.ao_database:
-                where_clauses.append(f"database_name = '{self.ao_database}'")
+                where_clauses.append("database_name = ?")
+                params.append(self.ao_database)
             query['query'] += f" where {' and '.join(where_clauses)}"
+            query['params'] = tuple(params)
         if self.major_version >= 12:
             # This column only supported in SQL Server 2014 and later
             is_primary_replica = "is_primary_replica"

sqlserver/datadog_checks/sqlserver/database_metrics/database_replication_stats_metrics.py

@@ -53,11 +53,14 @@ def queries(self):
         query = DATABASE_REPLICATION_STATS_METRICS_QUERY.copy()
         if self.availability_group or self.only_emit_local:
             where_clauses = []
+            params = []
             if self.availability_group:
-                where_clauses.append(f"resource_group_id = '{self.availability_group}'")
+                where_clauses.append("resource_group_id = ?")
+                params.append(self.availability_group)
             if self.only_emit_local:
                 where_clauses.append("is_local = 1")
             query['query'] += f" where {' and '.join(where_clauses)}"
+            query['params'] = tuple(params)
         return [query]
 
     def __repr__(self) -> str:

sqlserver/datadog_checks/sqlserver/database_metrics/db_fragmentation_metrics.py

@@ -21,7 +21,7 @@
         DDIPS.avg_fragment_size_in_pages as avg_fragment_size_in_pages,
         DDIPS.page_count as page_count,
         DDIPS.avg_fragmentation_in_percent as avg_fragmentation_in_percent
-        FROM sys.dm_db_index_physical_stats (DB_ID('{db}'),null,null,null,null) as DDIPS
+        FROM sys.dm_db_index_physical_stats (DB_ID(?),null,null,null,null) as DDIPS
         INNER JOIN sys.indexes as I WITH (NOLOCK) ON I.object_id = DDIPS.object_id
         AND DDIPS.index_id = I.index_id
         WHERE DDIPS.fragment_count is not null
@@ -112,14 +112,19 @@ def __repr__(self) -> str:
 
     def _build_query_executors(self):
         executors = []
+        if self.db_fragmentation_object_names:
+            placeholders = ','.join(['?'] * len(self.db_fragmentation_object_names))
+            object_name_filter = f" AND OBJECT_NAME(DDIPS.object_id, DDIPS.database_id) IN ({placeholders})"
+        else:
+            object_name_filter = None
         for database in self.databases:
             queries = copy.deepcopy(self.queries)
             for query in queries:
-                query['query'] = query['query'].format(db=database)
-                if self.db_fragmentation_object_names:
-                    query['query'] += " AND OBJECT_NAME(DDIPS.object_id, DDIPS.database_id) IN ({})".format(
-                        ','.join(["'{}'".format(name) for name in self.db_fragmentation_object_names])
-                    )
+                params = [database]
+                if object_name_filter:
+                    query['query'] += object_name_filter
+                    params.extend(self.db_fragmentation_object_names)
+                query['params'] = tuple(params)
             executor = self.new_query_executor(
                 queries,
                 executor=functools.partial(self.execute_query_handler, db=database),

sqlserver/datadog_checks/sqlserver/sqlserver.py

@@ -1064,13 +1064,16 @@ def collect_metrics(self):
                 with self.connection.get_managed_cursor(KEY_PREFIX) as cursor:
                     cursor.execute("SET NOCOUNT OFF")
 
-    def execute_query_raw(self, query, db=None):
+    def execute_query_raw(self, query, db=None, params=None):
         with self.connection.get_managed_cursor(KEY_PREFIX) as cursor:
             if db:
                 ctx = construct_use_statement(db)
                 self.log.debug("changing cursor context via use statement: %s", ctx)
                 cursor.execute(ctx)
-            cursor.execute(query)
+            if params is not None:
+                cursor.execute(query, params)
+            else:
+                cursor.execute(query)
             return cursor.fetchall()
 
     def do_stored_procedure_check(self):

sqlserver/pyproject.toml

@@ -28,7 +28,7 @@ classifiers = [
     "Private :: Do Not Upload",
 ]
 dependencies = [
-    "datadog-checks-base>=37.34.1",
+    "datadog-checks-base>=37.36.0",
 ]
 dynamic = [
     "version",

sqlserver/tests/common.py

@@ -37,7 +37,7 @@ def get_local_driver():
     we need to define the 'FreeTDS' driver in odbcinst.ini
     """
     if ON_MACOS:
-        return '/opt/homebrew/Cellar/freetds/1.4.26/lib/libtdsodbc.so'
+        return '/opt/homebrew/lib/libtdsodbc.so'
     elif ON_WINDOWS:
         return '{ODBC Driver 18 for SQL Server}'
     else:

sqlserver/tests/test_database_metrics.py

@@ -255,6 +255,59 @@ def test_sqlserver_ao_metrics(
                 aggregator.assert_metric(metric_name, value=metric_value, tags=expected_tags)
 
 
+def test_sqlserver_availability_groups_query_uses_parameterized_queries(init_config, instance_docker_metrics):
+    ag_with_quotes = "AG1'; DROP TABLE foo; --"
+    instance_docker_metrics['database_metrics'] = {
+        'ao_metrics': {'enabled': True, 'availability_group': ag_with_quotes},
+    }
+    sqlserver_check = SQLServer(CHECK_NAME, init_config, [instance_docker_metrics])
+    metrics = SqlserverAvailabilityGroupsMetrics(
+        config=sqlserver_check._config,
+        new_query_executor=sqlserver_check._new_query_executor,
+        server_static_info=STATIC_SERVER_INFO,
+        execute_query_handler=mock.MagicMock(),
+    )
+    q = metrics.queries[0]
+    assert "resource_group_id = ?" in q['query']
+    assert q['params'] == (ag_with_quotes,)
+
+
+def test_sqlserver_database_replication_stats_query_uses_parameterized_queries(init_config, instance_docker_metrics):
+    ag_with_quotes = "AG1'; DROP TABLE foo; --"
+    instance_docker_metrics['database_metrics'] = {
+        'ao_metrics': {'enabled': True, 'availability_group': ag_with_quotes},
+    }
+    sqlserver_check = SQLServer(CHECK_NAME, init_config, [instance_docker_metrics])
+    metrics = SqlserverDatabaseReplicationStatsMetrics(
+        config=sqlserver_check._config,
+        new_query_executor=sqlserver_check._new_query_executor,
+        server_static_info=STATIC_SERVER_INFO,
+        execute_query_handler=mock.MagicMock(),
+    )
+    q = metrics.queries[0]
+    assert "resource_group_id = ?" in q['query']
+    assert q['params'] == (ag_with_quotes,)
+
+
+def test_sqlserver_availability_replicas_query_uses_parameterized_queries(init_config, instance_docker_metrics):
+    ag_with_quotes = "AG1'; DROP TABLE foo; --"
+    db_with_quotes = "mydb'; DROP TABLE bar; --"
+    instance_docker_metrics['database_metrics'] = {
+        'ao_metrics': {'enabled': True, 'availability_group': ag_with_quotes, 'ao_database': db_with_quotes},
+    }
+    sqlserver_check = SQLServer(CHECK_NAME, init_config, [instance_docker_metrics])
+    metrics = SqlserverAvailabilityReplicasMetrics(
+        config=sqlserver_check._config,
+        new_query_executor=sqlserver_check._new_query_executor,
+        server_static_info=STATIC_SERVER_INFO,
+        execute_query_handler=mock.MagicMock(),
+    )
+    q = metrics.queries[0]
+    assert "resource_group_id = ?" in q['query']
+    assert "database_name = ?" in q['query']
+    assert q['params'] == (ag_with_quotes, db_with_quotes)
+
+
 @pytest.mark.integration
 @pytest.mark.usefixtures('dd_environment')
 @pytest.mark.parametrize('include_ao_metrics', [True, False])
@@ -287,7 +340,7 @@ def test_sqlserver_availability_groups_metrics(
 
     sqlserver_check = SQLServer(CHECK_NAME, init_config, [instance_docker_metrics])
 
-    def execute_query_handler_mocked(query, db=None):
+    def execute_query_handler_mocked(query, db=None, params=None):
         return mocked_results
 
     availability_groups_metrics = SqlserverAvailabilityGroupsMetrics(
@@ -298,9 +351,9 @@ def execute_query_handler_mocked(query, db=None):
     )
 
     if availability_group:
-        assert availability_groups_metrics.queries[0]['query'].endswith(
-            f" where resource_group_id = '{availability_group}'"
-        )
+        q = availability_groups_metrics.queries[0]
+        assert "resource_group_id = ?" in q['query']
+        assert q.get('params') == (availability_group,)
 
     sqlserver_check._database_metrics = [availability_groups_metrics]
 
@@ -389,7 +442,7 @@ def test_sqlserver_database_replication_stats_metrics(
 
     sqlserver_check = SQLServer(CHECK_NAME, init_config, [instance_docker_metrics])
 
-    def execute_query_handler_mocked(query, db=None):
+    def execute_query_handler_mocked(query, db=None, params=None):
         return mocked_results
 
     database_replication_stats_metrics = SqlserverDatabaseReplicationStatsMetrics(
@@ -400,7 +453,8 @@ def execute_query_handler_mocked(query, db=None):
     )
 
     if availability_group:
-        assert f"resource_group_id = '{availability_group}'" in database_replication_stats_metrics.queries[0]['query']
+        assert "resource_group_id = ?" in database_replication_stats_metrics.queries[0]['query']
+        assert availability_group in database_replication_stats_metrics.queries[0].get('params', ())
     if only_emit_local:
         assert "is_local = 1" in database_replication_stats_metrics.queries[0]['query']
 
@@ -530,7 +584,7 @@ def test_sqlserver_availability_replicas_metrics(
 
     sqlserver_check = SQLServer(CHECK_NAME, init_config, [instance_docker_metrics])
 
-    def execute_query_handler_mocked(query, db=None):
+    def execute_query_handler_mocked(query, db=None, params=None):
         return mocked_results
 
     availability_replicas_metrics = SqlserverAvailabilityReplicasMetrics(
@@ -541,11 +595,13 @@ def execute_query_handler_mocked(query, db=None):
     )
 
     if availability_group:
-        assert f"resource_group_id = '{availability_group}'" in availability_replicas_metrics.queries[0]['query']
+        assert "resource_group_id = ?" in availability_replicas_metrics.queries[0]['query']
+        assert availability_group in availability_replicas_metrics.queries[0].get('params', ())
     if only_emit_local:
         assert "is_local = 1" in availability_replicas_metrics.queries[0]['query']
     if ao_database:
-        assert f"database_name = '{ao_database}'" in availability_replicas_metrics.queries[0]['query']
+        assert "database_name = ?" in availability_replicas_metrics.queries[0]['query']
+        assert ao_database in availability_replicas_metrics.queries[0].get('params', ())
 
     sqlserver_check._database_metrics = [availability_replicas_metrics]
 
@@ -1024,6 +1080,69 @@ def test_sqlserver_index_usage_metrics(
         aggregator.assert_metric(metric_name, count=0)
 
 
+def test_sqlserver_db_fragmentation_query_uses_parameterized_queries(init_config, instance_docker_metrics):
+    instance_docker_metrics['database_autodiscovery'] = True
+    instance_docker_metrics['database_metrics'] = {
+        'db_fragmentation_metrics': {'enabled': True, 'enabled_tempdb': False},
+    }
+    db_with_quotes = "legit_db'; DROP TABLE foo; --"
+
+    sqlserver_check = SQLServer(CHECK_NAME, init_config, [instance_docker_metrics])
+
+    captured_queries = []
+
+    def capture_new_query_executor(queries, **kwargs):
+        captured_queries.extend(queries)
+        return mock.MagicMock()
+
+    db_fragmentation_metrics = SqlserverDBFragmentationMetrics(
+        config=sqlserver_check._config,
+        new_query_executor=capture_new_query_executor,
+        server_static_info=STATIC_SERVER_INFO,
+        execute_query_handler=mock.MagicMock(),
+        databases=[db_with_quotes],
+    )
+
+    db_fragmentation_metrics._build_query_executors()
+
+    assert len(captured_queries) == 1
+    q = captured_queries[0]
+    assert "DB_ID(?)" in q['query']
+    assert q['params'] == (db_with_quotes,)
+
+
+def test_sqlserver_db_fragmentation_object_names_uses_parameterized_queries(init_config, instance_docker_metrics):
+    obj_with_quotes = "my_obj'; DROP TABLE foo; --"
+    instance_docker_metrics['database_autodiscovery'] = True
+    instance_docker_metrics['database_metrics'] = {
+        'db_fragmentation_metrics': {'enabled': True, 'enabled_tempdb': False},
+    }
+    instance_docker_metrics['db_fragmentation_object_names'] = [obj_with_quotes]
+
+    sqlserver_check = SQLServer(CHECK_NAME, init_config, [instance_docker_metrics])
+
+    captured_queries = []
+
+    def capture_new_query_executor(queries, **kwargs):
+        captured_queries.extend(queries)
+        return mock.MagicMock()
+
+    db_fragmentation_metrics = SqlserverDBFragmentationMetrics(
+        config=sqlserver_check._config,
+        new_query_executor=capture_new_query_executor,
+        server_static_info=STATIC_SERVER_INFO,
+        execute_query_handler=mock.MagicMock(),
+        databases=['master'],
+    )
+
+    db_fragmentation_metrics._build_query_executors()
+
+    assert len(captured_queries) == 1
+    q = captured_queries[0]
+    assert "IN (?)" in q['query']
+    assert q['params'] == ('master', obj_with_quotes)
+
+
 @pytest.mark.integration
 @pytest.mark.usefixtures('dd_environment')
 @pytest.mark.parametrize('include_db_fragmentation_metrics', [True, False])