micropython · dpgeorge · Dec 12, 2023 · Dec 12, 2023 · Dec 12, 2023
diff --git a/extmod/modssl_mbedtls.c b/extmod/modssl_mbedtls.c
@@ -554,6 +554,7 @@ STATIC mp_obj_t ssl_socket_make_new(mp_obj_ssl_context_t *ssl_context, mp_obj_t
     mbedtls_raise_error(ret);
 }
 
+#if defined(MBEDTLS_SSL_KEEP_PEER_CERTIFICATE)
 STATIC mp_obj_t mod_ssl_getpeercert(mp_obj_t o_in, mp_obj_t binary_form) {
     mp_obj_ssl_socket_t *o = MP_OBJ_TO_PTR(o_in);
     if (!mp_obj_is_true(binary_form)) {
@@ -566,6 +567,7 @@ STATIC mp_obj_t mod_ssl_getpeercert(mp_obj_t o_in, mp_obj_t binary_form) {
     return mp_obj_new_bytes(peer_cert->raw.p, peer_cert->raw.len);
 }
 STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_ssl_getpeercert_obj, mod_ssl_getpeercert);
+#endif
 
 STATIC mp_obj_t mod_ssl_cipher(mp_obj_t o_in) {
     mp_obj_ssl_socket_t *o = MP_OBJ_TO_PTR(o_in);
@@ -726,7 +728,9 @@ STATIC const mp_rom_map_elem_t ssl_socket_locals_dict_table[] = {
     #if MICROPY_UNIX_COVERAGE
     { MP_ROM_QSTR(MP_QSTR_ioctl), MP_ROM_PTR(&mp_stream_ioctl_obj) },
     #endif
+    #if defined(MBEDTLS_SSL_KEEP_PEER_CERTIFICATE)
     { MP_ROM_QSTR(MP_QSTR_getpeercert), MP_ROM_PTR(&mod_ssl_getpeercert_obj) },
+    #endif
     { MP_ROM_QSTR(MP_QSTR_cipher), MP_ROM_PTR(&mod_ssl_cipher_obj) },
 };
 STATIC MP_DEFINE_CONST_DICT(ssl_socket_locals_dict, ssl_socket_locals_dict_table);

diff --git a/tests/multi_net/sslcontext_getpeercert.py b/tests/multi_net/sslcontext_getpeercert.py
@@ -1,6 +1,7 @@
 # Test creating an SSL connection and getting the peer certificate.
 
 try:
+    import io
     import os
     import socket
     import ssl
@@ -42,14 +43,20 @@ def instance0():
 
 # Client
 def instance1():
+    s_test = ssl.wrap_socket(io.BytesIO(), server_side=True, do_handshake=False)
+    s_test.close()
+    if not hasattr(s_test, "getpeercert"):
+        print("SKIP")
+        raise SystemExit
+
     multitest.next()
     s = socket.socket()
     s.connect(socket.getaddrinfo(IP, PORT)[0][-1])
     client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
     client_ctx.verify_mode = ssl.CERT_REQUIRED
     client_ctx.load_verify_locations(cafile=cafile)
     s = client_ctx.wrap_socket(s, server_hostname="micropython.local")
-    print(s.getpeercert(True))
+    print(s.getpeercert(True).hex())
     s.write(b"client to server")
     print(s.read(16))
     s.close()
diff --git a/tests/multi_net/sslcontext_getpeercert.py.exp b/tests/multi_net/sslcontext_getpeercert.py.exp
@@ -1,5 +1,5 @@
 --- instance0 ---
 b'client to server'
 --- instance1 ---
-None
+3082058930820371a00302010202141b3da08b15005eea265d0b57b8ba99812ab274cb300d06092a864886f70d01010b05003054310b30090603550406130241553113301106035504080c0a536f6d652d537461746531143012060355040a0c0b4d6963726f507974686f6e311a301806035504030c116d6963726f707974686f6e2e6c6f63616c301e170d3233313131393032323932375a170d3238313131373032323932375a3054310b30090603550406130241553113301106035504080c0a536f6d652d537461746531143012060355040a0c0b4d6963726f507974686f6e311a301806035504030c116d6963726f707974686f6e2e6c6f63616c30820222300d06092a864886f70d01010105000382020f003082020a0282020100deee37780ebca47e0f414ba033ebe692d2bc374a0eb1f42556bf266ad704208116ee0d8b2fd0b518074ad0981c4fd1322de81696ccea838884b06f56d3ebe49cf0561050f6f8ced5f7f4f13d086b28779a9647bbfae6c3f3ad68a5b28ee8a1ceb260d87ea300316599c4dd9f6082f89164b590df8695add518339f6730dec4f05b1ef63548329b0a48823035b23737f3303b56aa251dd8dcf0c20e6c1d291374c185ae657b349c20721c7c01a1b393c96d4c5f2bc8e2dfca7dab896e2fa84dee53d2bb6dbd1056970fa1812315e8ee9d92b3cb93e0b563d274bf07dd79600ef403b91d4ce814418b28cfaeb2b7d8401e64f6d4f39283df3204f2fe01f2fd289f5d2078d9ee2f96b6de1fd4284d9274fa38b0ad9ffcce8ffe66673be2cf304ee1b27c7cacaaf4ca76f1e84419e6e80f540add3e91cd469903e9ceb6bd2b1c33caa59acb5516ce8ac00e73d7a551bb65d39bd6af04411e81c20e6bd474d797a0bcd498e26720bd60ae4f900bb1afa59c7ac7a336273c7734ca5874ea63fb8ec787ab702041442da11a922baf5fbeb9eeea4f9f49cb1f659b561806d2169dbed07c43558c908c94e16491fe1a22cd92b8f33c1184353bdc985c88722f65e48024910f723035c0d33b789928296fb193cec6350884243b00bf51422ad09fb7012bd9cad4716803422be0d111deace913fac8cb2be1e96fa8449068430e5424bd0bd10203010001a3533051301d0603551d0e041604147d392a82ab464936fd7d74226694556a2945fd8d301f0603551d230418301680147d392a82ab464936fd7d74226694556a2945fd8d300f0603551d130101ff040530030101ff300d06092a864886f70d01010b05000382020100ae40c015e3eade8dabc84ee357ac9d694e7cd69ce4a1b265880273d16257119aa72fb2aa8b841e2899bea3e8690146f24d963a37825c93bf745447dc6ab09b5f2947671dca13b1e71f5c3e43011d74cdc688ed1215b3016071ae7235d77f79d7bb81f097bb04a08ccf400717721b29e2ea913eb23614610597deee477ed716db7e8ebe11aed39c7035f48259dfa54d88871c1f67159d52ce11eb111fa00708a7d7081c07fd92d54abbaec7ff1b50ce2f6f358857d2f55d1c7b5aa6dd66b9c3c2e654397e2d5330aca9834ff8fd749ce968c706fe3bb1b8510a379ec1910d7ece0212c34d56a2073fb7f25c88fe298568e448d03ec30b348f7d9a8836390216a6da7a8efed50dfb8c21a8531efc158e7f4398f87af18d1bd2926d08d34364bf5d85e88040dff3d3f1da6268dbc0cafa64f544c065380fa695a8d015b385aed0a1fd9ff1d7c2b28a549e04c1132b421f85a85640acac11c69416859fb9b461eeddffa92ae303b35c7233537077068de558dd02715e25aee976a97879038d2952be0d327892ab2fc78716b0d7aab4b923d5d79905f7f8b6a18c42e466fec62f84b6e5957deae0964dab8436b0e0cd4e08012661bafb9588fbfd7068fd6c08ab79101a4bdfe21d95cd0ee0aad7dd8a3ed128071c0ec2d063dc6dfa63189e51bf5d9259e776d7623f745a73f4e12e5c2b90493de1c6436b339e1400891e3e35c31057
 b'server to client'
diff --git a/tests/net_hosted/connect_nonblock_xfer.py b/tests/net_hosted/connect_nonblock_xfer.py
@@ -27,11 +27,12 @@ def do_connect(peer_addr, tls, handshake):
             print("  got", er.errno)
     # wrap with ssl/tls if desired
     if tls:
+        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        if hasattr(ssl_context, "check_hostname"):
+            ssl_context.check_hostname = False
+
         try:
-            if sys.implementation.name == "micropython":
-                s = ssl.wrap_socket(s, do_handshake=handshake)
-            else:
-                s = ssl.wrap_socket(s, do_handshake_on_connect=handshake)
+            s = ssl_context.wrap_socket(s, do_handshake_on_connect=handshake)
             print("wrap: True")
         except Exception as e:
             dp(e)

diff --git a/tests/net_hosted/ssl_getpeercert.py b/tests/net_hosted/ssl_getpeercert.py
@@ -1,8 +1,15 @@
 # test ssl.getpeercert() method
 
+import io
 import socket
 import ssl
 
+s_test = ssl.wrap_socket(io.BytesIO(), server_side=True, do_handshake=False)
+s_test.close()
+if not hasattr(s_test, "getpeercert"):
+    print("SKIP")
+    raise SystemExit
+
 
 def test(peer_addr):
     s = socket.socket()

diff --git a/tests/net_inet/ssl_errors.py b/tests/net_inet/ssl_errors.py
@@ -1,12 +1,12 @@
 # test that socket.connect() on a non-blocking socket raises EINPROGRESS
 # and that an immediate write/send/read/recv does the right thing
 
-import sys, errno, socket, ssl
+import sys, errno, select, socket, ssl
 
 
 def test(addr, hostname, block=True):
-    print("---", hostname or addr)
-    s = socket.socket()
+    print("---", hostname)
+    s = socket.socket(socket.AF_INET)
     s.setblocking(block)
     try:
         s.connect(addr)
@@ -16,11 +16,15 @@ def test(addr, hostname, block=True):
             raise
         print("EINPROGRESS")
 
+    if sys.implementation.name != "micropython":
+        # in CPython we have to wait, otherwise wrap_socket is not happy
+        select.select([], [s], [])
+
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ssl_context.verify_mode = ssl.CERT_REQUIRED
+
     try:
-        if sys.implementation.name == "micropython":
-            s = ssl.wrap_socket(s, do_handshake=block)
-        else:
-            s = ssl.wrap_socket(s, do_handshake_on_connect=block)
+        s = ssl_context.wrap_socket(s, do_handshake_on_connect=block, server_hostname=hostname)
         print("wrap: True")
     except OSError:
         print("wrap: error")
@@ -36,11 +40,11 @@ def test(addr, hostname, block=True):
 
 if __name__ == "__main__":
     # connect to plain HTTP port, oops!
-    addr = socket.getaddrinfo("micropython.org", 80)[0][-1]
-    test(addr, None)
+    addr = socket.getaddrinfo("micropython.org", 80, socket.AF_INET)[0][-1]
+    test(addr, "micropython.org")
     # connect to plain HTTP port, oops!
-    addr = socket.getaddrinfo("micropython.org", 80)[0][-1]
-    test(addr, None, False)
+    addr = socket.getaddrinfo("micropython.org", 80, socket.AF_INET)[0][-1]
+    test(addr, "micropython.org", False)
     # connect to server with self-signed cert, oops!
-    addr = socket.getaddrinfo("test.mosquitto.org", 8883)[0][-1]
+    addr = socket.getaddrinfo("test.mosquitto.org", 8883, socket.AF_INET)[0][-1]
     test(addr, "test.mosquitto.org")
diff --git a/tests/net_inet/test_tls_nonblock.py b/tests/net_inet/test_tls_nonblock.py
@@ -2,12 +2,12 @@
 
 
 def test_one(site, opts):
-    ai = socket.getaddrinfo(site, 443)
+    ai = socket.getaddrinfo(site, 443, socket.AF_INET)
     addr = ai[0][-1]
-    print(addr)
+    print(site)
 
     # Connect the raw socket
-    s = socket.socket()
+    s = socket.socket(socket.AF_INET)
     s.setblocking(False)
     try:
         s.connect(addr)
@@ -16,17 +16,22 @@ def test_one(site, opts):
         if e.errno != errno.EINPROGRESS:
             raise
 
+    # Create SSLContext.
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+
+    # CPython compatibility:
+    # - disable check_hostname
+    # - load default system certificate chain
+    # - must wait for socket to be writable before calling wrap_socket
     if sys.implementation.name != "micropython":
-        # in CPython we have to wait, otherwise wrap_socket is not happy
+        ssl_context.check_hostname = False
+        ssl_context.load_default_certs()
         select.select([], [s], [])
 
     try:
         # Wrap with SSL
         try:
-            if sys.implementation.name == "micropython":
-                s = ssl.wrap_socket(s, do_handshake=False)
-            else:
-                s = ssl.wrap_socket(s, do_handshake_on_connect=False)
+            s = ssl_context.wrap_socket(s, do_handshake_on_connect=False)
         except OSError as e:
             if e.errno != errno.EINPROGRESS:
                 raise
@@ -87,8 +92,7 @@ def test_one(site, opts):
 
 
 SITES = [
-    "google.com",
-    {"host": "www.google.com"},
+    "www.github.com",
     "micropython.org",
     "pypi.org",
     {"host": "api.pushbullet.com", "sni": True},
@@ -105,7 +109,7 @@ def main():
             test_one(site, opts)
             print(site, "ok")
         except Exception as e:
-            print(site, "error")
+            print(site, "error", e)
     print("DONE")
 
 

diff --git a/tests/net_inet/test_tls_sites.py b/tests/net_inet/test_tls_sites.py
@@ -1,24 +1,34 @@
+import sys
+import select
 import socket
 import ssl
 
-# CPython only supports server_hostname with SSLContext
-if hasattr(ssl, "SSLContext"):
-    ssl = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
-
 
 def test_one(site, opts):
-    ai = socket.getaddrinfo(site, 443)
+    ai = socket.getaddrinfo(site, 443, socket.AF_INET)
     addr = ai[0][-1]
 
-    s = socket.socket()
+    s = socket.socket(socket.AF_INET)
+
+    # Create SSLContext.
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+
+    # CPython compatibility:
+    # - disable check_hostname
+    # - load default system certificate chain
+    # - must wait for socket to be writable before calling wrap_socket
+    if sys.implementation.name != "micropython":
+        ssl_context.check_hostname = False
+        ssl_context.load_default_certs()
+        select.select([], [s], [])
 
     try:
         s.connect(addr)
 
         if "sni" in opts:
-            s = ssl.wrap_socket(s, server_hostname=opts["host"])
+            s = ssl_context.wrap_socket(s, server_hostname=opts["host"])
         else:
-            s = ssl.wrap_socket(s)
+            s = ssl_context.wrap_socket(s)
 
         s.write(b"GET / HTTP/1.0\r\nHost: %s\r\n\r\n" % bytes(site, "latin"))
         resp = s.read(4096)
@@ -31,8 +41,7 @@ def test_one(site, opts):
 
 
 SITES = [
-    "google.com",
-    "www.google.com",
+    "www.github.com",
     "micropython.org",
     "pypi.org",
     {"host": "api.pushbullet.com", "sni": True},

diff --git a/tests/net_inet/test_tls_sites.py.exp b/tests/net_inet/test_tls_sites.py.exp
diff --git a/tests/net_inet/tls_text_errors.py b/tests/net_inet/tls_text_errors.py
@@ -1,13 +1,17 @@
 # test that modtls produces a text error message
 
-import socket, ssl, sys
+import socket, ssl
 
 
 def test(addr):
     s = socket.socket()
     s.connect(addr)
     try:
-        s = ssl.wrap_socket(s)
+        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        if hasattr(ssl_context, "check_hostname"):
+            # Disable hostname check on CPython.
+            ssl_context.check_hostname = False
+        s = ssl_context.wrap_socket(s)
         print("wrap: no exception")
     except OSError as e:
         # mbedtls produces "mbedtls -0x7200: SSL - An invalid SSL record was received"