fix(s7commplus): renew legacy session keys

s7commplus/client.py

@@ -126,6 +126,7 @@ def connect(
         tls_key: Optional[str] = None,
         tls_ca: Optional[str] = None,
         password: Optional[str] = None,
+        legacy_session_key_refresh_interval: Optional[float] = 25 * 60.0,
     ) -> None:
         """Connect to an S7-1200/1500 PLC using S7CommPlus.
 
@@ -139,6 +140,8 @@ def connect(
             tls_key: Path to client private key (PEM)
             tls_ca: Path to CA certificate for PLC verification (PEM)
             password: PLC password for legitimation (V2+ with TLS)
+            legacy_session_key_refresh_interval: Seconds between legacy
+                SessionKey renewals, or ``None`` to disable them.
         """
         self._connect_params = {
             "host": host,
@@ -148,6 +151,7 @@ def connect(
             "tls_key": tls_key,
             "tls_ca": tls_ca,
             "password": password,
+            "legacy_session_key_refresh_interval": legacy_session_key_refresh_interval,
         }
         self._open_connection()
 
@@ -163,6 +167,7 @@ def _open_connection(self) -> None:
             tls_key=p["tls_key"],
             tls_ca=p["tls_ca"],
             password=p["password"] or "",
+            legacy_session_key_refresh_interval=p["legacy_session_key_refresh_interval"],
         )
         if p["password"] is not None and self._connection.tls_active and not self._connection.requires_substreamed:
             logger.info("Performing PLC legitimation (password authentication)")

s7commplus/connection.py

@@ -190,6 +190,7 @@ def _log_create_object_return_value(return_value: int, tls_active: bool) -> None
 _MAX_SYSTEM_EVENTS_PER_RESPONSE = 16
 _MAX_STALE_RESPONSES_PER_REQUEST = 16
 _SYSTEM_EVENT_RETURN_VALUE_ID = 40305
+_DEFAULT_LEGACY_SESSION_KEY_REFRESH_INTERVAL = 25 * 60.0
 
 
 def _system_event_return_value(payload: bytes) -> Optional[int]:
@@ -550,6 +551,10 @@ def __init__(
         self._session_key: Optional[bytes] = None
         self._session_auth_public_key: bytes = b""
         self._session_auth_family: int = 0
+        self._session_key_refresh_interval: Optional[float] = _DEFAULT_LEGACY_SESSION_KEY_REFRESH_INTERVAL
+        self._session_key_refresh_timer: Optional[threading.Timer] = None
+        self._session_key_refresh_generation = 0
+        self._session_key_refresh_error: Optional[Exception] = None
 
         # V2+ IntegrityId tracking
         self._integrity_id_read: int = 0
@@ -562,7 +567,10 @@ def __init__(
         # Password for post-auth legitimation (V1-initial PLCs)
         self._connect_password: str = ""
         self._notification_frames: deque[bytes] = deque()
-        self._request_lock = threading.Lock()
+        # Reentrant because integrity failures disconnect from inside a
+        # serialized request. Disconnect itself also takes this lock so a
+        # renewal cannot race transport teardown.
+        self._request_lock = threading.RLock()
 
         # Effective protection level, read once the session is up
         self._protection_level: Optional[int] = None
@@ -635,6 +643,7 @@ def connect(
         tls_key: Optional[str] = None,
         tls_ca: Optional[str] = None,
         password: str = "",
+        legacy_session_key_refresh_interval: Optional[float] = _DEFAULT_LEGACY_SESSION_KEY_REFRESH_INTERVAL,
     ) -> None:
         """Establish S7CommPlus connection.
 
@@ -652,7 +661,14 @@ def connect(
             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)
+            legacy_session_key_refresh_interval: Seconds between legacy
+                SessionKey renewals. Defaults to 25 minutes; pass ``None`` to
+                disable automatic renewal.
         """
+        if legacy_session_key_refresh_interval is not None and legacy_session_key_refresh_interval <= 0:
+            raise ValueError("legacy_session_key_refresh_interval must be positive or None")
+        self._session_key_refresh_interval = legacy_session_key_refresh_interval
+        self._session_key_refresh_error = None
         self._connect_password = password
         try:
             # Step 1: COTP connection (same TSAP for all S7CommPlus versions)
@@ -718,6 +734,7 @@ def connect(
                     logger.info(f"PLC reports protection level: {self._protection_level}")
 
             self._connected = True
+            self._schedule_session_key_refresh()
 
             logger.info(
                 f"S7CommPlus connected to {self.host}:{self.port}, "
@@ -931,6 +948,13 @@ def collect_explore_frames(self, first_payload: bytes) -> bytes:
 
     def disconnect(self) -> None:
         """Disconnect from PLC."""
+        self._stop_session_key_refresh()
+        with self._request_lock:
+            self._session_key_refresh_error = None
+            self._disconnect()
+
+    def _disconnect(self) -> None:
+        """Clear connection state without changing a stored refresh failure."""
         if self._session_ready and self._session_id:
             try:
                 self._delete_session()
@@ -963,6 +987,76 @@ def disconnect(self) -> None:
         self._notification_frames.clear()
         self._iso_conn.disconnect()
 
+    def _stop_session_key_refresh(self) -> None:
+        """Cancel pending legacy SessionKey renewal activity."""
+        self._session_key_refresh_generation += 1
+        timer = self._session_key_refresh_timer
+        self._session_key_refresh_timer = None
+        if timer is not None:
+            timer.cancel()
+
+    def _schedule_session_key_refresh(self) -> None:
+        """Schedule one renewal for an authenticated legacy session."""
+        interval = self._session_key_refresh_interval
+        if interval is None or self._session_key is None or not self._connected:
+            return
+        generation = self._session_key_refresh_generation
+        timer = threading.Timer(interval, self._session_key_refresh_callback, args=(generation,))
+        timer.daemon = True
+        self._session_key_refresh_timer = timer
+        timer.start()
+
+    def _session_key_refresh_callback(self, generation: int) -> None:
+        """Renew under the request lock, or make a failed renewal terminal."""
+        try:
+            with self._request_lock:
+                if generation != self._session_key_refresh_generation or not self._connected:
+                    return
+                self._session_key_refresh_timer = None
+                self._renew_session_key_locked()
+                if generation == self._session_key_refresh_generation:
+                    self._schedule_session_key_refresh()
+        except Exception as exc:
+            from snap7.error import S7ConnectionError
+
+            failure = S7ConnectionError(f"Legacy SessionKey renewal failed: {exc}")
+            logger.error("%s", failure)
+            self._session_key_refresh_error = failure
+            self._stop_session_key_refresh()
+            # Do not send DeleteObject on a stream whose key may have expired.
+            self._session_ready = False
+            self._session_id = 0
+            self._disconnect()
+
+    def _renew_session_key_locked(self) -> None:
+        """Perform the challenge/SecurityKey exchange while the old key is active."""
+        from snap7.error import S7ConnectionError
+
+        if self._session_key is None or not self._session_auth_public_key:
+            raise S7ConnectionError("Legacy SessionKey renewal prerequisites are unavailable")
+
+        from .session_auth.keys import KeyFamily
+
+        integrity_tail = 3 if self._session_auth_family == KeyFamily.S7_1200 else 4
+        challenge_payload = self._build_get_var_substreamed(self._session_id, LegitimationId.SERVER_SESSION_REQUEST)
+        challenge_response = self._send_request(FunctionCode.GET_VAR_SUBSTREAMED, challenge_payload, integrity_tail, False)
+        challenge = _parse_get_var_substreamed_response(challenge_response)
+        if len(challenge) != 20:
+            raise S7ConnectionError(f"SessionKey renewal returned an unexpected {len(challenge)}-byte challenge")
+
+        from .session_auth.legacy_auth import authenticate_real_plc
+
+        blob, new_session_key = authenticate_real_plc(challenge, self._session_auth_public_key, self._session_auth_family)
+        security_key = self._encode_security_key_struct(blob, new_session_key)
+        renewal_payload = _build_set_variable_payload(self._session_id, LegitimationId.SESSION_SETUP_LEGITIMATION, security_key)
+        # _send_request signs and verifies with self._session_key. Keep the old
+        # key installed until the PLC has accepted this write and its response
+        # has passed HMAC verification.
+        renewal_response = self._send_request(FunctionCode.SET_VARIABLE, renewal_payload, 4, False)
+        _check_set_variable_response(renewal_response)
+        self._session_key = new_session_key
+        logger.info("Legacy SessionKey renewed successfully")
+
     def _invalidate_integrity_failure(self) -> None:
         """Close an untrusted stream without sending protocol data on it."""
         self._session_ready = False
@@ -1027,6 +1121,8 @@ def _send_request(self, function_code: int, payload: bytes, integrity_tail: int,
         Returns:
             Response payload (after the 10-byte response header)
         """
+        if self._session_key_refresh_error is not None:
+            raise self._session_key_refresh_error
         if not (self._connected or self._session_ready):
             from snap7.error import S7ConnectionError
 

s7commplus/session_auth/ARCHITECTURE.md

@@ -52,6 +52,19 @@ After SetupSession, all data frames use **V3 framing** with HMAC-SHA256
 (keyed by the first 24 bytes of the session key). No intermediate
 activation sequence is needed — data reads work immediately.
 
+Legacy SessionKeys are renewed every 25 minutes by default, before the PLC's
+key expiry window. Renewal reads a fresh challenge from address 303 and writes
+a new SecurityKey to address 1830 while holding the same lock as application
+requests. The PLC's response is authenticated with the old key; the new key is
+installed only after that response is verified and accepted. A renewal failure
+closes the connection instead of continuing with an expired or ambiguous key.
+
+The interval is configurable in seconds through
+`S7CommPlusClient.connect(legacy_session_key_refresh_interval=...)` (or the
+low-level connection method). Pass `None` to disable automatic renewal. This
+timer applies only to legacy V1-initial SessionKey sessions; TLS sessions do not
+start it.
+
 Note: TIA Portal sends SET_VARIABLE attr 323 + finalize reads before
 data operations, but this is TIA-specific behavior. The HarpoS7
 reference implementation skips it, and V1-initial PLCs reject the

tests/test_s7_v2.py

@@ -1282,6 +1282,95 @@ def test_tls_v2_does_not_attempt_session_key_auth(self) -> None:
         assert conn._try_session_key_auth() is None
         assert conn._session_key is None
 
+
+class TestLegacySessionKeyRefresh:
+    @staticmethod
+    def _authenticated_connection() -> S7CommPlusConnection:
+        from s7commplus.session_auth.keys import KeyFamily
+
+        conn = S7CommPlusConnection("127.0.0.1")
+        conn._connected = True
+        conn._session_ready = True
+        conn._session_id = 0x70000FDC
+        conn._session_key = b"o" * 24
+        conn._session_auth_public_key = b"p" * 40
+        conn._session_auth_family = KeyFamily.S7_1500
+        return conn
+
+    def test_renewal_installs_key_only_after_accepted_old_key_response(self) -> None:
+        conn = self._authenticated_connection()
+        challenge = bytes(range(20))
+        challenge_response = bytes([0x00, 0x00, 0x10, DataType.USINT, len(challenge)]) + challenge
+        old_key = conn._session_key
+        new_key = b"n" * 24
+        keys_during_exchange: list[bytes | None] = []
+
+        def exchange(*_args: object) -> bytes:
+            keys_during_exchange.append(conn._session_key)
+            return challenge_response if len(keys_during_exchange) == 1 else b"\x00"
+
+        conn._send_request = MagicMock(side_effect=exchange)
+        with patch("s7commplus.session_auth.legacy_auth.authenticate_real_plc", return_value=(b"b" * 180, new_key)):
+            conn._renew_session_key_locked()
+
+        assert keys_during_exchange == [old_key, old_key]
+        assert conn._session_key == new_key
+        renewal_call = conn._send_request.call_args_list[1]
+        assert renewal_call.args[0] == FunctionCode.SET_VARIABLE
+        assert encode_uint32_vlq(LegitimationId.SESSION_SETUP_LEGITIMATION) in renewal_call.args[1]
+
+    def test_rejected_renewal_never_installs_generated_key(self) -> None:
+        conn = self._authenticated_connection()
+        challenge = bytes(range(20))
+        challenge_response = bytes([0x00, 0x00, 0x10, DataType.USINT, len(challenge)]) + challenge
+        old_key = conn._session_key
+        conn._send_request = MagicMock(side_effect=[challenge_response, encode_uint32_vlq(0x8104)])
+
+        with (
+            patch("s7commplus.session_auth.legacy_auth.authenticate_real_plc", return_value=(b"b" * 180, b"n" * 24)),
+            pytest.raises(S7ConnectionError, match="return_value=0x8104"),
+        ):
+            conn._renew_session_key_locked()
+
+        assert conn._session_key == old_key
+
+    def test_short_interval_renews_while_requests_remain_usable(self) -> None:
+        conn = self._authenticated_connection()
+        conn._session_key_refresh_interval = 0.01
+        renewed = threading.Event()
+        conn._renew_session_key_locked = MagicMock(side_effect=renewed.set)
+        conn._schedule_session_key_refresh()
+
+        assert renewed.wait(1)
+        conn._send_request = MagicMock(return_value=b"read result")
+        assert conn.send_request(FunctionCode.GET_VARIABLE) == b"read result"
+
+        next_timer = conn._session_key_refresh_timer
+        conn.disconnect()
+        if next_timer is not None:
+            next_timer.join(1)
+            assert not next_timer.is_alive()
+        assert conn._session_key_refresh_timer is None
+
+    def test_refresh_failure_is_terminal_and_surfaces_on_next_request(self) -> None:
+        conn = self._authenticated_connection()
+        conn._iso_conn.disconnect = MagicMock()
+        conn._renew_session_key_locked = MagicMock(side_effect=S7ConnectionError("PLC rejected key"))
+
+        conn._session_key_refresh_callback(conn._session_key_refresh_generation)
+
+        assert not conn.connected
+        assert conn._session_key is None
+        with pytest.raises(S7ConnectionError, match="Legacy SessionKey renewal failed: PLC rejected key"):
+            conn.send_request(FunctionCode.GET_VARIABLE)
+
+    def test_refresh_interval_must_be_positive(self) -> None:
+        conn = S7CommPlusConnection("127.0.0.1")
+        with pytest.raises(ValueError, match="must be positive"):
+            conn.connect(legacy_session_key_refresh_interval=0)
+
+
+class TestSessionKeyDescriptors:
     def test_security_key_descriptor_uses_pending_generated_key(self) -> None:
         from s7commplus.session_auth.keys import KeyFamily, get_public_key
         from s7commplus.session_auth.utils import derive_key_id