Files
shanxia2.0/backend/venv/lib/python3.9/site-packages/Crypto/IO/_PBES.py
T
2026-10-05 12:57:51 +08:00

595 lines
22 KiB
Python

#
# 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"
#: Default maximum iteration count for PBKDF2 and scrypt cost parameter
#: in PBES1/PBES2 decryption. This guards against denial-of-service
#: attacks where a crafted blob declares an astronomically high count.
_DEFAULT_MAX_ITERATION_COUNT = 50000000
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, max_iteration_count=None):
"""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.
max_iteration_count : integer
The maximum value allowed for the PBKDF2 iteration count.
An attacker could craft a malicious key blob that requires an
extremely high number of iterations, tying up the CPU for hours.
The default value is None (enforcing the default limit of 50,000,000).
Set it to ``0`` to disable the check.
:Returns:
The decrypted data, as a binary string.
"""
if max_iteration_count is None:
max_iteration_count = _DEFAULT_MAX_ITERATION_COUNT
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]
if max_iteration_count and iterations > max_iteration_count:
raise PbesError(
"Iteration count too high (%d > %d). "
"Use the max_iteration_count parameter to increase the limit."
% (iterations, max_iteration_count)
)
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, max_iteration_count=None):
"""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.
max_iteration_count : integer
The maximum value allowed for the PBKDF2 iteration count
or the scrypt cost parameter (*N*).
An attacker could craft a malicious key blob that requires an
extremely high number of iterations, tying up the CPU for hours.
The default value is None (enforcing the default limit of 50,000,000).
Set it to ``0`` to disable the check.
:Returns:
The decrypted data, as a binary string.
"""
if max_iteration_count is None:
max_iteration_count = _DEFAULT_MAX_ITERATION_COUNT
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]
if max_iteration_count and iteration_count > max_iteration_count:
raise PbesError(
"PBKDF2 iteration count too high (%d > %d). "
"Use the max_iteration_count parameter to increase "
"the limit."
% (iteration_count, max_iteration_count)
)
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 max_iteration_count and iteration_count > max_iteration_count:
raise PbesError(
"scrypt cost parameter too high (%d > %d). "
"Use the max_iteration_count parameter to increase "
"the limit."
% (iteration_count, max_iteration_count)
)
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