Message Chunk, as described in OPC UA specs Part 6, 6.7.2.
| 10 | |
| 11 | |
| 12 | class MessageChunk(ua.FrozenClass): |
| 13 | """ |
| 14 | Message Chunk, as described in OPC UA specs Part 6, 6.7.2. |
| 15 | """ |
| 16 | def __init__(self, security_policy, body=b'', msg_type=ua.MessageType.SecureMessage, chunk_type=ua.ChunkType.Single): |
| 17 | self.MessageHeader = ua.Header(msg_type, chunk_type) |
| 18 | if msg_type in (ua.MessageType.SecureMessage, ua.MessageType.SecureClose): |
| 19 | self.SecurityHeader = ua.SymmetricAlgorithmHeader() |
| 20 | elif msg_type == ua.MessageType.SecureOpen: |
| 21 | self.SecurityHeader = ua.AsymmetricAlgorithmHeader() |
| 22 | else: |
| 23 | raise ua.UaError("Unsupported message type: {0}".format(msg_type)) |
| 24 | self.SequenceHeader = ua.SequenceHeader() |
| 25 | self.Body = body |
| 26 | self.security_policy = security_policy |
| 27 | |
| 28 | @staticmethod |
| 29 | def from_binary(security_policy, data): |
| 30 | h = header_from_binary(data) |
| 31 | return MessageChunk.from_header_and_body(security_policy, h, data) |
| 32 | |
| 33 | @staticmethod |
| 34 | def from_header_and_body(security_policy, header, buf): |
| 35 | assert len(buf) >= header.body_size, 'Full body expected here' |
| 36 | data = buf.copy(header.body_size) |
| 37 | buf.skip(header.body_size) |
| 38 | if header.MessageType in (ua.MessageType.SecureMessage, ua.MessageType.SecureClose): |
| 39 | security_header = struct_from_binary(ua.SymmetricAlgorithmHeader, data) |
| 40 | crypto = security_policy.symmetric_cryptography |
| 41 | elif header.MessageType == ua.MessageType.SecureOpen: |
| 42 | security_header = struct_from_binary(ua.AsymmetricAlgorithmHeader, data) |
| 43 | crypto = security_policy.asymmetric_cryptography |
| 44 | else: |
| 45 | raise ua.UaError("Unsupported message type: {0}".format(header.MessageType)) |
| 46 | obj = MessageChunk(crypto) |
| 47 | obj.MessageHeader = header |
| 48 | obj.SecurityHeader = security_header |
| 49 | decrypted = crypto.decrypt(data.read(len(data))) |
| 50 | signature_size = crypto.vsignature_size() |
| 51 | if signature_size > 0: |
| 52 | signature = decrypted[-signature_size:] |
| 53 | decrypted = decrypted[:-signature_size] |
| 54 | crypto.verify(header_to_binary(obj.MessageHeader) + struct_to_binary(obj.SecurityHeader) + decrypted, signature) |
| 55 | data = ua.utils.Buffer(crypto.remove_padding(decrypted)) |
| 56 | obj.SequenceHeader = struct_from_binary(ua.SequenceHeader, data) |
| 57 | obj.Body = data.read(len(data)) |
| 58 | return obj |
| 59 | |
| 60 | def encrypted_size(self, plain_size): |
| 61 | size = plain_size + self.security_policy.signature_size() |
| 62 | pbs = self.security_policy.plain_block_size() |
| 63 | if size % pbs != 0: |
| 64 | print("ENC", plain_size, size, pbs) |
| 65 | raise ua.UaError("Encryption error") |
| 66 | return size // pbs * self.security_policy.encrypted_block_size() |
| 67 | |
| 68 | def to_binary(self): |
| 69 | security = struct_to_binary(self.SecurityHeader) |
no outgoing calls
no test coverage detected