espressif/esp-protocols
1
0
Fork 1
mirror of https://github.com/espressif/esp-protocols.git synced 2025-07-25 08:17:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(websocket): Expanded example to demonstrate the transfer over TLS

Browse Source
This commit is contained in:
Suren Gabrielyan
2024-02-06 22:55:06 +04:00
parent 38a3631a27
commit 0d0630ed76
14 changed files with 331 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
components/esp_websocket_client/examples/target/pytest_websocket.py
View File

@ -1,13 +1,16 @@
# SPDX-FileCopyrightText: 2022-2023 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2022-2024 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Unlicense OR CC0-1.0
import json
import random
import re
import socket
import ssl
import string
import sys
from threading import Event, Thread
from SimpleWebSocketServer import SimpleWebSocketServer, WebSocket
from SimpleWebSocketServer import (SimpleSSLWebSocketServer,
SimpleWebSocketServer, WebSocket)
def get_my_ip():
@ -27,7 +30,7 @@ class WebsocketTestEcho(WebSocket):
def handleMessage(self):
self.sendMessage(self.data)
print('Server sent: {}'.format(self.data))
print('\n Server sent: {}\n'.format(self.data))
def handleConnected(self):
print('Connection from: {}'.format(self.address))
@ -44,12 +47,23 @@ class Websocket(object):
conn.sendMessage(data)
def run(self):
self.server = SimpleWebSocketServer('', self.port, WebsocketTestEcho)
if self.use_tls is True:
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_context.load_cert_chain(certfile='main/certs/server_cert.pem', keyfile='main/certs/server_key.pem')
if self.client_verify is True:
ssl_context.load_verify_locations(cafile='main/certs/ca_cert.pem')
ssl_context.verify = ssl.CERT_REQUIRED
ssl_context.check_hostname = False
self.server = SimpleSSLWebSocketServer('', self.port, WebsocketTestEcho, ssl_context=ssl_context)
else:
self.server = SimpleWebSocketServer('', self.port, WebsocketTestEcho)
while not self.exit_event.is_set():
self.server.serveonce()
def __init__(self, port):
def __init__(self, port, use_tls, verify):
self.port = port
self.use_tls = use_tls
self.client_verify = verify
self.exit_event = Event()
self.thread = Thread(target=self.run)
self.thread.start()
@ -77,6 +91,7 @@ def test_examples_protocol_websocket(dut):
for i in range(0, 5):
dut.expect(re.compile(b'Received=hello (\\d)'))
print('All echos received')
sys.stdout.flush()
def test_close(dut):
code = dut.expect(
@ -103,7 +118,8 @@ def test_examples_protocol_websocket(dut):
match = dut.expect(
re.compile(b'Json=({[a-zA-Z0-9]*).*}')).group(0).decode()[5:]
if match == str(data[0]):
print('Sent message and received message are equal')
print('Sent message and received message are equal \n')
sys.stdout.flush()
else:
raise ValueError(
'DUT received string do not match sent string, \nexpected: {}\nwith length {}\
@ -126,7 +142,8 @@ def test_examples_protocol_websocket(dut):
recv_msg += match
if recv_msg == send_msg:
print('Sent message and received message are equal')
print('Sent message and received message are equal \n')
sys.stdout.flush()
else:
raise ValueError(
'DUT received string do not match sent string, \nexpected: {}\nwith length {}\
@ -145,14 +162,24 @@ def test_examples_protocol_websocket(dut):
uri = dut.app.sdkconfig['WEBSOCKET_URI']
uri_from_stdin = False
if dut.app.sdkconfig.get('WS_OVER_TLS_MUTUAL_AUTH') is True:
use_tls = True
client_verify = True
else:
use_tls = False
client_verify = False
except Exception:
print('ENV_TEST_FAILURE: Cannot find uri settings in sdkconfig')
raise
if uri_from_stdin:
server_port = 8080
with Websocket(server_port) as ws:
uri = 'ws://{}:{}'.format(get_my_ip(), server_port)
with Websocket(server_port, use_tls, client_verify) as ws:
if use_tls is True:
uri = 'wss://{}:{}'.format(get_my_ip(), server_port)
else:
uri = 'ws://{}:{}'.format(get_my_ip(), server_port)
print('DUT connecting to {}'.format(uri))
dut.expect('Please enter uri of websocket endpoint', timeout=30)
dut.write(uri)