Merge pull request #2 from Mastercard/feature/first_release

Code commit and tests

Author: pqlMC

client_encryption/api_encryption.py

@@ -0,0 +1,90 @@
+import json
+from functools import wraps
+from client_encryption.field_level_encryption_config import FieldLevelEncryptionConfig
+from client_encryption.session_key_params import SessionKeyParams
+from client_encryption.field_level_encryption import encrypt_payload, decrypt_payload
+
+
+class ApiEncryption(object):
+
+    def __init__(self, encryption_conf_file):
+        """Load and initialize FieldLevelEncryptionConfig object."""
+
+        if type(encryption_conf_file) is dict:
+            self._encryption_conf = FieldLevelEncryptionConfig(encryption_conf_file)
+        else:
+            with open(encryption_conf_file, encoding='utf-8') as json_file:
+                self._encryption_conf = FieldLevelEncryptionConfig(json_file.read())
+
+    def field_encryption(self, func):
+        """Decorator for API request. func is APIClient.request"""
+
+        @wraps(func)
+        def request_function(*args, **kwargs):
+            """Wrap request and add field encryption layer to it."""
+
+            in_body = kwargs.get("body", None)
+            kwargs["body"] = self._encrypt_payload(kwargs.get("headers", None), in_body) if in_body else in_body
+
+            response = func(*args, **kwargs)
+
+            if type(response.data) is not str:
+                response_body = self._decrypt_payload(response.getheaders(), response.json())
+                response._content = json.dumps(response_body, indent=4).encode('utf-8')
+
+            return response
+
+        return request_function
+
+    def _encrypt_payload(self, headers, body):
+        """Encryption enforcement based on configuration - encrypt and add session key params to header or body"""
+
+        conf = self._encryption_conf
+
+        if conf.use_http_headers:
+            params = SessionKeyParams.generate(conf)
+
+            headers[conf.iv_field_name] = params.iv_value
+            headers[conf.encrypted_key_field_name] = params.encrypted_key_value
+            headers[conf.encryption_certificate_fingerprint_field_name] = conf.encryption_certificate_fingerprint
+            headers[conf.encryption_key_fingerprint_field_name] = conf.encryption_key_fingerprint
+            headers[conf.oaep_padding_digest_algorithm_field_name] = conf.oaep_padding_digest_algorithm
+
+            encrypted_payload = encrypt_payload(body, conf, params)
+        else:
+            encrypted_payload = encrypt_payload(body, conf)
+
+        return encrypted_payload
+
+    def _decrypt_payload(self, headers, body):
+        """Encryption enforcement based on configuration - decrypt using session key params from header or body"""
+
+        conf = self._encryption_conf
+
+        if conf.use_http_headers:
+            if conf.iv_field_name in headers and conf.encrypted_key_field_name in headers:
+                iv = headers.pop(conf.iv_field_name)
+                encrypted_key = headers.pop(conf.encrypted_key_field_name)
+                oaep_digest_algo = headers.pop(conf.oaep_padding_digest_algorithm_field_name) \
+                    if conf.oaep_padding_digest_algorithm_field_name in headers else None
+                if conf.encryption_certificate_fingerprint_field_name in headers:
+                    del headers[conf.encryption_certificate_fingerprint_field_name]
+                if conf.encryption_key_fingerprint_field_name in headers:
+                    del headers[conf.encryption_key_fingerprint_field_name]
+
+                params = SessionKeyParams(conf, encrypted_key, iv, oaep_digest_algo)
+                payload = decrypt_payload(body, conf, params)
+            else:
+                # skip decryption if not iv nor key is in headers
+                payload = body
+        else:
+            payload = decrypt_payload(body, conf)
+
+        return payload
+
+
+def add_encryption_layer(api_client, encryption_conf_file):
+    """Decorate APIClient.request with field level encryption"""
+
+    api_encryption = ApiEncryption(encryption_conf_file)
+    api_client.request = api_encryption.field_encryption(api_client.request)

client_encryption/encoding_utils.py

@@ -0,0 +1,40 @@
+import base64
+from enum import Enum
+from client_encryption.encryption_exception import EncodingError
+
+
+def encode_bytes(_bytes, encoding):
+    """Encode byte sequence to Hex or Base64."""
+
+    if type(_bytes) is bytes:
+        if encoding == Encoding.HEX:
+            encoded = _bytes.hex()
+        elif encoding == Encoding.BASE64:
+            encoded = base64.b64encode(_bytes).decode('utf-8')
+        else:
+            raise EncodingError("Encode: Invalid encoding.")
+
+        return encoded
+    else:
+        raise ValueError("Encode: Invalid or missing input bytes.")
+
+
+def decode_value(value, encoding):
+    """Decode Hex or Base64 string to byte sequence."""
+
+    if type(value) is str:
+        if encoding == Encoding.HEX:
+            decoded = bytes.fromhex(value)
+        elif encoding == Encoding.BASE64:
+            decoded = base64.b64decode(value)
+        else:
+            raise EncodingError("Decode: Invalid encoding.")
+
+        return decoded
+    else:
+        raise ValueError("Decode: Invalid or missing input string.")
+
+
+class Encoding(Enum):
+    BASE64 = 'BASE64'
+    HEX = 'HEX'

client_encryption/encryption_exception.py

@@ -0,0 +1,28 @@
+class EncryptionError(Exception):
+    """Encryption related exception for client-encryption module."""
+    pass
+
+
+class EncodingError(Exception):
+    """Encoding not supported or invalid."""
+    pass
+
+
+class CertificateError(Exception):
+    """Certificate exception for client-encryption module."""
+    pass
+
+
+class PrivateKeyError(Exception):
+    """Private key exception for client-encryption module."""
+    pass
+
+
+class HashAlgorithmError(Exception):
+    """Hash algorithm exception for client-encryption module."""
+    pass
+
+
+class KeyWrappingError(Exception):
+    """Encryption exception on wrapping/unwrapping session key for client-encryption module."""
+    pass

client_encryption/encryption_utils.py

@@ -0,0 +1,68 @@
+from Crypto.PublicKey import RSA
+from Crypto.Hash import SHA1, SHA224, SHA256, SHA384, SHA512
+from OpenSSL.crypto import load_certificate, load_pkcs12, dump_privatekey, FILETYPE_PEM, FILETYPE_ASN1, Error
+from client_encryption.encryption_exception import CertificateError, PrivateKeyError, HashAlgorithmError
+
+
+_SUPPORTED_HASH = {"SHA1": SHA1, "SHA224": SHA224, "SHA256": SHA256, "SHA384": SHA384, "SHA512": SHA512}
+
+
+def load_encryption_certificate(certificate_path):
+    """Load X509 encryption certificate data at the given file path."""
+
+    try:
+        with open(certificate_path, "rb") as cert_content:
+            certificate = cert_content.read()
+            x509 = load_certificate(__get_crypto_file_type(certificate), certificate)
+
+        return x509
+    except IOError:
+        raise CertificateError("Unable to load certificate.")
+    except (ValueError, Error):
+        raise CertificateError("Wrong encryption certificate format.")
+
+
+def load_decryption_key(key_file_path, decryption_key_password=None):
+    """Load a RSA decryption key."""
+
+    try:
+        with open(key_file_path, "rb") as key_content:
+            private_key = key_content.read()
+
+            # if key format is p12 (decryption_key_password is populated) then we have to retrieve the private key
+            if decryption_key_password is not None:
+                private_key = __load_pkcs12_private_key(private_key, decryption_key_password)
+
+        return RSA.importKey(private_key)
+    except (Error, IOError):
+        raise PrivateKeyError("Unable to load key file.")
+    except ValueError:
+        raise PrivateKeyError("Wrong decryption key format.")
+
+
+def __load_pkcs12_private_key(pkcs12_key, password):
+    """Load a private key in ASN1 format out of a PKCS#12 container."""
+
+    pkcs12 = load_pkcs12(pkcs12_key, password.encode("utf-8")).get_privatekey()
+    return dump_privatekey(FILETYPE_ASN1, pkcs12)
+
+
+def __get_crypto_file_type(file_content):
+    if file_content.startswith(b"-----BEGIN "):
+        return FILETYPE_PEM
+    else:
+        return FILETYPE_ASN1
+
+
+def load_hash_algorithm(algo_str):
+    """Load a hash algorithm object of Crypto.Hash from a list of supported ones."""
+
+    if algo_str:
+        algo_key = algo_str.replace("-", "").upper()
+
+        if algo_key in _SUPPORTED_HASH:
+            return _SUPPORTED_HASH[algo_key]
+        else:
+            raise HashAlgorithmError("Hash algorithm invalid or not supported.")
+    else:
+        raise HashAlgorithmError("No hash algorithm provided.")

client_encryption/field_level_encryption.py

@@ -0,0 +1,126 @@
+import json
+import copy
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import pad, unpad
+from client_encryption.session_key_params import SessionKeyParams
+from client_encryption.encoding_utils import encode_bytes, decode_value
+from client_encryption.json_path_utils import get_node, pop_node, update_node, cleanup_node
+from client_encryption.encryption_exception import EncryptionError
+
+
+def encrypt_payload(payload, config, _params=None):
+    """Encrypt some fields of a JSON payload using the given configuration."""
+
+    try:
+        if type(payload) is str:
+            json_payload = json.loads(payload)
+        else:
+            json_payload = copy.deepcopy(payload)
+
+        for elem, target in config.paths["$"].to_encrypt.items():
+            if not _params:
+                params = SessionKeyParams.generate(config)
+            else:
+                params = _params
+
+            try:
+                value = pop_node(json_payload, elem)
+
+                try:
+                    encrypted_value = _encrypt_value(params.key, params.iv_spec, value)
+                    crypto_node = get_node(json_payload, target, create=True)
+                    crypto_node[config.encrypted_value_field_name] = encode_bytes(encrypted_value, config.data_encoding)
+
+                    if not _params:
+                        _populate_node_with_key_params(crypto_node, config, params)
+
+                except KeyError:
+                    raise EncryptionError("Field " + target + " not found!")
+
+            except KeyError:
+                pass  # data-to-encrypt node not found, nothing to encrypt
+
+        return json_payload
+
+    except (IOError, ValueError, TypeError) as e:
+        raise EncryptionError("Payload encryption failed!", e)
+
+
+def decrypt_payload(payload, config, _params=None):
+    """Decrypt some fields of a JSON payload using the given configuration."""
+
+    try:
+        if type(payload) is str:
+            json_payload = json.loads(payload)
+        else:
+            json_payload = payload
+
+        for elem, target in config.paths["$"].to_decrypt.items():
+            try:
+                node = get_node(json_payload, elem)
+
+                cipher_text = decode_value(node.pop(config.encrypted_value_field_name), config.data_encoding)
+
+                if not _params:
+                    try:
+                        encrypted_key = node.pop(config.encrypted_key_field_name)
+                        iv = node.pop(config.iv_field_name)
+                    except KeyError:
+                        raise EncryptionError("Encryption field(s) missing in payload.")
+
+                    oaep_digest_algo = node.pop(config.oaep_padding_digest_algorithm_field_name,
+                                                config.oaep_padding_digest_algorithm)
+
+                    _remove_fingerprint_from_node(node, config)
+
+                    params = SessionKeyParams(config, encrypted_key, iv, oaep_digest_algo)
+                else:
+                    params = _params
+
+                cleanup_node(json_payload, elem)
+
+                try:
+                    update_node(json_payload, target, _decrypt_bytes(params.key, params.iv_spec, cipher_text))
+                except KeyError:
+                    raise EncryptionError("Field '" + target + "' not found!")
+
+            except KeyError:
+                pass  # encrypted data node not found, nothing to decrypt
+
+        return json_payload
+
+    except (IOError, ValueError, TypeError) as e:
+        raise EncryptionError("Payload encryption failed!", e)
+
+
+def _encrypt_value(_key, iv, node_str):
+    padded_node = pad(node_str.encode('utf-8'), AES.block_size)
+
+    aes = AES.new(_key, AES.MODE_CBC, iv)
+    return aes.encrypt(padded_node)
+
+
+def _decrypt_bytes(_key, iv, _bytes):
+    aes = AES.new(_key, AES.MODE_CBC, iv)
+    plain_bytes = aes.decrypt(_bytes)
+
+    return unpad(plain_bytes, AES.block_size).decode('utf-8')
+
+
+def _populate_node_with_key_params(node, config, params):
+    node[config.encrypted_key_field_name] = params.encrypted_key_value
+    node[config.iv_field_name] = params.iv_value
+    if config.oaep_padding_digest_algorithm_field_name:
+        node[config.oaep_padding_digest_algorithm_field_name] = params.oaep_padding_digest_algorithm_value
+    if config.encryption_certificate_fingerprint_field_name:
+        node[config.encryption_certificate_fingerprint_field_name] = config.encryption_certificate_fingerprint
+    if config.encryption_key_fingerprint_field_name:
+        node[config.encryption_key_fingerprint_field_name] = config.encryption_key_fingerprint
+
+
+def _remove_fingerprint_from_node(node, config):
+    if config.encryption_certificate_fingerprint_field_name in node:
+        del node[config.encryption_certificate_fingerprint_field_name]
+    if config.encryption_key_fingerprint_field_name in node:
+        del node[config.encryption_key_fingerprint_field_name]
+