diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 93f4b4aa..501ba9ed 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -40,6 +40,7 @@ _build_get_var_substreamed_payload, _build_set_variable_payload, _check_set_variable_response, + _log_create_object_return_value, _parse_get_var_substreamed_response, _parse_protection_level_response, _set_s7_groups, @@ -939,8 +940,7 @@ async def _create_session(self) -> None: 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)") + _log_create_object_return_value(return_value, self._tls_active) self._server_session_version = parse_server_session_version(response[10 + obj_end :]) if self._server_session_version is not None: diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 756fd5e3..1478b3c8 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -68,6 +68,25 @@ logger = logging.getLogger(__name__) + +def _log_create_object_return_value(return_value: int, tls_active: bool) -> None: + """Log a non-zero CreateObject status without guessing at TLS requirements.""" + if return_value == 0: + return + if tls_active: + # Some firmware (e.g. S7-1200 FW V4.1) returns a non-zero value on a + # usable TLS session, so keep this informational. + logger.debug( + "CreateObject returned non-zero 0x%X on an active TLS session; continuing to parse the returned session data", + return_value, + ) + return + logger.warning( + "CreateObject returned non-zero 0x%X; continuing to parse the returned session data", + return_value, + ) + + # TLS cipher suites for S7 PLC compatibility. # ECDHE suites are preferred (forward secrecy); RSA-kx kept as fallback for # older firmware. The key to Siemens PLC compatibility is restricting the @@ -1093,13 +1112,7 @@ def _wstring_attr(attr_id: int, s: str) -> bytes: logger.debug(f"CreateObject response: return_value={return_value} object_ids={[hex(i) for i in object_ids]}") logger.debug(f"Session created: id=0x{self._session_id:08X} ({self._session_id}), version=V{version}") - if return_value != 0: - if self._tls_active: - # Some firmware (e.g. S7-1200 FW V4.1) returns a non-zero CreateObject - # value on a perfectly usable TLS session, so this is informational only. - logger.debug(f"CreateObject returned non-zero 0x{return_value:X} on an active TLS session (session still usable)") - else: - logger.warning(f"CreateObject returned error 0x{return_value:X} — PLC may require TLS (use_tls=True)") + _log_create_object_return_value(return_value, self._tls_active) # Parse remaining payload (the ResponseObject tree) for session attributes attrs = parse_create_object_attributes(response[offset:]) diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index a93efa6f..f803dd96 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -5,6 +5,7 @@ """ import hashlib +import logging import struct from unittest.mock import AsyncMock, MagicMock @@ -17,6 +18,7 @@ _build_get_var_substreamed_payload, _build_set_variable_payload, _check_set_variable_response, + _log_create_object_return_value, _parse_get_var_substreamed_response, _parse_protection_level_response, ) @@ -534,3 +536,20 @@ def test_authenticate_requires_tls(self) -> None: conn._tls_active = False with pytest.raises(S7ConnectionError, match="requires TLS"): conn.authenticate("password") + + +class TestCreateObjectStatusLogging: + """A CreateObject status alone does not identify a TLS requirement.""" + + def test_plain_connection_does_not_recommend_tls(self, caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.WARNING, logger="s7commplus.connection"): + _log_create_object_return_value(0x4000800000000011, tls_active=False) + + assert "continuing to parse the returned session data" in caplog.text + assert "TLS" not in caplog.text + + def test_success_is_silent(self, caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.DEBUG, logger="s7commplus.connection"): + _log_create_object_return_value(0, tls_active=False) + + assert caplog.text == ""