From f85c6a81be7f2d7d1b5b6ba858cefd9f50736d61 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Fri, 11 Sep 2026 11:54:55 +0200 Subject: [PATCH] fix(s7commplus): renew legacy session keys --- s7commplus/client.py | 5 ++ s7commplus/connection.py | 98 ++++++++++++++++++++++++- s7commplus/session_auth/ARCHITECTURE.md | 13 ++++ tests/test_s7_v2.py | 89 ++++++++++++++++++++++ 4 files changed, 204 insertions(+), 1 deletion(-) diff --git a/s7commplus/client.py b/s7commplus/client.py index b1879452..d0432c2c 100644 --- a/s7commplus/client.py +++ b/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)") diff --git a/s7commplus/connection.py b/s7commplus/connection.py index f20ef01a..df5fed7b 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -165,6 +165,7 @@ def _log_create_object_return_value(return_value: int, tls_active: bool) -> None _MAX_SYSTEM_EVENTS_PER_RESPONSE = 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]: @@ -525,6 +526,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 @@ -537,7 +542,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 @@ -610,6 +618,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. @@ -627,7 +636,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) @@ -693,6 +709,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}, " @@ -906,6 +923,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() @@ -938,6 +962,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 @@ -1002,6 +1096,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 diff --git a/s7commplus/session_auth/ARCHITECTURE.md b/s7commplus/session_auth/ARCHITECTURE.md index 6cd8357e..ba3c9ada 100644 --- a/s7commplus/session_auth/ARCHITECTURE.md +++ b/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 diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index b1d4ed18..9f1bf5a9 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -1276,6 +1276,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