Fenrir fixes (2026-06-23)

examples/client.py

@@ -89,6 +89,12 @@ def build_arg_parser():
         help="Disable client cert check"
     )
 
+    parser.add_argument(
+        "-n", action="store_true",
+        help="Disable server hostname check "
+             "(needed for IP literals or test certificates)"
+    )
+
     parser.add_argument(
         "-g", action="store_true",
         help="Send server HTTP GET"
@@ -125,6 +131,33 @@ def get_DTLSmethod(index):
         wolfssl.PROTOCOL_DTLSv1_3
     )[index]
 
+def configure_verification(context, args):
+    """
+    Configure peer certificate and hostname verification on the context
+    according to the parsed arguments. Returns the server_hostname to pass
+    to wrap_socket() (None when no hostname check should be performed).
+
+    When certificate verification is enabled (the default), hostname
+    verification is enabled too so that a CA-trusted certificate issued for
+    a different host is rejected. Pass -n to opt out explicitly (e.g. when
+    connecting to an IP literal or using test certificates).
+    """
+    if args.d:
+        context.verify_mode = wolfssl.CERT_NONE
+        context.check_hostname = False
+        return None
+
+    context.verify_mode = wolfssl.CERT_REQUIRED
+    context.load_verify_locations(args.A)
+
+    if args.n:
+        context.check_hostname = False
+        return None
+
+    context.check_hostname = True
+    return args.h
+
+
 def main():
     args = build_arg_parser().parse_args()
 
@@ -147,18 +180,15 @@ def main():
 
     context.load_cert_chain(args.c, args.k)
 
-    if args.d:
-        context.verify_mode = wolfssl.CERT_NONE
-    else:
-        context.verify_mode = wolfssl.CERT_REQUIRED
-        context.load_verify_locations(args.A)
+    server_hostname = configure_verification(context, args)
 
     if args.l:
         context.set_ciphers(args.l)
 
     secure_socket = None
     try:
-        secure_socket = context.wrap_socket(bind_socket)
+        secure_socket = context.wrap_socket(
+            bind_socket, server_hostname=server_hostname)
 
         if not args.C:
             secure_socket.enable_crl(1)

examples/server.py

@@ -115,6 +115,21 @@ def get_DTLSmethod(index):
     )[index]
 
 
+# Large enough to peek a DTLS ClientHello source address.
+PEEK_BUFSIZE = 1500
+
+
+def peek_peer_address(sock):
+    """
+    Return the source address of the next pending datagram without removing
+    it from the socket queue. MSG_PEEK leaves the datagram (the DTLS
+    ClientHello) intact so wolfSSL_accept() can consume it during the
+    handshake.
+    """
+    _, from_addr = sock.recvfrom(PEEK_BUFSIZE, socket.MSG_PEEK)
+    return from_addr
+
+
 def main():
     args = build_arg_parser().parse_args()
     # DTLS connection over UDP
@@ -124,7 +139,6 @@ def main():
             args.v = 1
         bind_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
         bind_socket.bind(("" if args.b else "localhost", args.p))
-        data, from_addr = bind_socket.recvfrom(1)
         context = wolfssl.SSLContext(get_DTLSmethod(args.v), server_side=True)
     # SSL/TLS connection over TCP
     else:
@@ -156,6 +170,9 @@ def main():
         try:
             secure_socket = None
             if args.u:
+                # Peek the client's address for this connection without
+                # consuming the ClientHello datagram needed by the handshake.
+                from_addr = peek_peer_address(bind_socket)
                 secure_socket = context.wrap_socket(bind_socket)
             else:
                 new_socket, from_addr = bind_socket.accept()

tests/test_client_example.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+#
+# test_client_example.py
+#
+# Copyright (C) 2006-2020 wolfSSL Inc.
+#
+# This file is part of wolfSSL. (formerly known as CyaSSL)
+#
+# wolfSSL is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# wolfSSL is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+
+# pylint: disable=missing-docstring, invalid-name, import-error
+
+import os
+import sys
+
+import wolfssl
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "examples"))
+import client as client_example  # noqa: E402
+
+
+def _args(argv):
+    return client_example.build_arg_parser().parse_args(argv)
+
+
+def _ctx():
+    return wolfssl.SSLContext(wolfssl.PROTOCOL_TLSv1_2)
+
+
+def test_verification_enables_hostname_check_by_default():
+    """
+    F-5621: with cert verification on (the default), the client must also
+    verify the peer's hostname and pass server_hostname to wrap_socket.
+    """
+    args = _args(["-h", "example.com"])
+    ctx = _ctx()
+
+    server_hostname = client_example.configure_verification(ctx, args)
+
+    assert ctx.verify_mode == wolfssl.CERT_REQUIRED
+    assert ctx.check_hostname is True
+    assert server_hostname == "example.com"
+
+
+def test_disable_cert_check_skips_hostname():
+    args = _args(["-d"])
+    ctx = _ctx()
+    # Simulate a reused context that previously had hostname checking on:
+    # -d must clear it, not leave it dangling against CERT_NONE.
+    ctx.verify_mode = wolfssl.CERT_REQUIRED
+    ctx.check_hostname = True
+
+    server_hostname = client_example.configure_verification(ctx, args)
+
+    assert ctx.verify_mode == wolfssl.CERT_NONE
+    assert ctx.check_hostname is False
+    assert server_hostname is None
+
+
+def test_hostname_check_can_be_opted_out():
+    """An explicit opt-out is provided for IP literals / test certs."""
+    args = _args(["-n"])
+    ctx = _ctx()
+    # Reused context with hostname checking previously enabled: -n must
+    # actively turn it back off.
+    ctx.verify_mode = wolfssl.CERT_REQUIRED
+    ctx.check_hostname = True
+
+    server_hostname = client_example.configure_verification(ctx, args)
+
+    assert ctx.verify_mode == wolfssl.CERT_REQUIRED
+    assert ctx.check_hostname is False
+    assert server_hostname is None

tests/test_dtls_handshake_once.py

@@ -0,0 +1,105 @@
+# -*- coding: utf-8 -*-
+#
+# test_dtls_handshake_once.py
+#
+# Copyright (C) 2006-2020 wolfSSL Inc.
+#
+# This file is part of wolfSSL. (formerly known as CyaSSL)
+#
+# wolfSSL is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# wolfSSL is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+
+# pylint: disable=missing-docstring, invalid-name, import-error
+# pylint: disable=protected-access
+
+"""
+F-4136: for DTLS, write()/read()/recv_into() used to call do_handshake() on
+every single call. Once the handshake has completed, re-running it is wasteful
+and, on a non-blocking socket, wolfSSL_accept/connect can raise
+SSLWantReadError and abort an otherwise valid I/O. The handshake must only be
+driven until it completes.
+"""
+
+from types import SimpleNamespace
+
+import pytest
+import wolfssl
+
+
+class _OkLib:
+    """_lib stub whose I/O calls always succeed."""
+
+    def wolfSSL_write(self, ssl, data, length):
+        return length
+
+    def wolfSSL_read(self, ssl, data, length):
+        return length
+
+    def wolfSSL_get_error(self, ssl, ret):  # pragma: no cover
+        return 0
+
+
+def _make_dtls_socket(handshake_complete):
+    sock = wolfssl.SSLSocket.__new__(wolfssl.SSLSocket)
+    sock.native_object = object()
+    sock._connected = True
+    sock._server_side = True
+    sock._context = SimpleNamespace(protocol=wolfssl.PROTOCOL_DTLSv1_2)
+    sock._handshake_complete = handshake_complete
+    sock._release_native_object = lambda: None
+    return sock
+
+
+@pytest.fixture
+def spy_handshake(monkeypatch):
+    monkeypatch.setattr(wolfssl, "_lib", _OkLib())
+    calls = []
+
+    def _record(sock):
+        calls.append(True)
+        sock._handshake_complete = True
+
+    return calls, _record
+
+
+@pytest.mark.parametrize("op", ["write", "read", "recv_into"])
+def test_dtls_io_does_not_redrive_completed_handshake(spy_handshake, op):
+    calls, record = spy_handshake
+    sock = _make_dtls_socket(handshake_complete=True)
+    sock.do_handshake = lambda block=False: record(sock)
+
+    if op == "write":
+        sock.write(b"payload")
+    elif op == "read":
+        sock.read(8)
+    else:
+        sock.recv_into(bytearray(8))
+
+    assert calls == [], "do_handshake() must not run once the handshake is done"
+
+
+@pytest.mark.parametrize("op", ["write", "read", "recv_into"])
+def test_dtls_io_drives_handshake_until_complete(spy_handshake, op):
+    calls, record = spy_handshake
+    sock = _make_dtls_socket(handshake_complete=False)
+    sock.do_handshake = lambda block=False: record(sock)
+
+    if op == "write":
+        sock.write(b"payload")
+    elif op == "read":
+        sock.read(8)
+    else:
+        sock.recv_into(bytearray(8))
+
+    assert calls == [True], "first DTLS I/O must drive the handshake once"

tests/test_dtls_server_example.py

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+#
+# test_dtls_server_example.py
+#
+# Copyright (C) 2006-2020 wolfSSL Inc.
+#
+# This file is part of wolfSSL. (formerly known as CyaSSL)
+#
+# wolfSSL is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# wolfSSL is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+
+# pylint: disable=missing-docstring, invalid-name, import-error
+
+import os
+import sys
+import socket
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "examples"))
+import server as server_example  # noqa: E402
+
+
+@pytest.fixture
+def udp_pair():
+    srv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    srv.bind(("localhost", 0))
+    cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    try:
+        yield srv, cli, srv.getsockname()
+    finally:
+        srv.close()
+        cli.close()
+
+
+def test_peek_peer_address_returns_source(udp_pair):
+    srv, cli, srv_addr = udp_pair
+    cli.bind(("localhost", 0))
+    cli.sendto(b"clienthello-payload", srv_addr)
+
+    addr = server_example.peek_peer_address(srv)
+
+    assert addr == cli.getsockname()
+
+
+def test_peek_peer_address_does_not_consume_datagram(udp_pair):
+    """
+    Regression test for F-3481: peeking the client's address before the
+    DTLS handshake must leave the ClientHello datagram intact. The previous
+    example used recvfrom(1), which consumed the datagram and discarded
+    everything past the first byte, breaking the handshake.
+    """
+    srv, cli, srv_addr = udp_pair
+    payload = b"X" * 256  # stand-in for a DTLS ClientHello record
+    cli.sendto(payload, srv_addr)
+
+    server_example.peek_peer_address(srv)
+
+    # The datagram must still be fully available for wolfSSL_accept().
+    srv.settimeout(2)
+    data, _ = srv.recvfrom(4096)
+    assert data == payload