mirror of
https://github.com/micropython/micropython.git
synced 2025-07-21 21:11:12 +02:00
f3d1495fd3
Changes include: - Some mbedtls source files renamed or deprecated. - Our `mbedtls_config.h` files are renamed to `mbedtls_config_port.h`, so they don't clash with mbedtls's new default configuration file named `mbedtls_config.h`. - MBEDTLS_TLS_DEFAULT_ALLOW_SHA1_IN_KEY_EXCHANGE is deprecated. - MBEDTLS_HAVE_TIME now requires an `mbedtls_ms_time` function to be defined but it's only used for TLSv1.3 (currently not enabled in MicroPython so there is a lazy implementation, i.e. seconds * 1000). - `tests/multi_net/ssl_data.py` is removed (due to deprecation of MBEDTLS_TLS_DEFAULT_ALLOW_SHA1_IN_KEY_EXCHANGE), there are the existing `ssl_cert_rsa.py` and `sslcontext_server_client.py` tests which do very similar, simple SSL data transfer. - Tests now use an EC key by default (they are smaller and faster), and the RSA key has been regenerated due to the old PKCS encoding used by openssl rsa command, see https://stackoverflow.com/questions/40822328/openssl-rsa-key-pem-and-der-conversion-does-not-match (and `tests/README.md` has been updated accordingly). Signed-off-by: Carlos Gil <carlosgilglez@gmail.com>
78 lines
1.8 KiB
Python
78 lines
1.8 KiB
Python
# Test asyncio TCP server and client with TLS, using readline() to read data.
|
|
|
|
try:
|
|
import os
|
|
import asyncio
|
|
import ssl
|
|
except ImportError:
|
|
print("SKIP")
|
|
raise SystemExit
|
|
|
|
PORT = 8000
|
|
|
|
# These are test certificates. See tests/README.md for details.
|
|
cert = cafile = "ec_cert.der"
|
|
key = "ec_key.der"
|
|
|
|
try:
|
|
os.stat(cafile)
|
|
os.stat(key)
|
|
except OSError:
|
|
print("SKIP")
|
|
raise SystemExit
|
|
|
|
|
|
async def handle_connection(reader, writer):
|
|
data = await reader.readline()
|
|
print("echo:", data)
|
|
data2 = await reader.readline()
|
|
print("echo:", data2)
|
|
writer.write(data + data2)
|
|
await writer.drain()
|
|
|
|
print("close")
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
print("done")
|
|
ev.set()
|
|
|
|
|
|
async def tcp_server():
|
|
global ev
|
|
|
|
server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
server_ctx.load_cert_chain(cert, key)
|
|
ev = asyncio.Event()
|
|
server = await asyncio.start_server(handle_connection, "0.0.0.0", PORT, ssl=server_ctx)
|
|
print("server running")
|
|
multitest.next()
|
|
async with server:
|
|
await asyncio.wait_for(ev.wait(), 10)
|
|
|
|
|
|
async def tcp_client(message):
|
|
client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
client_ctx.verify_mode = ssl.CERT_REQUIRED
|
|
client_ctx.load_verify_locations(cafile=cafile)
|
|
reader, writer = await asyncio.open_connection(
|
|
IP, PORT, ssl=client_ctx, server_hostname="micropython.local"
|
|
)
|
|
print("write:", message)
|
|
writer.write(message)
|
|
await writer.drain()
|
|
data = await reader.readline()
|
|
print("read:", data)
|
|
data2 = await reader.readline()
|
|
print("read:", data2)
|
|
|
|
|
|
def instance0():
|
|
multitest.globals(IP=multitest.get_network_ip())
|
|
asyncio.run(tcp_server())
|
|
|
|
|
|
def instance1():
|
|
multitest.next()
|
|
asyncio.run(tcp_client(b"client data\nclient data2\n"))
|