slight update

This commit is contained in:
2024-12-01 04:19:04 +09:00
parent 00b0afd17a
commit 4dbe1bee11
3130 changed files with 508084 additions and 0 deletions

View File

@ -0,0 +1,189 @@
#
# Util/PEM.py : Privacy Enhanced Mail utilities
#
# ===================================================================
#
# Copyright (c) 2014, Legrandin <helderijs@gmail.com>
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in
# the documentation and/or other materials provided with the
# distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
# ===================================================================
__all__ = ['encode', 'decode']
import re
from binascii import a2b_base64, b2a_base64, hexlify, unhexlify
from Crypto.Hash import MD5
from Crypto.Util.Padding import pad, unpad
from Crypto.Cipher import DES, DES3, AES
from Crypto.Protocol.KDF import PBKDF1
from Crypto.Random import get_random_bytes
from Crypto.Util.py3compat import tobytes, tostr
def encode(data, marker, passphrase=None, randfunc=None):
"""Encode a piece of binary data into PEM format.
Args:
data (byte string):
The piece of binary data to encode.
marker (string):
The marker for the PEM block (e.g. "PUBLIC KEY").
Note that there is no official master list for all allowed markers.
Still, you can refer to the OpenSSL_ source code.
passphrase (byte string):
If given, the PEM block will be encrypted. The key is derived from
the passphrase.
randfunc (callable):
Random number generation function; it accepts an integer N and returns
a byte string of random data, N bytes long. If not given, a new one is
instantiated.
Returns:
The PEM block, as a string.
.. _OpenSSL: https://github.com/openssl/openssl/blob/master/include/openssl/pem.h
"""
if randfunc is None:
randfunc = get_random_bytes
out = "-----BEGIN %s-----\n" % marker
if passphrase:
# We only support 3DES for encryption
salt = randfunc(8)
key = PBKDF1(passphrase, salt, 16, 1, MD5)
key += PBKDF1(key + passphrase, salt, 8, 1, MD5)
objenc = DES3.new(key, DES3.MODE_CBC, salt)
out += "Proc-Type: 4,ENCRYPTED\nDEK-Info: DES-EDE3-CBC,%s\n\n" %\
tostr(hexlify(salt).upper())
# Encrypt with PKCS#7 padding
data = objenc.encrypt(pad(data, objenc.block_size))
elif passphrase is not None:
raise ValueError("Empty password")
# Each BASE64 line can take up to 64 characters (=48 bytes of data)
# b2a_base64 adds a new line character!
chunks = [tostr(b2a_base64(data[i:i + 48]))
for i in range(0, len(data), 48)]
out += "".join(chunks)
out += "-----END %s-----" % marker
return out
def _EVP_BytesToKey(data, salt, key_len):
d = [ b'' ]
m = (key_len + 15 ) // 16
for _ in range(m):
nd = MD5.new(d[-1] + data + salt).digest()
d.append(nd)
return b"".join(d)[:key_len]
def decode(pem_data, passphrase=None):
"""Decode a PEM block into binary.
Args:
pem_data (string):
The PEM block.
passphrase (byte string):
If given and the PEM block is encrypted,
the key will be derived from the passphrase.
Returns:
A tuple with the binary data, the marker string, and a boolean to
indicate if decryption was performed.
Raises:
ValueError: if decoding fails, if the PEM file is encrypted and no passphrase has
been provided or if the passphrase is incorrect.
"""
# Verify Pre-Encapsulation Boundary
r = re.compile(r"\s*-----BEGIN (.*)-----\s+")
m = r.match(pem_data)
if not m:
raise ValueError("Not a valid PEM pre boundary")
marker = m.group(1)
# Verify Post-Encapsulation Boundary
r = re.compile(r"-----END (.*)-----\s*$")
m = r.search(pem_data)
if not m or m.group(1) != marker:
raise ValueError("Not a valid PEM post boundary")
# Removes spaces and slit on lines
lines = pem_data.replace(" ", '').split()
# Decrypts, if necessary
if lines[1].startswith('Proc-Type:4,ENCRYPTED'):
if not passphrase:
raise ValueError("PEM is encrypted, but no passphrase available")
DEK = lines[2].split(':')
if len(DEK) != 2 or DEK[0] != 'DEK-Info':
raise ValueError("PEM encryption format not supported.")
algo, salt = DEK[1].split(',')
salt = unhexlify(tobytes(salt))
padding = True
if algo == "DES-CBC":
key = _EVP_BytesToKey(passphrase, salt, 8)
objdec = DES.new(key, DES.MODE_CBC, salt)
elif algo == "DES-EDE3-CBC":
key = _EVP_BytesToKey(passphrase, salt, 24)
objdec = DES3.new(key, DES3.MODE_CBC, salt)
elif algo == "AES-128-CBC":
key = _EVP_BytesToKey(passphrase, salt[:8], 16)
objdec = AES.new(key, AES.MODE_CBC, salt)
elif algo == "AES-192-CBC":
key = _EVP_BytesToKey(passphrase, salt[:8], 24)
objdec = AES.new(key, AES.MODE_CBC, salt)
elif algo == "AES-256-CBC":
key = _EVP_BytesToKey(passphrase, salt[:8], 32)
objdec = AES.new(key, AES.MODE_CBC, salt)
elif algo.lower() == "id-aes256-gcm":
key = _EVP_BytesToKey(passphrase, salt[:8], 32)
objdec = AES.new(key, AES.MODE_GCM, nonce=salt)
padding = False
else:
raise ValueError("Unsupport PEM encryption algorithm (%s)." % algo)
lines = lines[2:]
else:
objdec = None
# Decode body
data = a2b_base64(''.join(lines[1:-1]))
enc_flag = False
if objdec:
if padding:
data = unpad(objdec.decrypt(data), objdec.block_size)
else:
# There is no tag, so we don't use decrypt_and_verify
data = objdec.decrypt(data)
enc_flag = True
return (data, marker, enc_flag)

View File

@ -0,0 +1,10 @@
from typing import Tuple, Optional, Callable
def encode(data: bytes,
marke: str,
passphrase: Optional[bytes] = ...,
randfunc: Optional[Callable[[int],bytes]] = ...) -> str: ...
def decode(pem_data: str,
passphrase: Optional[bytes] = ...) -> Tuple[bytes, str, bool]: ...

View File

@ -0,0 +1,226 @@
#
# PublicKey/PKCS8.py : PKCS#8 functions
#
# ===================================================================
#
# Copyright (c) 2014, Legrandin <helderijs@gmail.com>
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in
# the documentation and/or other materials provided with the
# distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
# ===================================================================
from Crypto.Util.py3compat import *
from Crypto.Util.asn1 import (
DerNull,
DerSequence,
DerObjectId,
DerOctetString,
)
from Crypto.IO._PBES import PBES1, PBES2, PbesError
__all__ = ['wrap', 'unwrap']
def wrap(private_key, key_oid, passphrase=None, protection=None,
prot_params=None, key_params=DerNull(), randfunc=None):
"""Wrap a private key into a PKCS#8 blob (clear or encrypted).
Args:
private_key (bytes):
The private key encoded in binary form. The actual encoding is
algorithm specific. In most cases, it is DER.
key_oid (string):
The object identifier (OID) of the private key to wrap.
It is a dotted string, like ``'1.2.840.113549.1.1.1'`` (for RSA keys)
or ``'1.2.840.10045.2.1'`` (for ECC keys).
Keyword Args:
passphrase (bytes or string):
The secret passphrase from which the wrapping key is derived.
Set it only if encryption is required.
protection (string):
The identifier of the algorithm to use for securely wrapping the key.
Refer to :ref:`the encryption parameters<enc_params>` .
The default value is ``'PBKDF2WithHMAC-SHA1AndDES-EDE3-CBC'``.
prot_params (dictionary):
Parameters for the key derivation function (KDF).
Refer to :ref:`the encryption parameters<enc_params>` .
key_params (DER object or None):
The ``parameters`` field to use in the ``AlgorithmIdentifier``
SEQUENCE. If ``None``, no ``parameters`` field will be added.
By default, the ASN.1 type ``NULL`` is used.
randfunc (callable):
Random number generation function; it should accept a single integer
N and return a string of random data, N bytes long.
If not specified, a new RNG will be instantiated
from :mod:`Crypto.Random`.
Returns:
bytes: The PKCS#8-wrapped private key (possibly encrypted).
"""
#
# PrivateKeyInfo ::= SEQUENCE {
# version Version,
# privateKeyAlgorithm PrivateKeyAlgorithmIdentifier,
# privateKey PrivateKey,
# attributes [0] IMPLICIT Attributes OPTIONAL
# }
#
if key_params is None:
algorithm = DerSequence([DerObjectId(key_oid)])
else:
algorithm = DerSequence([DerObjectId(key_oid), key_params])
pk_info = DerSequence([
0,
algorithm,
DerOctetString(private_key)
])
pk_info_der = pk_info.encode()
if passphrase is None:
return pk_info_der
if not passphrase:
raise ValueError("Empty passphrase")
# Encryption with PBES2
passphrase = tobytes(passphrase)
if protection is None:
protection = 'PBKDF2WithHMAC-SHA1AndDES-EDE3-CBC'
return PBES2.encrypt(pk_info_der, passphrase,
protection, prot_params, randfunc)
def unwrap(p8_private_key, passphrase=None):
"""Unwrap a private key from a PKCS#8 blob (clear or encrypted).
Args:
p8_private_key (bytes):
The private key wrapped into a PKCS#8 container, DER encoded.
Keyword Args:
passphrase (byte string or string):
The passphrase to use to decrypt the blob (if it is encrypted).
Return:
A tuple containing
#. the algorithm identifier of the wrapped key (OID, dotted string)
#. the private key (bytes, DER encoded)
#. the associated parameters (bytes, DER encoded) or ``None``
Raises:
ValueError : if decoding fails
"""
if passphrase is not None:
passphrase = tobytes(passphrase)
found = False
try:
p8_private_key = PBES1.decrypt(p8_private_key, passphrase)
found = True
except PbesError as e:
error_str = "PBES1[%s]" % str(e)
except ValueError:
error_str = "PBES1[Invalid]"
if not found:
try:
p8_private_key = PBES2.decrypt(p8_private_key, passphrase)
found = True
except PbesError as e:
error_str += ",PBES2[%s]" % str(e)
except ValueError:
error_str += ",PBES2[Invalid]"
if not found:
raise ValueError("Error decoding PKCS#8 (%s)" % error_str)
pk_info = DerSequence().decode(p8_private_key, nr_elements=(2, 3, 4, 5))
if len(pk_info) == 2 and not passphrase:
raise ValueError("Not a valid clear PKCS#8 structure "
"(maybe it is encrypted?)")
# RFC5208, PKCS#8, version is v1(0)
#
# PrivateKeyInfo ::= SEQUENCE {
# version Version,
# privateKeyAlgorithm PrivateKeyAlgorithmIdentifier,
# privateKey PrivateKey,
# attributes [0] IMPLICIT Attributes OPTIONAL
# }
#
# RFC5915, Asymmetric Key Package, version is v2(1)
#
# OneAsymmetricKey ::= SEQUENCE {
# version Version,
# privateKeyAlgorithm PrivateKeyAlgorithmIdentifier,
# privateKey PrivateKey,
# attributes [0] Attributes OPTIONAL,
# ...,
# [[2: publicKey [1] PublicKey OPTIONAL ]],
# ...
# }
if pk_info[0] == 0:
if len(pk_info) not in (3, 4):
raise ValueError("Not a valid PrivateKeyInfo SEQUENCE")
elif pk_info[0] == 1:
if len(pk_info) not in (3, 4, 5):
raise ValueError("Not a valid PrivateKeyInfo SEQUENCE")
else:
raise ValueError("Not a valid PrivateKeyInfo SEQUENCE")
algo = DerSequence().decode(pk_info[1], nr_elements=(1, 2))
algo_oid = DerObjectId().decode(algo[0]).value
if len(algo) == 1:
algo_params = None
else:
try:
DerNull().decode(algo[1])
algo_params = None
except:
algo_params = algo[1]
# PrivateKey ::= OCTET STRING
private_key = DerOctetString().decode(pk_info[2]).payload
# We ignore attributes and (for v2 only) publickey
return (algo_oid, private_key, algo_params)

View File

@ -0,0 +1,17 @@
from typing import Tuple, Optional, Union, Callable
from typing_extensions import NotRequired
from Crypto.Util.asn1 import DerObject
from Crypto.IO._PBES import ProtParams
def wrap(private_key: bytes,
key_oid: str,
passphrase: Union[bytes, str] = ...,
protection: str = ...,
prot_params: Optional[ProtParams] = ...,
key_params: Optional[DerObject] = ...,
randfunc: Optional[Callable[[int], str]] = ...) -> bytes: ...
def unwrap(p8_private_key: bytes, passphrase: Optional[Union[bytes, str]] = ...) -> Tuple[str, bytes, Optional[bytes]]: ...

View File

@ -0,0 +1,546 @@
#
# PublicKey/_PBES.py : Password-Based Encryption functions
#
# ===================================================================
#
# Copyright (c) 2014, Legrandin <helderijs@gmail.com>
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in
# the documentation and/or other materials provided with the
# distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
# ===================================================================
import re
from Crypto import Hash
from Crypto import Random
from Crypto.Util.asn1 import (
DerSequence, DerOctetString,
DerObjectId, DerInteger,
)
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.Protocol.KDF import PBKDF1, PBKDF2, scrypt
_OID_PBE_WITH_MD5_AND_DES_CBC = "1.2.840.113549.1.5.3"
_OID_PBE_WITH_MD5_AND_RC2_CBC = "1.2.840.113549.1.5.6"
_OID_PBE_WITH_SHA1_AND_DES_CBC = "1.2.840.113549.1.5.10"
_OID_PBE_WITH_SHA1_AND_RC2_CBC = "1.2.840.113549.1.5.11"
_OID_PBES2 = "1.2.840.113549.1.5.13"
_OID_PBKDF2 = "1.2.840.113549.1.5.12"
_OID_SCRYPT = "1.3.6.1.4.1.11591.4.11"
_OID_HMAC_SHA1 = "1.2.840.113549.2.7"
_OID_DES_EDE3_CBC = "1.2.840.113549.3.7"
_OID_AES128_CBC = "2.16.840.1.101.3.4.1.2"
_OID_AES192_CBC = "2.16.840.1.101.3.4.1.22"
_OID_AES256_CBC = "2.16.840.1.101.3.4.1.42"
_OID_AES128_GCM = "2.16.840.1.101.3.4.1.6"
_OID_AES192_GCM = "2.16.840.1.101.3.4.1.26"
_OID_AES256_GCM = "2.16.840.1.101.3.4.1.46"
class PbesError(ValueError):
pass
# These are the ASN.1 definitions used by the PBES1/2 logic:
#
# EncryptedPrivateKeyInfo ::= SEQUENCE {
# encryptionAlgorithm EncryptionAlgorithmIdentifier,
# encryptedData EncryptedData
# }
#
# EncryptionAlgorithmIdentifier ::= AlgorithmIdentifier
#
# EncryptedData ::= OCTET STRING
#
# AlgorithmIdentifier ::= SEQUENCE {
# algorithm OBJECT IDENTIFIER,
# parameters ANY DEFINED BY algorithm OPTIONAL
# }
#
# PBEParameter ::= SEQUENCE {
# salt OCTET STRING (SIZE(8)),
# iterationCount INTEGER
# }
#
# PBES2-params ::= SEQUENCE {
# keyDerivationFunc AlgorithmIdentifier {{PBES2-KDFs}},
# encryptionScheme AlgorithmIdentifier {{PBES2-Encs}}
# }
#
# PBKDF2-params ::= SEQUENCE {
# salt CHOICE {
# specified OCTET STRING,
# otherSource AlgorithmIdentifier {{PBKDF2-SaltSources}}
# },
# iterationCount INTEGER (1..MAX),
# keyLength INTEGER (1..MAX) OPTIONAL,
# prf AlgorithmIdentifier {{PBKDF2-PRFs}} DEFAULT algid-hmacWithSHA1
# }
#
# PBKDF2-PRFs ALGORITHM-IDENTIFIER ::= {
# {NULL IDENTIFIED BY id-hmacWithSHA1},
# {NULL IDENTIFIED BY id-hmacWithSHA224},
# {NULL IDENTIFIED BY id-hmacWithSHA256},
# {NULL IDENTIFIED BY id-hmacWithSHA384},
# {NULL IDENTIFIED BY id-hmacWithSHA512},
# {NULL IDENTIFIED BY id-hmacWithSHA512-224},
# {NULL IDENTIFIED BY id-hmacWithSHA512-256},
# ...
# }
# scrypt-params ::= SEQUENCE {
# salt OCTET STRING,
# costParameter INTEGER (1..MAX),
# blockSize INTEGER (1..MAX),
# parallelizationParameter INTEGER (1..MAX),
# keyLength INTEGER (1..MAX) OPTIONAL
# }
class PBES1(object):
"""Deprecated encryption scheme with password-based key derivation
(originally defined in PKCS#5 v1.5, but still present in `v2.0`__).
.. __: http://www.ietf.org/rfc/rfc2898.txt
"""
@staticmethod
def decrypt(data, passphrase):
"""Decrypt a piece of data using a passphrase and *PBES1*.
The algorithm to use is automatically detected.
:Parameters:
data : byte string
The piece of data to decrypt.
passphrase : byte string
The passphrase to use for decrypting the data.
:Returns:
The decrypted data, as a binary string.
"""
enc_private_key_info = DerSequence().decode(data)
encrypted_algorithm = DerSequence().decode(enc_private_key_info[0])
encrypted_data = DerOctetString().decode(enc_private_key_info[1]).payload
pbe_oid = DerObjectId().decode(encrypted_algorithm[0]).value
cipher_params = {}
if pbe_oid == _OID_PBE_WITH_MD5_AND_DES_CBC:
# PBE_MD5_DES_CBC
from Crypto.Hash import MD5
from Crypto.Cipher import DES
hashmod = MD5
module = DES
elif pbe_oid == _OID_PBE_WITH_MD5_AND_RC2_CBC:
# PBE_MD5_RC2_CBC
from Crypto.Hash import MD5
from Crypto.Cipher import ARC2
hashmod = MD5
module = ARC2
cipher_params['effective_keylen'] = 64
elif pbe_oid == _OID_PBE_WITH_SHA1_AND_DES_CBC:
# PBE_SHA1_DES_CBC
from Crypto.Hash import SHA1
from Crypto.Cipher import DES
hashmod = SHA1
module = DES
elif pbe_oid == _OID_PBE_WITH_SHA1_AND_RC2_CBC:
# PBE_SHA1_RC2_CBC
from Crypto.Hash import SHA1
from Crypto.Cipher import ARC2
hashmod = SHA1
module = ARC2
cipher_params['effective_keylen'] = 64
else:
raise PbesError("Unknown OID for PBES1")
pbe_params = DerSequence().decode(encrypted_algorithm[1], nr_elements=2)
salt = DerOctetString().decode(pbe_params[0]).payload
iterations = pbe_params[1]
key_iv = PBKDF1(passphrase, salt, 16, iterations, hashmod)
key, iv = key_iv[:8], key_iv[8:]
cipher = module.new(key, module.MODE_CBC, iv, **cipher_params)
pt = cipher.decrypt(encrypted_data)
return unpad(pt, cipher.block_size)
class PBES2(object):
"""Encryption scheme with password-based key derivation
(defined in `PKCS#5 v2.0`__).
.. __: http://www.ietf.org/rfc/rfc2898.txt."""
@staticmethod
def encrypt(data, passphrase, protection, prot_params=None, randfunc=None):
"""Encrypt a piece of data using a passphrase and *PBES2*.
:Parameters:
data : byte string
The piece of data to encrypt.
passphrase : byte string
The passphrase to use for encrypting the data.
protection : string
The identifier of the encryption algorithm to use.
The default value is '``PBKDF2WithHMAC-SHA1AndDES-EDE3-CBC``'.
prot_params : dictionary
Parameters of the protection algorithm.
+------------------+-----------------------------------------------+
| Key | Description |
+==================+===============================================+
| iteration_count | The KDF algorithm is repeated several times to|
| | slow down brute force attacks on passwords |
| | (called *N* or CPU/memory cost in scrypt). |
| | |
| | The default value for PBKDF2 is 1 000. |
| | The default value for scrypt is 16 384. |
+------------------+-----------------------------------------------+
| salt_size | Salt is used to thwart dictionary and rainbow |
| | attacks on passwords. The default value is 8 |
| | bytes. |
+------------------+-----------------------------------------------+
| block_size | *(scrypt only)* Memory-cost (r). The default |
| | value is 8. |
+------------------+-----------------------------------------------+
| parallelization | *(scrypt only)* CPU-cost (p). The default |
| | value is 1. |
+------------------+-----------------------------------------------+
randfunc : callable
Random number generation function; it should accept
a single integer N and return a string of random data,
N bytes long. If not specified, a new RNG will be
instantiated from ``Crypto.Random``.
:Returns:
The encrypted data, as a binary string.
"""
if prot_params is None:
prot_params = {}
if randfunc is None:
randfunc = Random.new().read
pattern = re.compile(r'^(PBKDF2WithHMAC-([0-9A-Z-]+)|scrypt)And([0-9A-Z-]+)$')
res = pattern.match(protection)
if res is None:
raise ValueError("Unknown protection %s" % protection)
if protection.startswith("PBKDF"):
pbkdf = "pbkdf2"
pbkdf2_hmac_algo = res.group(2)
enc_algo = res.group(3)
else:
pbkdf = "scrypt"
enc_algo = res.group(3)
aead = False
if enc_algo == 'DES-EDE3-CBC':
from Crypto.Cipher import DES3
key_size = 24
module = DES3
cipher_mode = DES3.MODE_CBC
enc_oid = _OID_DES_EDE3_CBC
enc_param = {'iv': randfunc(8)}
elif enc_algo == 'AES128-CBC':
key_size = 16
module = AES
cipher_mode = AES.MODE_CBC
enc_oid = _OID_AES128_CBC
enc_param = {'iv': randfunc(16)}
elif enc_algo == 'AES192-CBC':
key_size = 24
module = AES
cipher_mode = AES.MODE_CBC
enc_oid = _OID_AES192_CBC
enc_param = {'iv': randfunc(16)}
elif enc_algo == 'AES256-CBC':
key_size = 32
module = AES
cipher_mode = AES.MODE_CBC
enc_oid = _OID_AES256_CBC
enc_param = {'iv': randfunc(16)}
elif enc_algo == 'AES128-GCM':
key_size = 16
module = AES
cipher_mode = AES.MODE_GCM
enc_oid = _OID_AES128_GCM
enc_param = {'nonce': randfunc(12)}
aead = True
elif enc_algo == 'AES192-GCM':
key_size = 24
module = AES
cipher_mode = AES.MODE_GCM
enc_oid = _OID_AES192_GCM
enc_param = {'nonce': randfunc(12)}
aead = True
elif enc_algo == 'AES256-GCM':
key_size = 32
module = AES
cipher_mode = AES.MODE_GCM
enc_oid = _OID_AES256_GCM
enc_param = {'nonce': randfunc(12)}
aead = True
else:
raise ValueError("Unknown encryption mode '%s'" % enc_algo)
iv_nonce = list(enc_param.values())[0]
salt = randfunc(prot_params.get("salt_size", 8))
# Derive key from password
if pbkdf == 'pbkdf2':
count = prot_params.get("iteration_count", 1000)
digestmod = Hash.new(pbkdf2_hmac_algo)
key = PBKDF2(passphrase,
salt,
key_size,
count,
hmac_hash_module=digestmod)
pbkdf2_params = DerSequence([
DerOctetString(salt),
DerInteger(count)
])
if pbkdf2_hmac_algo != 'SHA1':
try:
hmac_oid = Hash.HMAC.new(b'', digestmod=digestmod).oid
except KeyError:
raise ValueError("No OID for HMAC hash algorithm")
pbkdf2_params.append(DerSequence([DerObjectId(hmac_oid)]))
kdf_info = DerSequence([
DerObjectId(_OID_PBKDF2), # PBKDF2
pbkdf2_params
])
elif pbkdf == 'scrypt':
count = prot_params.get("iteration_count", 16384)
scrypt_r = prot_params.get('block_size', 8)
scrypt_p = prot_params.get('parallelization', 1)
key = scrypt(passphrase, salt, key_size,
count, scrypt_r, scrypt_p)
kdf_info = DerSequence([
DerObjectId(_OID_SCRYPT), # scrypt
DerSequence([
DerOctetString(salt),
DerInteger(count),
DerInteger(scrypt_r),
DerInteger(scrypt_p)
])
])
else:
raise ValueError("Unknown KDF " + res.group(1))
# Create cipher and use it
cipher = module.new(key, cipher_mode, **enc_param)
if aead:
ct, tag = cipher.encrypt_and_digest(data)
encrypted_data = ct + tag
else:
encrypted_data = cipher.encrypt(pad(data, cipher.block_size))
enc_info = DerSequence([
DerObjectId(enc_oid),
DerOctetString(iv_nonce)
])
# Result
enc_private_key_info = DerSequence([
# encryptionAlgorithm
DerSequence([
DerObjectId(_OID_PBES2),
DerSequence([
kdf_info,
enc_info
]),
]),
DerOctetString(encrypted_data)
])
return enc_private_key_info.encode()
@staticmethod
def decrypt(data, passphrase):
"""Decrypt a piece of data using a passphrase and *PBES2*.
The algorithm to use is automatically detected.
:Parameters:
data : byte string
The piece of data to decrypt.
passphrase : byte string
The passphrase to use for decrypting the data.
:Returns:
The decrypted data, as a binary string.
"""
enc_private_key_info = DerSequence().decode(data, nr_elements=2)
enc_algo = DerSequence().decode(enc_private_key_info[0])
encrypted_data = DerOctetString().decode(enc_private_key_info[1]).payload
pbe_oid = DerObjectId().decode(enc_algo[0]).value
if pbe_oid != _OID_PBES2:
raise PbesError("Not a PBES2 object")
pbes2_params = DerSequence().decode(enc_algo[1], nr_elements=2)
# Key Derivation Function selection
kdf_info = DerSequence().decode(pbes2_params[0], nr_elements=2)
kdf_oid = DerObjectId().decode(kdf_info[0]).value
kdf_key_length = None
# We only support PBKDF2 or scrypt
if kdf_oid == _OID_PBKDF2:
pbkdf2_params = DerSequence().decode(kdf_info[1], nr_elements=(2, 3, 4))
salt = DerOctetString().decode(pbkdf2_params[0]).payload
iteration_count = pbkdf2_params[1]
left = len(pbkdf2_params) - 2
idx = 2
if left > 0:
try:
# Check if it's an INTEGER
kdf_key_length = pbkdf2_params[idx] - 0
left -= 1
idx += 1
except TypeError:
# keyLength is not present
pass
# Default is HMAC-SHA1
pbkdf2_prf_oid = _OID_HMAC_SHA1
if left > 0:
pbkdf2_prf_algo_id = DerSequence().decode(pbkdf2_params[idx])
pbkdf2_prf_oid = DerObjectId().decode(pbkdf2_prf_algo_id[0]).value
elif kdf_oid == _OID_SCRYPT:
scrypt_params = DerSequence().decode(kdf_info[1], nr_elements=(4, 5))
salt = DerOctetString().decode(scrypt_params[0]).payload
iteration_count, scrypt_r, scrypt_p = [scrypt_params[x]
for x in (1, 2, 3)]
if len(scrypt_params) > 4:
kdf_key_length = scrypt_params[4]
else:
kdf_key_length = None
else:
raise PbesError("Unsupported PBES2 KDF")
# Cipher selection
enc_info = DerSequence().decode(pbes2_params[1])
enc_oid = DerObjectId().decode(enc_info[0]).value
aead = False
if enc_oid == _OID_DES_EDE3_CBC:
# DES_EDE3_CBC
from Crypto.Cipher import DES3
module = DES3
cipher_mode = DES3.MODE_CBC
key_size = 24
cipher_param = 'iv'
elif enc_oid == _OID_AES128_CBC:
module = AES
cipher_mode = AES.MODE_CBC
key_size = 16
cipher_param = 'iv'
elif enc_oid == _OID_AES192_CBC:
module = AES
cipher_mode = AES.MODE_CBC
key_size = 24
cipher_param = 'iv'
elif enc_oid == _OID_AES256_CBC:
module = AES
cipher_mode = AES.MODE_CBC
key_size = 32
cipher_param = 'iv'
elif enc_oid == _OID_AES128_GCM:
module = AES
cipher_mode = AES.MODE_GCM
key_size = 16
cipher_param = 'nonce'
aead = True
elif enc_oid == _OID_AES192_GCM:
module = AES
cipher_mode = AES.MODE_GCM
key_size = 24
cipher_param = 'nonce'
aead = True
elif enc_oid == _OID_AES256_GCM:
module = AES
cipher_mode = AES.MODE_GCM
key_size = 32
cipher_param = 'nonce'
aead = True
else:
raise PbesError("Unsupported PBES2 cipher " + enc_algo)
if kdf_key_length and kdf_key_length != key_size:
raise PbesError("Mismatch between PBES2 KDF parameters"
" and selected cipher")
iv_nonce = DerOctetString().decode(enc_info[1]).payload
# Create cipher
if kdf_oid == _OID_PBKDF2:
try:
hmac_hash_module_oid = Hash.HMAC._hmac2hash_oid[pbkdf2_prf_oid]
except KeyError:
raise PbesError("Unsupported HMAC %s" % pbkdf2_prf_oid)
hmac_hash_module = Hash.new(hmac_hash_module_oid)
key = PBKDF2(passphrase, salt, key_size, iteration_count,
hmac_hash_module=hmac_hash_module)
else:
key = scrypt(passphrase, salt, key_size, iteration_count,
scrypt_r, scrypt_p)
cipher = module.new(key, cipher_mode, **{cipher_param:iv_nonce})
# Decrypt data
if len(encrypted_data) < cipher.block_size:
raise ValueError("Too little data to decrypt")
if aead:
tag_len = cipher.block_size
pt = cipher.decrypt_and_verify(encrypted_data[:-tag_len],
encrypted_data[-tag_len:])
else:
pt_padded = cipher.decrypt(encrypted_data)
pt = unpad(pt_padded, cipher.block_size)
return pt

View File

@ -0,0 +1,26 @@
from typing import Optional, Callable, TypedDict
from typing_extensions import NotRequired
class PbesError(ValueError):
...
class PBES1(object):
@staticmethod
def decrypt(data: bytes, passphrase: bytes) -> bytes: ...
class ProtParams(TypedDict):
iteration_count: NotRequired[int]
salt_size: NotRequired[int]
block_size: NotRequired[int]
parallelization: NotRequired[int]
class PBES2(object):
@staticmethod
def encrypt(data: bytes,
passphrase: bytes,
protection: str,
prot_params: Optional[ProtParams] = ...,
randfunc: Optional[Callable[[int],bytes]] = ...) -> bytes: ...
@staticmethod
def decrypt(data:bytes, passphrase: bytes) -> bytes: ...

View File

@ -0,0 +1,31 @@
# ===================================================================
#
# Copyright (c) 2014, Legrandin <helderijs@gmail.com>
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in
# the documentation and/or other materials provided with the
# distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
# ===================================================================
__all__ = ['PEM', 'PKCS8']