[AGENT-12451] Add ciphers option for fetch_intermediate_certs (#19469)

* Add ciphers option for fetch_intermediate_certs * Add changelog * Update __slots__ * Rename attribute to tls_cipherss_allowed * Add parsing for cipher list * Add tests * Run formatter
DataDog · Jan 24, 2025 · 238a1b6 · 238a1b6
1 parent d3b9825
commit 238a1b6
Show file tree

Hide file tree

Showing 3 changed files with 46 additions and 0 deletions.
diff --git a/datadog_checks_base/changelog.d/19469.added b/datadog_checks_base/changelog.d/19469.added
@@ -0,0 +1 @@
+Add tls_ciphers option for http
diff --git a/datadog_checks_base/datadog_checks/base/utils/http.py b/datadog_checks_base/datadog_checks/base/utils/http.py
@@ -94,6 +94,7 @@
     'tls_private_key': None,
     'tls_protocols_allowed': DEFAULT_PROTOCOL_VERSIONS,
     'tls_verify': True,
+    'tls_ciphers': 'ALL',
     'timeout': DEFAULT_TIMEOUT,
     'use_legacy_auth_encoding': True,
     'username': None,
@@ -155,6 +156,7 @@ class RequestsWrapper(object):
         'auth_token_handler',
         'request_size',
         'tls_protocols_allowed',
+        'tls_ciphers_allowed',
     )
 
     def __init__(self, instance, init_config, remapper=None, logger=None, session=None):
@@ -349,6 +351,14 @@ def __init__(self, instance, init_config, remapper=None, logger=None, session=No
         if config['kerberos_cache']:
             self.request_hooks.append(lambda: handle_kerberos_cache(config['kerberos_cache']))
 
+        ciphers = config.get('tls_ciphers')
+        if ciphers:
+            if 'ALL' in ciphers:
+                updated_ciphers = "ALL"
+            else:
+                updated_ciphers = ":".join(ciphers)
+        self.tls_ciphers_allowed = updated_ciphers
+
     def get(self, url, **options):
         return self._request('get', url, options)
 
@@ -467,6 +477,7 @@ def fetch_intermediate_certs(self, hostname, port=443):
             try:
                 context = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS)
                 context.verify_mode = ssl.CERT_NONE
+                context.set_ciphers(self.tls_ciphers_allowed)
 
                 with context.wrap_socket(sock, server_hostname=hostname) as secure_sock:
                     der_cert = secure_sock.getpeercert(binary_form=True)

diff --git a/datadog_checks_base/tests/base/utils/http/test_http.py b/datadog_checks_base/tests/base/utils/http/test_http.py
@@ -2,6 +2,7 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 import logging
+import ssl
 
 import mock
 import pytest
@@ -26,6 +27,39 @@ def test_activate(self):
         assert isinstance(check.http, RequestsWrapper)
 
 
+class TestTLSCiphers:
+    @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_cipher_construction(self, instance, expected_ciphers):
+        init_config = {}
+        http = RequestsWrapper(instance, init_config)
+        mock_socket = mock.MagicMock()
+
+        with mock.patch.object(ssl.SSLContext, 'set_ciphers') as mock_set_ciphers, mock.patch(
+            'datadog_checks.base.utils.http.create_socket_connection', return_value=mock_socket
+        ):
+            http.fetch_intermediate_certs('https://www.google.com')
+            mock_set_ciphers.assert_called_once_with(expected_ciphers)
+
+
 class TestRequestSize:
     def test_behavior_correct(self, mock_http_response):
         instance = {'request_size': 0.5}