chenjintao
551a7bb1e6
Prioritize X25519 while retaining enabled curve fallbacks and log the negotiated ECDHE group. Extend ABI checks, crypto benchmarks and backend group probes, and document the updated APP package.
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)
|