Allow SSLContext ciphers to be customized

datadog_checks_base/changelog.d/19312.added

@@ -0,0 +1 @@
+Allow for Ciphers to be customizable in the SSLContext creation

datadog_checks_base/datadog_checks/base/utils/tls.py

@@ -28,6 +28,7 @@
     'tls_private_key': None,
     'tls_private_key_password': None,
     'tls_validate_hostname': True,
+    'tls_ciphers': 'ALL',
 }
 
 
@@ -115,6 +116,15 @@ def _create_tls_context(self):
         else:
             context.check_hostname = False
 
+        ciphers = self.config.get('tls_ciphers')
+        if ciphers:
+            if 'ALL' in ciphers:
+                updated_ciphers = "ALL"
+            else:
+                updated_ciphers = ":".join(ciphers)
+
+            context.set_ciphers(updated_ciphers)
+
         # https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_verify_locations
         # https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_default_certs
         ca_cert = self.config['tls_ca_cert']

datadog_checks_base/tests/base/utils/test_tls.py

@@ -299,6 +299,64 @@ def test_client_key_expanded_tls_verify_false(self):
             check.get_tls_context()
             mock_expand.assert_called_with('~/foo')
 
+    @pytest.mark.parametrize(
+        'instance,expected_ciphers',
+        [
+            pytest.param(
+                {'tls_verify': False},
+                'ALL',
+                id="Construct ciphers with no config",
+            ),
+            pytest.param(
+                {'tls_ciphers': ['TLS_RSA_WITH_SEED_CBC_SHA', 'TLS_SM4_GCM_SM3']},
+                'TLS_RSA_WITH_SEED_CBC_SHA:TLS_SM4_GCM_SM3',
+                id='Construct ciphers with specific ciphers',
+            ),
+            pytest.param(
+                {'tls_ciphers': ['ALL']},
+                'ALL',
+                id="Construct Ciphers with 'ALL' ciphers",
+            ),
+        ],
+    )
+    def test_cipher_construction(self, instance, expected_ciphers):
+        with patch.object(ssl.SSLContext, 'set_ciphers') as mock_set_ciphers:
+            check = AgentCheck('test', {}, [instance])
+            check.get_tls_context()
+            mock_set_ciphers.assert_called_once_with(expected_ciphers)
+
+    @pytest.mark.parametrize(
+        'instance,expected_ciphers',
+        [
+            pytest.param(
+                {'tls_verify': False},
+                'ALL',
+                id="No Ciphers, default to 'ALL'",
+            ),
+            pytest.param(
+                {'tls_ciphers': ['PSK-CAMELLIA128-SHA256', 'DHE-PSK-CAMELLIA128-SHA256']},
+                'PSK-CAMELLIA128-SHA256:DHE-PSK-CAMELLIA128-SHA256',
+                id='Add specific ciphers only',
+            ),
+            pytest.param(
+                {'tls_ciphers': ['ALL']},
+                'ALL',
+                id="'ALL' manually",
+            ),
+        ],
+    )
+    def test_ciphers(self, instance, expected_ciphers):
+        check = AgentCheck('test', {}, [instance])
+        context = check.get_tls_context()
+
+        expected_context = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT)
+        expected_context.set_ciphers(expected_ciphers)
+
+        actual_ciphers = sorted(cipher['name'] for cipher in context.get_ciphers())
+        expected_ciphers_list = sorted(cipher['name'] for cipher in expected_context.get_ciphers())
+
+        assert actual_ciphers == expected_ciphers_list
+
 
 class TestTLSContextOverrides:
     def test_override_context(self):