diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 93f4b4aa..fb218f6f 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -45,6 +45,7 @@ _set_s7_groups, ) from .protocol import ( + FLAGS_34_FUNCTION_CODES, READ_FUNCTION_CODES, S7COMMPLUS_LOCAL_TSAP, S7COMMPLUS_REMOTE_TSAP, @@ -737,8 +738,8 @@ async def _send_request( 0x0000, seq_num, self._session_id, - # Transport flags: 0x34 for GetMultiVariables and Explore, 0x36 otherwise. - 0x34 if function_code in (FunctionCode.GET_MULTI_VARIABLES, FunctionCode.EXPLORE) else 0x36, + # Transport flags: 0x34 for the function codes the reference sends with 0x34. + 0x34 if function_code in FLAGS_34_FUNCTION_CODES else 0x36, ) integrity_id_bytes = b"" @@ -992,7 +993,7 @@ async def _delete_session(self) -> None: 0x0000, seq_num, self._session_id, - 0x36, + 0x34, ) request += struct.pack(">I", 0) diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 756fd5e3..861e091c 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -52,6 +52,7 @@ from .codec import decode_header, encode_header, encode_object_qualifier, parse_create_object_attributes from .protocol import ( + FLAGS_34_FUNCTION_CODES, READ_FUNCTION_CODES, S7COMMPLUS_LOCAL_TSAP, S7COMMPLUS_REMOTE_TSAP, @@ -716,10 +717,8 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: seq_num, self._session_id, # Transport flags: 0x34 after SessionKey auth (matches TIA Portal), - # also for GetMultiVariables and Explore; 0x36 for other V1/TLS requests. - 0x34 - if self._session_key is not None or function_code in (FunctionCode.GET_MULTI_VARIABLES, FunctionCode.EXPLORE) - else 0x36, + # and for the function codes the reference sends with 0x34. + 0x34 if self._session_key is not None or function_code in FLAGS_34_FUNCTION_CODES else 0x36, ) integrity_id_bytes = b"" @@ -1470,7 +1469,7 @@ def _delete_session(self) -> None: 0x0000, seq_num, self._session_id, - 0x36, + 0x34, ) request += struct.pack(">I", 0) diff --git a/s7commplus/protocol.py b/s7commplus/protocol.py index 4d869630..bc05d20d 100644 --- a/s7commplus/protocol.py +++ b/s7commplus/protocol.py @@ -226,6 +226,26 @@ class Ids(IntEnum): } ) +# Function codes whose requests carry transport flags 0x34. The reference sets +# this per request class rather than by read/write, so it is a different split +# than READ_FUNCTION_CODES: the writes SetVariable, SetMultiVariables and +# DeleteObject use 0x34 too. Only CreateObject (0x36) and InitSSL (0x30) differ, +# and a session-setup CreateObject sent with 0x34 makes the PLC reset the +# connection. Subscription and alarm CreateObjects are the documented exception: +# the reference overrides those to 0x34. +# +# Reference: TransportFlags in thomas-v2/S7CommPlusDriver/Core/*Request.cs +FLAGS_34_FUNCTION_CODES: frozenset[int] = frozenset( + { + FunctionCode.DELETE_OBJECT, + FunctionCode.EXPLORE, + FunctionCode.GET_MULTI_VARIABLES, + FunctionCode.GET_VAR_SUBSTREAMED, + FunctionCode.SET_MULTI_VARIABLES, + FunctionCode.SET_VARIABLE, + } +) + class AccessLevel(IntEnum): """Protection levels reported by `Ids.EFFECTIVE_PROTECTION_LEVEL`. diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index a93efa6f..98b5e746 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -27,12 +27,15 @@ derive_legitimation_key, ) from s7commplus.protocol import ( + FLAGS_34_FUNCTION_CODES, READ_FUNCTION_CODES, AccessLevel, DataType, FunctionCode, Ids, LegitimationId, + ObjectId, + Opcode, ProtocolVersion, ) from s7commplus.vlq import decode_uint32_vlq, encode_uint32_vlq @@ -73,6 +76,43 @@ def test_delete_object_is_write(self) -> None: assert FunctionCode.DELETE_OBJECT not in READ_FUNCTION_CODES +class TestFlags34FunctionCodes: + """Test FLAGS_34_FUNCTION_CODES classification.""" + + def test_delete_object_uses_flags_34(self) -> None: + assert FunctionCode.DELETE_OBJECT in FLAGS_34_FUNCTION_CODES + + def test_explore_uses_flags_34(self) -> None: + assert FunctionCode.EXPLORE in FLAGS_34_FUNCTION_CODES + + def test_get_multi_variables_uses_flags_34(self) -> None: + assert FunctionCode.GET_MULTI_VARIABLES in FLAGS_34_FUNCTION_CODES + + def test_get_var_substreamed_uses_flags_34(self) -> None: + assert FunctionCode.GET_VAR_SUBSTREAMED in FLAGS_34_FUNCTION_CODES + + def test_set_multi_variables_uses_flags_34(self) -> None: + assert FunctionCode.SET_MULTI_VARIABLES in FLAGS_34_FUNCTION_CODES + + def test_set_variable_uses_flags_34(self) -> None: + assert FunctionCode.SET_VARIABLE in FLAGS_34_FUNCTION_CODES + + def test_create_object_uses_flags_36(self) -> None: + assert FunctionCode.CREATE_OBJECT not in FLAGS_34_FUNCTION_CODES + + def test_init_ssl_uses_flags_36(self) -> None: + assert FunctionCode.INIT_SSL not in FLAGS_34_FUNCTION_CODES + + def test_get_variable_uses_flags_36(self) -> None: + assert FunctionCode.GET_VARIABLE not in FLAGS_34_FUNCTION_CODES + + def test_get_variables_address_uses_flags_36(self) -> None: + assert FunctionCode.GET_VARIABLES_ADDRESS not in FLAGS_34_FUNCTION_CODES + + def test_get_link_uses_flags_36(self) -> None: + assert FunctionCode.GET_LINK not in FLAGS_34_FUNCTION_CODES + + class TestLegitimationId: """Test legitimation ID constants.""" @@ -224,6 +264,9 @@ def test_tls_v2_response_application_payload_is_not_stripped(self) -> None: assert conn.send_request(FunctionCode.GET_MULTI_VARIABLES, bytes(4)) == application_payload + # GetMultiVariables is in FLAGS_34_FUNCTION_CODES + assert conn._send_s7_data.call_args[0][0][17] == 0x34 + class TestIntegrityIdVlqEncoding: """Test VLQ encoding used for IntegrityId values.""" @@ -331,6 +374,116 @@ async def test_async_challenge_uses_protocol_request_shape(self) -> None: ) +class TestCreateSessionRequest: + """The CreateObject request that opens an S7CommPlus session.""" + + def test_sync_request_shape(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._send_s7_data = MagicMock() + # Frame header declaring a zero-length body: _create_session bails out on the + # length check, by which point the request is already on the wire. + conn._recv_s7_data = MagicMock(return_value=bytes.fromhex("72010000")) + + with pytest.raises(S7ConnectionError, match="CreateObject response too short"): + conn._create_session() + + frame = conn._send_s7_data.call_args[0][0] + request = struct.pack( + ">BHHHHIB", + Opcode.REQUEST, + 0x0000, + FunctionCode.CREATE_OBJECT, + 0x0000, + 0, # first sequence number on a fresh connection + ObjectId.OBJECT_NULL_SERVER_SESSION, + 0x36, + ) + request += struct.pack(">I", ObjectId.OBJECT_SERVER_SESSION_CONTAINER) + expected = encode_header(ProtocolVersion.V1, len(frame) - 8) + request + assert frame[: len(expected)] == expected + assert frame[-4:] == struct.pack(">BBH", 0x72, ProtocolVersion.V1, 0x0000) + + @pytest.mark.asyncio + async def test_async_request_shape(self) -> None: + client = S7CommPlusAsyncClient() + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=bytes.fromhex("72010000")) + + with pytest.raises(RuntimeError, match="CreateObject response too short"): + await client._create_session() + + client._send_cotp_dt.assert_awaited_once() + assert client._send_cotp_dt.await_args is not None + frame = client._send_cotp_dt.await_args[0][0] + request = struct.pack( + ">BHHHHIB", + Opcode.REQUEST, + 0x0000, + FunctionCode.CREATE_OBJECT, + 0x0000, + 0, + ObjectId.OBJECT_NULL_SERVER_SESSION, + 0x36, + ) + request += struct.pack(">I", ObjectId.OBJECT_SERVER_SESSION_CONTAINER) + expected = encode_header(ProtocolVersion.V1, len(frame) - 8) + request + assert frame[: len(expected)] == expected + assert frame[-4:] == struct.pack(">BBH", 0x72, ProtocolVersion.V1, 0x0000) + + +class TestDeleteSessionRequest: + """The DeleteObject request that closes an S7CommPlus session.""" + + def test_sync_request_shape(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._protocol_version = ProtocolVersion.V2 + conn._session_id = 0x70000001 + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(side_effect=OSError("no reply")) + + conn._delete_session() + + request = struct.pack( + ">BHHHHIB", + Opcode.REQUEST, + 0x0000, + FunctionCode.DELETE_OBJECT, + 0x0000, + 0, # first sequence number on a fresh connection + 0x70000001, + 0x34, + ) + request += struct.pack(">I", 0) + expected = encode_header(ProtocolVersion.V2, len(request)) + request + expected += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0x0000) + conn._send_s7_data.assert_called_once_with(expected) + + @pytest.mark.asyncio + async def test_async_request_shape(self) -> None: + client = S7CommPlusAsyncClient() + client._protocol_version = ProtocolVersion.V2 + client._session_id = 0x70000001 + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(side_effect=OSError("no reply")) + + await client._delete_session() + + request = struct.pack( + ">BHHHHIB", + Opcode.REQUEST, + 0x0000, + FunctionCode.DELETE_OBJECT, + 0x0000, + 0, + 0x70000001, + 0x34, + ) + request += struct.pack(">I", 0) + expected = encode_header(ProtocolVersion.V2, len(request)) + request + expected += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0x0000) + client._send_cotp_dt.assert_awaited_once_with(expected) + + class TestProtectionLevel: """The effective protection level read that precedes legitimation.""" @@ -393,6 +546,39 @@ async def test_async_read_uses_protocol_request_shape(self) -> None: ) +class TestSessionKeyTransportFlags: + """After SessionKey auth, requests use V3 HMAC framing and transport flags 0x34.""" + + def test_session_key_request_frame_structure(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + conn._session_id = 0x70000001 + conn._session_key = bytes(32) + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(side_effect=OSError("no reply")) + + with pytest.raises(OSError, match="no reply"): + conn.send_request(FunctionCode.GET_VARIABLE, bytes(4)) + + frame = conn._send_s7_data.call_args[0][0] + request = struct.pack( + ">BHHHHIB", + Opcode.REQUEST, + 0x0000, + FunctionCode.GET_VARIABLE, + 0x0000, + 0, # first sequence number on a fresh connection + 0x70000001, + 0x34, # the session key forces 0x34 even for a function code outside FLAGS_34_FUNCTION_CODES + ) + request += bytes(4) + assert frame[:4] == encode_header(ProtocolVersion.V3, len(frame) - 8) + assert frame[4] == 0x20 # hash-length marker before the 32-byte HMAC digest + assert frame[37:-4] == request + assert frame[-4:] == struct.pack(">BBH", 0x72, ProtocolVersion.V3, 0x0000) + + class TestSessionKeySelection: def test_tls_v2_does_not_attempt_session_key_auth(self) -> None: conn = S7CommPlusConnection("127.0.0.1")