Fix greengrass discovery hostname verification

AWSIoTPythonSDK/core/greengrass/discovery/providers.py

@@ -261,82 +261,15 @@
             ssl_sock = ssl_context.wrap_socket(sock, server_hostname=self._host, do_handshake_on_connect=False)
             ssl_sock.do_handshake()
         else:
-            # To keep the SSL Context update minimal, only apply forced ssl context to python3.12+
-            force_ssl_context = sys.version_info[0] > 3 or (sys.version_info[0] == 3 and sys.version_info[1] >= 12)
-            if force_ssl_context:
-                ssl_context = ssl.SSLContext(ssl_protocol_version)
-                ssl_context.load_cert_chain(self._cert_path, self._key_path)
-                ssl_context.load_verify_locations(self._ca_path)
-                ssl_context.verify_mode = ssl.CERT_REQUIRED
-
-                ssl_sock = ssl_context.wrap_socket(sock)
-            else:
-                ssl_sock = ssl.wrap_socket(sock,
-                                           certfile=self._cert_path,
-                                           keyfile=self._key_path,
-                                           ca_certs=self._ca_path,
-                                           cert_reqs=ssl.CERT_REQUIRED,
-                                           ssl_version=ssl_protocol_version)
-
-        self._logger.debug("Matching host name...")
-        if sys.version_info[0] < 3 or (sys.version_info[0] == 3 and sys.version_info[1] < 2):
-            self._tls_match_hostname(ssl_sock)
-        elif sys.version_info[0] == 3 and sys.version_info[1] < 7:
-            # host name verification is handled internally in Python3.7+
-            ssl.match_hostname(ssl_sock.getpeercert(), self._host)
+            ssl_context = ssl.SSLContext(ssl_protocol_version)
+            ssl_context.load_cert_chain(self._cert_path, self._key_path)
+            ssl_context.load_verify_locations(self._ca_path)
+            ssl_context.verify_mode = ssl.CERT_REQUIRED
+            ssl_context.check_hostname = True
+            ssl_sock = ssl_context.wrap_socket(sock, server_hostname=self._host)
 
         return ssl_sock
 
-    def _tls_match_hostname(self, ssl_sock):
-        try:
-            cert = ssl_sock.getpeercert()
-        except AttributeError:
-            # the getpeercert can throw Attribute error: object has no attribute 'peer_certificate'
-            # Don't let that crash the whole client. See also: http://bugs.python.org/issue13721
-            raise ssl.SSLError('Not connected')
-
-        san = cert.get('subjectAltName')
-        if san:
-            have_san_dns = False
-            for (key, value) in san:
-                if key == 'DNS':
-                    have_san_dns = True
-                    if self._host_matches_cert(self._host.lower(), value.lower()) == True:
-                        return
-                if key == 'IP Address':
-                    have_san_dns = True
-                    if value.lower() == self._host.lower():
-                        return
-
-            if have_san_dns:
-                # Only check subject if subjectAltName dns not found.
-                raise ssl.SSLError('Certificate subject does not match remote hostname.')
-        subject = cert.get('subject')
-        if subject:
-            for ((key, value),) in subject:
-                if key == 'commonName':
-                    if self._host_matches_cert(self._host.lower(), value.lower()) == True:
-                        return
-
-        raise ssl.SSLError('Certificate subject does not match remote hostname.')
-
-    def _host_matches_cert(self, host, cert_host):
-        if cert_host[0:2] == "*.":
-            if cert_host.count("*") != 1:
-                return False
-
-            host_match = host.split(".", 1)[1]
-            cert_match = cert_host.split(".", 1)[1]
-            if host_match == cert_match:
-                return True
-            else:
-                return False
-        else:
-            if host == cert_host:
-                return True
-            else:
-                return False
-
     def _send_discovery_request(self, ssl_sock, thing_name):
         request = self.REQUEST_TYPE_PREFIX + \
                   self.PAYLOAD_PREFIX + \

test-integration/IntegrationTests/IntegrationTestDiscoveryHostnameVerification.py

@@ -0,0 +1,122 @@
+# This integration test verifies that TLS hostname verification is enforced
+# when connecting to the Greengrass discovery endpoint on port 8443.
+#
+# It tests two scenarios:
+# 1. A discovery request to the correct endpoint hostname completes the TLS handshake.
+# 2. A discovery request using the endpoint's IP address (hostname mismatch) fails
+#    because the certificate's CN/SAN contains the DNS name but not the IP address.
+#
+# Scenario 2 first performs a control handshake to the same IP with hostname checking
+# disabled. If that succeeds, the server's certificate chain is trusted, so a rejection
+# by the SDK can only come from hostname verification. This avoids depending on
+# TLS-backend-specific error codes or messages.
+#
+# Without the hostname verification fix, scenario 2 would have succeeded on Python 3.7+
+# because only the CA chain was validated, not the hostname.
+
+
+import ssl
+import socket
+import sys
+sys.path.insert(0, "./test-integration/IntegrationTests/TestToolLibrary")
+sys.path.insert(0, "./test-integration/IntegrationTests/TestToolLibrary/SDKPackage")
+
+from TestToolLibrary.SDKPackage.AWSIoTPythonSDK.core.greengrass.discovery.providers import DiscoveryInfoProvider
+from TestToolLibrary.SDKPackage.AWSIoTPythonSDK.exception.AWSIoTExceptions import DiscoveryFailure
+from TestToolLibrary.SDKPackage.AWSIoTPythonSDK.exception.AWSIoTExceptions import DiscoveryTimeoutException
+from TestToolLibrary.checkInManager import checkInManager
+from TestToolLibrary.skip import skip_when_match
+from TestToolLibrary.skip import ModeIsWebSocket
+
+
+PORT = 8443
+CA = "./test-integration/Credentials/rootCA.crt"
+CERT = "./test-integration/Credentials/certificate_drs.pem.crt"
+KEY = "./test-integration/Credentials/privateKey_drs.pem.key"
+TIME_OUT_SEC = 30
+THING_NAME = "DRS_GGAD_0kegiNGA_0"
+
+
+def create_discovery_info_provider(endpoint):
+    discovery_info_provider = DiscoveryInfoProvider()
+    discovery_info_provider.configureEndpoint(endpoint, PORT)
+    discovery_info_provider.configureCredentials(CA, CERT, KEY)
+    discovery_info_provider.configureTimeout(TIME_OUT_SEC)
+    return discovery_info_provider
+
+
+def handshake_without_hostname_check(ip):
+    # Same CA and client credentials as the SDK, but only the certificate chain is verified
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ssl_context.check_hostname = False
+    ssl_context.verify_mode = ssl.CERT_REQUIRED
+    ssl_context.load_verify_locations(CA)
+    ssl_context.load_cert_chain(CERT, KEY)
+    sock = socket.create_connection((ip, PORT), timeout=TIME_OUT_SEC)
+    try:
+        ssl_context.wrap_socket(sock, server_hostname=ip).close()
+    finally:
+        sock.close()
+
+
+############################################################################
+# Main #
+# Check inputs
+my_check_in_manager = checkInManager(2)
+my_check_in_manager.verify(sys.argv)
+mode = my_check_in_manager.mode
+host = my_check_in_manager.host
+
+# GG Discovery only applies mutual auth with cert
+skip_when_match(ModeIsWebSocket(mode), "This test is not applicable for mode: %s. Skipping..." % mode)
+
+############################################################################
+# Test 1: Discovery against the correct hostname should complete the TLS handshake
+############################################################################
+print("=" * 60)
+print("Test 1: Discovery against correct hostname should SUCCEED")
+print("=" * 60)
+
+try:
+    create_discovery_info_provider(host).discover(THING_NAME)
+    print("PASSED: Discovery succeeded against: " + host)
+except ssl.SSLError as e:
+    print("FAILED: TLS handshake with correct hostname was rejected: " + str(e))
+    exit(4)
+except (DiscoveryFailure, DiscoveryTimeoutException) as e:
+    # The TLS handshake completed; the service-level failure is covered by IntegrationTestDiscovery
+    print("PASSED: TLS handshake succeeded against: " + host + " (discovery returned: " + type(e).__name__ + ")")
+
+############################################################################
+# Test 2: Discovery using the IP address (hostname mismatch) should FAIL
+############################################################################
+print("")
+print("=" * 60)
+print("Test 2: Discovery using IP address (hostname mismatch) should FAIL")
+print("=" * 60)
+
+try:
+    real_ip = socket.gethostbyname(host)
+    print("Resolved " + host + " to IP: " + real_ip)
+except socket.gaierror:
+    print("SKIPPED: Could not resolve hostname.")
+    exit(0)
+
+try:
+    handshake_without_hostname_check(real_ip)
+    print("Control: certificate chain from " + real_ip + " is trusted when the hostname is not checked")
+except (ssl.SSLError, socket.error) as e:
+    print("FAILED: Control handshake to " + real_ip + " failed, so a hostname mismatch cannot be isolated: " + str(e))
+    exit(4)
+
+try:
+    create_discovery_info_provider(real_ip).discover(THING_NAME)
+    print("FAILED: Discovery using IP address should have been rejected by hostname verification!")
+    exit(4)
+except ssl.SSLError as e:
+    print("PASSED: Discovery correctly rejected by hostname verification: " + str(e))
+
+print("")
+print("=" * 60)
+print("ALL TESTS PASSED: Discovery hostname verification is working correctly.")
+print("=" * 60)

test-integration/run/run.sh

@@ -117,7 +117,7 @@ else
             echo "[SUB] Running test: ${file}..."
 
             Scale=10
-            Host=TestHost
+            Host=${TestHost}
             case "$file" in
                 "IntegrationTestMQTTConnection.py") Scale=$2
                 ;;
@@ -141,9 +141,12 @@ else
                 "IntegrationTestJobsClient.py") Scale=""
                 ;;
                 "IntegrationTestHostnameVerification.py") Scale=""
+                ;;
+                "IntegrationTestDiscoveryHostnameVerification.py") Scale=""
+                Host=${GreengrassHost}
             esac
 
-            python ${TEST_DIR}${file} ${TestMode} ${TestHost} ${Scale}
+            python ${TEST_DIR}${file} ${TestMode} ${Host} ${Scale}
             currentTestStatus=$?
             echo "[SUB] Test: ${file} completed. Exiting with status: ${currentTestStatus}"
             if [ ${currentTestStatus} -ne 0 ]; then