diff --git a/AWSIoTPythonSDK/core/greengrass/discovery/providers.py b/AWSIoTPythonSDK/core/greengrass/discovery/providers.py index 192f71a..0d312d8 100644 --- a/AWSIoTPythonSDK/core/greengrass/discovery/providers.py +++ b/AWSIoTPythonSDK/core/greengrass/discovery/providers.py @@ -261,82 +261,15 @@ def _create_ssl_connection(self, sock): 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 + \ diff --git a/test-integration/IntegrationTests/IntegrationTestDiscoveryHostnameVerification.py b/test-integration/IntegrationTests/IntegrationTestDiscoveryHostnameVerification.py new file mode 100644 index 0000000..cc96006 --- /dev/null +++ b/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) diff --git a/test-integration/run/run.sh b/test-integration/run/run.sh index bea6fc4..d7203c0 100755 --- a/test-integration/run/run.sh +++ b/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