opcua.ua.ErrorMessage.to_binary() - Code Metrics - Inspection of "Merge pull request #101 from huazh/subrace" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 5a9df5...39e3a6 )

by Olivier

created 2016-01-04 13:47 UTC

opcua.ua.ErrorMessage.to_binary() A

↳ Parent: opcua.ua.ErrorMessage

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.512

Metric	Value
cc	1
dl	0
loc	5
ccs	1
cts	5
cp	0.2
crap	1.512
rs	9.4286

import struct
import logging
import hashlib
from enum import IntEnum

from opcua.ua import uaprotocol_auto as auto
from opcua.ua import uatypes 
from opcua.ua.uatypes import uatype_UInt32
from opcua.common import utils 

logger = logging.getLogger('opcua.uaprotocol')

OPC_TCP_SCHEME = 'opc.tcp'


class ValueRank(IntEnum):
    """
    Defines dimensions of a variable. 
    This enum does not support all cases since ValueRank support any n>0
    but since it is an IntEnum it can be replace by a normal int
    """
    ScalarOrOneDimension = -3
    Any = -2
    Scalar = -1
    OneOrMoreDimensions = 0
    OneDimension = 1
    # the next names are not in spec but so common we express them here
    TwoDimensions = 2
    ThreeDimensions = 3
    FourDimensions = 3


class AccessLevelMask(object):
    """
    Used by AccessLevel and UserAccessLevel
    """
    CurrentRead = 0
    CurrentWrite = 1
    HistoryRead = 2
    HistoryWrite = 3
    SemanticChange = 4


class Hello(uatypes.FrozenClass):

    def __init__(self):
        self.ProtocolVersion = 0
        self.ReceiveBufferSize = 65536
        self.SendBufferSize = 65536
        self.MaxMessageSize = 0
        self.MaxChunkCount = 0
        self.EndpointUrl = ""
        self._freeze = True

    def to_binary(self):
        b = []
        b.append(uatype_UInt32.pack(self.ProtocolVersion))
        b.append(uatype_UInt32.pack(self.ReceiveBufferSize))
        b.append(uatype_UInt32.pack(self.SendBufferSize))
        b.append(uatype_UInt32.pack(self.MaxMessageSize))
        b.append(uatype_UInt32.pack(self.MaxChunkCount))
        b.append(uatypes.pack_string(self.EndpointUrl))
        return b"".join(b)

    @staticmethod
    def from_binary(data):
        hello = Hello()
        hello.ProtocolVersion = uatype_UInt32.unpack(data.read(4))[0]
        hello.ReceiveBufferSize = uatype_UInt32.unpack(data.read(4))[0]
        hello.SendBufferSize = uatype_UInt32.unpack(data.read(4))[0]
        hello.MaxMessageSize = uatype_UInt32.unpack(data.read(4))[0]
        hello.MaxChunkCount = uatype_UInt32.unpack(data.read(4))[0]
        hello.EndpointUrl = uatypes.unpack_string(data)
        return hello


class MessageType(object):
    Invalid = b"INV"  # FIXME: check value
    Hello = b"HEL"
    Acknowledge = b"ACK"
    Error = b"ERR"
    SecureOpen = b"OPN"
    SecureClose = b"CLO"
    SecureMessage = b"MSG"


class ChunkType(object):
    Invalid = b"0"  # FIXME check
    Single = b"F"
    Intermediate = b"C"
    Abort = b"A"    # when an error occurred and the Message is aborted (body is ErrorMessage)


class Header(uatypes.FrozenClass):

    def __init__(self, msgType=None, chunkType=None, channelid=0):
        self.MessageType = msgType
        self.ChunkType = chunkType
        self.ChannelId = channelid
        self.body_size = 0
        self.packet_size = 0
        self._freeze = True

    def add_size(self, size):
        self.body_size += size

    def to_binary(self):
        b = []
        b.append(struct.pack("<3ss", self.MessageType, self.ChunkType))
        size = self.body_size + 8
        if self.MessageType in (MessageType.SecureOpen, MessageType.SecureClose, MessageType.SecureMessage):
            size += 4
        b.append(uatype_UInt32.pack(size))
        if self.MessageType in (MessageType.SecureOpen, MessageType.SecureClose, MessageType.SecureMessage):
            b.append(uatype_UInt32.pack(self.ChannelId))
        return b"".join(b)

    @staticmethod
    def from_string(data):
        hdr = Header()
        hdr.MessageType, hdr.ChunkType, hdr.packet_size = struct.unpack("<3scI", data.read(8))
        hdr.body_size = hdr.packet_size - 8
        if hdr.MessageType in (MessageType.SecureOpen, MessageType.SecureClose, MessageType.SecureMessage):
            hdr.body_size -= 4
            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__


class ErrorMessage(uatypes.FrozenClass):

    def __init__(self):
        self.Error = uatypes.StatusCode()
        self.Reason = ""
        self._freeze = True

    def to_binary(self):
        b = []
        b.append(self.Error.to_binary())
        b.append(uatypes.pack_string(self.Reason))
        return b"".join(b)

    @staticmethod
    def from_binary(data):
        ack = ErrorMessage()
        ack.Error = uatypes.StatusCode.from_binary(data)
        ack.Reason = uatypes.unpack_string(data)
        return ack

    def __str__(self):
        return "MessageAbort(error:{}, reason:{})".format(self.Error, self.Reason)
    __repr__ = __str__


class Acknowledge(uatypes.FrozenClass):

    def __init__(self):
        self.ProtocolVersion = 0
        self.ReceiveBufferSize = 65536
        self.SendBufferSize = 65536
        self.MaxMessageSize = 0  # No limits
        self.MaxChunkCount = 0  # No limits
        self._freeze = True

    def to_binary(self):
        return struct.pack(
            "<5I",
            self.ProtocolVersion,
            self.ReceiveBufferSize,
            self.SendBufferSize,
            self.MaxMessageSize,
            self.MaxChunkCount)

    @staticmethod
    def from_binary(data):
        ack = Acknowledge()
        ack.ProtocolVersion, ack.ReceiveBufferSize, ack.SendBufferSize, ack.MaxMessageSize, ack.MaxChunkCount \
            = struct.unpack("<5I", data.read(20))
        return ack


class AsymmetricAlgorithmHeader(uatypes.FrozenClass):

    def __init__(self):
        self.SecurityPolicyURI = "http://opcfoundation.org/UA/SecurityPolicy#None"
        self.SenderCertificate = b""
        self.ReceiverCertificateThumbPrint = b""
        self._freeze = True

    def to_binary(self):
        b = []
        b.append(uatypes.pack_string(self.SecurityPolicyURI))
        b.append(uatypes.pack_string(self.SenderCertificate))
        b.append(uatypes.pack_string(self.ReceiverCertificateThumbPrint))
        return b"".join(b)

    @staticmethod
    def from_binary(data):
        hdr = AsymmetricAlgorithmHeader()
        hdr.SecurityPolicyURI = uatypes.unpack_bytes(data)
        hdr.SenderCertificate = uatypes.unpack_bytes(data)
        hdr.ReceiverCertificateThumbPrint = uatypes.unpack_bytes(data)
        return hdr

    def __str__(self):
        return "{}(SecurityPolicy:{}, certificatesize:{}, receiverCertificatesize:{} )".format(self.__class__.__name__, self.SecurityPolicyURI, len(self.SenderCertificate), len(self.ReceiverCertificateThumbPrint))
    __repr__ = __str__


class SymmetricAlgorithmHeader(uatypes.FrozenClass):

    def __init__(self):
        self.TokenId = 0
        self._freeze = True

    @staticmethod
    def from_binary(data):
        obj = SymmetricAlgorithmHeader()
        obj.TokenId = uatype_UInt32.unpack(data.read(4))[0]
        return obj

    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__


class SequenceHeader(uatypes.FrozenClass):

    def __init__(self):
        self.SequenceNumber = None
        self.RequestId = None
        self._freeze = True

    @staticmethod
    def from_binary(data):
        obj = SequenceHeader()
        obj.SequenceNumber = uatype_UInt32.unpack(data.read(4))[0]
        obj.RequestId = uatype_UInt32.unpack(data.read(4))[0]
        return obj

    def to_binary(self):
        b = []
        b.append(uatype_UInt32.pack(self.SequenceNumber))
        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):
        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
    """
    URI = "http://opcfoundation.org/UA/SecurityPolicy#None"
    signature_key_size = 0
    symmetric_key_size = 0

    def __init__(self):
        self.asymmetric_cryptography = CryptographyNone()
        self.symmetric_cryptography = CryptographyNone()
        self.Mode = auto.MessageSecurityMode.None_
        self.server_certificate = b""
        self.client_certificate = b""

    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__


def tcp_message_from_header_and_body(security_policy, header, body):
    """
    Convert binary stream to OPC UA TCP message (see OPC UA specs Part 6, 7.1)
    The only supported message types are Hello, Acknowledge and Error
    """
    if header.MessageType in (MessageType.SecureOpen, MessageType.SecureClose, MessageType.SecureMessage):
        return MessageChunk.from_header_and_body(security_policy, header, body)
    elif header.MessageType == MessageType.Hello:
        return Hello.from_binary(body)
    elif header.MessageType == MessageType.Acknowledge:
        return Acknowledge.from_binary(body)
    elif header.MessageType == MessageType.Error:
        return ErrorMessage.from_binary(body)
    else:
        raise Exception("Unsupported message type {}".format(header.MessageType))


def tcp_message_from_socket(security_policy, socket):
    logger.debug("Waiting for header")
    header = Header.from_string(socket)
    logger.info("received header: %s", header)
    body = socket.read(header.body_size)
    if len(body) != header.body_size:
        raise Exception("{} bytes expected, {} available".format(header.body_size, len(body)))
    return tcp_message_from_header_and_body(security_policy, header, utils.Buffer(body))


# FIXES for missing switchfield in NodeAttributes classes
ana = auto.NodeAttributesMask


class ObjectAttributes(auto.ObjectAttributes):

    def __init__(self):
        auto.ObjectAttributes.__init__(self)
        self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.EventNotifier


class ObjectTypeAttributes(auto.ObjectTypeAttributes):

    def __init__(self):
        auto.ObjectTypeAttributes.__init__(self)
        self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.IsAbstract


class VariableAttributes(auto.VariableAttributes):

    def __init__(self):
        auto.VariableAttributes.__init__(self)
        self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.Value | ana.DataType | ana.ValueRank | ana.ArrayDimensions | ana.AccessLevel | ana.UserAccessLevel | ana.MinimumSamplingInterval | ana.Historizing
        self.Historizing = False


class VariableTypeAttributes(auto.VariableTypeAttributes):

    def __init__(self):
        auto.VariableTypeAttributes.__init__(self)
        self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.Value | ana.DataType | ana.ValueRank | ana.ArrayDimensions | ana.IsAbstract


class MethodAttributes(auto.MethodAttributes):

    def __init__(self):
        auto.MethodAttributes.__init__(self)
        self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.Executable | ana.UserExecutable


class ReferenceTypeAttributes(auto.ReferenceTypeAttributes):

    def __init__(self):
        auto.ReferenceTypeAttributes.__init__(self)
        self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.IsAbstract | ana.Symmetric | ana.InverseName


class DataTypeAttributes(auto.DataTypeAttributes):

    def __init__(self):
        auto.DataTypeAttributes.__init__(self)
        self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.IsAbstract


class ViewAttributes(auto.ViewAttributes):

    def __init__(self):
        auto.ViewAttributes.__init__(self)
        self.SpecifiedAttributes = ana.DisplayName | ana.Description | ana.WriteMask | ana.UserWriteMask | ana.ContainsNoLoops | ana.EventNotifier


class Argument(auto.Argument):

    def __init__(self):
        auto.Argument.__init__(self)
        self.ValueRank = -2


#AttributeIdsInv = {v: k for k, v in AttributeIds.__dict__.items()}


1	1	import struct
2	1	import logging
3	1	import hashlib
4	1	from enum import IntEnum
5
6	1	from opcua.ua import uaprotocol_auto as auto
7	1	from opcua.ua import uatypes
8	1	from opcua.ua.uatypes import uatype_UInt32
9	1	from opcua.common import utils
10
11	1	logger = logging.getLogger('opcua.uaprotocol')
12
13	1	OPC_TCP_SCHEME = 'opc.tcp'
14
15
16	1	class ValueRank(IntEnum):
17		"""
18		Defines dimensions of a variable.
19		This enum does not support all cases since ValueRank support any n>0
20		but since it is an IntEnum it can be replace by a normal int
21		"""
22	1	ScalarOrOneDimension = -3
23	1	Any = -2
24	1	Scalar = -1
25	1	OneOrMoreDimensions = 0
26	1	OneDimension = 1
27		# the next names are not in spec but so common we express them here
28	1	TwoDimensions = 2
29	1	ThreeDimensions = 3
30	1	FourDimensions = 3
31
32
33	1	class AccessLevelMask(object):
34		"""
35		Used by AccessLevel and UserAccessLevel
36		"""
37	1	CurrentRead = 0
38	1	CurrentWrite = 1
39	1	HistoryRead = 2
40	1	HistoryWrite = 3
41	1	SemanticChange = 4
42
43
44	1	class Hello(uatypes.FrozenClass):
45
46	1	def __init__(self):
47	1	self.ProtocolVersion = 0
48	1	self.ReceiveBufferSize = 65536
49	1	self.SendBufferSize = 65536
50	1	self.MaxMessageSize = 0
51	1	self.MaxChunkCount = 0
52	1	self.EndpointUrl = ""
53	1	self._freeze = True
54
55	1	def to_binary(self):
56	1	b = []
57	1	b.append(uatype_UInt32.pack(self.ProtocolVersion))
58	1	b.append(uatype_UInt32.pack(self.ReceiveBufferSize))
59	1	b.append(uatype_UInt32.pack(self.SendBufferSize))
60	1	b.append(uatype_UInt32.pack(self.MaxMessageSize))
61	1	b.append(uatype_UInt32.pack(self.MaxChunkCount))
62	1	b.append(uatypes.pack_string(self.EndpointUrl))
63	1	return b"".join(b)
64
65	1	@staticmethod
66		def from_binary(data):
67	1	hello = Hello()
68	1	hello.ProtocolVersion = uatype_UInt32.unpack(data.read(4))[0]
69	1	hello.ReceiveBufferSize = uatype_UInt32.unpack(data.read(4))[0]
70	1	hello.SendBufferSize = uatype_UInt32.unpack(data.read(4))[0]
71	1	hello.MaxMessageSize = uatype_UInt32.unpack(data.read(4))[0]
72	1	hello.MaxChunkCount = uatype_UInt32.unpack(data.read(4))[0]
73	1	hello.EndpointUrl = uatypes.unpack_string(data)
74	1	return hello
75
76
77	1	class MessageType(object):
78	1	Invalid = b"INV" # FIXME: check value
79	1	Hello = b"HEL"
80	1	Acknowledge = b"ACK"
81	1	Error = b"ERR"
82	1	SecureOpen = b"OPN"
83	1	SecureClose = b"CLO"
84	1	SecureMessage = b"MSG"
85
86
87	1	class ChunkType(object):
88	1	Invalid = b"0" # FIXME check
89	1	Single = b"F"
90	1	Intermediate = b"C"
91	1	Abort = b"A" # when an error occurred and the Message is aborted (body is ErrorMessage)
92
93
94	1	class Header(uatypes.FrozenClass):
95
96	1	def __init__(self, msgType=None, chunkType=None, channelid=0):
97	1	self.MessageType = msgType
98	1	self.ChunkType = chunkType
99	1	self.ChannelId = channelid
100	1	self.body_size = 0
101	1	self.packet_size = 0
102	1	self._freeze = True
103
104	1	def add_size(self, size):
105	1	self.body_size += size
106
107	1	def to_binary(self):
108	1	b = []
109	1	b.append(struct.pack("<3ss", self.MessageType, self.ChunkType))
110	1	size = self.body_size + 8
111	1	if self.MessageType in (MessageType.SecureOpen, MessageType.SecureClose, MessageType.SecureMessage):
112	1	size += 4
113	1	b.append(uatype_UInt32.pack(size))
114	1	if self.MessageType in (MessageType.SecureOpen, MessageType.SecureClose, MessageType.SecureMessage):
115	1	b.append(uatype_UInt32.pack(self.ChannelId))
116	1	return b"".join(b)
117
118	1	@staticmethod
119		def from_string(data):
120	1	hdr = Header()
121	1	hdr.MessageType, hdr.ChunkType, hdr.packet_size = struct.unpack("<3scI", data.read(8))
122	1	hdr.body_size = hdr.packet_size - 8
123	1	if hdr.MessageType in (MessageType.SecureOpen, MessageType.SecureClose, MessageType.SecureMessage):
124	1	hdr.body_size -= 4
125	1	hdr.ChannelId = uatype_UInt32.unpack(data.read(4))[0]
126	1	return hdr
127
128	1	@staticmethod
129		def max_size():
130	1	return struct.calcsize("<3scII")
131
132	1	def __str__(self):
133		return "Header(type:{}, chunk_type:{}, body_size:{}, channel:{})".format(self.MessageType, self.ChunkType, self.body_size, self.ChannelId)
134	1	__repr__ = __str__
135
136
137	1	class ErrorMessage(uatypes.FrozenClass):
138
139	1	def __init__(self):
140		self.Error = uatypes.StatusCode()
141		self.Reason = ""
142		self._freeze = True
143
144	1	def to_binary(self):
145		b = []
146		b.append(self.Error.to_binary())
147		b.append(uatypes.pack_string(self.Reason))
148		return b"".join(b)
149
150	1	@staticmethod
151		def from_binary(data):
152		ack = ErrorMessage()
153		ack.Error = uatypes.StatusCode.from_binary(data)
154		ack.Reason = uatypes.unpack_string(data)
155		return ack
156
157	1	def __str__(self):
158		return "MessageAbort(error:{}, reason:{})".format(self.Error, self.Reason)
159	1	__repr__ = __str__
160
161
162	1	class Acknowledge(uatypes.FrozenClass):
163
164	1	def __init__(self):
165	1	self.ProtocolVersion = 0
166	1	self.ReceiveBufferSize = 65536
167	1	self.SendBufferSize = 65536
168	1	self.MaxMessageSize = 0 # No limits
169	1	self.MaxChunkCount = 0 # No limits
170	1	self._freeze = True
171
172	1	def to_binary(self):
173	1	return struct.pack(
174		"<5I",
175		self.ProtocolVersion,
176		self.ReceiveBufferSize,
177		self.SendBufferSize,
178		self.MaxMessageSize,
179		self.MaxChunkCount)
180
181	1	@staticmethod
182		def from_binary(data):
183	1	ack = Acknowledge()
184	1	ack.ProtocolVersion, ack.ReceiveBufferSize, ack.SendBufferSize, ack.MaxMessageSize, ack.MaxChunkCount \
185		= struct.unpack("<5I", data.read(20))
186	1	return ack
187
188
189	1	class AsymmetricAlgorithmHeader(uatypes.FrozenClass):
190
191	1	def __init__(self):
192	1	self.SecurityPolicyURI = "http://opcfoundation.org/UA/SecurityPolicy#None"
193	1	self.SenderCertificate = b""
194	1	self.ReceiverCertificateThumbPrint = b""
195	1	self._freeze = True
196
197	1	def to_binary(self):
198	1	b = []
199	1	b.append(uatypes.pack_string(self.SecurityPolicyURI))
200	1	b.append(uatypes.pack_string(self.SenderCertificate))
201	1	b.append(uatypes.pack_string(self.ReceiverCertificateThumbPrint))
202	1	return b"".join(b)
203
204	1	@staticmethod
205		def from_binary(data):
206	1	hdr = AsymmetricAlgorithmHeader()
207	1	hdr.SecurityPolicyURI = uatypes.unpack_bytes(data)
208	1	hdr.SenderCertificate = uatypes.unpack_bytes(data)
209	1	hdr.ReceiverCertificateThumbPrint = uatypes.unpack_bytes(data)
210	1	return hdr
211
212	1	def __str__(self):
213		return "{}(SecurityPolicy:{}, certificatesize:{}, receiverCertificatesize:{} )".format(self.__class__.__name__, self.SecurityPolicyURI, len(self.SenderCertificate), len(self.ReceiverCertificateThumbPrint))
214	1	__repr__ = __str__
215
216
217	1	class SymmetricAlgorithmHeader(uatypes.FrozenClass):
218
219	1	def __init__(self):
220	1	self.TokenId = 0
221	1	self._freeze = True
222
223	1	@staticmethod
224		def from_binary(data):
225	1	obj = SymmetricAlgorithmHeader()
226	1	obj.TokenId = uatype_UInt32.unpack(data.read(4))[0]
227	1	return obj
228
229	1	def to_binary(self):
230	1	return uatype_UInt32.pack(self.TokenId)
231
232	1	@staticmethod
233		def max_size():
234	1	return struct.calcsize("<I")
235
236	1	def __str__(self):
237		return "{}(TokenId:{} )".format(self.__class__.__name__, self.TokenId)
238	1	__repr__ = __str__
239
240
241	1	class SequenceHeader(uatypes.FrozenClass):
242
243	1	def __init__(self):
244	1	self.SequenceNumber = None
245	1	self.RequestId = None
246	1	self._freeze = True
247
248	1	@staticmethod
249		def from_binary(data):
250	1	obj = SequenceHeader()
251	1	obj.SequenceNumber = uatype_UInt32.unpack(data.read(4))[0]
252	1	obj.RequestId = uatype_UInt32.unpack(data.read(4))[0]
253	1	return obj
254
255	1	def to_binary(self):
256	1	b = []
257	1	b.append(uatype_UInt32.pack(self.SequenceNumber))
258	1	b.append(uatype_UInt32.pack(self.RequestId))
259	1	return b"".join(b)
260
261	1	@staticmethod
262		def max_size():
263	1	return struct.calcsize("<II")
264
265	1	def __str__(self):
266		return "{}(SequenceNumber:{}, RequestId:{} )".format(self.__class__.__name__, self.SequenceNumber, self.RequestId)
267	1	__repr__ = __str__
268
269
270	1	class CryptographyNone:
271		"""
272		Base class for symmetric/asymmetric cryprography
273		"""
274	1	def __init__(self):
275	1	pass
276
277	1	def plain_block_size(self):
278		"""
279		Size of plain text block for block cipher.
280		"""
281	1	return 1
282
283	1	def encrypted_block_size(self):
284		"""
285		Size of encrypted text block for block cipher.
286		"""
287	1	return 1
288
289	1	def padding(self, size):
290		"""
291		Create padding for a block of given size.
292		plain_size = size + len(padding) + signature_size()
293		plain_size = N * plain_block_size()
294		"""
295	1	return b''
296
297	1	def min_padding_size(self):
298	1	return 0
299
300	1	def signature_size(self):
301	1	return 0
302
303	1	def signature(self, data):
304	1	return b''
305
306	1	def encrypt(self, data):
307	1	return data
308
309	1	def decrypt(self, data):
310	1	return data
311
312	1	def vsignature_size(self):
313	1	return 0
314
315	1	def verify(self, data, signature):
316		"""
317		Verify signature and raise exception if signature is invalid
318		"""
319	1	pass
320
321	1	def remove_padding(self, data):
322	1	return data
323
324
325	1	class SecurityPolicy:
326		"""
327		Base class for security policy
328		"""
329	1	URI = "http://opcfoundation.org/UA/SecurityPolicy#None"
330	1	signature_key_size = 0
331	1	symmetric_key_size = 0
332
333	1	def __init__(self):
334	1	self.asymmetric_cryptography = CryptographyNone()
335	1	self.symmetric_cryptography = CryptographyNone()
336	1	self.Mode = auto.MessageSecurityMode.None_
337	1	self.server_certificate = b""
338	1	self.client_certificate = b""
339
340	1	def make_symmetric_key(self, a, b):
341	1	pass
342
343
344	1	class MessageChunk(uatypes.FrozenClass):
345		"""
346		Message Chunk, as described in OPC UA specs Part 6, 6.7.2.
347		"""
348	1	def __init__(self, security_policy, body=b'', msg_type=MessageType.SecureMessage, chunk_type=ChunkType.Single):
349	1	self.MessageHeader = Header(msg_type, chunk_type)
350	1	if msg_type in (MessageType.SecureMessage, MessageType.SecureClose):
351	1	self.SecurityHeader = SymmetricAlgorithmHeader()
352	1	elif msg_type == MessageType.SecureOpen:
353	1	self.SecurityHeader = AsymmetricAlgorithmHeader()
354		else:
355		raise Exception("Unsupported message type: {}".format(msg_type))
356	1	self.SequenceHeader = SequenceHeader()
357	1	self.Body = body
358	1	self._security_policy = security_policy
359
360	1	@staticmethod
361		def from_binary(security_policy, data):
362	1	h = Header.from_string(data)
363	1	return MessageChunk.from_header_and_body(security_policy, h, data)
364
365	1	@staticmethod
366		def from_header_and_body(security_policy, header, data):
367	1	if header.MessageType in (MessageType.SecureMessage, MessageType.SecureClose):
368	1	security_header = SymmetricAlgorithmHeader.from_binary(data)
369	1	crypto = security_policy.symmetric_cryptography
370	1	elif header.MessageType == MessageType.SecureOpen:
371	1	security_header = AsymmetricAlgorithmHeader.from_binary(data)
372	1	crypto = security_policy.asymmetric_cryptography
373		else:
374		raise Exception("Unsupported message type: {}".format(header.MessageType))
375	1	obj = MessageChunk(crypto)
376	1	obj.MessageHeader = header
377	1	obj.SecurityHeader = security_header
378	1	decrypted = crypto.decrypt(data.read(len(data)))
379	1	signature_size = crypto.vsignature_size()
380	1	if signature_size > 0:
381		signature = decrypted[-signature_size:]
382		decrypted = decrypted[:-signature_size]
383		crypto.verify(obj.MessageHeader.to_binary() + obj.SecurityHeader.to_binary() + decrypted, signature)
384	1	data = utils.Buffer(crypto.remove_padding(decrypted))
385	1	obj.SequenceHeader = SequenceHeader.from_binary(data)
386	1	obj.Body = data.read(len(data))
387	1	return obj
388
389	1	def encrypted_size(self, plain_size):
390	1	size = plain_size + self._security_policy.signature_size()
391	1	pbs = self._security_policy.plain_block_size()
392	1	assert(size % pbs == 0)
393	1	return size // pbs * self._security_policy.encrypted_block_size()
394
395	1	def to_binary(self):
396	1	security = self.SecurityHeader.to_binary()
397	1	encrypted_part = self.SequenceHeader.to_binary() + self.Body
398	1	encrypted_part += self._security_policy.padding(len(encrypted_part))
399	1	self.MessageHeader.body_size = len(security) + self.encrypted_size(len(encrypted_part))
400	1	header = self.MessageHeader.to_binary()
401	1	encrypted_part += self._security_policy.signature(header + security + encrypted_part)
402	1	return header + security + self._security_policy.encrypt(encrypted_part)
403
404	1	@staticmethod
405		def max_body_size(crypto, max_chunk_size):
406	1	max_encrypted_size = max_chunk_size - Header.max_size() - SymmetricAlgorithmHeader.max_size()
407	1	max_plain_size = (max_encrypted_size // crypto.encrypted_block_size()) * crypto.plain_block_size()
408	1	return max_plain_size - SequenceHeader.max_size() - crypto.signature_size() - crypto.min_padding_size()
409
410	1	@staticmethod
411	1	def message_to_chunks(security_policy, body, max_chunk_size, message_type=MessageType.SecureMessage, channel_id=1, request_id=1, token_id=1):
412		"""
413		Pack message body (as binary string) into one or more chunks.
414		Size of each chunk will not exceed max_chunk_size.
415		Returns a list of MessageChunks. SequenceNumber is not initialized here,
416		it must be set by Secure Channel driver.
417		"""
418	1	if message_type == MessageType.SecureOpen:
419		# SecureOpen message must be in a single chunk (specs, Part 6, 6.7.2)
420	1	chunk = MessageChunk(security_policy.asymmetric_cryptography, body, message_type, ChunkType.Single)
421	1	chunk.SecurityHeader.SecurityPolicyURI = security_policy.URI
422	1	if security_policy.client_certificate:
423		chunk.SecurityHeader.SenderCertificate = security_policy.client_certificate
424	1	if security_policy.server_certificate:
425		chunk.SecurityHeader.ReceiverCertificateThumbPrint = hashlib.sha1(security_policy.server_certificate).digest()
426	1	chunk.MessageHeader.ChannelId = channel_id
427	1	chunk.SequenceHeader.RequestId = request_id
428	1	return [chunk]
429
430	1	crypto = security_policy.symmetric_cryptography
431	1	max_size = MessageChunk.max_body_size(crypto, max_chunk_size)
432
433	1	chunks = []
434	1	for i in range(0, len(body), max_size):
435	1	part = body[i:i+max_size]
436	1	if i+max_size >= len(body):
437	1	chunk_type = ChunkType.Single
438		else:
439	1	chunk_type = ChunkType.Intermediate
440	1	chunk = MessageChunk(crypto, part, message_type, chunk_type)
441	1	chunk.SecurityHeader.TokenId = token_id
442	1	chunk.MessageHeader.ChannelId = channel_id
443	1	chunk.SequenceHeader.RequestId = request_id
444	1	chunks.append(chunk)
445	1	return chunks
446
447	1	def __str__(self):
448		return "{}({}, {}, {}, {} bytes)".format(self.__class__.__name__,
449		self.MessageHeader, self.SequenceHeader, self.SecurityHeader, len(self.Body))
450	1	__repr__ = __str__
451
452
453	1	def tcp_message_from_header_and_body(security_policy, header, body):
454		"""
455		Convert binary stream to OPC UA TCP message (see OPC UA specs Part 6, 7.1)
456		The only supported message types are Hello, Acknowledge and Error
457		"""
458	1	if header.MessageType in (MessageType.SecureOpen, MessageType.SecureClose, MessageType.SecureMessage):
459	1	return MessageChunk.from_header_and_body(security_policy, header, body)
460	1	elif header.MessageType == MessageType.Hello:
461		return Hello.from_binary(body)
462	1	elif header.MessageType == MessageType.Acknowledge:
463	1	return Acknowledge.from_binary(body)
464		elif header.MessageType == MessageType.Error:
465		return ErrorMessage.from_binary(body)
466		else:
467		raise Exception("Unsupported message type {}".format(header.MessageType))
468
469
470	1	def tcp_message_from_socket(security_policy, socket):
471	1	logger.debug("Waiting for header")
472	1	header = Header.from_string(socket)
473	1	logger.info("received header: %s", header)
474	1	body = socket.read(header.body_size)
475	1	if len(body) != header.body_size:
476		raise Exception("{} bytes expected, {} available".format(header.body_size, len(body)))
477	1	return tcp_message_from_header_and_body(security_policy, header, utils.Buffer(body))
478
479
480		# FIXES for missing switchfield in NodeAttributes classes
481	1	ana = auto.NodeAttributesMask
482
483
484	1	class ObjectAttributes(auto.ObjectAttributes):
485
486	1	def __init__(self):
487	1	auto.ObjectAttributes.__init__(self)
488	1	self.SpecifiedAttributes = ana.DisplayName \| ana.Description \| ana.WriteMask \| ana.UserWriteMask \| ana.EventNotifier
489
490
491	1	class ObjectTypeAttributes(auto.ObjectTypeAttributes):
492
493	1	def __init__(self):
494	1	auto.ObjectTypeAttributes.__init__(self)
495	1	self.SpecifiedAttributes = ana.DisplayName \| ana.Description \| ana.WriteMask \| ana.UserWriteMask \| ana.IsAbstract
496
497
498	1	class VariableAttributes(auto.VariableAttributes):
499
500	1	def __init__(self):
501	1	auto.VariableAttributes.__init__(self)
502	1	self.SpecifiedAttributes = ana.DisplayName \| ana.Description \| ana.WriteMask \| ana.UserWriteMask \| ana.Value \| ana.DataType \| ana.ValueRank \| ana.ArrayDimensions \| ana.AccessLevel \| ana.UserAccessLevel \| ana.MinimumSamplingInterval \| ana.Historizing
503	1	self.Historizing = False
504
505
506	1	class VariableTypeAttributes(auto.VariableTypeAttributes):
507
508	1	def __init__(self):
509	1	auto.VariableTypeAttributes.__init__(self)
510	1	self.SpecifiedAttributes = ana.DisplayName \| ana.Description \| ana.WriteMask \| ana.UserWriteMask \| ana.Value \| ana.DataType \| ana.ValueRank \| ana.ArrayDimensions \| ana.IsAbstract
511
512
513	1	class MethodAttributes(auto.MethodAttributes):
514
515	1	def __init__(self):
516	1	auto.MethodAttributes.__init__(self)
517	1	self.SpecifiedAttributes = ana.DisplayName \| ana.Description \| ana.WriteMask \| ana.UserWriteMask \| ana.Executable \| ana.UserExecutable
518
519
520	1	class ReferenceTypeAttributes(auto.ReferenceTypeAttributes):
521
522	1	def __init__(self):
523	1	auto.ReferenceTypeAttributes.__init__(self)
524	1	self.SpecifiedAttributes = ana.DisplayName \| ana.Description \| ana.WriteMask \| ana.UserWriteMask \| ana.IsAbstract \| ana.Symmetric \| ana.InverseName
525
526
527	1	class DataTypeAttributes(auto.DataTypeAttributes):
528
529	1	def __init__(self):
530	1	auto.DataTypeAttributes.__init__(self)
531	1	self.SpecifiedAttributes = ana.DisplayName \| ana.Description \| ana.WriteMask \| ana.UserWriteMask \| ana.IsAbstract
532
533
534	1	class ViewAttributes(auto.ViewAttributes):
535
536	1	def __init__(self):
537		auto.ViewAttributes.__init__(self)
538		self.SpecifiedAttributes = ana.DisplayName \| ana.Description \| ana.WriteMask \| ana.UserWriteMask \| ana.ContainsNoLoops \| ana.EventNotifier
539
540
541	1	class Argument(auto.Argument):
542
543	1	def __init__(self):
544	1	auto.Argument.__init__(self)
545	1	self.ValueRank = -2
546
547
548		#AttributeIdsInv = {v: k for k, v in AttributeIds.__dict__.items()}
549

FreeOpcUa / python-opcua

Push — master ( 5a9df5...39e3a6 )

opcua.ua.ErrorMessage.to_binary() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like