chenjintao
3deb51d838
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.
76 lines
3.7 KiB
Python
76 lines
3.7 KiB
Python
"""Test the actual timed handshake loop with scripted SDK steps and 5 ms ticks."""
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
root = Path(__file__).resolve().parents[1]
|
|
source = (root / 'ql-application/threadx/interface/custom_app/lib/librws/src/rws_socketpriv.c').read_bytes().decode('latin1')
|
|
body = source[source.index('static const char *rws_tls_phase_name('):].split('int rws_ssl_conn(', 1)[0]
|
|
harness = r'''
|
|
#include <assert.h>
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
#include <stdarg.h>
|
|
#include "mbedtls/ssl_internal.h"
|
|
typedef struct { int ret, next_state; unsigned int duration; } step;
|
|
static step steps[16];
|
|
static int count, calls, sleeps;
|
|
static unsigned int ticks;
|
|
static void test_log(const char *fmt, ...) { (void)fmt; }
|
|
#define LOG_I test_log
|
|
static unsigned int ql_rtos_get_systicks(void) { return ticks; }
|
|
static void rws_thread_sleep(unsigned int ms) { ticks += ms / 5; ++sleeps; }
|
|
int mbedtls_ssl_handshake_step(mbedtls_ssl_context *ctx) {
|
|
step s;
|
|
assert(ctx->state != MBEDTLS_SSL_HANDSHAKE_OVER);
|
|
if (calls >= count) { ++calls; return MBEDTLS_ERR_SSL_WANT_READ; }
|
|
s = steps[calls++]; ticks += s.duration; ctx->state = s.next_state; return s.ret;
|
|
}
|
|
'''
|
|
harness += body
|
|
harness += r'''
|
|
static void reset(mbedtls_ssl_context *ctx) {
|
|
memset(ctx, 0, sizeof(*ctx));
|
|
ctx->state = MBEDTLS_SSL_CLIENT_HELLO;
|
|
count = calls = sleeps = 0; ticks = 0; memset(steps, 0, sizeof(steps));
|
|
}
|
|
int main(void) {
|
|
mbedtls_ssl_context ctx;
|
|
mbedtls_ssl_handshake_params handshake = {0};
|
|
reset(&ctx);
|
|
steps[0] = (step){0, MBEDTLS_SSL_SERVER_HELLO, 1};
|
|
steps[1] = (step){MBEDTLS_ERR_SSL_WANT_READ, MBEDTLS_SSL_SERVER_HELLO, 0};
|
|
steps[2] = (step){0, MBEDTLS_SSL_SERVER_CERTIFICATE, 2};
|
|
steps[3] = (step){0, MBEDTLS_SSL_CLIENT_KEY_EXCHANGE, 10};
|
|
steps[4] = (step){MBEDTLS_ERR_SSL_WANT_WRITE, MBEDTLS_SSL_CLIENT_KEY_EXCHANGE, 1};
|
|
steps[5] = (step){0, MBEDTLS_SSL_SERVER_FINISHED, 5};
|
|
steps[6] = (step){0, MBEDTLS_SSL_HANDSHAKE_OVER, 1}; count = 7;
|
|
assert(rws_tls_handshake_timed(&ctx) == 0 && calls == 7 && sleeps == 2 && ticks == 24);
|
|
reset(&ctx); ctx.handshake = &handshake; ctx.state = MBEDTLS_SSL_SERVER_KEY_EXCHANGE;
|
|
handshake.ecdh_ctx.grp_id = MBEDTLS_ECP_DP_CURVE25519;
|
|
steps[0] = (step){0, MBEDTLS_SSL_CLIENT_KEY_EXCHANGE, 1};
|
|
steps[1] = (step){0, MBEDTLS_SSL_HANDSHAKE_OVER, 1}; count = 2;
|
|
assert(rws_tls_handshake_timed(&ctx) == 0 && calls == 2 && sleeps == 0);
|
|
reset(&ctx); steps[0] = (step){-99, MBEDTLS_SSL_SERVER_CERTIFICATE, 1}; count = 1;
|
|
assert(rws_tls_handshake_timed(&ctx) == -99 && calls == 1 && sleeps == 0);
|
|
reset(&ctx);
|
|
assert(rws_tls_handshake_timed(&ctx) == MBEDTLS_ERR_SSL_TIMEOUT && ticks == 4000 && calls < 2100);
|
|
reset(&ctx); ticks = (unsigned int)-1000;
|
|
assert(rws_tls_handshake_timed(&ctx) == MBEDTLS_ERR_SSL_TIMEOUT && calls < 2100);
|
|
reset(&ctx); steps[0] = (step){0, MBEDTLS_SSL_SERVER_CERTIFICATE, 4001}; count = 1;
|
|
assert(rws_tls_handshake_timed(&ctx) == MBEDTLS_ERR_SSL_TIMEOUT && calls == 1 && sleeps == 0);
|
|
reset(&ctx); ctx.state = MBEDTLS_SSL_HANDSHAKE_OVER;
|
|
assert(rws_tls_handshake_timed(&ctx) == 0 && calls == 0);
|
|
puts("PASS: timed SDK handshake loop: progress, WANT_READ/WRITE, no progress sleeps, fatal, deadline, tick wrap, complete state");
|
|
return 0;
|
|
}
|
|
'''
|
|
output = root / 'artifacts/wss'
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
cfile = output / 'test_wss_handshake.c'
|
|
cfile.write_text(harness, encoding='utf-8')
|
|
exe = output / 'test_wss_handshake.exe'
|
|
subprocess.run(['cl', '/nologo', '/W4', '/WX', '/TC', str(cfile),
|
|
'/I' + str(root / 'ql-application/threadx/common/include'),
|
|
'/Fo' + str(output / 'test_wss_handshake.obj'), '/Fe' + str(exe)], check=True)
|
|
subprocess.run([str(exe)], check=True)
|