feat(sqlserver): support client certificates on the ODBC connection string

encrypt and trust_server_certificate only cover server-side encryption,
so a client certificate could not be presented at all. Add
client_certificate, client_key, client_key_password and
host_name_in_certificate to SQLServerConnectionConfig and emit them as the
ClientCertificate/ClientKey/ClientKeyPassword/HostNameInCertificate ODBC
keywords from build_connection_string(), with the same four names accepted
as backend kwargs.

validate_config() now rejects a client_key given without a
client_certificate, so the mismatch surfaces as a clear ValueError instead
of an opaque driver failure at connect time.

Author: vistart

changelog.d/5.fixed.md

@@ -0,0 +1 @@
+Add client-certificate (mutual TLS) support to the ODBC connection string. `SQLServerConnectionConfig` could only express server-side encryption via `encrypt` and `trust_server_certificate`; there was no way to present a client certificate. The new `client_certificate`, `client_key`, `client_key_password` and `host_name_in_certificate` fields are emitted as the `ClientCertificate`, `ClientKey`, `ClientKeyPassword` and `HostNameInCertificate` ODBC keywords by `build_connection_string()`, and are accepted as backend kwargs. `validate_config()` now rejects a `client_key` supplied without a `client_certificate`, so the mismatch surfaces as a clear error instead of an opaque driver failure.

src/rhosocial/activerecord/backend/impl/sqlserver/backend/async_backend.py

@@ -100,7 +100,10 @@ def __init__(self, **kwargs):
             sqlserver_params = [
                 'host', 'port', 'database', 'deployment_target', 'username', 'password',
                 'trusted_connection', 'driver', 'encrypt',
-                'trust_server_certificate', 'timeout', 'query_timeout',
+                'trust_server_certificate',
+                'client_certificate', 'client_key', 'client_key_password',
+                'host_name_in_certificate',
+                'timeout', 'query_timeout',
                 'autocommit', 'charset', 'pool_size', 'pool_timeout',
             ]
 

src/rhosocial/activerecord/backend/impl/sqlserver/backend/backend.py

@@ -122,7 +122,10 @@ def __init__(self, **kwargs):
             sqlserver_params = [
                 'host', 'port', 'database', 'deployment_target', 'username', 'password',
                 'trusted_connection', 'driver', 'encrypt',
-                'trust_server_certificate', 'timeout', 'query_timeout',
+                'trust_server_certificate',
+                'client_certificate', 'client_key', 'client_key_password',
+                'host_name_in_certificate',
+                'timeout', 'query_timeout',
                 'autocommit', 'charset', 'pool_size', 'pool_timeout',
             ]
 

src/rhosocial/activerecord/backend/impl/sqlserver/config.py

@@ -72,6 +72,15 @@ class SQLServerConnectionConfig(ConnectionConfig):
     encrypt_connection: Optional[bool] = None
     trust_server_certificate: bool = False
 
+    # Client certificate (mutual TLS). The ODBC driver reads these straight
+    # from the filesystem, so they are PEM/PFX paths rather than material.
+    client_certificate: Optional[str] = None
+    client_key: Optional[str] = None
+    client_key_password: Optional[str] = None
+    # Overrides the name matched against the server certificate. Required when
+    # the certificate CN is not the host name being connected to.
+    host_name_in_certificate: Optional[str] = None
+
     timeout: int = 30
     query_timeout: int = 0
 
@@ -136,7 +145,16 @@ def build_connection_string(self) -> str:
 
         if self.charset and self.charset.upper() != "UTF-8":
             parts.append(f"CharacterSet={self.charset}")
-        
+
+        if self.client_certificate:
+            parts.append(f"ClientCertificate={self.client_certificate}")
+        if self.client_key:
+            parts.append(f"ClientKey={self.client_key}")
+        if self.client_key_password:
+            parts.append(f"ClientKeyPassword={self.client_key_password}")
+        if self.host_name_in_certificate:
+            parts.append(f"HostNameInCertificate={self.host_name_in_certificate}")
+
         for key, value in self.options.items():
             if value is not None:
                 parts.append(f"{key}={value}")
@@ -164,6 +182,11 @@ def validate_config(self) -> None:
                 UserWarning,
                 stacklevel=2
             )
+
+        if self.client_key and not self.client_certificate:
+            raise ValueError(
+                "client_key requires client_certificate for SQL Server connections"
+            )
 
     def __repr__(self) -> str:
         """Return safe string representation (password masked)."""

tests/rhosocial/activerecord_sqlserver_test/feature/backend/test_ssl_connection_params.py

@@ -0,0 +1,65 @@
+# tests/rhosocial/activerecord_sqlserver_test/feature/backend/test_ssl_connection_params.py
+"""Regression tests: SQL Server TLS settings must reach the ODBC connection string.
+
+``build_connection_string()`` is the only place ``pyodbc`` gets its
+configuration, so anything the config exposes for TLS has to appear there.
+"""
+
+import pytest
+
+from rhosocial.activerecord.backend.impl.sqlserver.config import SQLServerConnectionConfig
+
+
+def make_config(**kwargs):
+    defaults = dict(host="db.example.com", database="test_db", username="sa", password="secret")
+    defaults.update(kwargs)
+    return SQLServerConnectionConfig(**defaults)
+
+
+def test_encryption_is_on_by_default():
+    assert "Encrypt=yes" in make_config().build_connection_string()
+    assert "TrustServerCertificate=no" in make_config().build_connection_string()
+
+
+def test_encrypt_can_be_disabled():
+    conn_str = make_config(encrypt=False).build_connection_string()
+    assert "Encrypt=no" in conn_str
+
+
+def test_trust_server_certificate_is_forwarded():
+    conn_str = make_config(trust_server_certificate=True).build_connection_string()
+    assert "TrustServerCertificate=yes" in conn_str
+
+
+def test_client_certificate_is_forwarded():
+    conn_str = make_config(
+        client_certificate="/certs/client.pem",
+        client_key="/certs/client.key",
+    ).build_connection_string()
+    assert "ClientCertificate=/certs/client.pem" in conn_str
+    assert "ClientKey=/certs/client.key" in conn_str
+
+
+def test_host_name_in_certificate_is_forwarded():
+    """Needed when the server certificate CN is not the connect host."""
+    conn_str = make_config(host_name_in_certificate="db.internal").build_connection_string()
+    assert "HostNameInCertificate=db.internal" in conn_str
+
+
+def test_client_key_password_is_forwarded():
+    conn_str = make_config(
+        client_certificate="/certs/client.pfx",
+        client_key_password="pfx-secret",
+    ).build_connection_string()
+    assert "ClientKeyPassword=pfx-secret" in conn_str
+
+
+def test_no_client_tls_keys_when_unset():
+    conn_str = make_config().build_connection_string()
+    for key in ("ClientCertificate", "ClientKey", "ClientKeyPassword", "HostNameInCertificate"):
+        assert key not in conn_str
+
+
+def test_client_key_without_certificate_is_rejected():
+    with pytest.raises(ValueError, match="client_key requires client_certificate"):
+        make_config(client_key="/certs/client.key").validate_config()