Escape SQL Server passwords in connection strings (#24138)

* Escape SQL Server passwords in connection strings

* Add changelog for SQL Server password escaping

* Fix SQL Server semicolon password test cleanup

* Document SQL Server password escaping rules

* Clean up SQL Server password test sessions

* Avoid SQL Server password test cleanup races

* Clarify SQL Server password escaping comments

* Cover SQL Server password delimiter characters

* Stabilize SQL Server password escape test

* Use one SQL Server special-character password test

* Handle legacy FreeTDS password brace parsing

* Select FreeTDS password escape directly

* Scope FreeTDS password escape to configured driver

* Reject unsupported FreeTDS password sequence

Author: lu-zhengda

sqlserver/changelog.d/24138.fixed

@@ -0,0 +1 @@
+Escape SQL Server passwords when building connection strings.

sqlserver/datadog_checks/sqlserver/connection.py

@@ -137,6 +137,47 @@ def parse_connection_string_properties(cs):
     return params
 
 
+def _escape_odbc_connection_string_value(value: str) -> str:
+    """Return a value escaped for an ODBC connection string."""
+    # SQL Server passwords may contain symbols, but ODBC connection strings use
+    # semicolons to separate key/value pairs. The ODBC string spec wraps such
+    # values in braces and escapes a literal right brace as "}}"; the outer
+    # braces are delimiters, not part of the password.
+    # Password: pa;ss}word
+    # Connection string: Server=host;UID=datadog;PWD={pa;ss}}word};
+    # Sources: https://learn.microsoft.com/en-us/sql/relational-databases/security/strong-passwords
+    # https://sqlprotocoldoc.z19.web.core.windows.net/MS-ODBCSTR/%5BMS-ODBCSTR%5D-100604.pdf
+    return '{{{}}}'.format(value.replace('}', '}}'))
+
+
+def _escape_legacy_freetds_odbc_connection_string_value(value: str) -> str:
+    """Return a value escaped for legacy FreeTDS ODBC connection strings."""
+    # FreeTDS 1.3.6 does not decode "}}" inside braced values. It treats "}" as
+    # part of the password unless the next character is ";", so the FreeTDS path
+    # keeps the brace raw.
+    # Password: pa;ss}word
+    # Connection string: Driver=FreeTDS;Server=host;UID=datadog;PWD={pa;ss}word};
+    # Source: https://github.com/FreeTDS/freetds/blob/v1.3.6/src/odbc/connectparams.c
+    return '{{{}}}'.format(value)
+
+
+def _is_freetds_driver(driver: str) -> bool:
+    """Return whether an ODBC driver value points to FreeTDS."""
+    return 'freetds' in driver.lower() or 'libtdsodbc' in driver.lower()
+
+
+def _escape_adodbapi_connection_string_value(value: str) -> str:
+    """Return a value escaped for an adodbapi connection string."""
+    # adodbapi uses an ADO/OLE DB connection string, where semicolons separate
+    # key/value pairs. Quoting keeps semicolons inside the password value; a
+    # literal quote inside that value is escaped by doubling it. The outer quotes
+    # are delimiters, not part of the password.
+    # Password: pa;ss"word
+    # Connection string: Provider=MSOLEDBSQL;User ID=datadog;Password="pa;ss""word";
+    # Docs: https://learn.microsoft.com/en-us/sql/connect/oledb/applications/using-connection-string-keywords-with-oledb-driver-for-sql-server
+    return '"{}"'.format(value.replace('"', '""'))
+
+
 class Connection(object):
     """Manages the connection to a SQL Server instance."""
 
@@ -614,6 +655,11 @@ def _conn_string_odbc(self, db_key, conn_key=None, db_name=None):
         else:
             dsn, host, username, password, database, driver = self._get_access_info(db_key, db_name)
 
+        escape_func = _escape_odbc_connection_string_value
+        is_freetds_driver = driver and _is_freetds_driver(driver)
+        if is_freetds_driver:
+            escape_func = _escape_legacy_freetds_odbc_connection_string_value
+
         if self.managed_auth_enabled:
             # if managed_identity authentication is configured,
             # remove the username/password from the CS, if set
@@ -638,7 +684,12 @@ def _conn_string_odbc(self, db_key, conn_key=None, db_name=None):
             conn_str += 'UID={};'.format(username)
         self.log.debug("Connection string (before password) %s", conn_str)
         if password:
-            conn_str += 'PWD={};'.format(password)
+            if is_freetds_driver and '};' in password:
+                raise ConfigurationError(
+                    "SQL Server passwords containing the sequence '};' cannot be represented in FreeTDS ODBC "
+                    "connection strings. Use Microsoft ODBC Driver for SQL Server or change the password."
+                )
+            conn_str += 'PWD={};'.format(escape_func(password))
         return conn_str
 
     def _conn_string_adodbapi(self, db_key, conn_key=None, db_name=None):
@@ -659,7 +710,7 @@ def _conn_string_adodbapi(self, db_key, conn_key=None, db_name=None):
             conn_str += 'User ID={};'.format(username)
         self.log.debug("Connection string (before password) %s", conn_str)
         if password:
-            conn_str += 'Password={};'.format(password)
+            conn_str += 'Password={};'.format(_escape_adodbapi_connection_string_value(password))
         if not username and not password:
             conn_str += 'Integrated Security=SSPI;'
         return conn_str

sqlserver/tests/test_connection.py

@@ -26,6 +26,8 @@
 from .common import CHECK_NAME, SQLSERVER_YEAR
 
 KEY_PREFIX = "dbm-test-"
+SPECIAL_CHARACTERS_PASSWORD_LOGIN = 'datadog_special_characters_password'
+SPECIAL_CHARACTERS_PASSWORD = 'Pa;ss}"word123!'
 
 
 @pytest.mark.unit
@@ -286,6 +288,94 @@ def test_config_with_and_without_port(instance_minimal_defaults, host, port, exp
     assert result_host == expected_host
 
 
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    'connector,driver,password,expected_password_parameter',
+    [
+        pytest.param('odbc', None, SPECIAL_CHARACTERS_PASSWORD, 'PWD={Pa;ss}}"word123!};', id='odbc'),
+        pytest.param('odbc', 'FreeTDS', SPECIAL_CHARACTERS_PASSWORD, 'PWD={Pa;ss}"word123!};', id='odbc-freetds'),
+        pytest.param('adodbapi', None, SPECIAL_CHARACTERS_PASSWORD, 'Password="Pa;ss}""word123!";', id='adodbapi'),
+    ],
+)
+def test_connection_string_escapes_password_special_characters(
+    instance_minimal_defaults: dict[str, object],
+    connector: str,
+    driver: str | None,
+    password: str,
+    expected_password_parameter: str,
+) -> None:
+    instance_minimal_defaults.update({'connector': connector, 'password': password})
+    if driver:
+        instance_minimal_defaults['driver'] = driver
+    connection = Connection({}, instance_minimal_defaults, None)
+
+    if connector == 'odbc':
+        conn_str = connection._conn_string_odbc('database')
+    else:
+        conn_str = connection._conn_string_adodbapi('database')
+
+    assert expected_password_parameter in conn_str
+
+
+@pytest.mark.unit
+def test_freetds_password_with_closing_brace_semicolon_raises_limitation(
+    instance_minimal_defaults: dict[str, object],
+) -> None:
+    password = 'pass};word'
+    instance_minimal_defaults.update({'connector': 'odbc', 'driver': 'FreeTDS', 'password': password})
+    connection = Connection({}, instance_minimal_defaults, None)
+
+    with pytest.raises(ConfigurationError) as exc_info:
+        connection._conn_string_odbc('database')
+
+    error_message = str(exc_info.value)
+    assert "'};'" in error_message
+    assert 'FreeTDS' in error_message
+    assert 'Microsoft ODBC Driver for SQL Server' in error_message
+    assert password not in error_message
+
+
+@pytest.fixture
+def special_characters_password_login(sa_conn: object) -> tuple[str, str]:
+    with sa_conn.cursor() as cursor:
+        cursor.execute(
+            """
+            IF NOT EXISTS (
+                SELECT 1 FROM sys.server_principals WHERE name = N'{login}'
+            )
+                CREATE LOGIN [{login}] WITH PASSWORD = '{password}', CHECK_POLICY = OFF
+            ELSE
+                ALTER LOGIN [{login}] WITH PASSWORD = '{password}', CHECK_POLICY = OFF
+            """.format(login=SPECIAL_CHARACTERS_PASSWORD_LOGIN, password=SPECIAL_CHARACTERS_PASSWORD)
+        )
+        cursor.execute(
+            """
+            IF NOT EXISTS (
+                SELECT 1 FROM sys.database_principals WHERE name = N'{login}'
+            )
+                CREATE USER [{login}] FOR LOGIN [{login}]
+            """.format(login=SPECIAL_CHARACTERS_PASSWORD_LOGIN)
+        )
+    return SPECIAL_CHARACTERS_PASSWORD_LOGIN, SPECIAL_CHARACTERS_PASSWORD
+
+
+@pytest.mark.integration
+@pytest.mark.usefixtures('dd_environment')
+def test_connection_with_special_characters_in_password(
+    instance_docker: dict[str, object], special_characters_password_login: tuple[str, str]
+) -> None:
+    login_name, password = special_characters_password_login
+    instance_docker['username'] = login_name
+    instance_docker['password'] = password
+    check = SQLServer(CHECK_NAME, {}, [instance_docker])
+    check.initialize_connection()
+
+    with check.connection.open_managed_default_connection(KEY_PREFIX):
+        with check.connection.get_managed_cursor(KEY_PREFIX) as cursor:
+            cursor.execute("select 1")
+            assert cursor.fetchall()
+
+
 @pytest.mark.flaky
 @pytest.mark.integration
 @pytest.mark.usefixtures('dd_environment')