You've already forked MicroPythonOS
mirror of
https://github.com/m5stack/MicroPythonOS.git
synced 2026-05-20 11:51:27 -07:00
ssl websocket debugging
This commit is contained in:
@@ -3,7 +3,7 @@ import ssl
|
||||
import ubinascii
|
||||
from websocket import websocket
|
||||
|
||||
# Connect to Wi-Fi
|
||||
# Connect to Wi-Fi (disabled as per your code)
|
||||
if False:
|
||||
wlan = network.WLAN(network.STA_IF)
|
||||
wlan.active(True)
|
||||
@@ -11,7 +11,7 @@ if False:
|
||||
while not wlan.isconnected():
|
||||
pass
|
||||
print('Connected:', wlan.ifconfig())
|
||||
|
||||
|
||||
# Resolve hostname
|
||||
host = 'echo.websocket.events' # Replace with your WSS server
|
||||
port = 443
|
||||
@@ -52,11 +52,71 @@ handshake = (
|
||||
'Sec-WebSocket-Version: 13\r\n'
|
||||
'\r\n'
|
||||
).format(host, key.decode())
|
||||
ssl_sock.write(handshake.encode())
|
||||
response = ssl_sock.read(1024).decode()
|
||||
if '101 Switching Protocols' not in response:
|
||||
|
||||
# Send handshake request
|
||||
try:
|
||||
bytes_written = ssl_sock.write(handshake.encode())
|
||||
print('Handshake sent, bytes written:', bytes_written)
|
||||
print('Handshake request:', handshake)
|
||||
except Exception as e:
|
||||
print('Failed to send handshake:', e)
|
||||
ssl_sock.close()
|
||||
raise Exception('Handshake failed: ' + response)
|
||||
raise
|
||||
|
||||
# Read response in chunks with debugging
|
||||
response_bytes = bytearray()
|
||||
max_read = 1024 # Maximum bytes to read
|
||||
read_timeout = 10 # Timeout in seconds (adjust as needed)
|
||||
import time
|
||||
|
||||
start_time = time.time()
|
||||
while len(response_bytes) < max_read:
|
||||
try:
|
||||
# Read a small chunk to avoid blocking too long
|
||||
chunk = ssl_sock.read(128)
|
||||
if not chunk: # EOF or connection closed
|
||||
print('No more data received (EOF)')
|
||||
break
|
||||
print('Received chunk, length:', len(chunk), 'bytes:', chunk)
|
||||
response_bytes.extend(chunk)
|
||||
print('Total bytes received:', len(response_bytes))
|
||||
except Exception as e:
|
||||
print('Error reading chunk:', e)
|
||||
break
|
||||
if time.time() - start_time > read_timeout:
|
||||
print('Read timeout reached')
|
||||
break
|
||||
|
||||
# Inspect raw bytes
|
||||
print('Raw response bytes:', response_bytes)
|
||||
print('Raw response hex:', response_bytes.hex())
|
||||
|
||||
# Attempt to decode response
|
||||
try:
|
||||
response = response_bytes.decode('utf-8')
|
||||
print('Decoded response:', response)
|
||||
except UnicodeError as e:
|
||||
print('UnicodeError during decode:', e)
|
||||
# Try decoding with 'ignore' to see partial response
|
||||
response = response_bytes.decode('utf-8', errors='ignore')
|
||||
print('Decoded with errors ignored:', response)
|
||||
# Try alternative encoding (e.g., latin-1)
|
||||
try:
|
||||
response = response_bytes.decode('latin-1')
|
||||
print('Decoded as latin-1:', response)
|
||||
except Exception as e:
|
||||
print('Latin-1 decode failed:', e)
|
||||
# Dump printable characters
|
||||
printable = ''.join(c if 32 <= ord(c) < 127 else '.' for c in response_bytes.decode('latin-1'))
|
||||
print('Printable characters:', printable)
|
||||
ssl_sock.close()
|
||||
raise Exception('Failed to decode response')
|
||||
|
||||
# Check for valid WebSocket handshake
|
||||
if '101 Switching Protocols' not in response:
|
||||
print('Handshake response:', response)
|
||||
ssl_sock.close()
|
||||
raise Exception('Handshake failed')
|
||||
|
||||
# Create WebSocket object
|
||||
ws = websocket(ssl_sock, True)
|
||||
|
||||
Reference in New Issue
Block a user