90 lines
4.5 KiB
Python
90 lines
4.5 KiB
Python
"""Test the actual timed handshake loop with scripted SDK steps and 5 ms ticks."""
|
|
from pathlib import Path
|
|
import subprocess
|
|
import os
|
|
|
|
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 char diagnostic[1024];
|
|
static void test_log(const char *fmt, ...) {
|
|
va_list args;
|
|
if (!strstr(fmt, "[WSS-DIAG]")) return;
|
|
va_start(args, fmt); vsnprintf(diagnostic, sizeof(diagnostic), fmt, args); va_end(args);
|
|
}
|
|
#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);
|
|
assert(strstr(diagnostic, "result=failed phase=server certificate") && strstr(diagnostic, "ret=-99"));
|
|
reset(&ctx); ctx.state = MBEDTLS_SSL_SERVER_HELLO;
|
|
assert(rws_tls_handshake_timed(&ctx) == MBEDTLS_ERR_SSL_TIMEOUT && ticks == 4000 && calls < 2100);
|
|
assert(strstr(diagnostic, "result=timeout phase=server hello") && strstr(diagnostic, "elapsed_ms=20000"));
|
|
assert(strstr(diagnostic, "last_step=-26880") && strstr(diagnostic, "want_write=0"));
|
|
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'
|
|
if os.name == "nt":
|
|
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)
|
|
else:
|
|
subprocess.run(['cc', '-std=c99', '-Wall', '-Wextra', '-Werror',
|
|
'-I' + str(root / 'ql-application/threadx/common/include'),
|
|
str(cfile), '-o', str(exe)], check=True)
|
|
subprocess.run([str(exe)], check=True)
|