MCPcopy Create free account
hub / github.com/FreeOpcUa/python-opcua / message_to_binary

Method message_to_binary

opcua/common/connection.py:231–244  ·  view source on GitHub ↗

Convert OPC UA secure message to binary. The only supported types are SecureOpen, SecureMessage, SecureClose if message_type is SecureMessage, the AlgoritmHeader should be passed as arg

(self, message, message_type=ua.MessageType.SecureMessage, request_id=0)

Source from the content-addressed store, hash-verified

229 self.security_policy.make_local_symmetric_key(self.remote_nonce, self.local_nonce)
230
231 def message_to_binary(self, message, message_type=ua.MessageType.SecureMessage, request_id=0):
232 """
233 Convert OPC UA secure message to binary.
234 The only supported types are SecureOpen, SecureMessage, SecureClose
235 if message_type is SecureMessage, the AlgoritmHeader should be passed as arg
236 """
237 chunks = MessageChunk.message_to_chunks(self.security_policy, message, self._max_chunk_size, message_type=message_type, channel_id=self.security_token.ChannelId, request_id=request_id, token_id=self.security_token.TokenId)
238 for chunk in chunks:
239 self._sequence_number += 1
240 if self._sequence_number >= (1 << 32):
241 logger.debug("Wrapping sequence number: %d -> 1", self._sequence_number)
242 self._sequence_number = 1
243 chunk.SequenceHeader.SequenceNumber = self._sequence_number
244 return b"".join([chunk.to_binary() for chunk in chunks])
245
246 def _check_sym_header(self, securityHeader):
247 """

Callers 2

_send_requestMethod · 0.80
send_responseMethod · 0.80

Calls 2

message_to_chunksMethod · 0.80
to_binaryMethod · 0.45

Tested by

no test coverage detected