asyncua.crypto.security_policies.SecurityPolicyBasic256.make_local_symmetric_key() - Code Metrics - FreeOpcUa/opcua-asyncio - Measure and Improve Code Quality continuously with Scrutinizer

SecurityPolicyBasic256.make_local_symmetric_key() A
last analyzed 2020-12-22 17:52 UTC

↳ Parent: asyncua.crypto.security_policies

Complexity

Conditions

Size

Total Lines	7
Code Lines	5

Duplication

Lines	7
Ratio	100 %

Importance

Changes

Metric	Value
cc	1
eloc	5
nop	3
dl	7
loc	7
rs	10
c	0
b	0
f	0

import logging
import struct

from abc import ABCMeta, abstractmethod
from ..ua import CryptographyNone, SecurityPolicy, MessageSecurityMode, UaError

try:
    from ..crypto import uacrypto
    CRYPTOGRAPHY_AVAILABLE = True
except ImportError:
    CRYPTOGRAPHY_AVAILABLE = False


POLICY_NONE_URI = 'http://opcfoundation.org/UA/SecurityPolicy#None'


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


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
        if mode not in (MessageSecurityMode.Sign,
                        MessageSecurityMode.SignAndEncrypt):
            raise ValueError(f"unknown security mode {mode}")
        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()
        extrapad_size = 2 if self.Encryptor.encrypted_block_size() > 256 else 1
        rem = (size + self.signature_size() + extrapad_size) % block_size
        if rem != 0:
            rem = block_size - rem
        data = bytes(bytearray([rem % 256])) * (rem + 1)
        if self.Encryptor.encrypted_block_size() > 256:
            data = data + bytes(bytearray([rem >> 8]))
        return data

    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:
            if not len(data) % self.Encryptor.plain_block_size() == 0:
                raise ValueError
            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:
            if self.Decryptor.encrypted_block_size() > 256:
                pad_size = struct.unpack('<h', data[-2:])[0] + 2
            else:
                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.hmac_sha1(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.hmac_sha1(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 SignerSha256(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_sha256(self.client_pk, data)


class VerifierSha256(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_sha256(self.server_cert, data, signature)


class SignerHMac256(Signer):

    def __init__(self, key):
        require_cryptography(self)
        self.key = key

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

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


class VerifierHMac256(Verifier):

    def __init__(self, key):
        require_cryptography(self)
        self.key = key

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

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


class SecurityPolicyBasic128Rsa15(SecurityPolicy):

    """
    DEPRECATED, do not use anymore!

    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
    AsymmetricEncryptionURI = "http://www.w3.org/2001/04/xmlenc#rsa-1_5"
    AsymmetricSignatureURI = "http://www.w3.org/2000/09/xmldsig#rsa-sha1"

    @staticmethod
    def encrypt_asymmetric(pubkey, data):
        return uacrypto.encrypt_rsa15(pubkey, data)

    def __init__(self, server_cert, client_cert, client_pk, mode):
        logger = logging.getLogger(__name__)
        logger.warning("DEPRECATED! Do not use SecurityPolicyBasic128Rsa15 anymore!")

        require_cryptography(self)
        if isinstance(server_cert, bytes):
            server_cert = uacrypto.x509_from_der(server_cert)
        # 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_local_symmetric_key(self, secret, seed):
        key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)

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

    def make_remote_symmetric_key(self, secret, seed):
        key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)

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


class SecurityPolicyBasic256(SecurityPolicy):

    """
    DEPRECATED, do not use anymore!

    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
    AsymmetricEncryptionURI = "http://www.w3.org/2001/04/xmlenc#rsa-oaep"
    AsymmetricSignatureURI = "http://www.w3.org/2000/09/xmldsig#rsa-sha1"

    @staticmethod
    def encrypt_asymmetric(pubkey, data):
        return uacrypto.encrypt_rsa_oaep(pubkey, data)

    def __init__(self, server_cert, client_cert, client_pk, mode):
        logger = logging.getLogger(__name__)
        logger.warning("DEPRECATED! Do not use SecurityPolicyBasic256 anymore!")

        require_cryptography(self)
        if isinstance(server_cert, bytes):
            server_cert = uacrypto.x509_from_der(server_cert)
        # 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_local_symmetric_key(self, secret, seed):
        # specs part 6, 6.7.5
        key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)

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

    def make_remote_symmetric_key(self, secret, seed):

        # specs part 6, 6.7.5
        key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)

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


class SecurityPolicyBasic256Sha256(SecurityPolicy):
    """
    Security Basic 256Sha256
    A suite of algorithms that uses Sha256 as Key-Wrap-algorithm
    and 256-Bit (32 bytes) for encryption algorithms.

    - SymmetricSignatureAlgorithm_HMAC-SHA2-256
      https://tools.ietf.org/html/rfc4634
    - SymmetricEncryptionAlgorithm_AES256-CBC
      http://www.w3.org/2001/04/xmlenc#aes256-cbc
    - AsymmetricSignatureAlgorithm_RSA-PKCS15-SHA2-256
      http://www.w3.org/2001/04/xmldsig-more#rsa-sha256
    - AsymmetricEncryptionAlgorithm_RSA-OAEP-SHA1
      http://www.w3.org/2001/04/xmlenc#rsa-oaep
    - KeyDerivationAlgorithm_P-SHA2-256
      http://docs.oasis-open.org/ws-sx/ws-secureconversation/200512/dk/p_sha256
    - CertificateSignatureAlgorithm_RSA-PKCS15-SHA2-256
      http://www.w3.org/2001/04/xmldsig-more#rsa-sha256
    - Basic256Sha256_Limits
        -> DerivedSignatureKeyLength: 256 bits
        -> MinAsymmetricKeyLength: 2048 bits
        -> MaxAsymmetricKeyLength: 4096 bits
        -> SecureChannelNonceLength: 32 bytes
    """

    URI = "http://opcfoundation.org/UA/SecurityPolicy#Basic256Sha256"
    signature_key_size = 32
    symmetric_key_size = 32
    AsymmetricEncryptionURI = "http://www.w3.org/2001/04/xmlenc#rsa-oaep"
    AsymmetricSignatureURI = "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256"

    @staticmethod
    def encrypt_asymmetric(pubkey, data):
        return uacrypto.encrypt_rsa_oaep(pubkey, data)

    def __init__(self, peer_cert, host_cert, client_pk, mode,
                 permission_ruleset=None):
        require_cryptography(self)
        if isinstance(peer_cert, bytes):
            peer_cert = uacrypto.x509_from_der(peer_cert)
        # 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 = SignerSha256(client_pk)
        self.asymmetric_cryptography.Verifier = VerifierSha256(peer_cert)
        self.asymmetric_cryptography.Encryptor = EncryptorRsa(
            peer_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.peer_certificate = uacrypto.der_from_x509(peer_cert)
        self.host_certificate = uacrypto.der_from_x509(host_cert)
        if permission_ruleset is None:
            from asyncua.crypto.permission_rules import SimpleRoleRuleset
            permission_ruleset = SimpleRoleRuleset()
        
        self.permissions = permission_ruleset

    def make_local_symmetric_key(self, secret, seed):
        # specs part 6, 6.7.5
        key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)

        (sigkey, key, init_vec) = uacrypto.p_sha256(secret, seed, key_sizes)
        self.symmetric_cryptography.Signer = SignerHMac256(sigkey)
        self.symmetric_cryptography.Encryptor = EncryptorAesCbc(key, init_vec)

    def make_remote_symmetric_key(self, secret, seed):

        # specs part 6, 6.7.5
        key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)

        (sigkey, key, init_vec) = uacrypto.p_sha256(secret, seed, key_sizes)
        self.symmetric_cryptography.Verifier = VerifierHMac256(sigkey)
        self.symmetric_cryptography.Decryptor = DecryptorAesCbc(key, init_vec)


def encrypt_asymmetric(pubkey, data, policy_uri):
    """
    Encrypt data with pubkey using an asymmetric algorithm.
    The algorithm is selected by policy_uri.
    Returns a tuple (encrypted_data, algorithm_uri)
    """
    for cls in [SecurityPolicyBasic256Sha256, SecurityPolicyBasic256, SecurityPolicyBasic128Rsa15]:
        if policy_uri == cls.URI:
            return (cls.encrypt_asymmetric(pubkey, data),
                    cls.AsymmetricEncryptionURI)
    if not policy_uri or policy_uri == POLICY_NONE_URI:
        return data, ''
    raise UaError(f"Unsupported security policy `{policy_uri}`")


1		import logging
2		import struct
3
4		from abc import ABCMeta, abstractmethod
5		from ..ua import CryptographyNone, SecurityPolicy, MessageSecurityMode, UaError
6
7		try:
8		from ..crypto import uacrypto
9		CRYPTOGRAPHY_AVAILABLE = True
10		except ImportError:
11		CRYPTOGRAPHY_AVAILABLE = False
12
13
14		POLICY_NONE_URI = 'http://opcfoundation.org/UA/SecurityPolicy#None'
15
16
17		def require_cryptography(obj):
18		"""
19		Raise exception if cryptography module is not available.
20		Call this function in constructors.
21		"""
22		if not CRYPTOGRAPHY_AVAILABLE:
23		raise UaError(f"Can't use {obj.__class__.__name__}, cryptography module is not installed")
24
25
26		class Signer(object):
27		"""
28		Abstract base class for cryptographic signature algorithm
29		"""
30
31		__metaclass__ = ABCMeta
32
33		@abstractmethod
34		def signature_size(self):
35		pass
36
37		@abstractmethod
38		def signature(self, data):
39		pass
40
41
42		class Verifier(object):
43		"""
44		Abstract base class for cryptographic signature verification
45		"""
46
47		__metaclass__ = ABCMeta
48
49		@abstractmethod
50		def signature_size(self):
51		pass
52
53		@abstractmethod
54		def verify(self, data, signature):
55		pass
56
57
58		class Encryptor(object):
59		"""
60		Abstract base class for encryption algorithm
61		"""
62
63		__metaclass__ = ABCMeta
64
65		@abstractmethod
66		def plain_block_size(self):
67		pass
68
69		@abstractmethod
70		def encrypted_block_size(self):
71		pass
72
73		@abstractmethod
74		def encrypt(self, data):
75		pass
76
77
78		class Decryptor(object):
79		"""
80		Abstract base class for decryption algorithm
81		"""
82
83		__metaclass__ = ABCMeta
84
85		@abstractmethod
86		def plain_block_size(self):
87		pass
88
89		@abstractmethod
90		def encrypted_block_size(self):
91		pass
92
93		@abstractmethod
94		def decrypt(self, data):
95		pass
96
97
98		class Cryptography(CryptographyNone):
99		"""
100		Security policy: Sign or SignAndEncrypt
101		"""
102
103		def __init__(self, mode=MessageSecurityMode.Sign):
104		self.Signer = None
105		self.Verifier = None
106		self.Encryptor = None
107		self.Decryptor = None
108		if mode not in (MessageSecurityMode.Sign,
109		MessageSecurityMode.SignAndEncrypt):
110		raise ValueError(f"unknown security mode {mode}")
111		self.is_encrypted = (mode == MessageSecurityMode.SignAndEncrypt)
112
113		def plain_block_size(self):
114		"""
115		Size of plain text block for block cipher.
116		"""
117		if self.is_encrypted:
118		return self.Encryptor.plain_block_size()
119		return 1
120
121		def encrypted_block_size(self):
122		"""
123		Size of encrypted text block for block cipher.
124		"""
125		if self.is_encrypted:
126		return self.Encryptor.encrypted_block_size()
127		return 1
128
129		def padding(self, size):
130		"""
131		Create padding for a block of given size.
132		plain_size = size + len(padding) + signature_size()
133		plain_size = N * plain_block_size()
134		"""
135		if not self.is_encrypted:
136		return b''
137		block_size = self.Encryptor.plain_block_size()
138		extrapad_size = 2 if self.Encryptor.encrypted_block_size() > 256 else 1
139		rem = (size + self.signature_size() + extrapad_size) % block_size
140		if rem != 0:
141		rem = block_size - rem
142		data = bytes(bytearray([rem % 256])) * (rem + 1)
143		if self.Encryptor.encrypted_block_size() > 256:
144		data = data + bytes(bytearray([rem >> 8]))
145		return data
146
147		def min_padding_size(self):
148		if self.is_encrypted:
149		return 1
150		return 0
151
152		def signature_size(self):
153		return self.Signer.signature_size()
154
155		def signature(self, data):
156		return self.Signer.signature(data)
157
158		def vsignature_size(self):
159		return self.Verifier.signature_size()
160
161		def verify(self, data, sig):
162		self.Verifier.verify(data, sig)
163
164		def encrypt(self, data):
165		if self.is_encrypted:
166		if not len(data) % self.Encryptor.plain_block_size() == 0:
167		raise ValueError
168		return self.Encryptor.encrypt(data)
169		return data
170
171		def decrypt(self, data):
172		if self.is_encrypted:
173		return self.Decryptor.decrypt(data)
174		return data
175
176		def remove_padding(self, data):
177		if self.is_encrypted:
178		if self.Decryptor.encrypted_block_size() > 256:
179		pad_size = struct.unpack('<h', data[-2:])[0] + 2
180		else:
181		pad_size = bytearray(data[-1:])[0] + 1
182		return data[:-pad_size]
183		return data
184
185
186		class SignerRsa(Signer):
187
188		def __init__(self, client_pk):
189		require_cryptography(self)
190		self.client_pk = client_pk
191		self.key_size = self.client_pk.key_size // 8
192
193		def signature_size(self):
194		return self.key_size
195
196		def signature(self, data):
197		return uacrypto.sign_sha1(self.client_pk, data)
198
199
200		class VerifierRsa(Verifier):
201
202		def __init__(self, server_cert):
203		require_cryptography(self)
204		self.server_cert = server_cert
205		self.key_size = self.server_cert.public_key().key_size // 8
206
207		def signature_size(self):
208		return self.key_size
209
210		def verify(self, data, signature):
211		uacrypto.verify_sha1(self.server_cert, data, signature)
212
213
214		class EncryptorRsa(Encryptor):
215
216		def __init__(self, server_cert, enc_fn, padding_size):
217		require_cryptography(self)
218		self.server_cert = server_cert
219		self.key_size = self.server_cert.public_key().key_size // 8
220		self.encryptor = enc_fn
221		self.padding_size = padding_size
222
223		def plain_block_size(self):
224		return self.key_size - self.padding_size
225
226		def encrypted_block_size(self):
227		return self.key_size
228
229		def encrypt(self, data):
230		encrypted = b''
231		block_size = self.plain_block_size()
232		for i in range(0, len(data), block_size):
233		encrypted += self.encryptor(self.server_cert.public_key(),
234		data[i: i + block_size])
235		return encrypted
236
237
238		class DecryptorRsa(Decryptor):
239
240		def __init__(self, client_pk, dec_fn, padding_size):
241		require_cryptography(self)
242		self.client_pk = client_pk
243		self.key_size = self.client_pk.key_size // 8
244		self.decryptor = dec_fn
245		self.padding_size = padding_size
246
247		def plain_block_size(self):
248		return self.key_size - self.padding_size
249
250		def encrypted_block_size(self):
251		return self.key_size
252
253		def decrypt(self, data):
254		decrypted = b''
255		block_size = self.encrypted_block_size()
256		for i in range(0, len(data), block_size):
257		decrypted += self.decryptor(self.client_pk,
258		data[i: i + block_size])
259		return decrypted
260
261
262		class SignerAesCbc(Signer):
263
264		def __init__(self, key):
265		require_cryptography(self)
266		self.key = key
267
268		def signature_size(self):
269		return uacrypto.sha1_size()
270
271		def signature(self, data):
272		return uacrypto.hmac_sha1(self.key, data)
273
274
275		class VerifierAesCbc(Verifier):
276
277		def __init__(self, key):
278		require_cryptography(self)
279		self.key = key
280
281		def signature_size(self):
282		return uacrypto.sha1_size()
283
284		def verify(self, data, signature):
285		expected = uacrypto.hmac_sha1(self.key, data)
286		if signature != expected:
287		raise uacrypto.InvalidSignature
288
289
290		class EncryptorAesCbc(Encryptor):
291
292		def __init__(self, key, init_vec):
293		require_cryptography(self)
294		self.cipher = uacrypto.cipher_aes_cbc(key, init_vec)
295
296		def plain_block_size(self):
297		return self.cipher.algorithm.key_size // 8
298
299		def encrypted_block_size(self):
300		return self.cipher.algorithm.key_size // 8
301
302		def encrypt(self, data):
303		return uacrypto.cipher_encrypt(self.cipher, data)
304
305
306		class DecryptorAesCbc(Decryptor):
307
308		def __init__(self, key, init_vec):
309		require_cryptography(self)
310		self.cipher = uacrypto.cipher_aes_cbc(key, init_vec)
311
312		def plain_block_size(self):
313		return self.cipher.algorithm.key_size // 8
314
315		def encrypted_block_size(self):
316		return self.cipher.algorithm.key_size // 8
317
318		def decrypt(self, data):
319		return uacrypto.cipher_decrypt(self.cipher, data)
320
321
322		class SignerSha256(Signer):
323
324		def __init__(self, client_pk):
325		require_cryptography(self)
326		self.client_pk = client_pk
327		self.key_size = self.client_pk.key_size // 8
328
329		def signature_size(self):
330		return self.key_size
331
332		def signature(self, data):
333		return uacrypto.sign_sha256(self.client_pk, data)
334
335
336		class VerifierSha256(Verifier):
337
338		def __init__(self, server_cert):
339		require_cryptography(self)
340		self.server_cert = server_cert
341		self.key_size = self.server_cert.public_key().key_size // 8
342
343		def signature_size(self):
344		return self.key_size
345
346		def verify(self, data, signature):
347		uacrypto.verify_sha256(self.server_cert, data, signature)
348
349
350		class SignerHMac256(Signer):
351
352		def __init__(self, key):
353		require_cryptography(self)
354		self.key = key
355
356		def signature_size(self):
357		return uacrypto.sha256_size()
358
359		def signature(self, data):
360		return uacrypto.hmac_sha256(self.key, data)
361
362
363		class VerifierHMac256(Verifier):
364
365		def __init__(self, key):
366		require_cryptography(self)
367		self.key = key
368
369		def signature_size(self):
370		return uacrypto.sha256_size()
371
372		def verify(self, data, signature):
373		expected = uacrypto.hmac_sha256(self.key, data)
374		if signature != expected:
375		raise uacrypto.InvalidSignature
376
377
378	View Code Duplication	class SecurityPolicyBasic128Rsa15(SecurityPolicy):
		0 ignored issues – show Duplication introduced 2019-03-15 08:18 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
379		"""
380		DEPRECATED, do not use anymore!
381
382		Security Basic 128Rsa15
383		A suite of algorithms that uses RSA15 as Key-Wrap-algorithm
384		and 128-Bit (16 bytes) for encryption algorithms.
385		- SymmetricSignatureAlgorithm - HmacSha1
386		(http://www.w3.org/2000/09/xmldsig#hmac-sha1)
387		- SymmetricEncryptionAlgorithm - Aes128
388		(http://www.w3.org/2001/04/xmlenc#aes128-cbc)
389		- AsymmetricSignatureAlgorithm - RsaSha1
390		(http://www.w3.org/2000/09/xmldsig#rsa-sha1)
391		- AsymmetricKeyWrapAlgorithm - KwRsa15
392		(http://www.w3.org/2001/04/xmlenc#rsa-1_5)
393		- AsymmetricEncryptionAlgorithm - Rsa15
394		(http://www.w3.org/2001/04/xmlenc#rsa-1_5)
395		- KeyDerivationAlgorithm - PSha1
396		(http://docs.oasis-open.org/ws-sx/ws-secureconversation/200512/dk/p_sha1)
397		- DerivedSignatureKeyLength - 128 (16 bytes)
398		- MinAsymmetricKeyLength - 1024 (128 bytes)
399		- MaxAsymmetricKeyLength - 2048 (256 bytes)
400		- CertificateSignatureAlgorithm - Sha1
401
402		If a certificate or any certificate in the chain is not signed with
403		a hash that is Sha1 or stronger then the certificate shall be rejected.
404		"""
405
406		URI = "http://opcfoundation.org/UA/SecurityPolicy#Basic128Rsa15"
407		signature_key_size = 16
408		symmetric_key_size = 16
409		AsymmetricEncryptionURI = "http://www.w3.org/2001/04/xmlenc#rsa-1_5"
410		AsymmetricSignatureURI = "http://www.w3.org/2000/09/xmldsig#rsa-sha1"
411
412		@staticmethod
413		def encrypt_asymmetric(pubkey, data):
414		return uacrypto.encrypt_rsa15(pubkey, data)
415
416		def __init__(self, server_cert, client_cert, client_pk, mode):
417		logger = logging.getLogger(__name__)
418		logger.warning("DEPRECATED! Do not use SecurityPolicyBasic128Rsa15 anymore!")
419
420		require_cryptography(self)
421		if isinstance(server_cert, bytes):
422		server_cert = uacrypto.x509_from_der(server_cert)
423		# even in Sign mode we need to asymmetrically encrypt secrets
424		# transmitted in OpenSecureChannel. So SignAndEncrypt here
425		self.asymmetric_cryptography = Cryptography(
426		MessageSecurityMode.SignAndEncrypt)
427		self.asymmetric_cryptography.Signer = SignerRsa(client_pk)
428		self.asymmetric_cryptography.Verifier = VerifierRsa(server_cert)
429		self.asymmetric_cryptography.Encryptor = EncryptorRsa(
430		server_cert, uacrypto.encrypt_rsa15, 11)
431		self.asymmetric_cryptography.Decryptor = DecryptorRsa(
432		client_pk, uacrypto.decrypt_rsa15, 11)
433		self.symmetric_cryptography = Cryptography(mode)
434		self.Mode = mode
435		self.server_certificate = uacrypto.der_from_x509(server_cert)
436		self.client_certificate = uacrypto.der_from_x509(client_cert)
437
438		def make_local_symmetric_key(self, secret, seed):
439		key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)
440
441		(sigkey, key, init_vec) = uacrypto.p_sha1(secret, seed, key_sizes)
442		self.symmetric_cryptography.Signer = SignerAesCbc(sigkey)
443		self.symmetric_cryptography.Encryptor = EncryptorAesCbc(key, init_vec)
444
445		def make_remote_symmetric_key(self, secret, seed):
446		key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)
447
448		(sigkey, key, init_vec) = uacrypto.p_sha1(secret, seed, key_sizes)
449		self.symmetric_cryptography.Verifier = VerifierAesCbc(sigkey)
450		self.symmetric_cryptography.Decryptor = DecryptorAesCbc(key, init_vec)
451
452
453	View Code Duplication	class SecurityPolicyBasic256(SecurityPolicy):
		0 ignored issues – show Duplication introduced 2019-12-03 21:37 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
454		"""
455		DEPRECATED, do not use anymore!
456
457		Security Basic 256
458		A suite of algorithms that are for 256-Bit (32 bytes) encryption,
459		algorithms include:
460		- SymmetricSignatureAlgorithm - HmacSha1
461		(http://www.w3.org/2000/09/xmldsig#hmac-sha1)
462		- SymmetricEncryptionAlgorithm - Aes256
463		(http://www.w3.org/2001/04/xmlenc#aes256-cbc)
464		- AsymmetricSignatureAlgorithm - RsaSha1
465		(http://www.w3.org/2000/09/xmldsig#rsa-sha1)
466		- AsymmetricKeyWrapAlgorithm - KwRsaOaep
467		(http://www.w3.org/2001/04/xmlenc#rsa-oaep-mgf1p)
468		- AsymmetricEncryptionAlgorithm - RsaOaep
469		(http://www.w3.org/2001/04/xmlenc#rsa-oaep)
470		- KeyDerivationAlgorithm - PSha1
471		(http://docs.oasis-open.org/ws-sx/ws-secureconversation/200512/dk/p_sha1)
472		- DerivedSignatureKeyLength - 192 (24 bytes)
473		- MinAsymmetricKeyLength - 1024 (128 bytes)
474		- MaxAsymmetricKeyLength - 2048 (256 bytes)
475		- CertificateSignatureAlgorithm - Sha1
476
477		If a certificate or any certificate in the chain is not signed with
478		a hash that is Sha1 or stronger then the certificate shall be rejected.
479		"""
480
481		URI = "http://opcfoundation.org/UA/SecurityPolicy#Basic256"
482		signature_key_size = 24
483		symmetric_key_size = 32
484		AsymmetricEncryptionURI = "http://www.w3.org/2001/04/xmlenc#rsa-oaep"
485		AsymmetricSignatureURI = "http://www.w3.org/2000/09/xmldsig#rsa-sha1"
486
487		@staticmethod
488		def encrypt_asymmetric(pubkey, data):
489		return uacrypto.encrypt_rsa_oaep(pubkey, data)
490
491		def __init__(self, server_cert, client_cert, client_pk, mode):
492		logger = logging.getLogger(__name__)
493		logger.warning("DEPRECATED! Do not use SecurityPolicyBasic256 anymore!")
494
495		require_cryptography(self)
496		if isinstance(server_cert, bytes):
497		server_cert = uacrypto.x509_from_der(server_cert)
498		# even in Sign mode we need to asymmetrically encrypt secrets
499		# transmitted in OpenSecureChannel. So SignAndEncrypt here
500		self.asymmetric_cryptography = Cryptography(
501		MessageSecurityMode.SignAndEncrypt)
502		self.asymmetric_cryptography.Signer = SignerRsa(client_pk)
503		self.asymmetric_cryptography.Verifier = VerifierRsa(server_cert)
504		self.asymmetric_cryptography.Encryptor = EncryptorRsa(
505		server_cert, uacrypto.encrypt_rsa_oaep, 42)
506		self.asymmetric_cryptography.Decryptor = DecryptorRsa(
507		client_pk, uacrypto.decrypt_rsa_oaep, 42)
508		self.symmetric_cryptography = Cryptography(mode)
509		self.Mode = mode
510		self.server_certificate = uacrypto.der_from_x509(server_cert)
511		self.client_certificate = uacrypto.der_from_x509(client_cert)
512
513		def make_local_symmetric_key(self, secret, seed):
514		# specs part 6, 6.7.5
515		key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)
516
517		(sigkey, key, init_vec) = uacrypto.p_sha1(secret, seed, key_sizes)
518		self.symmetric_cryptography.Signer = SignerAesCbc(sigkey)
519		self.symmetric_cryptography.Encryptor = EncryptorAesCbc(key, init_vec)
520
521		def make_remote_symmetric_key(self, secret, seed):
522
523		# specs part 6, 6.7.5
524		key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)
525
526		(sigkey, key, init_vec) = uacrypto.p_sha1(secret, seed, key_sizes)
527		self.symmetric_cryptography.Verifier = VerifierAesCbc(sigkey)
528		self.symmetric_cryptography.Decryptor = DecryptorAesCbc(key, init_vec)
529
530
531		class SecurityPolicyBasic256Sha256(SecurityPolicy):
532		"""
533		Security Basic 256Sha256
534		A suite of algorithms that uses Sha256 as Key-Wrap-algorithm
535		and 256-Bit (32 bytes) for encryption algorithms.
536
537		- SymmetricSignatureAlgorithm_HMAC-SHA2-256
538		https://tools.ietf.org/html/rfc4634
539		- SymmetricEncryptionAlgorithm_AES256-CBC
540		http://www.w3.org/2001/04/xmlenc#aes256-cbc
541		- AsymmetricSignatureAlgorithm_RSA-PKCS15-SHA2-256
542		http://www.w3.org/2001/04/xmldsig-more#rsa-sha256
543		- AsymmetricEncryptionAlgorithm_RSA-OAEP-SHA1
544		http://www.w3.org/2001/04/xmlenc#rsa-oaep
545		- KeyDerivationAlgorithm_P-SHA2-256
546		http://docs.oasis-open.org/ws-sx/ws-secureconversation/200512/dk/p_sha256
547		- CertificateSignatureAlgorithm_RSA-PKCS15-SHA2-256
548		http://www.w3.org/2001/04/xmldsig-more#rsa-sha256
549		- Basic256Sha256_Limits
550		-> DerivedSignatureKeyLength: 256 bits
551		-> MinAsymmetricKeyLength: 2048 bits
552		-> MaxAsymmetricKeyLength: 4096 bits
553		-> SecureChannelNonceLength: 32 bytes
554		"""
555
556		URI = "http://opcfoundation.org/UA/SecurityPolicy#Basic256Sha256"
557		signature_key_size = 32
558		symmetric_key_size = 32
559		AsymmetricEncryptionURI = "http://www.w3.org/2001/04/xmlenc#rsa-oaep"
560		AsymmetricSignatureURI = "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256"
561
562		@staticmethod
563		def encrypt_asymmetric(pubkey, data):
564		return uacrypto.encrypt_rsa_oaep(pubkey, data)
565
566		def __init__(self, peer_cert, host_cert, client_pk, mode,
567		permission_ruleset=None):
568		require_cryptography(self)
569		if isinstance(peer_cert, bytes):
570		peer_cert = uacrypto.x509_from_der(peer_cert)
571		# even in Sign mode we need to asymmetrically encrypt secrets
572		# transmitted in OpenSecureChannel. So SignAndEncrypt here
573		self.asymmetric_cryptography = Cryptography(
574		MessageSecurityMode.SignAndEncrypt)
575		self.asymmetric_cryptography.Signer = SignerSha256(client_pk)
576		self.asymmetric_cryptography.Verifier = VerifierSha256(peer_cert)
577		self.asymmetric_cryptography.Encryptor = EncryptorRsa(
578		peer_cert, uacrypto.encrypt_rsa_oaep, 42)
579		self.asymmetric_cryptography.Decryptor = DecryptorRsa(
580		client_pk, uacrypto.decrypt_rsa_oaep, 42)
581		self.symmetric_cryptography = Cryptography(mode)
582		self.Mode = mode
583		self.peer_certificate = uacrypto.der_from_x509(peer_cert)
584		self.host_certificate = uacrypto.der_from_x509(host_cert)
585		if permission_ruleset is None:
586		from asyncua.crypto.permission_rules import SimpleRoleRuleset
587		permission_ruleset = SimpleRoleRuleset()
588
589		self.permissions = permission_ruleset
590
591		def make_local_symmetric_key(self, secret, seed):
592		# specs part 6, 6.7.5
593		key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)
594
595		(sigkey, key, init_vec) = uacrypto.p_sha256(secret, seed, key_sizes)
596		self.symmetric_cryptography.Signer = SignerHMac256(sigkey)
597		self.symmetric_cryptography.Encryptor = EncryptorAesCbc(key, init_vec)
598
599		def make_remote_symmetric_key(self, secret, seed):
600
601		# specs part 6, 6.7.5
602		key_sizes = (self.signature_key_size, self.symmetric_key_size, 16)
603
604		(sigkey, key, init_vec) = uacrypto.p_sha256(secret, seed, key_sizes)
605		self.symmetric_cryptography.Verifier = VerifierHMac256(sigkey)
606		self.symmetric_cryptography.Decryptor = DecryptorAesCbc(key, init_vec)
607
608
609		def encrypt_asymmetric(pubkey, data, policy_uri):
610		"""
611		Encrypt data with pubkey using an asymmetric algorithm.
612		The algorithm is selected by policy_uri.
613		Returns a tuple (encrypted_data, algorithm_uri)
614		"""
615		for cls in [SecurityPolicyBasic256Sha256, SecurityPolicyBasic256, SecurityPolicyBasic128Rsa15]:
616		if policy_uri == cls.URI:
617		return (cls.encrypt_asymmetric(pubkey, data),
618		cls.AsymmetricEncryptionURI)
619		if not policy_uri or policy_uri == POLICY_NONE_URI:
620		return data, ''
621		raise UaError(f"Unsupported security policy `{policy_uri}`")
622

FreeOpcUa / opcua-asyncio

SecurityPolicyBasic256.make_local_symmetric_key() A last analyzed 2020-12-22 17:52 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

SecurityPolicyBasic256.make_local_symmetric_key() A
last analyzed 2020-12-22 17:52 UTC