TCP over TLS
证书生成
openssl ecparam -genkey -name prime256v1 -out key_sec1.pem
openssl pkcs8 -topk8 -nocrypt -in key_sec1.pem -out server_key.pem
openssl req -new -x509 -key server_key.pem -out server_cert.pem -days 36500 -subj "/CN=SEM Meter HTTPS Server" -addext "subjectAltName=DNS:eastron-meter.local" -addext "basicConstraints=critical,CA:FALSE" -addext "keyUsage=critical,digitalSignature" -addext "extendedKeyUsage=serverAuth"
设备端代码
#include <string.h>
#include <sys/param.h>
#include <sys/time.h>
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "esp_system.h"
#include "esp_wifi.h"
#include "esp_event.h"
#include "esp_log.h"
#include "esp_idf_version.h"
#include "nvs_flash.h"
#include "esp_netif.h"
#include "protocol_examples_common.h"
#include "lwip/err.h"
#include "lwip/sockets.h"
#include "lwip/sys.h"
#include <lwip/netdb.h>
#include "esp_tls.h"
#include "mbedtls/ssl.h"
#define PORT CONFIG_EXAMPLE_PORT
#define KEEPALIVE_IDLE CONFIG_EXAMPLE_KEEPALIVE_IDLE
#define KEEPALIVE_INTERVAL CONFIG_EXAMPLE_KEEPALIVE_INTERVAL
#define KEEPALIVE_COUNT CONFIG_EXAMPLE_KEEPALIVE_COUNT
#define SERVER_TASK_STACK_SIZE 8192
static const char *TAG = "example";
extern const uint8_t server_crt_start[] asm("_binary_server_crt_start");
extern const uint8_t server_crt_end[] asm("_binary_server_crt_end");
extern const uint8_t server_key_start[] asm("_binary_server_key_start");
extern const uint8_t server_key_end[] asm("_binary_server_key_end");
static esp_tls_cfg_server_t s_tls_cfg;
static void do_retransmit_tls(esp_tls_t *tls) {
int len;
char rx_buffer[128];
while (1) {
// TLS 4、TLS 会话读取
len = esp_tls_conn_read(tls, rx_buffer, sizeof(rx_buffer) - 1);
if (len == ESP_TLS_ERR_SSL_WANT_READ || len == ESP_TLS_ERR_SSL_WANT_WRITE) {
continue;
} else if (len < 0) {
ESP_LOGE(TAG, "Error occurred during receiving: 0x%x", -len);
break;
} else if (len == 0) {
ESP_LOGW(TAG, "Connection closed");
break;
}
rx_buffer[len] = 0;
ESP_LOGI(TAG, "Received %d bytes: %s", len, rx_buffer);
int to_write = len;
while (to_write > 0) {
// TLS 5、TLS 会话写入
int written = esp_tls_conn_write(tls, rx_buffer + (len - to_write), to_write);
if (written == ESP_TLS_ERR_SSL_WANT_WRITE || written == ESP_TLS_ERR_SSL_WANT_READ) {
continue;
}
if (written <= 0) {
ESP_LOGE(TAG, "Error occurred during sending: 0x%x", -written);
return;
}
to_write -= written;
}
}
}
static void tls_session_serve(int sock) {
struct timeval tv = {
.tv_sec = 60 * 10,
.tv_usec = 0,
};
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
esp_tls_t *tls = esp_tls_init();
if (tls == NULL) {
ESP_LOGE(TAG, "Unable to allocate esp_tls context");
close(sock);
return;
}
// TLS 3、TLS 会话创建，并进行握手
if (esp_tls_server_session_create(&s_tls_cfg, sock, tls) != 0) {
ESP_LOGE(TAG, "TLS handshake failed, dropping connection");
esp_tls_server_session_delete(tls);
close(sock);
return;
}
ESP_LOGI(TAG, "TLS session established");
do_retransmit_tls(tls);
// TLS 6、TLS 会话关闭通知
mbedtls_ssl_context *ssl = (mbedtls_ssl_context *) esp_tls_get_ssl_context(tls);
if (ssl != NULL) {
mbedtls_ssl_close_notify(ssl);
}
// TLS 7、TLS 会话删除
esp_tls_server_session_delete(tls);
// TLS 8、关闭套接字
close(sock);
}
static void tcp_server_task(void *pvParameters) {
char addr_str[128];
int addr_family = (int) pvParameters;
int ip_protocol = 0;
int keepAlive = 1;
int keepIdle = KEEPALIVE_IDLE;
int keepInterval = KEEPALIVE_INTERVAL;
int keepCount = KEEPALIVE_COUNT;
struct sockaddr_storage dest_addr;
if (addr_family == AF_INET) {
struct sockaddr_in *dest_addr_ip4 = (struct sockaddr_in *) &dest_addr;
dest_addr_ip4->sin_addr.s_addr = htonl(INADDR_ANY);
dest_addr_ip4->sin_family = AF_INET;
dest_addr_ip4->sin_port = htons(PORT);
ip_protocol = IPPROTO_IP;
}
int listen_sock = socket(addr_family, SOCK_STREAM, ip_protocol);
if (listen_sock < 0) {
ESP_LOGE(TAG, "Unable to create socket: errno %d", errno);
vTaskDelete(NULL);
return;
}
int opt = 1;
setsockopt(listen_sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
ESP_LOGI(TAG, "Socket created");
int err = bind(listen_sock, (struct sockaddr *) &dest_addr, sizeof(dest_addr));
if (err != 0) {
ESP_LOGE(TAG, "Socket unable to bind: errno %d", errno);
ESP_LOGE(TAG, "IPPROTO: %d", addr_family);
goto CLEAN_UP;
}
ESP_LOGI(TAG, "Socket bound, port %d", PORT);
err = listen(listen_sock, 1);
if (err != 0) {
ESP_LOGE(TAG, "Error occurred during listen: errno %d", errno);
goto CLEAN_UP;
}
while (1) {
ESP_LOGI(TAG, "Socket listening");
struct sockaddr_storage source_addr;
socklen_t addr_len = sizeof(source_addr);
int sock = accept(listen_sock, (struct sockaddr *) &source_addr, &addr_len);
if (sock < 0) {
ESP_LOGE(TAG, "Unable to accept connection: errno %d", errno);
break;
}
setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, &keepAlive, sizeof(int));
setsockopt(sock, IPPROTO_TCP, TCP_KEEPIDLE, &keepIdle, sizeof(int));
setsockopt(sock, IPPROTO_TCP, TCP_KEEPINTVL, &keepInterval, sizeof(int));
setsockopt(sock, IPPROTO_TCP, TCP_KEEPCNT, &keepCount, sizeof(int));
if (source_addr.ss_family == PF_INET) {
inet_ntoa_r(((struct sockaddr_in *) &source_addr)->sin_addr, addr_str, sizeof(addr_str) - 1);
}
ESP_LOGI(TAG, "Socket accepted ip address: %s", addr_str);
tls_session_serve(sock);
}
CLEAN_UP:
close(listen_sock);
vTaskDelete(NULL);
}
void app_main(void) {
ESP_ERROR_CHECK(nvs_flash_init());
ESP_ERROR_CHECK(esp_netif_init());
ESP_ERROR_CHECK(esp_event_loop_create_default());
ESP_ERROR_CHECK(example_connect());
// TLS 1、配置服务器证书和密钥缓冲区
s_tls_cfg.servercert_buf = server_crt_start;
s_tls_cfg.servercert_bytes = (unsigned int) (server_crt_end - server_crt_start);
s_tls_cfg.serverkey_buf = server_key_start;
s_tls_cfg.serverkey_bytes = (unsigned int) (server_key_end - server_key_start);
xTaskCreate(tcp_server_task, "tcp_server", SERVER_TASK_STACK_SIZE, (void *) AF_INET, 5, NULL);
}
客户端测试
测试脚本ls_client.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ESP32 TLS TCP server 上位机测试客户端（命令行版）。
对 docs/README 中的 ESP32 示例做 TLS 回声测试：打印协商结果与对端证书信息，
按轮次发送数据并校验回声，输出 RTT 统计，可选写 JSON 报告。
用法示例:
# 正常校验（用 gen_certs.sh 生成的 ca.crt）
python3 tools/tls_client.py --host 192.168.0.167 --port 3333 --ca certs/ca.crt
# 只想确认能不能连上（跳过证书校验，仅调试）
python3 tools/tls_client.py --host 192.168.0.167 --insecure
# 连发 100 轮，每轮间隔 20ms，并写报告
python3 tools/tls_client.py --host 192.168.0.167 --ca certs/ca.crt \
--count 100 --interval 20 --json report.json
# 交互式收发（相当于支持 TLS 的 nc）
python3 tools/tls_client.py --host 192.168.0.167 --ca certs/ca.crt --interactive
只依赖 Python 标准库，Python 3.8+ 均可运行。
"""
from __future__ import annotations
import argparse
import json
import socket
import ssl
import sys
import time
from typing import Dict, List, Optional, Tuple
DEFAULT_PORT = 3333
DEFAULT_MESSAGE = "Data to ESP"
DEFAULT_TIMEOUT = 5.0
EXIT_OK = 0
EXIT_MISMATCH = 1
EXIT_CONNECTION_ERROR = 2
# --------------------------------------------------------------------------- #
# 异常
# --------------------------------------------------------------------------- #
class ClientError(Exception):
"""本工具主动抛出的、可读的错误。"""
class ConnError(ClientError):
"""TCP/TLS 建连失败。"""
class ClosedError(ClientError):
"""对端关闭了连接。"""
class RecvTimeout(ClientError):
"""等待数据超时。"""
# --------------------------------------------------------------------------- #
# TLS 上下文与证书展示（GUI 也会 import 这里，不要在本文件顶层做副作用操作）
# --------------------------------------------------------------------------- #
def build_ssl_context(ca: Optional[str] = None,
insecure: bool = False,
cert: Optional[str] = None,
key: Optional[str] = None) -> ssl.SSLContext:
"""构造客户端 SSLContext。
insecure=True 时跳过证书与主机名校验（仅用于调试）；
否则用 ca 指定的根证书校验服务器证书，并校验主机名/SAN。
"""
if insecure:
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
else:
ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=ca)
# ESP32 默认只开 TLS1.2（CONFIG_MBEDTLS_SSL_PROTO_TLS1_3 默认 n），
# 这里抬高下限避免协商到 1.0/1.1。
try:
ctx.minimum_version = ssl.TLSVersion.TLSv1_2
except (AttributeError, ValueError):
pass
if cert:
ctx.load_cert_chain(certfile=cert, keyfile=key)
return ctx
def describe_peer_cert(tls: ssl.SSLSocket) -> Dict[str, object]:
"""提取对端证书摘要。--insecure 时 getpeercert() 返回空字典，只能返回 {}。"""
try:
cert = tls.getpeercert()
except (ValueError, OSError):
return {}
if not cert:
return {}
def _flat(name: str) -> Dict[str, str]:
# cert['subject'] 形如 ((('commonName', 'x'),), (('organizationName','y'),),)
# 每个 RDN 内部还可能有多项，逐层展开比 dict(x[0] ...) 更稳妥
out: Dict[str, str] = {}
for rdn in cert.get(name, ()):
for key, value in rdn:
out[key] = value
return out
subject = _flat("subject")
issuer = _flat("issuer")
return {
"subject": subject.get("commonName", "?"),
"issuer": issuer.get("commonName", "?"),
"not_after": cert.get("notAfter", ""),
"san": ["{}={}".format(k, v) for k, v in cert.get("subjectAltName", ())],
}
def describe_session(tls: ssl.SSLSocket) -> Dict[str, object]:
"""协商结果 + 证书摘要。"""
cipher = tls.cipher() or ("?", "?", 0)
info: Dict[str, object] = {
"tls_version": tls.version() or "?",
"cipher": cipher[0],
"cipher_bits": cipher[2] if len(cipher) > 2 else 0,
"peer": describe_peer_cert(tls),
}
return info
# --------------------------------------------------------------------------- #
# 连接与收发
# --------------------------------------------------------------------------- #
def open_connection(host: str, port: int, timeout: float,
ctx: ssl.SSLContext,
sni: Optional[str] = None) -> ssl.SSLSocket:
"""TCP 建连 + TLS 握手，失败统一转成 ConnError。"""
try:
raw = socket.create_connection((host, port), timeout=timeout)
except OSError as exc:
raise ConnError("TCP 连接 {}:{} 失败: {}".format(host, port, exc)) from exc
raw.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) # 测 RTT 时关掉 Nagle
server_hostname = sni or host
try:
tls = ctx.wrap_socket(raw, server_hostname=server_hostname)
except ssl.SSLCertVerificationError as exc:
raw.close()
raise ConnError("证书校验失败: {}".format(
getattr(exc, "verify_message", None) or exc)) from exc
except ssl.SSLError as exc:
raw.close()
raise ConnError("TLS 握手失败: {}".format(exc)) from exc
except OSError as exc:
raw.close()
raise ConnError("TLS 握手异常: {}".format(exc)) from exc
tls.settimeout(timeout)
return tls
def recv_exact(tls: ssl.SSLSocket, want: int, timeout: float) -> bytes:
"""收满 want 字节；对端关闭/超时分别抛 ClosedError / RecvTimeout。"""
buf = bytearray()
deadline = time.monotonic() + timeout
while len(buf) < want:
left = deadline - time.monotonic()
if left <= 0:
raise RecvTimeout("等待回声超时（已收到 {} 字节，期望 {}）".format(len(buf), want))
tls.settimeout(max(0.05, left))
try:
chunk = tls.recv(want - len(buf))
except ssl.SSLEOFError as exc:
raise ClosedError("对端未发 close_notify 就断开（裸 EOF）: {}".format(exc)) from exc
except ssl.SSLError as exc:
raise ClosedError("TLS 读失败: {}".format(exc)) from exc
except socket.timeout as exc:
raise RecvTimeout("等待回声超时（已收到 {} 字节，期望 {}）".format(len(buf), want)) from exc
except ConnectionResetError as exc:
raise ClosedError("连接被对端重置: {}".format(exc)) from exc
if not chunk:
raise ClosedError("对端已关闭连接（收到 {} / {} 字节）".format(len(buf), want))
buf += chunk
return bytes(buf)
def drain(tls: ssl.SSLSocket, wait: float = 0.2) -> bytes:
"""把当前能读到的数据都读出来（交互模式用）。"""
out = bytearray()
end = time.monotonic() + wait
while True:
left = end - time.monotonic()
if left <= 0:
break
tls.settimeout(left)
try:
chunk = tls.recv(4096)
except (socket.timeout, ssl.SSLWantReadError):
break
except (ssl.SSLEOFError, ssl.SSLError, OSError):
break
if not chunk:
break
out += chunk
return bytes(out)
def close_cleanly(tls: ssl.SSLSocket) -> None:
"""发 close_notify 并等对端的 close_notify（走完整的 TLS 关闭握手）。"""
try:
tls.settimeout(2.0)
tls.unwrap()
except Exception:
pass
finally:
try:
tls.close()
except Exception:
pass
# --------------------------------------------------------------------------- #
# 输出
# --------------------------------------------------------------------------- #
def fmt_bytes(data: bytes, use_hex: bool) -> str:
if use_hex:
return data.hex(" ")
return data.decode("utf-8", "replace")
def print_session(info: Dict[str, object], target: str) -> None:
print("已连接 {} -> TLS版本: {} 密码套件: {} ({} bits)".format(
target, info["tls_version"], info["cipher"], info["cipher_bits"]))
peer = info.get("peer") or {}
if peer:
print("对端证书: CN={} 签发者={} 有效期至={}".format(
peer.get("subject"), peer.get("issuer"), peer.get("not_after")))
print("证书 SAN: {}".format(", ".join(peer.get("san", [])) or "(无)"))
else:
print("对端证书: 未校验（--insecure 模式不提供证书信息）")
# --------------------------------------------------------------------------- #
# 主流程
# --------------------------------------------------------------------------- #
def run_echo_rounds(tls: ssl.SSLSocket, payload: bytes, count: int,
interval_ms: float, timeout: float,
use_hex: bool) -> Tuple[List[float], int]:
"""跑 count 轮回声，返回 (每轮 RTT 毫秒, 总收发字节)。"""
rtts: List[float] = []
total = 0
for idx in range(1, count + 1):
start = time.monotonic()
tls.sendall(payload)
echoed = recv_exact(tls, len(payload), timeout)
rtt = (time.monotonic() - start) * 1000.0
rtts.append(rtt)
total += len(echoed)
if echoed != payload:
print("第 {} 轮回声不一致！".format(idx))
print(" 发送: {}".format(fmt_bytes(payload, use_hex)))
print(" 收到: {}".format(fmt_bytes(echoed, use_hex)))
raise ClientError("回声数据不匹配")
if count <= 10 or idx % max(1, count // 10) == 0 or idx == count:
print("第 {}/{} 轮 OK RTT {:.2f} ms 回声 {} 字节".format(
idx, count, rtt, len(echoed)))
if interval_ms > 0 and idx < count:
time.sleep(interval_ms / 1000.0)
return rtts, total
def run_interactive(tls: ssl.SSLSocket, use_hex: bool, timeout: float) -> None:
print("交互模式：输入文本回车发送，Ctrl-D / Ctrl-C 退出。")
while True:
try:
line = input("> ")
except (EOFError, KeyboardInterrupt):
print()
break
if not line:
continue
payload = bytes.fromhex(line.replace(" ", "")) if use_hex else line.encode("utf-8")
try:
tls.sendall(payload)
echoed = recv_exact(tls, len(payload), timeout)
except ClientError as exc:
print("收发失败: {}".format(exc))
break
print("<- {}".format(fmt_bytes(echoed, use_hex)))
extra = drain(tls, 0.1)
if extra:
print("<- (附加 {} 字节) {}".format(len(extra), fmt_bytes(extra, use_hex)))
def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="ESP32 TLS TCP server 上位机测试客户端",
formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--host", required=True, help="ESP32 的 IP 或主机名")
parser.add_argument("--port", type=int, default=DEFAULT_PORT,
help="端口，默认 {}".format(DEFAULT_PORT))
g = parser.add_argument_group("证书与校验")
g.add_argument("--ca", help="用于校验服务器证书的 CA 文件（certs/ca.crt）")
g.add_argument("--insecure", action="store_true",
help="跳过证书/主机名校验，仅用于调试")
g.add_argument("--cert", help="客户端证书（服务端开启双向认证时才需要）")
g.add_argument("--key", help="客户端私钥（配合 --cert）")
g.add_argument("--sni", help="覆盖用于 SNI/主机名校验的名字，默认用 --host")
g = parser.add_argument_group("测试数据")
g.add_argument("--message", default=DEFAULT_MESSAGE,
help="每轮发送的内容，默认 {!r}".format(DEFAULT_MESSAGE))
g.add_argument("--count", type=int, default=1, help="发送轮数，默认 1")
g.add_argument("--interval", type=float, default=0.0,
help="轮间隔毫秒数，默认 0")
g.add_argument("--hex", action="store_true",
help="按十六进制收发（--message 按 hex 解析）")
g.add_argument("--interactive", action="store_true", help="交互式收发")
g.add_argument("--hold", type=float, default=0.0,
help="回声测试后保持连接 N 秒再关闭（观察服务端日志用）")
g = parser.add_argument_group("其它")
g.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT,
help="单次收发超时秒数，默认 {}".format(DEFAULT_TIMEOUT))
g.add_argument("--json", dest="json_path", help="把测试结果写成 JSON 报告")
g.add_argument("--quiet", action="store_true", help="只输出错误")
return parser.parse_args(argv)
def main(argv: Optional[List[str]] = None) -> int:
args = parse_args(argv)
quiet = args.quiet
def log(msg: str) -> None:
if not quiet:
print(msg)
if not args.insecure and not args.ca:
print("提示: 未指定 --ca，将使用系统信任库校验服务器证书；"
"自签证书会被拒绝。调试可加 --insecure。", file=sys.stderr)
target = "{}:{}".format(args.host, args.port)
report: Dict[str, object] = {"target": target, "result": "unknown"}
try:
ctx = build_ssl_context(ca=args.ca, insecure=args.insecure,
cert=args.cert, key=args.key)
except (OSError, ssl.SSLError) as exc:
print("加载证书/私钥失败: {}".format(exc), file=sys.stderr)
return EXIT_CONNECTION_ERROR
try:
tls = open_connection(args.host, args.port, args.timeout, ctx, args.sni)
except ClientError as exc:
print("连接失败: {}".format(exc), file=sys.stderr)
report["result"] = "connection_error"
report["error"] = str(exc)
if args.json_path:
_write_json(args.json_path, report)
return EXIT_CONNECTION_ERROR
exit_code = EXIT_OK
try:
info = describe_session(tls)
report["session"] = info
if not quiet:
print_session(info, target)
if args.interactive:
run_interactive(tls, args.hex, args.timeout)
report["result"] = "interactive"
else:
payload = (bytes.fromhex(args.message.replace(" ", ""))
if args.hex else args.message.encode("utf-8"))
if args.count < 1:
raise ClientError("--count 必须 >= 1")
rtts, total = run_echo_rounds(tls, payload, args.count,
args.interval, args.timeout, args.hex)
report.update({
"result": "ok",
"rounds": args.count,
"payload_bytes": len(payload),
"echoed_bytes": total,
"rtt_ms": {
"min": round(min(rtts), 3),
"avg": round(sum(rtts) / len(rtts), 3),
"max": round(max(rtts), 3),
},
})
log("回声测试通过: {} 轮，RTT min/avg/max = {:.2f}/{:.2f}/{:.2f} ms".format(
args.count, min(rtts), sum(rtts) / len(rtts), max(rtts)))
if args.hold > 0:
log("保持连接 {:.1f} 秒...".format(args.hold))
time.sleep(args.hold)
except ClientError as exc:
print("测试失败: {}".format(exc), file=sys.stderr)
report["result"] = "failed"
report["error"] = str(exc)
exit_code = EXIT_MISMATCH
finally:
close_cleanly(tls)
log("连接已关闭")
if args.json_path:
_write_json(args.json_path, report)
return exit_code
def _write_json(path: str, report: Dict[str, object]) -> None:
try:
with open(path, "w", encoding="utf-8") as fp:
json.dump(report, fp, ensure_ascii=False, indent=2)
print("报告已写入 {}".format(path))
except OSError as exc:
print("写报告失败: {}".format(exc), file=sys.stderr)
if __name__ == "__main__":
sys.exit(main())
bash 命令
python3 tls_client.py --host 172.16.9.55 --port 3333 --ca server_cert.pem --sni eastron-meter.local --message 666 --count 3600 --interval 1000
License:
CC BY 4.0