Adapt WSS to new boards and use MQTT domain for eSIM connectivity
Preserve main GNSS and volume behavior while porting verified TLS, receive fixes and full-handshake optimizations. Add MQTT connection diagnostics and route MQTT through banban.api.talkingq.com:1883. Include migration documentation and regression tools. Firmware builds and package checks passed; user confirmed new-board connectivity after switching the MQTT host to the domain.
This commit is contained in:
@@ -0,0 +1,36 @@
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import socket
|
||||
import ssl
|
||||
from pathlib import Path
|
||||
|
||||
root = Path(__file__).resolve().parents[1]
|
||||
source = (root / 'ql-application/threadx/interface/custom_app/talkingq_root_ca.c').read_text()
|
||||
pem = ''.join(json.loads('"' + value + '"') for value in re.findall(r'"(.*?)"', source) if 'CERTIFICATE' in value or value.endswith(r'\n'))
|
||||
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
||||
context.load_verify_locations(cadata=pem)
|
||||
context.minimum_version = ssl.TLSVersion.TLSv1_2
|
||||
context.maximum_version = ssl.TLSVersion.TLSv1_2
|
||||
server = 'banban.api.talkingq.com'
|
||||
report = {'root_sha256': hashlib.sha256(ssl.PEM_cert_to_DER_cert(pem)).hexdigest(), 'trust': 'only the compiled root CA; no OS trust or intermediate download'}
|
||||
with socket.create_connection((server, 443), timeout=15) as tcp:
|
||||
with context.wrap_socket(tcp, server_hostname=server) as tls:
|
||||
report['tls'] = tls.version()
|
||||
report['cipher'] = tls.cipher()
|
||||
report['root_only_validation'] = 'PASS'
|
||||
try:
|
||||
with socket.create_connection((server, 443), timeout=15) as tcp:
|
||||
with context.wrap_socket(tcp, server_hostname='wrong.invalid'):
|
||||
raise AssertionError('wrong hostname accepted')
|
||||
except ssl.SSLCertVerificationError:
|
||||
report['wrong_hostname_rejected'] = True
|
||||
empty = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
||||
try:
|
||||
with socket.create_connection((server, 443), timeout=15) as tcp:
|
||||
with empty.wrap_socket(tcp, server_hostname=server):
|
||||
raise AssertionError('untrusted CA accepted')
|
||||
except ssl.SSLCertVerificationError:
|
||||
report['untrusted_ca_rejected'] = True
|
||||
(root / 'artifacts/wss/root-only-probe.json').write_text(json.dumps(report, indent=2))
|
||||
print(json.dumps(report, indent=2))
|
||||
Reference in New Issue
Block a user