Split messages into MessageChunks

Client and server should respect BufferSize values and
should not send chunks bigger than peer's BufferSize.
Also, added base class for SecurityPolicy. Only None policy
(no signing, no encryption) is implemented now.

opcua/binary_client.py

@@ -24,7 +24,7 @@ class UASocketClient(object):
     handle socket connection and send ua messages
     timeout is the timeout used while waiting for an ua answer from server
     """
-    def __init__(self, timeout=1):
+    def __init__(self, timeout=1, security_policy=ua.SecurityPolicy()):
         self.logger = logging.getLogger(__name__ + "Socket")
         self._thread = None
         self._lock = Lock()
@@ -37,6 +37,8 @@ class UASocketClient(object):
         self._request_id = 0
         self._request_handle = 0
         self._callbackmap = {}
+        self._security_policy = security_policy
+        self._max_chunk_size = 65536
 
     def start(self):
         """
@@ -47,28 +49,43 @@ class UASocketClient(object):
         self._thread = Thread(target=self._run)
         self._thread.start()
 
-    def send_request(self, request, callback=None, timeout=1000):
+    def _send_request(self, request, callback=None, timeout=1000, message_type=ua.MessageType.SecureMessage):
         """
-        send request to server.
+        send request to server, lower-level method
         timeout is the timeout written in ua header
+        returns future
         """
         with self._lock:
             request.RequestHeader = self._create_request_header(timeout)
             try:
-                cachedreq = CachedRequest(request.to_binary())
+                binreq = request.to_binary()
             except:
                 # reset reqeust handle if any error
                 # see self._create_request_header
                 self._request_handle -= 1
                 raise
-            hdr = ua.Header(ua.MessageType.SecureMessage, ua.ChunkType.Single, self._security_token.ChannelId)
-            symhdr = self._create_sym_algo_header()
-            seqhdr = self._create_sequence_header()
+            self._request_id += 1
             future = Future()
             if callback:
                 future.add_done_callback(callback)
-            self._callbackmap[seqhdr.RequestId] = future
-            self._write_socket(hdr, symhdr, seqhdr, cachedreq)
+            self._callbackmap[self._request_id] = future
+            for chunk in ua.MessageChunk.message_to_chunks(self._security_policy, binreq, self._max_chunk_size,
+                    message_type=message_type,
+                    channel_id=self._security_token.ChannelId,
+                    request_id=self._request_id,
+                    token_id=self._security_token.TokenId):
+                self._sequence_number += 1
+                chunk.SequenceHeader.SequenceNumber = self._sequence_number
+                self._socket.write(chunk.to_binary())
+        return future
+
+    def send_request(self, request, callback=None, timeout=1000, message_type=ua.MessageType.SecureMessage):
+        """
+        send request to server.
+        timeout is the timeout written in ua header
+        returns response object if no callback is provided
+        """
+        future = self._send_request(request, callback, timeout, message_type)
         if not callback:
             data = future.result(self.timeout)
             self.check_answer(data, " in response to " + request.__class__.__name__)
@@ -179,19 +196,6 @@ class UASocketClient(object):
         hdr.TimeoutHint = timeout
         return hdr
 
-    def _create_sym_algo_header(self):
-        hdr = ua.SymmetricAlgorithmHeader()
-        hdr.TokenId = self._security_token.TokenId
-        return hdr
-
-    def _create_sequence_header(self):
-        hdr = ua.SequenceHeader()
-        self._sequence_number += 1
-        hdr.SequenceNumber = self._sequence_number
-        self._request_id += 1
-        hdr.RequestId = self._request_id
-        return hdr
-
     def connect_socket(self, host, port):
         """
         connect to server socket and start receiving thread
@@ -215,22 +219,15 @@ class UASocketClient(object):
         with self._lock:
             self._callbackmap[0] = future
         self._write_socket(header, hello)
-        return ua.Acknowledge.from_binary(future.result(self.timeout))
+        ack = ua.Acknowledge.from_binary(future.result(self.timeout))
+        self._max_chunk_size = ack.SendBufferSize	# client shouldn't send chunks larger than this
+        return ack
 
     def open_secure_channel(self, params):
         self.logger.info("open_secure_channel")
-        with self._lock:
         request = ua.OpenSecureChannelRequest()
         request.Parameters = params
-            request.RequestHeader = self._create_request_header()
-
-            hdr = ua.Header(ua.MessageType.SecureOpen, ua.ChunkType.Single, self._security_token.ChannelId)
-            asymhdr = ua.AsymmetricAlgorithmHeader()
-            seqhdr = self._create_sequence_header()
-
-            future = Future()
-            self._callbackmap[seqhdr.RequestId] = future
-            self._write_socket(hdr, asymhdr, seqhdr, request)
+        future = self._send_request(request, message_type=ua.MessageType.SecureOpen)
 
         response = ua.OpenSecureChannelResponse.from_binary(future.result(self.timeout))
         response.ResponseHeader.ServiceResult.check()
@@ -240,17 +237,16 @@ class UASocketClient(object):
     def close_secure_channel(self):
         """
         close secure channel. It seems to trigger a shutdown of socket
-        in most servers, so be prepare to reconnect
+        in most servers, so be prepare to reconnect.
+        OPC UA specs Part 6, 7.1.4 say that Server does not send a CloseSecureChannel response and should just close socket
         """
-        self.logger.info("get_endpoint")
-        with self._lock:
+        self.logger.info("close_secure_channel")
         request = ua.CloseSecureChannelRequest()
-            request.RequestHeader = self._create_request_header()
-
-            hdr = ua.Header(ua.MessageType.SecureClose, ua.ChunkType.Single, self._security_token.ChannelId)
-            symhdr = self._create_sym_algo_header()
-            seqhdr = self._create_sequence_header()
-            self._write_socket(hdr, symhdr, seqhdr, request)
+        future = self._send_request(request, message_type=ua.MessageType.SecureClose)
+        with self._lock:
+            # don't expect any more answers
+            future.cancel()
+            self._callbackmap.clear()
 
         # some servers send a response here, most do not ... so we ignore
 

opcua/uaprocessor.py

@@ -28,17 +28,20 @@ class UAProcessor(object):
         self._socketlock = Lock()
         self._datalock = Lock()
         self._publishdata_queue = []
-        self._seq_number = 1
+        self._seq_number = 0
+        self._security_policy = ua.SecurityPolicy()
+        self._max_chunk_size = 65536
 
     def send_response(self, requesthandle, algohdr, seqhdr, response, msgtype=ua.MessageType.SecureMessage):
         with self._socketlock:
             response.ResponseHeader.RequestHandle = requesthandle
-            seqhdr.SequenceNumber = self._seq_number
+            for chunk in ua.MessageChunk.message_to_chunks(self._security_policy, response.to_binary(), self._max_chunk_size, msgtype,
+                    channel_id=self.channel.SecurityToken.ChannelId,
+                    token_id=self.channel.SecurityToken.TokenId,
+                    request_id=seqhdr.RequestId):
                 self._seq_number += 1
-            hdr = ua.Header(msgtype, ua.ChunkType.Single, self.channel.SecurityToken.ChannelId)
-            if isinstance(algohdr, ua.SymmetricAlgorithmHeader):
-                algohdr.TokenId = self.channel.SecurityToken.TokenId
-            self._write_socket(hdr, algohdr, seqhdr, response)
+                chunk.SequenceHeader.SequenceNumber = self._seq_number
+                self.socket.write(chunk.to_binary())
 
     def _write_socket(self, hdr, *args):
         alle = []
@@ -81,6 +84,7 @@ class UAProcessor(object):
             hello = ua.Hello.from_binary(body)
             hdr = ua.Header(ua.MessageType.Acknowledge, ua.ChunkType.Single)
             ack = ua.Acknowledge()
+            self._max_chunk_size = hello.ReceiveBufferSize
             ack.ReceiveBufferSize = hello.ReceiveBufferSize
             ack.SendBufferSize = hello.SendBufferSize
             self._write_socket(hdr, ack)

opcua/uaprotocol_hand.py

 import struct
 import logging
+import hashlib
 
 import opcua.uaprotocol_auto as auto
 import opcua.uatypes as uatypes
@@ -70,7 +71,7 @@ class ChunkType(object):
     Invalid = b"0"  # FIXME check
     Single = b"F"
     Intermediate = b"C"
-    Final = b"A"
+    Abort = b"A"    # when an error occurred and the Message is aborted (body is ErrorMessage)
 
 
 class Header(uatypes.FrozenClass):
@@ -107,6 +108,10 @@ class Header(uatypes.FrozenClass):
             hdr.ChannelId = uatype_UInt32.unpack(data.read(4))[0]
         return hdr
 
+    @staticmethod
+    def max_size():
+        return struct.calcsize("<3scII")
+
     def __str__(self):
         return "Header(type:{}, chunk_type:{}, body_size:{}, channel:{})".format(self.MessageType, self.ChunkType, self.body_size, self.ChannelId)
     __repr__ = __str__
@@ -207,6 +212,10 @@ class SymmetricAlgorithmHeader(uatypes.FrozenClass):
     def to_binary(self):
         return uatype_UInt32.pack(self.TokenId)
 
+    @staticmethod
+    def max_size():
+        return struct.calcsize("<I")
+
     def __str__(self):
         return "{}(TokenId:{} )".format(self.__class__.__name__, self.TokenId)
     __repr__ = __str__
@@ -232,10 +241,198 @@ class SequenceHeader(uatypes.FrozenClass):
         b.append(uatype_UInt32.pack(self.RequestId))
         return b"".join(b)
 
+    @staticmethod
+    def max_size():
+        return struct.calcsize("<II")
+
     def __str__(self):
         return "{}(SequenceNumber:{}, RequestId:{} )".format(self.__class__.__name__, self.SequenceNumber, self.RequestId)
     __repr__ = __str__
 
+
+class CryptographyNone:
+    """
+    Base class for symmetric/asymmetric cryprography
+    """
+    def __init__(self, mode=auto.MessageSecurityMode.None_):
+        pass
+
+    def plain_block_size(self):
+        """
+        Size of plain text block for block cipher.
+        """
+        return 1
+
+    def encrypted_block_size(self):
+        """
+        Size of encrypted text block for block cipher.
+        """
+        return 1
+
+    def padding(self, size):
+        """
+        Create padding for a block of given size.
+        plain_size = size + len(padding) + signature_size()
+        plain_size = N * plain_block_size()
+        """
+        return b''
+
+    def min_padding_size(self):
+        return 0
+
+    def signature_size(self):
+        return 0
+
+    def signature(self, data):
+        return b''
+
+    def encrypt(self, data):
+        return data
+
+    def decrypt(self, data):
+        return data
+
+    def vsignature_size(self):
+        return 0
+
+    def verify(self, data, signature):
+        """
+        Verify signature and raise exception if signature is invalid
+        """
+        pass
+
+    def remove_padding(self, data):
+        return data
+
+
+class SecurityPolicy:
+    """
+    Base class for security policy
+    """
+    def __init__(self):
+        self.asymmetric_cryptography = CryptographyNone()
+        self.symmetric_cryptography = CryptographyNone()
+        self.Mode = auto.MessageSecurityMode.None_
+        self.URI = "http://opcfoundation.org/UA/SecurityPolicy#None"
+        self.server_certificate = b""
+        self.client_certificate = b""
+
+    def symmetric_key_size(self):
+        return 0
+
+    def make_symmetric_key(self, a, b):
+        pass
+
+
+class MessageChunk(uatypes.FrozenClass):
+    """
+    Message Chunk, as described in OPC UA specs Part 6, 6.7.2.
+    """
+    def __init__(self, security_policy, body=b'', msg_type=MessageType.SecureMessage, chunk_type=ChunkType.Single):
+        self.MessageHeader = Header(msg_type, chunk_type)
+        if msg_type in (MessageType.SecureMessage, MessageType.SecureClose):
+            self.SecurityHeader = SymmetricAlgorithmHeader()
+        elif msg_type == MessageType.SecureOpen:
+            self.SecurityHeader = AsymmetricAlgorithmHeader()
+        else:
+            raise Exception("Unsupported message type: {}".format(msg_type))
+        self.SequenceHeader = SequenceHeader()
+        self.Body = body
+        self._security_policy = security_policy
+
+    @staticmethod
+    def from_binary(security_policy, data):
+        h = Header.from_string(data)
+        return MessageChunk.from_header_and_body(security_policy, h, data)
+
+    @staticmethod
+    def from_header_and_body(security_policy, header, data):
+        if header.MessageType in (MessageType.SecureMessage, MessageType.SecureClose):
+            security_header = SymmetricAlgorithmHeader.from_binary(data)
+            crypto = security_policy.symmetric_cryptography
+        elif header.MessageType == MessageType.SecureOpen:
+            security_header = AsymmetricAlgorithmHeader.from_binary(data)
+            crypto = security_policy.asymmetric_cryptography
+        else:
+            raise Exception("Unsupported message type: {}".format(header.MessageType))
+        obj = MessageChunk(crypto)
+        obj.MessageHeader = header
+        obj.SecurityHeader = security_header
+        decrypted = crypto.decrypt(data.read(len(data)))
+        signature_size = crypto.vsignature_size()
+        if signature_size > 0:
+            signature = decrypted[-signature_size:]
+            decrypted = decrypted[:-signature_size]
+            crypto.verify(obj.MessageHeader.to_binary() + obj.SecurityHeader.to_binary() + decrypted, signature)
+        data = utils.Buffer(crypto.remove_padding(decrypted))
+        obj.SequenceHeader = SequenceHeader.from_binary(data)
+        obj.Body = data.read(len(data))
+        return obj
+
+    def encrypted_size(self, plain_size):
+        size = plain_size + self._security_policy.signature_size()
+        pbs = self._security_policy.plain_block_size()
+        assert(size % pbs == 0)
+        return size // pbs * self._security_policy.encrypted_block_size()
+
+    def to_binary(self):
+        security = self.SecurityHeader.to_binary()
+        encrypted_part = self.SequenceHeader.to_binary() + self.Body
+        encrypted_part += self._security_policy.padding(len(encrypted_part))
+        self.MessageHeader.body_size = len(security) + self.encrypted_size(len(encrypted_part))
+        header = self.MessageHeader.to_binary()
+        encrypted_part += self._security_policy.signature(header + security + encrypted_part)
+        return header + security + self._security_policy.encrypt(encrypted_part)
+
+    @staticmethod
+    def max_body_size(crypto, max_chunk_size):
+        max_encrypted_size = max_chunk_size - Header.max_size() - SymmetricAlgorithmHeader.max_size()
+        max_plain_size = (max_encrypted_size // crypto.encrypted_block_size()) * crypto.plain_block_size()
+        return max_plain_size - SequenceHeader.max_size() - crypto.signature_size() - crypto.min_padding_size()
+
+    @staticmethod
+    def message_to_chunks(security_policy, body, max_chunk_size, message_type=MessageType.SecureMessage, channel_id=1, request_id=1, token_id=1):
+        """
+        Pack message body (as binary string) into one or more chunks.
+        Size of each chunk will not exceed max_chunk_size.
+        Returns a list of MessageChunks. SequenceNumber is not initialized here,
+        it must be set by Secure Channel driver.
+        """
+        if message_type == MessageType.SecureOpen:
+            # SecureOpen message must be in a single chunk (specs, Part 6, 6.7.2)
+            chunk = MessageChunk(security_policy.asymmetric_cryptography, body, message_type, ChunkType.Single)
+            chunk.SecurityHeader.SecurityPolicyURI = security_policy.URI
+            if security_policy.client_certificate:
+                chunk.SecurityHeader.SenderCertificate = security_policy.client_certificate
+            if security_policy.server_certificate:
+                chunk.SecurityHeader.ReceiverCertificateThumbPrint = hashlib.sha1(security_policy.server_certificate).digest()
+            chunk.MessageHeader.ChannelId = channel_id
+            chunk.SequenceHeader.RequestId = request_id
+            return [chunk]
+
+        crypto = security_policy.symmetric_cryptography
+        max_size = MessageChunk.max_body_size(crypto, max_chunk_size)
+
+        chunks = []
+        for i in range(0, len(body), max_size):
+            part = body[i:i+max_size]
+            if i+max_size >= len(body):
+                chunk_type = ChunkType.Single
+            else:
+                chunk_type = ChunkType.Intermediate
+            chunk = MessageChunk(crypto, part, message_type, chunk_type)
+            chunk.SecurityHeader.TokenId = token_id
+            chunk.MessageHeader.ChannelId = channel_id
+            chunk.SequenceHeader.RequestId = request_id
+            chunks.append(chunk)
+        return chunks
+
+    def __str__(self):
+        return "{}({}, {}, {}, {} bytes)".format(self.__class__.__name__,
+                self.MessageHeader, self.SequenceHeader, self.SecurityHeader, len(self.Body))
+    __repr__ = __str__
+
+
 # FIXES for missing switchfield in NodeAttributes classes
 ana = auto.NodeAttributesMask
 

tests/tests.py

@@ -266,6 +266,29 @@ class Unit(unittest.TestCase):
         t4 = ua.LocalizedText.from_binary(ua.utils.Buffer(t1.to_binary()))
         self.assertEqual(t1, t4)
 
+    def test_message_chunk(self):
+        pol = ua.SecurityPolicy()
+        chunks = ua.MessageChunk.message_to_chunks(pol, b'123', 65536)
+        self.assertEqual(len(chunks), 1)
+        seq = 0
+        for chunk in chunks:
+            seq += 1
+            chunk.SequenceHeader.SequenceNumber = seq
+        chunk2 = ua.MessageChunk.from_binary(pol, ua.utils.Buffer(chunks[0].to_binary()))
+        self.assertEqual(chunks[0].to_binary(), chunk2.to_binary())
+
+        # for policy None, MessageChunk overhead is 12+4+8 = 24 bytes
+        # Let's pack 11 bytes into 28-byte chunks. The message must be split as 4+4+3
+        chunks = ua.MessageChunk.message_to_chunks(pol, b'12345678901', 28)
+        self.assertEqual(len(chunks), 3)
+        self.assertEqual(chunks[0].Body, b'1234')
+        self.assertEqual(chunks[1].Body, b'5678')
+        self.assertEqual(chunks[2].Body, b'901')
+        for chunk in chunks:
+            seq += 1
+            chunk.SequenceHeader.SequenceNumber = seq
+            self.assertTrue(len(chunk.to_binary()) <= 28)
+
 
 class CommonTests(object):