opcua.SecurityPolicyBasic256 - Code Metrics - Inspection of "Merge pull request #94 from alkor/client-cleanup" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 49fbc0...e6a6db )

by Olivier

created 2015-12-30 20:46 UTC

opcua.SecurityPolicyBasic256 A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	57
Duplicated Lines	0 %

Test Coverage

Coverage

26.09%

Metric	Value
wmc	2
dl	0
loc	57
ccs	6
cts	23
cp	0.2609
rs	10

2 Methods

Rating	Name	Duplication	Size	Complexity
A	make_symmetric_key()	0	10	1
A	__init__()	0	16	1

from opcua.uaprotocol import CryptographyNone, SecurityPolicy
from opcua.uaprotocol import MessageSecurityMode
from abc import ABCMeta, abstractmethod
try:
    from opcua import uacrypto
    CRYPTOGRAPHY_AVAILABLE = True
except ImportError:
    CRYPTOGRAPHY_AVAILABLE = False


def require_cryptography(obj):
    """
    Raise exception if cryptography module is not available.
    Call this function in constructors.
    """
    if not CRYPTOGRAPHY_AVAILABLE:
        raise Exception("Can't use {}, cryptography module is not installed"
                        .format(obj.__class__.__name__))


class Signer(object):
    """
    Abstract base class for cryptographic signature algorithm
    """

    __metaclass__ = ABCMeta

    @abstractmethod
    def signature_size(self):
        pass

    @abstractmethod
    def signature(self, data):
        pass


class Verifier(object):
    """
    Abstract base class for cryptographic signature verification
    """

    __metaclass__ = ABCMeta

    @abstractmethod
    def signature_size(self):
        pass

    @abstractmethod
    def verify(self, data, signature):
        pass


class Encryptor(object):
    """
    Abstract base class for encryption algorithm
    """

    __metaclass__ = ABCMeta

    @abstractmethod
    def plain_block_size(self):
        pass

    @abstractmethod
    def encrypted_block_size(self):
        pass

    @abstractmethod
    def encrypt(self, data):
        pass


class Decryptor(object):
    """
    Abstract base class for decryption algorithm
    """

    __metaclass__ = ABCMeta

    @abstractmethod
    def plain_block_size(self):
        pass

    @abstractmethod
    def encrypted_block_size(self):
        pass

    @abstractmethod
    def decrypt(self, data):
        pass


class Cryptography(CryptographyNone):
    """
    Security policy: Sign or SignAndEncrypt
    """
    def __init__(self, mode=MessageSecurityMode.Sign):
        self.Signer = None
        self.Verifier = None
        self.Encryptor = None
        self.Decryptor = None
        assert mode in (MessageSecurityMode.Sign,
                        MessageSecurityMode.SignAndEncrypt)
        self.is_encrypted = (mode == MessageSecurityMode.SignAndEncrypt)

    def plain_block_size(self):
        """
        Size of plain text block for block cipher.
        """
        if self.is_encrypted:
            return self.Encryptor.plain_block_size()
        return 1

    def encrypted_block_size(self):
        """
        Size of encrypted text block for block cipher.
        """
        if self.is_encrypted:
            return self.Encryptor.encrypted_block_size()
        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()
        """
        if not self.is_encrypted:
            return b''
        block_size = self.Encryptor.plain_block_size()
        rem = (size + self.signature_size() + 1) % block_size
        if rem != 0:
            rem = block_size - rem
        return bytes(bytearray([rem])) * (rem + 1)

    def min_padding_size(self):
        if self.is_encrypted:
            return 1
        return 0

    def signature_size(self):
        return self.Signer.signature_size()

    def signature(self, data):
        return self.Signer.signature(data)

    def vsignature_size(self):
        return self.Verifier.signature_size()

    def verify(self, data, sig):
        self.Verifier.verify(data, sig)

    def encrypt(self, data):
        if self.is_encrypted:
            assert len(data) % self.Encryptor.plain_block_size() == 0
            return self.Encryptor.encrypt(data)
        return data

    def decrypt(self, data):
        if self.is_encrypted:
            return self.Decryptor.decrypt(data)
        return data

    def remove_padding(self, data):
        if self.is_encrypted:
            pad_size = bytearray(data[-1:])[0] + 1
            return data[:-pad_size]
        return data


class SignerRsa(Signer):
    def __init__(self, client_pk):
        require_cryptography(self)
        self.client_pk = client_pk
        self.key_size = self.client_pk.key_size // 8

    def signature_size(self):
        return self.key_size

    def signature(self, data):
        return uacrypto.sign_sha1(self.client_pk, data)


class VerifierRsa(Verifier):
    def __init__(self, server_cert):
        require_cryptography(self)
        self.server_cert = server_cert
        self.key_size = self.server_cert.public_key().key_size // 8

    def signature_size(self):
        return self.key_size

    def verify(self, data, signature):
        uacrypto.verify_sha1(self.server_cert, data, signature)


class EncryptorRsa(Encryptor):
    def __init__(self, server_cert, enc_fn, padding_size):
        require_cryptography(self)
        self.server_cert = server_cert
        self.key_size = self.server_cert.public_key().key_size // 8
        self.encryptor = enc_fn
        self.padding_size = padding_size

    def plain_block_size(self):
        return self.key_size - self.padding_size

    def encrypted_block_size(self):
        return self.key_size

    def encrypt(self, data):
        encrypted = b''
        block_size = self.plain_block_size()
        for i in range(0, len(data), block_size):
            encrypted += self.encryptor(self.server_cert.public_key(),
                    data[i : i+block_size])
        return encrypted


class DecryptorRsa(Decryptor):
    def __init__(self, client_pk, dec_fn, padding_size):
        require_cryptography(self)
        self.client_pk = client_pk
        self.key_size = self.client_pk.key_size // 8
        self.decryptor = dec_fn
        self.padding_size = padding_size

    def plain_block_size(self):
        return self.key_size - self.padding_size

    def encrypted_block_size(self):
        return self.key_size

    def decrypt(self, data):
        decrypted = b''
        block_size = self.encrypted_block_size()
        for i in range(0, len(data), block_size):
            decrypted += self.decryptor(self.client_pk,
                    data[i : i+block_size])
        return decrypted


class SignerAesCbc(Signer):
    def __init__(self, key):
        require_cryptography(self)
        self.key = key

    def signature_size(self):
        return uacrypto.sha1_size()

    def signature(self, data):
        return uacrypto.hash_hmac(self.key, data)


class VerifierAesCbc(Verifier):
    def __init__(self, key):
        require_cryptography(self)
        self.key = key

    def signature_size(self):
        return uacrypto.sha1_size()

    def verify(self, data, signature):
        expected = uacrypto.hash_hmac(self.key, data)
        if signature != expected:
            raise uacrypto.InvalidSignature


class EncryptorAesCbc(Encryptor):
    def __init__(self, key, init_vec):
        require_cryptography(self)
        self.cipher = uacrypto.cipher_aes_cbc(key, init_vec)

    def plain_block_size(self):
        return self.cipher.algorithm.key_size // 8

    def encrypted_block_size(self):
        return self.cipher.algorithm.key_size // 8

    def encrypt(self, data):
        return uacrypto.cipher_encrypt(self.cipher, data)


class DecryptorAesCbc(Decryptor):
    def __init__(self, key, init_vec):
        require_cryptography(self)
        self.cipher = uacrypto.cipher_aes_cbc(key, init_vec)

    def plain_block_size(self):
        return self.cipher.algorithm.key_size // 8

    def encrypted_block_size(self):
        return self.cipher.algorithm.key_size // 8

    def decrypt(self, data):
        return uacrypto.cipher_decrypt(self.cipher, data)


class SecurityPolicyBasic128Rsa15(SecurityPolicy):

    """
    Security Basic 128Rsa15
    A suite of algorithms that uses RSA15 as Key-Wrap-algorithm
    and 128-Bit (16 bytes) for encryption algorithms.
    - SymmetricSignatureAlgorithm - HmacSha1
      (http://www.w3.org/2000/09/xmldsig#hmac-sha1)
    - SymmetricEncryptionAlgorithm - Aes128
      (http://www.w3.org/2001/04/xmlenc#aes128-cbc)
    - AsymmetricSignatureAlgorithm - RsaSha1
      (http://www.w3.org/2000/09/xmldsig#rsa-sha1)
    - AsymmetricKeyWrapAlgorithm - KwRsa15
      (http://www.w3.org/2001/04/xmlenc#rsa-1_5)
    - AsymmetricEncryptionAlgorithm - Rsa15
      (http://www.w3.org/2001/04/xmlenc#rsa-1_5)
    - KeyDerivationAlgorithm - PSha1
      (http://docs.oasis-open.org/ws-sx/ws-secureconversation/200512/dk/p_sha1)
    - DerivedSignatureKeyLength - 128 (16 bytes)
    - MinAsymmetricKeyLength - 1024 (128 bytes)
    - MaxAsymmetricKeyLength - 2048 (256 bytes)
    - CertificateSignatureAlgorithm - Sha1

    If a certificate or any certificate in the chain is not signed with
    a hash that is Sha1 or stronger then the certificate shall be rejected.
    """

    URI = "http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15"
    signature_key_size = 16
    symmetric_key_size = 16

    def __init__(self, server_cert, client_cert, client_pk, mode):
        require_cryptography(self)
        # even in Sign mode we need to asymmetrically encrypt secrets
        # transmitted in OpenSecureChannel. So SignAndEncrypt here
        self.asymmetric_cryptography = Cryptography(
                MessageSecurityMode.SignAndEncrypt)
        self.asymmetric_cryptography.Signer = SignerRsa(client_pk)
        self.asymmetric_cryptography.Verifier = VerifierRsa(server_cert)
        self.asymmetric_cryptography.Encryptor = EncryptorRsa(
                server_cert, uacrypto.encrypt_rsa15, 11)
        self.asymmetric_cryptography.Decryptor = DecryptorRsa(
                client_pk, uacrypto.decrypt_rsa15, 11)
        self.symmetric_cryptography = Cryptography(mode)
        self.Mode = mode
        self.server_certificate = uacrypto.der_from_x509(server_cert)
        self.client_certificate = uacrypto.der_from_x509(client_cert)

    def make_symmetric_key(self, nonce1, nonce2):
        key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)

        (sigkey, key, init_vec) = uacrypto.p_sha1(nonce2, nonce1, key_sizes)
        self.symmetric_cryptography.Signer = SignerAesCbc(sigkey)
        self.symmetric_cryptography.Encryptor = EncryptorAesCbc(key, init_vec)

        (sigkey, key, init_vec) = uacrypto.p_sha1(nonce1, nonce2, key_sizes)
        self.symmetric_cryptography.Verifier = VerifierAesCbc(sigkey)
        self.symmetric_cryptography.Decryptor = DecryptorAesCbc(key, init_vec)


class SecurityPolicyBasic256(SecurityPolicy):

    """
    Security Basic 256
    A suite of algorithms that are for 256-Bit (32 bytes) encryption,
    algorithms include:
    - SymmetricSignatureAlgorithm - HmacSha1
      (http://www.w3.org/2000/09/xmldsig#hmac-sha1)
    - SymmetricEncryptionAlgorithm - Aes256
      (http://www.w3.org/2001/04/xmlenc#aes256-cbc)
    - AsymmetricSignatureAlgorithm - RsaSha1
      (http://www.w3.org/2000/09/xmldsig#rsa-sha1)
    - AsymmetricKeyWrapAlgorithm - KwRsaOaep
      (http://www.w3.org/2001/04/xmlenc#rsa-oaep-mgf1p)
    - AsymmetricEncryptionAlgorithm - RsaOaep
      (http://www.w3.org/2001/04/xmlenc#rsa-oaep)
    - KeyDerivationAlgorithm - PSha1
      (http://docs.oasis-open.org/ws-sx/ws-secureconversation/200512/dk/p_sha1)
    - DerivedSignatureKeyLength - 192 (24 bytes)
    - MinAsymmetricKeyLength - 1024 (128 bytes)
    - MaxAsymmetricKeyLength - 2048 (256 bytes)
    - CertificateSignatureAlgorithm - Sha1

    If a certificate or any certificate in the chain is not signed with
    a hash that is Sha1 or stronger then the certificate shall be rejected.
    """

    URI = "http://opcfoundation.org/UA/SecurityPolicy#Basic256"
    signature_key_size = 24
    symmetric_key_size = 32

    def __init__(self, server_cert, client_cert, client_pk, mode):
        require_cryptography(self)
        # even in Sign mode we need to asymmetrically encrypt secrets
        # transmitted in OpenSecureChannel. So SignAndEncrypt here
        self.asymmetric_cryptography = Cryptography(
                MessageSecurityMode.SignAndEncrypt)
        self.asymmetric_cryptography.Signer = SignerRsa(client_pk)
        self.asymmetric_cryptography.Verifier = VerifierRsa(server_cert)
        self.asymmetric_cryptography.Encryptor = EncryptorRsa(
                server_cert, uacrypto.encrypt_rsa_oaep, 42)
        self.asymmetric_cryptography.Decryptor = DecryptorRsa(
                client_pk, uacrypto.decrypt_rsa_oaep, 42)
        self.symmetric_cryptography = Cryptography(mode)
        self.Mode = mode
        self.server_certificate = uacrypto.der_from_x509(server_cert)
        self.client_certificate = uacrypto.der_from_x509(client_cert)

    def make_symmetric_key(self, nonce1, nonce2):
        key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)

        (sigkey, key, init_vec) = uacrypto.p_sha1(nonce2, nonce1, key_sizes)
        self.symmetric_cryptography.Signer = SignerAesCbc(sigkey)
        self.symmetric_cryptography.Encryptor = EncryptorAesCbc(key, init_vec)

        (sigkey, key, init_vec) = uacrypto.p_sha1(nonce1, nonce2, key_sizes)
        self.symmetric_cryptography.Verifier = VerifierAesCbc(sigkey)
        self.symmetric_cryptography.Decryptor = DecryptorAesCbc(key, init_vec)


1	1	from opcua.uaprotocol import CryptographyNone, SecurityPolicy
2	1	from opcua.uaprotocol import MessageSecurityMode
3	1	from abc import ABCMeta, abstractmethod
4	1	try:
5	1	from opcua import uacrypto
6		CRYPTOGRAPHY_AVAILABLE = True
7	1	except ImportError:
8	1	CRYPTOGRAPHY_AVAILABLE = False
9
10
11	1	def require_cryptography(obj):
12		"""
13		Raise exception if cryptography module is not available.
14		Call this function in constructors.
15		"""
16		if not CRYPTOGRAPHY_AVAILABLE:
17		raise Exception("Can't use {}, cryptography module is not installed"
18		.format(obj.__class__.__name__))
19
20
21	1	class Signer(object):
22		"""
23		Abstract base class for cryptographic signature algorithm
24		"""
25
26	1	__metaclass__ = ABCMeta
27
28	1	@abstractmethod
29		def signature_size(self):
30		pass
31
32	1	@abstractmethod
33		def signature(self, data):
34		pass
35
36
37	1	class Verifier(object):
38		"""
39		Abstract base class for cryptographic signature verification
40		"""
41
42	1	__metaclass__ = ABCMeta
43
44	1	@abstractmethod
45		def signature_size(self):
46		pass
47
48	1	@abstractmethod
49		def verify(self, data, signature):
50		pass
51
52
53	1	class Encryptor(object):
54		"""
55		Abstract base class for encryption algorithm
56		"""
57
58	1	__metaclass__ = ABCMeta
59
60	1	@abstractmethod
61		def plain_block_size(self):
62		pass
63
64	1	@abstractmethod
65		def encrypted_block_size(self):
66		pass
67
68	1	@abstractmethod
69		def encrypt(self, data):
70		pass
71
72
73	1	class Decryptor(object):
74		"""
75		Abstract base class for decryption algorithm
76		"""
77
78	1	__metaclass__ = ABCMeta
79
80	1	@abstractmethod
81		def plain_block_size(self):
82		pass
83
84	1	@abstractmethod
85		def encrypted_block_size(self):
86		pass
87
88	1	@abstractmethod
89		def decrypt(self, data):
90		pass
91
92
93	1	class Cryptography(CryptographyNone):
94		"""
95		Security policy: Sign or SignAndEncrypt
96		"""
97	1	def __init__(self, mode=MessageSecurityMode.Sign):
98		self.Signer = None
99		self.Verifier = None
100		self.Encryptor = None
101		self.Decryptor = None
102		assert mode in (MessageSecurityMode.Sign,
103		MessageSecurityMode.SignAndEncrypt)
104		self.is_encrypted = (mode == MessageSecurityMode.SignAndEncrypt)
105
106	1	def plain_block_size(self):
107		"""
108		Size of plain text block for block cipher.
109		"""
110		if self.is_encrypted:
111		return self.Encryptor.plain_block_size()
112		return 1
113
114	1	def encrypted_block_size(self):
115		"""
116		Size of encrypted text block for block cipher.
117		"""
118		if self.is_encrypted:
119		return self.Encryptor.encrypted_block_size()
120		return 1
121
122	1	def padding(self, size):
123		"""
124		Create padding for a block of given size.
125		plain_size = size + len(padding) + signature_size()
126		plain_size = N * plain_block_size()
127		"""
128		if not self.is_encrypted:
129		return b''
130		block_size = self.Encryptor.plain_block_size()
131		rem = (size + self.signature_size() + 1) % block_size
132		if rem != 0:
133		rem = block_size - rem
134		return bytes(bytearray([rem])) * (rem + 1)
135
136	1	def min_padding_size(self):
137		if self.is_encrypted:
138		return 1
139		return 0
140
141	1	def signature_size(self):
142		return self.Signer.signature_size()
143
144	1	def signature(self, data):
145		return self.Signer.signature(data)
146
147	1	def vsignature_size(self):
148		return self.Verifier.signature_size()
149
150	1	def verify(self, data, sig):
151		self.Verifier.verify(data, sig)
152
153	1	def encrypt(self, data):
154		if self.is_encrypted:
155		assert len(data) % self.Encryptor.plain_block_size() == 0
156		return self.Encryptor.encrypt(data)
157		return data
158
159	1	def decrypt(self, data):
160		if self.is_encrypted:
161		return self.Decryptor.decrypt(data)
162		return data
163
164	1	def remove_padding(self, data):
165		if self.is_encrypted:
166		pad_size = bytearray(data[-1:])[0] + 1
167		return data[:-pad_size]
168		return data
169
170
171	1	class SignerRsa(Signer):
172	1	def __init__(self, client_pk):
173		require_cryptography(self)
174		self.client_pk = client_pk
175		self.key_size = self.client_pk.key_size // 8
176
177	1	def signature_size(self):
178		return self.key_size
179
180	1	def signature(self, data):
181		return uacrypto.sign_sha1(self.client_pk, data)
182
183
184	1	class VerifierRsa(Verifier):
185	1	def __init__(self, server_cert):
186		require_cryptography(self)
187		self.server_cert = server_cert
188		self.key_size = self.server_cert.public_key().key_size // 8
189
190	1	def signature_size(self):
191		return self.key_size
192
193	1	def verify(self, data, signature):
194		uacrypto.verify_sha1(self.server_cert, data, signature)
195
196
197	1	class EncryptorRsa(Encryptor):
198	1	def __init__(self, server_cert, enc_fn, padding_size):
199		require_cryptography(self)
200		self.server_cert = server_cert
201		self.key_size = self.server_cert.public_key().key_size // 8
202		self.encryptor = enc_fn
203		self.padding_size = padding_size
204
205	1	def plain_block_size(self):
206		return self.key_size - self.padding_size
207
208	1	def encrypted_block_size(self):
209		return self.key_size
210
211	1	def encrypt(self, data):
212		encrypted = b''
213		block_size = self.plain_block_size()
214		for i in range(0, len(data), block_size):
215		encrypted += self.encryptor(self.server_cert.public_key(),
216		data[i : i+block_size])
217		return encrypted
218
219
220	1	class DecryptorRsa(Decryptor):
221	1	def __init__(self, client_pk, dec_fn, padding_size):
222		require_cryptography(self)
223		self.client_pk = client_pk
224		self.key_size = self.client_pk.key_size // 8
225		self.decryptor = dec_fn
226		self.padding_size = padding_size
227
228	1	def plain_block_size(self):
229		return self.key_size - self.padding_size
230
231	1	def encrypted_block_size(self):
232		return self.key_size
233
234	1	def decrypt(self, data):
235		decrypted = b''
236		block_size = self.encrypted_block_size()
237		for i in range(0, len(data), block_size):
238		decrypted += self.decryptor(self.client_pk,
239		data[i : i+block_size])
240		return decrypted
241
242
243	1	class SignerAesCbc(Signer):
244	1	def __init__(self, key):
245		require_cryptography(self)
246		self.key = key
247
248	1	def signature_size(self):
249		return uacrypto.sha1_size()
250
251	1	def signature(self, data):
252		return uacrypto.hash_hmac(self.key, data)
253
254
255	1	class VerifierAesCbc(Verifier):
256	1	def __init__(self, key):
257		require_cryptography(self)
258		self.key = key
259
260	1	def signature_size(self):
261		return uacrypto.sha1_size()
262
263	1	def verify(self, data, signature):
264		expected = uacrypto.hash_hmac(self.key, data)
265		if signature != expected:
266		raise uacrypto.InvalidSignature
267
268
269	1	class EncryptorAesCbc(Encryptor):
270	1	def __init__(self, key, init_vec):
271		require_cryptography(self)
272		self.cipher = uacrypto.cipher_aes_cbc(key, init_vec)
273
274	1	def plain_block_size(self):
275		return self.cipher.algorithm.key_size // 8
276
277	1	def encrypted_block_size(self):
278		return self.cipher.algorithm.key_size // 8
279
280	1	def encrypt(self, data):
281		return uacrypto.cipher_encrypt(self.cipher, data)
282
283
284	1	class DecryptorAesCbc(Decryptor):
285	1	def __init__(self, key, init_vec):
286		require_cryptography(self)
287		self.cipher = uacrypto.cipher_aes_cbc(key, init_vec)
288
289	1	def plain_block_size(self):
290		return self.cipher.algorithm.key_size // 8
291
292	1	def encrypted_block_size(self):
293		return self.cipher.algorithm.key_size // 8
294
295	1	def decrypt(self, data):
296		return uacrypto.cipher_decrypt(self.cipher, data)
297
298
299	1	class SecurityPolicyBasic128Rsa15(SecurityPolicy):
		0 ignored issues – show Duplication introduced 2015-12-30 15:36 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
300		"""
301		Security Basic 128Rsa15
302		A suite of algorithms that uses RSA15 as Key-Wrap-algorithm
303		and 128-Bit (16 bytes) for encryption algorithms.
304		- SymmetricSignatureAlgorithm - HmacSha1
305		(http://www.w3.org/2000/09/xmldsig#hmac-sha1)
306		- SymmetricEncryptionAlgorithm - Aes128
307		(http://www.w3.org/2001/04/xmlenc#aes128-cbc)
308		- AsymmetricSignatureAlgorithm - RsaSha1
309		(http://www.w3.org/2000/09/xmldsig#rsa-sha1)
310		- AsymmetricKeyWrapAlgorithm - KwRsa15
311		(http://www.w3.org/2001/04/xmlenc#rsa-1_5)
312		- AsymmetricEncryptionAlgorithm - Rsa15
313		(http://www.w3.org/2001/04/xmlenc#rsa-1_5)
314		- KeyDerivationAlgorithm - PSha1
315		(http://docs.oasis-open.org/ws-sx/ws-secureconversation/200512/dk/p_sha1)
316		- DerivedSignatureKeyLength - 128 (16 bytes)
317		- MinAsymmetricKeyLength - 1024 (128 bytes)
318		- MaxAsymmetricKeyLength - 2048 (256 bytes)
319		- CertificateSignatureAlgorithm - Sha1
320
321		If a certificate or any certificate in the chain is not signed with
322		a hash that is Sha1 or stronger then the certificate shall be rejected.
323		"""
324
325	1	URI = "http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15"
326	1	signature_key_size = 16
327	1	symmetric_key_size = 16
328
329	1	def __init__(self, server_cert, client_cert, client_pk, mode):
330		require_cryptography(self)
331		# even in Sign mode we need to asymmetrically encrypt secrets
332		# transmitted in OpenSecureChannel. So SignAndEncrypt here
333		self.asymmetric_cryptography = Cryptography(
334		MessageSecurityMode.SignAndEncrypt)
335		self.asymmetric_cryptography.Signer = SignerRsa(client_pk)
336		self.asymmetric_cryptography.Verifier = VerifierRsa(server_cert)
337		self.asymmetric_cryptography.Encryptor = EncryptorRsa(
338		server_cert, uacrypto.encrypt_rsa15, 11)
339		self.asymmetric_cryptography.Decryptor = DecryptorRsa(
340		client_pk, uacrypto.decrypt_rsa15, 11)
341		self.symmetric_cryptography = Cryptography(mode)
342		self.Mode = mode
343		self.server_certificate = uacrypto.der_from_x509(server_cert)
344		self.client_certificate = uacrypto.der_from_x509(client_cert)
345
346	1	def make_symmetric_key(self, nonce1, nonce2):
347		key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)
348
349		(sigkey, key, init_vec) = uacrypto.p_sha1(nonce2, nonce1, key_sizes)
350		self.symmetric_cryptography.Signer = SignerAesCbc(sigkey)
351		self.symmetric_cryptography.Encryptor = EncryptorAesCbc(key, init_vec)
352
353		(sigkey, key, init_vec) = uacrypto.p_sha1(nonce1, nonce2, key_sizes)
354		self.symmetric_cryptography.Verifier = VerifierAesCbc(sigkey)
355		self.symmetric_cryptography.Decryptor = DecryptorAesCbc(key, init_vec)
356
357
358	1	class SecurityPolicyBasic256(SecurityPolicy):
		0 ignored issues – show Duplication introduced 2015-12-30 15:36 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
359		"""
360		Security Basic 256
361		A suite of algorithms that are for 256-Bit (32 bytes) encryption,
362		algorithms include:
363		- SymmetricSignatureAlgorithm - HmacSha1
364		(http://www.w3.org/2000/09/xmldsig#hmac-sha1)
365		- SymmetricEncryptionAlgorithm - Aes256
366		(http://www.w3.org/2001/04/xmlenc#aes256-cbc)
367		- AsymmetricSignatureAlgorithm - RsaSha1
368		(http://www.w3.org/2000/09/xmldsig#rsa-sha1)
369		- AsymmetricKeyWrapAlgorithm - KwRsaOaep
370		(http://www.w3.org/2001/04/xmlenc#rsa-oaep-mgf1p)
371		- AsymmetricEncryptionAlgorithm - RsaOaep
372		(http://www.w3.org/2001/04/xmlenc#rsa-oaep)
373		- KeyDerivationAlgorithm - PSha1
374		(http://docs.oasis-open.org/ws-sx/ws-secureconversation/200512/dk/p_sha1)
375		- DerivedSignatureKeyLength - 192 (24 bytes)
376		- MinAsymmetricKeyLength - 1024 (128 bytes)
377		- MaxAsymmetricKeyLength - 2048 (256 bytes)
378		- CertificateSignatureAlgorithm - Sha1
379
380		If a certificate or any certificate in the chain is not signed with
381		a hash that is Sha1 or stronger then the certificate shall be rejected.
382		"""
383
384	1	URI = "http://opcfoundation.org/UA/SecurityPolicy#Basic256"
385	1	signature_key_size = 24
386	1	symmetric_key_size = 32
387
388	1	def __init__(self, server_cert, client_cert, client_pk, mode):
389		require_cryptography(self)
390		# even in Sign mode we need to asymmetrically encrypt secrets
391		# transmitted in OpenSecureChannel. So SignAndEncrypt here
392		self.asymmetric_cryptography = Cryptography(
393		MessageSecurityMode.SignAndEncrypt)
394		self.asymmetric_cryptography.Signer = SignerRsa(client_pk)
395		self.asymmetric_cryptography.Verifier = VerifierRsa(server_cert)
396		self.asymmetric_cryptography.Encryptor = EncryptorRsa(
397		server_cert, uacrypto.encrypt_rsa_oaep, 42)
398		self.asymmetric_cryptography.Decryptor = DecryptorRsa(
399		client_pk, uacrypto.decrypt_rsa_oaep, 42)
400		self.symmetric_cryptography = Cryptography(mode)
401		self.Mode = mode
402		self.server_certificate = uacrypto.der_from_x509(server_cert)
403		self.client_certificate = uacrypto.der_from_x509(client_cert)
404
405	1	def make_symmetric_key(self, nonce1, nonce2):
406		key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)
407
408		(sigkey, key, init_vec) = uacrypto.p_sha1(nonce2, nonce1, key_sizes)
409		self.symmetric_cryptography.Signer = SignerAesCbc(sigkey)
410		self.symmetric_cryptography.Encryptor = EncryptorAesCbc(key, init_vec)
411
412		(sigkey, key, init_vec) = uacrypto.p_sha1(nonce1, nonce2, key_sizes)
413		self.symmetric_cryptography.Verifier = VerifierAesCbc(sigkey)
414		self.symmetric_cryptography.Decryptor = DecryptorAesCbc(key, init_vec)
415

FreeOpcUa / python-opcua

Push — master ( 49fbc0...e6a6db )

opcua.SecurityPolicyBasic256 A

Complexity

Size/Duplication

Test Coverage

2 Methods

Duplication Side-by-Side

Filter issues like