fix(tls): allow TLS 1.2 for S7-1500 PLCs

doc/connecting.rst

@@ -152,29 +152,30 @@ with the ``s7commplus`` extra:
    ``browse()`` and other CommPlus-only operations are not yet
    supported on those firmwares — see issue #710.
 
-TLS handshake fails on OpenSSL 3.5+ (post-quantum key share)
--------------------------------------------------------------
-
-On systems with **OpenSSL ≥ 3.5** (e.g. Debian 13 and other recent
-distributions) the TLS 1.3 ``ClientHello`` advertises a post-quantum
-hybrid key-exchange group, ``X25519MLKEM768``, by default. Its key
-share is roughly 1.2 KB, and the S7-1500's TLS stack rejects the
-oversized ``ClientHello`` — it drops the connection mid-handshake, so
-``connect(use_tls=True)`` fails with a *connection reset by peer*.
-
-The PLC mandates TLS 1.3 (it refuses TLS 1.2 outright), and CPython's
-``ssl`` module exposes no API to restrict the TLS 1.3
-``supported_groups`` list. The fix is to restrict the offered groups
-through OpenSSL's own configuration — via the ``OPENSSL_CONF``
-environment variable — to the classic ECDHE curves that every
-S7-1200/1500 supports.
+TLS handshake rejected by the PLC (connection reset)
+------------------------------------------------------
+
+S7 PLCs have a minimal TLS stack that rejects ``ClientHello`` messages
+containing features it does not recognise. Two common causes:
+
+* **Post-quantum key share (OpenSSL ≥ 3.5)** — the default
+  ``ClientHello`` advertises the ``X25519MLKEM768`` hybrid group whose
+  ~1.2 KB key share the PLC drops.
+* **Modern signature algorithms** — OpenSSL advertises Ed25519, Ed448,
+  and RSA-PSS variants in the ``signature_algorithms`` extension.
+  Instead of ignoring unknown algorithms (as TLS 1.2 requires), the PLC
+  treats them as a fatal error and sends a TCP RST.
+
+CPython's ``ssl`` module exposes no API for either the
+``supported_groups`` or ``signature_algorithms`` lists. The fix is to
+restrict both through OpenSSL's own configuration via the
+``OPENSSL_CONF`` environment variable.
 
 1. Create an OpenSSL configuration file, e.g. ``s7-openssl.cnf``:
 
    .. code-block:: ini
 
-      # Restrict the TLS key-exchange groups to classic ECDHE curves so the
-      # ClientHello stays small enough for the S7-1500 to accept.
+      # Restrict TLS parameters for S7 PLC compatibility.
       openssl_conf = openssl_init
 
       [openssl_init]
@@ -184,7 +185,10 @@ S7-1200/1500 supports.
       system_default = system_default_sect
 
       [system_default_sect]
+      # Classic ECDHE curves only — no post-quantum groups.
       Groups = x25519:secp256r1:secp384r1
+      # Classic signature algorithms only — no Ed25519, Ed448, or RSA-PSS.
+      SignatureAlgorithms = RSA+SHA256:RSA+SHA384:RSA+SHA512:ECDSA+SHA256:ECDSA+SHA384
 
 2. Point ``OPENSSL_CONF`` at it **before** the Python process starts.
    OpenSSL reads this configuration once, when it initialises, so
@@ -200,14 +204,9 @@ S7-1200/1500 supports.
       # or for the whole shell session
       export OPENSSL_CONF=/path/to/s7-openssl.cnf
 
-With the groups restricted to classic curves, the TLS 1.3 handshake no
-longer offers ``X25519MLKEM768`` and the S7-1500 completes it normally.
-
-.. note::
-
-   This only affects OpenSSL ≥ 3.5. On older OpenSSL releases the
-   default key-exchange groups are already classic ECDHE curves, so no
-   extra configuration is needed.
+With both settings applied, the ``ClientHello`` contains only
+algorithms that S7 PLCs understand and the handshake completes
+normally.
 
 PLC Password Authentication
 ----------------------------

s7/_s7commplus_async_client.py

@@ -13,6 +13,7 @@
 import struct
 from typing import Any, Optional
 
+from .connection import _S7_CIPHERS
 from .protocol import (
     DataType,
     ElementID,
@@ -242,17 +243,20 @@ async def _activate_tls(
         tls_key: Optional[str] = None,
         tls_ca: Optional[str] = None,
     ) -> None:
-        """Activate TLS 1.3 over the COTP connection."""
+        """Activate TLS over the COTP connection."""
         if self._writer is None:
             from snap7.error import S7ConnectionError
 
             raise S7ConnectionError("Cannot activate TLS: not connected")
 
         ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
-        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+        ctx.maximum_version = ssl.TLSVersion.TLSv1_2
 
-        if hasattr(ctx, "set_ciphersuites"):
-            ctx.set_ciphersuites("TLS_AES_256_GCM_SHA384:TLS_AES_128_GCM_SHA256")
+        ctx.set_ciphers(_S7_CIPHERS)
+        ctx.options |= ssl.OP_NO_TICKET
+        ctx.options |= 0x00080000  # SSL_OP_NO_ENCRYPT_THEN_MAC
+        ctx.options |= 0x00000001  # SSL_OP_NO_EXTENDED_MASTER_SECRET (OpenSSL 3.0+)
 
         if tls_cert and tls_key:
             ctx.load_cert_chain(tls_cert, tls_key)
@@ -285,7 +289,7 @@ async def _activate_tls(
             logger.warning(f"Could not extract OMS exporter secret: {e}")
             self._oms_secret = None
 
-        logger.info("TLS 1.3 activated (tunneled inside COTP frames)")
+        logger.info("TLS activated (tunneled inside COTP frames)")
 
     async def _do_tls_handshake(self) -> None:
         """Perform the TLS handshake, tunneling records through COTP DT frames."""
@@ -821,13 +825,16 @@ async def _create_session(self) -> None:
         # Parse response body: ReturnValue(VLQ) + ObjectIdCount + ObjectIds(VLQ).
         # The usable session id is ObjectIds[0] (NOT the header SessionId field).
         body = response[14:]
-        object_ids, obj_end = parse_create_object_session_id(body)
+        object_ids, obj_end, return_value = parse_create_object_session_id(body)
         if object_ids:
             self._session_id = object_ids[0]
         else:
             self._session_id = struct.unpack_from(">I", response, 9)[0]
         self._protocol_version = version
 
+        if return_value != 0:
+            logger.warning(f"CreateObject returned error 0x{return_value:X} — PLC may require TLS (use_tls=True)")
+
         self._server_session_version = parse_server_session_version(response[14 + obj_end :])
         if self._server_session_version is not None:
             logger.info(f"ServerSessionVersion captured: {len(self._server_session_version)} bytes")

s7/_s7commplus_server.py

@@ -41,6 +41,7 @@
     READ_FUNCTION_CODES,
     SoftDataType,
 )
+from .connection import _S7_CIPHERS
 from .vlq import encode_uint32_vlq, decode_uint32_vlq, encode_uint64_vlq
 from .codec import (
     encode_header,
@@ -291,7 +292,10 @@ def start(
             if not tls_cert or not tls_key:
                 raise ValueError("TLS requires tls_cert and tls_key")
             self._ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
-            self._ssl_context.minimum_version = ssl.TLSVersion.TLSv1_3
+            self._ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
+            # Accept the same RSA-only ciphers the client uses (static RSA key exchange
+            # is excluded from the OpenSSL 3.x defaults but required by S7 PLCs).
+            self._ssl_context.set_ciphers(_S7_CIPHERS)
             self._ssl_context.load_cert_chain(tls_cert, tls_key)
             if tls_ca:
                 self._ssl_context.load_verify_locations(tls_ca)

s7/codec.py

@@ -612,15 +612,15 @@ def skip_typed_value(data: bytes, offset: int, datatype: int, flags: int) -> int
         return offset
 
 
-def parse_create_object_session_id(body: bytes) -> tuple[list[int], int]:
+def parse_create_object_session_id(body: bytes) -> tuple[list[int], int, int]:
     """Parse a CreateObject response body (after the 14-byte response header).
 
     Body layout: ReturnValue (UInt64 VLQ) + ObjectIdCount (1 byte) + ObjectIds (UInt32 VLQ
     each). The usable session id is ``ObjectIds[0]`` (not the header SessionId field).
 
-    Returns ``(object_ids, offset)`` where ``offset`` points just past the ObjectIds.
+    Returns ``(object_ids, offset, return_value)`` where ``offset`` points just past the ObjectIds.
     """
-    _return_value, consumed = decode_uint64_vlq(body, 0)
+    return_value, consumed = decode_uint64_vlq(body, 0)
     boff = consumed
     obj_count = body[boff] if boff < len(body) else 0
     boff += 1
@@ -629,7 +629,7 @@ def parse_create_object_session_id(body: bytes) -> tuple[list[int], int]:
         oid, c = decode_uint32_vlq(body, boff)
         boff += c
         object_ids.append(oid)
-    return object_ids, boff
+    return object_ids, boff, return_value
 
 
 def parse_server_session_version(payload: bytes) -> Optional[bytes]:

s7/connection.py

@@ -7,7 +7,7 @@
 - V1: Early S7-1200 (FW >= V4.0). Simple session handshake.
 - V2: Adds integrity checking and session authentication.
 - V3: Adds public-key-based key exchange.
-- V3 + TLS: TIA Portal V17+. Standard TLS 1.3 with per-device certificates.
+- V3 + TLS: TIA Portal V17+. TLS 1.2 with per-device certificates.
 
 The wire protocol (VLQ encoding, data types, function codes, object model) is
 the same across all versions -- only the session authentication layer differs.
@@ -68,6 +68,13 @@
 
 logger = logging.getLogger(__name__)
 
+# RSA-only cipher suites for S7 PLC compatibility.  ECDHE ciphers are excluded
+# because they force OpenSSL to include ec_point_formats and supported_groups
+# extensions in the ClientHello; S7-1500 PLCs send a bare TCP RST when they see
+# compressed EC point formats (which Python's ssl module always advertises with
+# ECDHE, and there is no API to suppress them).
+_S7_CIPHERS = "AES128-GCM-SHA256:AES256-GCM-SHA384:AES128-SHA256:AES256-SHA256"
+
 
 class S7CommPlusConnection:
     """S7CommPlus connection with multi-version support.
@@ -759,7 +766,7 @@ def _create_session(self) -> None:
         # Parse response body: ReturnValue(UInt64 VLQ) + ObjectIdCount + ObjectIds(VLQ).
         # The usable session id is ObjectIds[0] (NOT the header SessionId field).
         body = response[14:]
-        object_ids, obj_end = parse_create_object_session_id(body)
+        object_ids, obj_end, return_value = parse_create_object_session_id(body)
         if object_ids:
             self._session_id = object_ids[0]
         else:
@@ -778,6 +785,9 @@ def _create_session(self) -> None:
         logger.debug(f"CreateObject response payload: {response[14:].hex(' ')}")
         logger.debug(f"Session created: id=0x{self._session_id:08X} ({self._session_id}), version=V{version}")
 
+        if return_value != 0:
+            logger.warning(f"CreateObject returned error 0x{return_value:X} — PLC may require TLS (use_tls=True)")
+
         # Parse response payload to extract ServerSessionVersion
         self._server_session_version = parse_server_session_version(response[14 + obj_end :])
         if self._server_session_version is not None:
@@ -938,7 +948,7 @@ def _activate_tls(
         tls_key: Optional[str] = None,
         tls_ca: Optional[str] = None,
     ) -> None:
-        """Activate TLS 1.3 tunneled inside COTP data frames.
+        """Activate TLS tunneled inside COTP data frames.
 
         The S7CommPlus protocol transports TLS records as the payload
         of COTP DT frames — TPKT and COTP headers stay unencrypted on
@@ -955,7 +965,7 @@ def _activate_tls(
         Args:
             tls_cert: Path to client TLS certificate (PEM)
             tls_key: Path to client private key (PEM)
-            tls_ca: Path to CA certificate for PLC verification (PEM)
+            tls_ca: Path to PLC CA certificate (PEM)
         """
         ctx = self._setup_ssl_context(
             cert_path=tls_cert,
@@ -987,7 +997,7 @@ def _activate_tls(
             logger.warning(f"Could not extract OMS exporter secret: {e}")
             self._oms_secret = None
 
-        logger.info("TLS 1.3 activated (tunneled inside COTP frames)")
+        logger.info("TLS activated (tunneled inside COTP frames)")
 
     def _do_tls_handshake(self) -> None:
         """Perform TLS handshake, tunneling records through COTP."""
@@ -1020,12 +1030,13 @@ def _setup_ssl_context(
             Configured SSLContext
         """
         ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
-        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+        ctx.maximum_version = ssl.TLSVersion.TLSv1_2
 
-        # TLS 1.3 ciphersuites are configured differently from TLS 1.2
-        if hasattr(ctx, "set_ciphersuites"):
-            ctx.set_ciphersuites("TLS_AES_256_GCM_SHA384:TLS_AES_128_GCM_SHA256")
-        # If set_ciphersuites not available, TLS 1.3 uses its mandatory defaults
+        ctx.set_ciphers(_S7_CIPHERS)
+        ctx.options |= ssl.OP_NO_TICKET
+        ctx.options |= 0x00080000  # SSL_OP_NO_ENCRYPT_THEN_MAC
+        ctx.options |= 0x00000001  # SSL_OP_NO_EXTENDED_MASTER_SECRET (OpenSSL 3.0+)
 
         if cert_path and key_path:
             ctx.load_cert_chain(cert_path, key_path)