为何OpenSSL客户端MAC验证失败，python-mbedtls客户端却正常？

问题概况 #

在Ubuntu 20.04上基于python-mbedtls开发预共享密钥（PSK）模式的UDP DTLS服务器，用于接收嵌入式应用的UDP数据报：

• 服务器与python-mbedtls测试客户端交互正常
• 服务器与OpenSSL命令行s_client握手时，出现消息MAC验证失败
• 嵌入式DTLS客户端可与OpenSSL命令行s_server正常通信，但无法通过该python-mbedtls服务器的MAC验证
• 所有场景使用相同的预共享密钥和客户端标识，已排除密码套件问题

核心原因分析 #

1. PSK格式不匹配 #

OpenSSL的-psk参数默认将传入的字符串解析为十六进制字节，而python-mbedtls代码中直接使用b'010102030405060708090a0b0c0d0e0f'会被当作ASCII字节处理（例如b'0'对应0x30，b'1'对应0x31），导致两端实际使用的PSK值完全不同，最终MAC计算结果不匹配。

2. DTLS握手流程错误 #

服务器代码中在收到HelloVerifyRequest（HVR）后，通过cli0.accept()重新获取新的socket上下文并关闭原上下文，这种操作会破坏DTLS握手的状态连续性，导致后续握手阶段的密钥派生或MAC计算出现偏差。

诊断与修复建议 #

1. 修正PSK格式 #

将python-mbedtls配置中的PSK改为十六进制解码后的原始字节：

conf = DTLSConfiguration(
    ciphers=(
        "TLS-PSK-WITH-AES-256-CBC-SHA",
        "TLS-ECDHE-PSK-WITH-CHACHA20-POLY1305-SHA256",
        "TLS-RSA-PSK-WITH-CHACHA20-POLY1305-SHA256"
    ),
    # 使用bytes.fromhex将十六进制字符串转为原始字节
    pre_shared_key=('Client_identity', bytes.fromhex('010102030405060708090a0b0c0d0e0f')),
    validate_certificates=False
)

OpenSSL客户端命令保持不变，因为-psk参数默认处理十六进制字符串。

2. 修复DTLS握手流程 #

移除多余的socket上下文切换逻辑，在收到HVR后直接复用原上下文完成握手：

def echo_until(sock, end):
    cli, cli_address = sock.accept()
    cli.setcookieparam(cli_address[0].encode("ascii"))
    print("cli_address: {}".format(cli_address))
    try:
        cli.do_handshake()
    except HelloVerifyRequest:
        print("Received HelloVerifyRequest, retrying handshake")
        # 收到HVR后直接重试握手，无需重新accept
        cli.do_handshake()
    print(" .", "handshake", cli.negotiated_tls_version())

    while True:
        data = cli.recv(4096)
        print(" .", "R", data)
        nn = cli.send(data)
        print(" .", "S", nn, len(data))
        if data == end:
            break

    print(" .", "done")
    print(cli)
    cli.close()

3. 增强调试信息 #

• 将mbedtls调试级别设为最高（4）：_set_debug_level(4)，查看密钥派生、MAC算法选择等细节
• OpenSSL客户端命令添加-msg参数，输出完整握手消息：

  openssl s_client -psk "010102030405060708090a0b0c0d0e0f" -psk_identity "Client_identity" -dtls1_2 -port 9009 -host 149.28.170.96 -async -debug -msg
  

原服务器代码 #

#!/home/ron/venvs/udpserver/bin/python3

"""Example DTLS server"""

import sys
import time
from contextlib import suppress
from functools import partial
from typing import Any, Callable, NoReturn, Optional, Tuple, Union
import datetime as dt
import socket
import struct
from mbedtls.pk import RSA, ECC
from mbedtls.x509 import BasicConstraints, CRT, CSR
from mbedtls.tls import *
from mbedtls._tls import _enable_debug_output, _set_debug_level  # type: ignore
from mbedtls.tls import (
    DTLSConfiguration,
    HelloVerifyRequest,
    ServerContext,
    TLSWrappedSocket,
)
   
conf = DTLSConfiguration(
    ciphers=(
        "TLS-PSK-WITH-AES-256-CBC-SHA",
        "TLS-ECDHE-PSK-WITH-CHACHA20-POLY1305-SHA256",
        "TLS-RSA-PSK-WITH-CHACHA20-POLY1305-SHA256"
    ),
    pre_shared_key=('Client_identity',b'010102030405060708090a0b0c0d0e0f'),
    validate_certificates=False
    )

print("conf: {}".format(conf))

_enable_debug_output(conf)
_set_debug_level(3)

def echo_until(sock, end):
    cli0, cli_address0 = sock.accept()
    cli0.setcookieparam(cli_address0[0].encode("ascii"))
    print("cli_address0: {}".format(cli_address0))
    try:
        # block(cli0.do_handshake)
        cli0.do_handshake()
    except HelloVerifyRequest:
        print("HVR")

    cli1, cli_address1 = cli0.accept()
    cli0.close()
    cli1.setcookieparam(cli_address1[0].encode("ascii"))
    print("cli_address1: {}".format(cli_address1))
    # block(cli1.do_handshake)
    cli1.do_handshake()
    print(" .", "handshake", cli1.negotiated_tls_version())

    cli = cli1

    while True:
        # data = block(cli.recv, 4096)
        data = cli.recv(4096)
        print(" .", "R", data)
        # nn = block(cli.send, data)
        nn = cli.send(data)
        print(" .", "S", nn, len(data))
        if data == end:
            break

    print(" .", "done")
    print(cli)
    cli.close()
  
address = ("0.0.0.0", 9009)
host, port = address

ctx = ServerContext(conf)
srv = ctx.wrap_socket(
    socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

print(" .", "bind", srv, address)
srv.bind(address)

while True:
    print(" .", ">>>")
    echo_until(srv, b"\0")
    print(" .", "<<<")

原客户端代码 #

#!/media/fred/venvs/dtlsclient/bin/python3
"""An example DTLS PSK client.
"""

from __future__ import annotations

import socket
import sys
import time
from contextlib import suppress
from typing import Any, Optional, Tuple, Union

from mbedtls._tls import _enable_debug_output, _set_debug_level  # type: ignore
from mbedtls.exceptions import TLSError
from mbedtls.tls import (
    ClientContext,
    DTLSConfiguration,
    TLSWrappedSocket,
)

if sys.version_info < (3, 10):
    from typing_extensions import TypeAlias
else:
    from typing import TypeAlias

if sys.version_info < (3, 8):
    from typing_extensions import Final
else:
    from typing import Final

def _echo_dtls(sock: TLSWrappedSocket, buffer: bytes, chunksize: int) -> bytes:
    view = memoryview(buffer)
    received = bytearray()
    while len(received) != len(buffer):
        part = view[len(received) : len(received) + chunksize]
        sock.send(part)
        data, _addr = sock.recvfrom(chunksize)
        received += data
        if not data:
            # Avoid tight loop.
            time.sleep(0.01)
    return received


class Client:
    def __init__(
        self,
        cli_conf: DTLSConfiguration,
        proto: socket.SocketKind,
        srv_address: _Address,
        srv_hostname: Optional[str],
    ) -> None:
        super().__init__()
        self.cli_conf: Final = cli_conf
        self.proto: Final = proto
        self.srv_address: Final = srv_address
        self.srv_hostname: Final = srv_hostname
        self._sock: Optional[TLSWrappedSocket] = None
        self._echo: Final = {
            socket.SOCK_DGRAM: _echo_dtls,
        }[self.proto]

    def __enter__(self) -> Client:
        self.start()
        return self

    def __exit__(self, *exc_info: object) -> None:
        self.stop()

    def __del__(self) -> None:
        self.stop()

    @property
    def context(self) -> Optional[ClientContext]:
        if self._sock is None:
            return None
        assert isinstance(self._sock.context, ClientContext)
        return self._sock.context

    def do_handshake(self) -> None:
        if not self._sock:
            return

        self._sock.do_handshake()

    def echo(self, buffer: bytes, chunksize: int) -> bytes:
        if not self._sock:
            return b""

        return bytes(self._echo(self._sock, buffer, chunksize))

    def start(self) -> None:
        if self._sock:
            self.stop()

        self._sock = ClientContext(self.cli_conf).wrap_socket(
            socket.socket(socket.AF_INET, self.proto),
            server_hostname=self.srv_hostname,
        )
        self._sock.connect(self.srv_address)

    def stop(self) -> None:
        if not self._sock:
            return

        with suppress(TLSError, OSError):
            self._sock.close()
        self._sock = None

    def restart(self) -> None:
        self.stop()
        self.start()

def main() -> None:
    address = "149.28.170.96"
    port = 9009
    message = "Trundled off to the jungle"
    server_name = "dtlsserver"
    proto = socket.SOCK_DGRAM
    debug = 3
    conf = DTLSConfiguration(
        ciphers=(
        "TLS-PSK-WITH-AES-256-CBC-SHA",
        "TLS-ECDHE-PSK-WITH-CHACHA20-POLY1305-SHA256",
        "TLS-RSA-PSK-WITH-CHACHA20-POLY1305-SHA256"
        ),
        pre_shared_key=('Client_identity',b'010102030405060708090a0b0c0d0e0f'),
        validate_certificates=False
    )
    print(conf)

    if debug is not None:
        _enable_debug_output(conf)
        _set_debug_level(debug)

    with Client(
        conf, proto, (address, port), server_name
    ) as cli:
        cli.do_handshake()
        received = cli.echo(message.encode("utf-8"), 1024)
    print("Received:" + received.decode("utf-8"))


if __name__ == "__main__":
    main( )

失败的OpenSSL客户端命令 #

openssl s_client -psk "010102030405060708090a0b0c0d0e0f" -psk_identity "Client_identity" -dtls1_2 -port 9009 -host 149.28.170.96 -async -debug 

服务器调试报错信息 #

ssl_msg.c:3620: input record: msgtype = 22, version = [3:3], msglen = 68
ssl_msg.c:1131: => decrypt buf
ssl_msg.c:1342: using encrypt then mac
ssl_msg.c:1385: message mac does not match
ssl_msg.c:3755: ssl_decrypt_buf() returned -29056 (-0x7180)
ssl_msg.c:4921: => send alert message
ssl_msg.c:4922: send alert level=2 message=20
ssl_msg.c:2684: => write record
ssl_msg.c:2797: output record: msgtype = 21, version = [254:253], msglen = 2
ssl_msg.c:2087: => flush output
ssl_msg.c:2105: message length: 15, out_left: 15
ssl_msg.c:2112: ssl->f_send() returned 15 (-0xfffffff1)
ssl_msg.c:2140: <= flush output
ssl_msg.c:2853: <= write record
ssl_msg.c:4934: <= send alert message
ssl_msg.c:3917: ssl_get_next_record() returned -29056 (-0x7180)
ssl_tls.c:3650: mbedtls_ssl_read_record() returned -29056 (-0x7180)
ssl_msg.c:0072: set_timer to 0 ms
ssl_msg.c:0072: set_timer to 0 ms
Traceback (most recent call last):
  File "/home/ron/venvs/udpserver/./dtlsthreaded.py", line 96, in <module>
    echo_until(srv, b"\0")
  File "/home/ron/venvs/udpserver/./dtlsthreaded.py", line 64, in echo_until
    block(cli1.do_handshake)
  File "/home/ron/venvs/udpserver/./dtlsthreaded.py", line 28, in block
    result = cb(*args, **kwargs)
  File "/home/ron/venvs/udpserver/lib/python3.10/site-packages/mbedtls/tls.py", line 341, in do_handshake
    self._buffer.do_handshake()
  File "src/mbedtls/_tls.pyx", line 1404, in mbedtls._tls.MbedTLSBuffer.do_handshake
  File "src/mbedtls/_tls.pyx", line 1429, in mbedtls._tls.MbedTLSBuffer._handle_handshake_response
  File "src/mbedtls/exceptions.pyx", line 53, in mbedtls.exceptions.check_error
  File "src/mbedtls/exceptions.pyx", line 56, in mbedtls.exceptions.check_error
mbedtls.exceptions.TLSError: TLSError([0x7180] 'SSL - Verification of the message MAC failed')
ssl_tls.c:6800: => free

内容的提问来源于stack exchange，提问作者kiwironnie