chenjintao/banban
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Accelerate legacy WSS elliptic-curve operations and time TLS phases

Browse Source 
Enable mbedTLS NIST modular reduction and report per-phase handshake timings. Add handshake-loop regression tests and a certificate verification benchmark, and document the full kernel and APP package requirement.
This commit is contained in:
chenjintao
2026-09-16 14:38:43 +08:00
parent 5c3e7eee6d
commit 3a0714c735
11 changed files with 186 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/check_mbedtls_abi.py
+1 -1
View File
@@ -51,7 +51,7 @@ for target in ['kernel', 'app']:
or line.startswith('#define QUECTEL_MBEDTLS_'))
assert layouts['kernel'] == layouts['app'], json.dumps(layouts, indent=2)
assert macros['kernel'] == macros['app'], json.dumps({k: sorted(set(macros[k]) - set(macros['app' if k == 'kernel' else 'kernel'])) for k in macros})
for feature in ['ECDSA_C', 'KEY_EXCHANGE_ECDHE_ECDSA_ENABLED', 'GCM_C', 'HAVE_TIME_DATE', 'PLATFORM_TIME_ALT']:
for feature in ['ECDSA_C', 'ECP_NIST_OPTIM', 'KEY_EXCHANGE_ECDHE_ECDSA_ENABLED', 'GCM_C', 'HAVE_TIME_DATE', 'PLATFORM_TIME_ALT']:
assert any(line.startswith('#define MBEDTLS_' + feature + ' ') for line in macros['kernel'])
report = {'result': 'PASS', 'layout': layouts['kernel'], 'features': macros['kernel']}
(output / 'tls-abi.json').write_text(json.dumps(report, indent=2))
tools/test_mbedtls_cert.c
+17
View File
@@ -36,6 +36,8 @@ int main(int argc, char **argv)
char pem[16384];
FILE *file;
size_t length;
clock_t verify_started;
mbedtls_ecp_group group;
unsigned char key[16] = {0}, iv[12] = {0}, plain[16] = {0}, cipher[16], tag[16];
const unsigned char expected_cipher[16] = {0x03,0x88,0xda,0xce,0x60,0xb6,0xa3,0x92,0xf3,0x28,0xc2,0xb9,0x71,0xb2,0xfe,0x78};
const unsigned char expected_tag[16] = {0xab,0x6e,0x47,0xd4,0x2c,0xec,0x13,0xbd,0xf5,0x3a,0x67,0xb2,0x12,0x57,0xbd,0xdf};
@@ -60,6 +62,21 @@ int main(int argc, char **argv)
ret = mbedtls_x509_crt_verify(&chain, &root, NULL, "banban.api.talkingq.com", &flags, NULL, NULL, 0);
printf("SDK server chain + hostname verify: %d, flags=0x%08x\n", ret, (unsigned int)flags);
assert(ret == 0 && flags == 0);
verify_started = clock();
for (i = 0; i < 20; ++i) {
ret = mbedtls_x509_crt_verify(&chain, &root, NULL, "banban.api.talkingq.com", &flags, NULL, NULL, 0);
assert(ret == 0 && flags == 0);
}
printf("Host certificate chain verify average: %.2f ms (20 runs)\n",
1000.0 * (clock() - verify_started) / CLOCKS_PER_SEC / 20);
mbedtls_ecp_group_init(&group);
assert(mbedtls_ecp_group_load(&group, MBEDTLS_ECP_DP_SECP384R1) == 0);
#ifdef MBEDTLS_ECP_NIST_OPTIM
assert(group.modp != NULL);
#else
assert(group.modp == NULL);
#endif
mbedtls_ecp_group_free(&group);
ret = mbedtls_x509_crt_verify(&chain, &root, NULL, "wrong.invalid", &flags, NULL, NULL, 0);
assert(ret != 0 && (flags & MBEDTLS_X509_BADCERT_CN_MISMATCH));
ret = mbedtls_x509_crt_verify(&chain, NULL, NULL, "banban.api.talkingq.com", &flags, NULL, NULL, 0);
tools/test_mbedtls_cert.py
+13 -7
View File
@@ -2,13 +2,17 @@
Run from a Visual Studio developer shell. Chain PEMs come from the WSS probe.
"""
from pathlib import Path
import argparse
import subprocess
import sys
root = Path(__file__).resolve().parents[1]
output = root / 'artifacts/wss/ecc-fix/host-cert'
parser = argparse.ArgumentParser()
parser.add_argument('chain', nargs='?', default=str(root / 'artifacts/wss/retest-443'))
parser.add_argument('--compare-nist', action='store_true')
args = parser.parse_args()
output = root / 'artifacts/wss/tls-speed/host-cert'
output.mkdir(parents=True, exist_ok=True)
chain = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else root / 'artifacts/wss/retest-443'
chain = Path(args.chain).resolve()
sources = '''asn1parse asn1write bignum base64 pem oid pk pkparse pk_wrap rsa rsa_internal
ecp ecp_curves ecdsa md md5 sha1 sha256 sha512 aes des gcm cipher cipher_wrap
chacha20 chachapoly poly1305 constant_time ctr_drbg x509 x509_crt platform_util
@@ -21,7 +25,9 @@ command = ['cl', '/nologo', '/O2', '/Gy', '/TC', '/D_CRT_SECURE_NO_WARNINGS',
str(root / 'tools/test_mbedtls_cert.c'),
str(root / 'ql-application/threadx/interface/custom_app/talkingq_root_ca.c')]
command += [str(root / ('ql-kernel/threadx/quec_mbedtls/' + name + '.c')) for name in sources]
exe = output / 'test_mbedtls_cert.exe'
command += ['/Fe' + str(exe), '/link', '/OPT:REF']
subprocess.run(command, cwd=output, check=True)
subprocess.run([str(exe), str(chain)], check=True)
for variant in (['baseline', 'nist'] if args.compare_nist else ['nist']):
exe = output / ('test_mbedtls_cert_' + variant + '.exe')
flags = ['/DWSS_TEST_DISABLE_NIST_OPTIM'] if variant == 'baseline' else []
subprocess.run(command + flags + ['/Fe' + str(exe), '/link', '/OPT:REF'], cwd=output, check=True)
print('Certificate test variant:', variant, flush=True)
subprocess.run([str(exe), str(chain)], check=True)
tools/test_wss.cmd
+2
View File
@@ -12,4 +12,6 @@ if errorlevel 1 exit /b 1
ql-cross-tool\win32\owtoolchain\Python\python.exe tools\test_wss_io.py
if errorlevel 1 exit /b 1
ql-cross-tool\win32\owtoolchain\Python\python.exe tools\test_wss_receive.py
if errorlevel 1 exit /b 1
ql-cross-tool\win32\owtoolchain\Python\python.exe tools\test_wss_handshake.py
exit /b %errorlevel%
tools/test_wss_handshake.py
+74
View File
@@ -0,0 +1,74 @@
"""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>
enum { MBEDTLS_SSL_CLIENT_HELLO, MBEDTLS_SSL_SERVER_HELLO,
MBEDTLS_SSL_SERVER_CERTIFICATE, MBEDTLS_SSL_SERVER_KEY_EXCHANGE,
MBEDTLS_SSL_CLIENT_KEY_EXCHANGE, MBEDTLS_SSL_SERVER_FINISHED,
MBEDTLS_SSL_HANDSHAKE_OVER };
#define MBEDTLS_ERR_SSL_WANT_READ -2
#define MBEDTLS_ERR_SSL_WANT_WRITE -3
#define MBEDTLS_ERR_SSL_TIMEOUT -4
typedef struct { int state; } mbedtls_ssl_context;
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; }
static 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) {
ctx->state = MBEDTLS_SSL_CLIENT_HELLO;
count = calls = sleeps = 0; ticks = 0; memset(steps, 0, sizeof(steps));
}
int main(void) {
mbedtls_ssl_context ctx;
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); 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),
'/Fo' + str(output / 'test_wss_handshake.obj'), '/Fe' + str(exe)], check=True)
subprocess.run([str(exe)], check=True)
tools/verify_wss_package.py
+7 -5
View File
@@ -8,7 +8,7 @@ import zipfile
from pathlib import Path
root = Path(__file__).resolve().parents[1]
package = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else root / 'artifacts/wss/recv-fix/EG800AKCN_11LG-legacy-wss-recv.zip'
package = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else root / 'artifacts/wss/tls-speed/EG800AKCN_11LG-legacy-wss-fast.zip'
output = package.parent
output.mkdir(parents=True, exist_ok=True)
objdump = root / 'ql-cross-tool/win32/owtoolchain/gcc-arm-none-eabi/bin/arm-none-eabi-objdump.exe'
@@ -16,14 +16,15 @@ app = root / 'ql-application/threadx/build/app.elf'
symbols = subprocess.check_output([str(objdump.with_name('arm-none-eabi-nm.exe')), str(app)], text=True)
required = ['talkingq_parse_wss_url', 'talkingq_root_ca', 'rws_check_handshake', 'rws_ssl_conn',
'mbedtls_ssl_set_hostname', 'mbedtls_ssl_get_verify_result', 'mbedtls_ssl_conf_authmode',
'mbedtls_ctr_drbg_random', 'mbedtls_sha1', 'mbedtls_base64_encode']
'mbedtls_ctr_drbg_random', 'mbedtls_sha1', 'mbedtls_base64_encode', 'mbedtls_ssl_handshake_step']
for symbol in required:
assert re.search(r'\b' + symbol + r'$', symbols, re.M), 'Missing linked symbol: ' + symbol
kernel_symbols = subprocess.check_output([str(objdump.with_name('arm-none-eabi-nm.exe')),
str(root / 'ql-kernel/threadx/build/kernel.axf')], text=True)
kernel_required = ['ecdsa_verify_restartable', 'mbedtls_ecdsa_read_signature',
'mbedtls_gcm_crypt_and_tag', 'mbedtls_x509_time_is_future',
'mbedtls_x509_time_is_past', 'mbedtls_platform_gmtime_r', 'mbedtls_time']
'mbedtls_x509_time_is_past', 'mbedtls_platform_gmtime_r', 'mbedtls_time',
'ecp_mod_p256', 'ecp_mod_p384', 'ecp_mod_p521']
for symbol in kernel_required:
assert re.search(r'\b' + symbol + r'$', kernel_symbols, re.M), 'Missing kernel capability: ' + symbol
clock_body = subprocess.check_output([str(objdump), '-d', '--disassemble=x509_get_current_time',
@@ -32,7 +33,7 @@ assert '<mbedtls_platform_gmtime_r>' in clock_body and '<GetCurrentCompTime>' no
(output / 'x509-clock.disasm.txt').write_text(clock_body)
report = {'package': package.name, 'sha256': hashlib.sha256(package.read_bytes()).hexdigest(),
'linked_symbols': required, 'kernel_symbols': kernel_required,
'hardware_test': '2026-09-16 14:06 device log: ECC/RTC TLS verification passed; receive retry fix awaits device test',
'hardware_test': '2026-09-16 14:20 device log: TLS and HTTP 101 passed; NIST acceleration and handshake timings await device test',
'backend': '2026-09-16 11:32 CST: WSS 443 /ws HTTP 101; 8080 plaintext WS'}
with zipfile.ZipFile(package) as archive:
assert archive.testzip() is None, 'ZIP CRC failure'
@@ -42,6 +43,7 @@ with zipfile.ZipFile(package) as archive:
assert data == (root / source).read_bytes(), 'Stale image: ' + name
report[name] = {'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest()}
assert b'wss://banban.api.talkingq.com:443/ws' in archive.read('customer_app.bin')
assert b'WSS TLS handshake total=' in archive.read('customer_app.bin')
report['download_manifest'] = json.loads(archive.read('download.json'))
printf_address = int(re.search(r'^([0-9a-fA-F]+) T printf$', symbols, re.M).group(1), 16)
for symbol, binary, expected in [
@@ -51,7 +53,7 @@ for symbol, binary, expected in [
body = subprocess.check_output([str(objdump), '-d', '--disassemble=' + symbol, str(binary)], text=True)
assert '<' + symbol + '>:' in body and expected in body and 'task_create' not in body
(output / (symbol + '.disasm.txt')).write_text(body)
report['verification'] = 'PASS: ZIP CRC, current APP/kernel identity, GNSS bypass, WSS URL, ECDSA/GCM/RTC kernel symbols'
report['verification'] = 'PASS: ZIP CRC, current APP/kernel identity, GNSS bypass, WSS URL, ECDSA/GCM/RTC/NIST kernel symbols and handshake timing logs'
(output / 'verification.json').write_text(json.dumps(report, indent=2))
print(report['verification'])
print('SHA256:', report['sha256'])
tools/wss_tls_host_config.h
+3
View File
@@ -1,6 +1,9 @@
/* Host adapter: crypto/X.509 features are the firmware's shared profile.
* Only CPU assembly and the OS allocator/clock implementation differ. */
#include "../ql-config/config/talkingq_mbedtls_config.h"
#ifdef WSS_TEST_DISABLE_NIST_OPTIM
#undef MBEDTLS_ECP_NIST_OPTIM
#endif
#undef MBEDTLS_HAVE_ASM
#undef MBEDTLS_FS_IO
#define QUECTEL_MBEDTLS_SUPPORT