Python源码示例：Crypto.Cipher.AES.MODE_GCM
示例1
def decrypt(self, data):
key = self.session.get_session_key()
method = self.settings.get("pia.encryption_method")
if method == EncryptionMethod.AES_ECB:
aes = AES.new(key, AES.MODE_ECB)
return aes.decrypt(data)
else:
nonce = self.session.generate_nonce(self)
aes = AES.new(key, AES.MODE_GCM, nonce=nonce)
try:
data = aes.decrypt_and_verify(data, self.signature)
except ValueError:
logger.warning("Received incorrect AES-GCM tag")
return None
return data
示例2
def decrypt_secret_data(self, class_key):
key = AESUnwrap(class_key, self.encrypted_secret_data_wrapped_key)
if not key:
raise ValueError("Failed to unwrap key. Bad class key?")
plist = readPlistFromString(self.protobuf_item.encryptedSecretData.ciphertext)
authenticated = ns_keyed_unarchiver(plist)
#decrypted = gcm_decrypt(key,
# authenticated['SFInitializationVector'],
# authenticated['SFCiphertext'], '',
# authenticated['SFAuthenticationCode'])
gcm = AES.new(key, AES.MODE_GCM, authenticated['SFInitializationVector'])
decrypted = gcm.decrypt_and_verify(authenticated['SFCiphertext'], authenticated['SFAuthenticationCode'])
if not decrypted:
raise ValueError("Failed to decrypt")
return decrypted
示例3
def test_2(self):
key = unhexlify("843ffcf5d2b72694d19ed01d01249412")
iv = unhexlify("dbcca32ebf9b804617c3aa9e")
aad = unhexlify("00000000000000000000000000000000" +
"101112131415161718191a1b1c1d1e1f")
pt = unhexlify("000102030405060708090a0b0c0d0e0f" +
"101112131415161718191a1b1c1d1e1f" +
"202122232425262728292a2b2c2d2e2f" +
"303132333435363738393a3b3c3d3e3f" +
"404142434445464748494a4b4c4d4e4f")
ct = unhexlify("6268c6fa2a80b2d137467f092f657ac0" +
"4d89be2beaa623d61b5a868c8f03ff95" +
"d3dcee23ad2f1ab3a6c80eaf4b140eb0" +
"5de3457f0fbc111a6b43d0763aa422a3" +
"013cf1dc37fe417d1fbfc449b75d4cc5")
digest = unhexlify("3b629ccfbc1119b7319e1dce2cd6fd6d")
cipher = AES.new(key, AES.MODE_GCM, iv).update(aad)
ct2, digest2 = cipher.encrypt_and_digest(pt)
self.assertEqual(ct, ct2)
self.assertEqual(digest, digest2)
示例4
def decrypt(self, ciphertext: bytes) -> bytes:
"""Return plaintext for given ciphertext."""
# Split out the nonce, tag, and encrypted data.
nonce = ciphertext[:12]
if len(nonce) != 12:
raise DataIntegrityError("Cipher text is damaged: invalid nonce length")
tag = ciphertext[12:28]
if len(tag) != 16:
raise DataIntegrityError("Cipher text is damaged: invalid tag length")
encrypted = ciphertext[28:]
# Construct AES cipher, with old nonce.
cipher = AES.new(self.cipher_key, AES.MODE_GCM, nonce)
# Decrypt and verify.
try:
plaintext = cipher.decrypt_and_verify(encrypted, tag) # type: ignore
except ValueError as e:
raise DataIntegrityError("Cipher text is damaged: {}".format(e))
return plaintext
示例5
def gcm_decrypt(k, iv, encrypted, auth_data, tag):
aes = AES.new(k, AES.MODE_GCM)
h = aes.encrypt(chr(0) * aes.block_size)
if len(iv) == 12:
y0 = iv + "\x00\x00\x00\x01"
else:
y0 = ghash(h, '', iv)
decrypted = gctr(k, y0, encrypted)
s = ghash(h, auth_data, encrypted)
t = aes.encrypt(y0)
T = strxor.strxor(s, t)
if T != tag:
return '' # decrypted data is invalid
else:
return decrypted
示例6
def decode(key, data, debug=False):
"""Decode function used for both the old and new style encryption"""
# Raw data structure is IV[:12] + crypttext[12:-16] + auth_tag[-16:]
iv = data[:12]
crypttext = data[12:-16]
tag = data[-16:]
if debug:
print("Input bytes: %", bytes2Hex(data))
print("IV: %s" % bytes2Hex(iv))
print("Crypttext: %s" % bytes2Hex(crypttext))
print("Auth tag: %s" % bytes2Hex(tag))
aes = AES.new(key, AES.MODE_GCM, nonce=iv)
try:
dec = aes.decrypt_and_verify(crypttext, tag)
if debug:
print("Decrypted data: %s" % bytes2Hex(dec))
return dec.decode('UTF-8')
except ValueError as e:
print(e)
print("The passphrase was probably wrong")
return None
示例7
def __decrypt_bkey_v4(self):
key_salt = self._pwkey_salt[:16]
logging.debug('KEY_SALT[%s] = %s', len(key_salt),
binascii.hexlify(key_salt))
key = PBKDF2(self._upwd, key_salt, Decryptor.dklen, Decryptor.count,
Decryptor.prf)
logging.debug('KEY[%s] = %s', len(key), binascii.hexlify(key))
nonce = self._pwkey_salt[16:]
logging.debug('KEY NONCE[%s] = %s', len(nonce),
binascii.hexlify(nonce))
cipher = AES.new(key, mode=AES.MODE_GCM, nonce=nonce)
self._bkey = cipher.decrypt(self._e_perbackupkey)[:32]
logging.debug('BKEY[%s] = %s',
len(self._bkey), binascii.hexlify(self._bkey))
示例8
def test_output_param_neg(self):
pt = b'5' * 16
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
ct = cipher.encrypt(pt)
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
self.assertRaises(TypeError, cipher.encrypt, pt, output=b'0'*16)
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
self.assertRaises(TypeError, cipher.decrypt, ct, output=b'0'*16)
shorter_output = bytearray(15)
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
self.assertRaises(ValueError, cipher.encrypt, pt, output=shorter_output)
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
self.assertRaises(ValueError, cipher.decrypt, ct, output=shorter_output)
示例9
def test_valid_multiple_encrypt_or_decrypt(self):
for method_name in "encrypt", "decrypt":
for auth_data in (None, b"333", self.data_128,
self.data_128 + b"3"):
if auth_data is None:
assoc_len = None
else:
assoc_len = len(auth_data)
cipher = AES.new(self.key_128, AES.MODE_GCM,
nonce=self.nonce_96)
if auth_data is not None:
cipher.update(auth_data)
method = getattr(cipher, method_name)
method(self.data_128)
method(self.data_128)
method(self.data_128)
method(self.data_128)
示例10
def test_encrypt(self, tv):
self._id = "Wycheproof Encrypt GCM Test #" + str(tv.id)
try:
cipher = AES.new(tv.key, AES.MODE_GCM, tv.iv, mac_len=tv.tag_size,
**self._extra_params)
except ValueError as e:
if len(tv.iv) == 0 and "Nonce cannot be empty" in str(e):
return
raise e
cipher.update(tv.aad)
ct, tag = cipher.encrypt_and_digest(tv.msg)
if tv.valid:
self.assertEqual(ct, tv.ct)
self.assertEqual(tag, tv.tag)
self.warn(tv)
示例11
def test_mac_len(self):
# Invalid MAC length
self.assertRaises(ValueError, AES.new, self.key_128, AES.MODE_GCM,
nonce=self.nonce_96, mac_len=3)
self.assertRaises(ValueError, AES.new, self.key_128, AES.MODE_GCM,
nonce=self.nonce_96, mac_len=16+1)
# Valid MAC length
for mac_len in range(5, 16 + 1):
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96,
mac_len=mac_len)
_, mac = cipher.encrypt_and_digest(self.data_128)
self.assertEqual(len(mac), mac_len)
# Default MAC length
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
_, mac = cipher.encrypt_and_digest(self.data_128)
self.assertEqual(len(mac), 16)
示例12
def test_output_param_neg(self):
pt = b'5' * 16
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
ct = cipher.encrypt(pt)
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
self.assertRaises(TypeError, cipher.encrypt, pt, output=b'0'*16)
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
self.assertRaises(TypeError, cipher.decrypt, ct, output=b'0'*16)
shorter_output = bytearray(15)
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
self.assertRaises(ValueError, cipher.encrypt, pt, output=shorter_output)
cipher = AES.new(self.key_128, AES.MODE_GCM, nonce=self.nonce_96)
self.assertRaises(ValueError, cipher.decrypt, ct, output=shorter_output)
示例13
def test_valid_multiple_encrypt_or_decrypt(self):
for method_name in "encrypt", "decrypt":
for auth_data in (None, b"333", self.data_128,
self.data_128 + b"3"):
if auth_data is None:
assoc_len = None
else:
assoc_len = len(auth_data)
cipher = AES.new(self.key_128, AES.MODE_GCM,
nonce=self.nonce_96)
if auth_data is not None:
cipher.update(auth_data)
method = getattr(cipher, method_name)
method(self.data_128)
method(self.data_128)
method(self.data_128)
method(self.data_128)
示例14
def encrypt_AES_GCM(msg, password, kdf_salt=None, nonce=None):
"""Used for encryption of msg"""
kdf_salt = kdf_salt or os.urandom(16)
nonce = nonce or os.urandom(16)
# Encoding of message
msg = msg.encode()
secret_key = Crypto.generate_key(kdf_salt, password)
aes_cipher = AES.new(secret_key, AES.MODE_GCM, nonce=nonce)
ciphertext, auth_tag = aes_cipher.encrypt_and_digest(msg)
return (kdf_salt, ciphertext, nonce, auth_tag)
示例15
def decrypt_AES_GCM(encryptedMsg, password, kdf_salt=None, nonce=None):
"""Used for decryption of msg"""
(stored_kdf_salt, ciphertext, stored_nonce, auth_tag) = encryptedMsg
kdf_salt = kdf_salt or stored_kdf_salt
nonce = nonce or stored_nonce
secret_key = Crypto.generate_key(kdf_salt, password)
aes_cipher = AES.new(secret_key, AES.MODE_GCM, nonce=nonce)
plaintext = aes_cipher.decrypt_and_verify(ciphertext, auth_tag)
# decoding byte data to normal string data
plaintext = plaintext.decode("utf8")
return plaintext
示例16
def encrypt(self, plaintext: bytes) -> bytes:
"""Return ciphertext for given plaintext."""
# Construct AES-GCM cipher, with 96-bit nonce.
cipher = AES.new(self.cipher_key, AES.MODE_GCM, nonce=random_bytes(12))
# Encrypt and digest.
encrypted, tag = cipher.encrypt_and_digest(plaintext) # type: ignore
# Combine with nonce.
ciphertext = cipher.nonce + tag + encrypted # type: ignore
# Return ciphertext.
return ciphertext
示例17
def _decrypt(self, value, encrypted_value):
"""Decrypt encoded cookies
"""
if sys.platform == 'win32':
try:
return self._decrypt_windows_chrome(value, encrypted_value)
# Fix for change in Chrome 80
except RuntimeError: # Failed to decrypt the cipher text with DPAPI
if not self.key:
raise RuntimeError(
'Failed to decrypt the cipher text with DPAPI and no AES key.')
# Encrypted cookies should be prefixed with 'v10' according to the
# Chromium code. Strip it off.
encrypted_value = encrypted_value[3:]
nonce, tag = encrypted_value[:12], encrypted_value[-16:]
aes = AES.new(self.key, AES.MODE_GCM, nonce=nonce)
data = aes.decrypt_and_verify(encrypted_value[12:-16], tag)
return data.decode()
if value or (encrypted_value[:3] not in [b'v11', b'v10']):
return value
# Encrypted cookies should be prefixed with 'v10' according to the
# Chromium code. Strip it off.
encrypted_value = encrypted_value[3:]
encrypted_value_half_len = int(len(encrypted_value) / 2)
cipher = pyaes.Decrypter(
pyaes.AESModeOfOperationCBC(self.key, self.iv))
decrypted = cipher.feed(encrypted_value[:encrypted_value_half_len])
decrypted += cipher.feed(encrypted_value[encrypted_value_half_len:])
decrypted += cipher.feed()
return decrypted.decode("utf-8")
示例18
def generate_challenge_reply(self, nonce, key, challenge):
key = hmac.new(self.key, key, digestmod=hashlib.sha256).digest()[:16]
data = hmac.new(self.key, challenge, digestmod=hashlib.sha256).digest()[:16]
aes = AES.new(key, AES.MODE_GCM, nonce=nonce)
ciphertext, tag = aes.encrypt_and_digest(data)
return tag + ciphertext
示例19
def generate_challenge(self, key, challenge):
key = self.generate_challenge_key(key)
nonce = self.generate_nonce(self.nonce_counter)
aes = AES.new(key, AES.MODE_GCM, nonce=nonce)
ciphertext, tag = aes.encrypt_and_digest(challenge)
return tag + ciphertext
示例20
def parse_challenge(self, stream):
if self.settings.get("pia.lan_version") == 0:
return b""
version = stream.u8()
if version != self.settings.get("pia.lan_version"):
logger.warning("Browse request has unexpected version number")
return None
crypto = stream.bool()
nonce = stream.u64()
key = stream.read(16)
tag = stream.read(16)
challenge = stream.read(256)
if crypto:
challenge_key = self.generate_challenge_key(key)
nonce = self.generate_nonce(nonce)
aes = AES.new(challenge_key, AES.MODE_GCM, nonce=nonce)
try:
challenge = aes.decrypt_and_verify(challenge, tag)
except ValueError:
logger.warning("Challenge has incorrect authentication tag")
return None
new_key = secrets.token_bytes(16)
self.nonce_counter += 1
nonce = self.generate_nonce(self.nonce_counter)
reply = struct.pack(">Q", self.nonce_counter) + new_key
reply += self.generate_challenge_reply(nonce, new_key + key, challenge)
else:
reply = secrets.token_bytes(16 + 16 + 16)
return reply
示例21
def encrypt(self, data):
key = self.session.get_session_key()
method = self.settings.get("pia.encryption_method")
if method == EncryptionMethod.AES_ECB:
aes = AES.new(key, AES.MODE_ECB)
return aes.encrypt(data)
else:
if len(data) % 16:
data += b"\xFF" * (16 - len(data) % 16)
nonce = self.session.generate_nonce(self)
aes = AES.new(key, AES.MODE_GCM, nonce=nonce)
data, self.signature = aes.encrypt_and_digest(data)
return data
示例22
def gcm_encrypt(k, iv, plaintext, auth_data):
aes = AES.new(k, AES.MODE_GCM)
h = aes.encrypt(chr(0) * aes.block_size)
if len(iv) == 12:
y0 = iv + "\x00\x00\x00\x01"
else:
y0 = ghash(h, '', iv)
encrypted = gctr(k, y0, plaintext)
s = ghash(h, auth_data, encrypted)
t = aes.encrypt(y0)
T = strxor.strxor(s, t)
return (encrypted, T)
示例23
def decrypt_metadata(self, metadata_class_key):
wrapped_plist = readPlistFromString(self.protobuf_item.encryptedMetadata.wrappedKey)
wrapped_sf_params = ns_keyed_unarchiver(wrapped_plist)
#metadata_key = gcm_decrypt(metadata_class_key,
# wrapped_sf_params['SFInitializationVector'],
# wrapped_sf_params['SFCiphertext'], '',
# wrapped_sf_params['SFAuthenticationCode'])
gcm = AES.new(metadata_class_key, AES.MODE_GCM, wrapped_sf_params['SFInitializationVector'])
metadata_key = gcm.decrypt_and_verify(wrapped_sf_params['SFCiphertext'], wrapped_sf_params['SFAuthenticationCode'])
if not metadata_key:
raise ValueError("Failed to decrypt metadata key")
ciphertext = ns_keyed_unarchiver(readPlistFromString(
self.protobuf_item.encryptedMetadata.ciphertext))
#metadata = gcm_decrypt(metadata_key,
# ciphertext['SFInitializationVector'],
# ciphertext['SFCiphertext'], '',
# ciphertext['SFAuthenticationCode'])
gcm = AES.new(metadata_key, AES.MODE_GCM, ciphertext['SFInitializationVector'])
metadata = gcm.decrypt_and_verify(ciphertext['SFCiphertext'], ciphertext['SFAuthenticationCode'])
if not metadata:
raise ValueError("Failed to decrypt metadata")
return metadata
示例24
def new_cipher(self, subkey: bytes, nonce: bytes):
return AES.new(subkey, AES.MODE_GCM, nonce=nonce, mac_len=self.TAG_SIZE)
示例25
def aes_gcm(self,ciphertext,nounce,aes_key, hdr, tag_auth, decrypt=True):
cipher = AES.new(aes_key, AES.MODE_GCM, nounce)
if hdr!=None:
cipher.update(hdr)
if decrypt:
plaintext = cipher.decrypt(ciphertext)
try:
cipher.verify(tag_auth)
return plaintext
except ValueError:
return None
示例26
def create_meta_cipher(secret):
meta_key = hkdf(16, secret, info=b'metadata')
return AES.new(meta_key, AES.MODE_GCM, b'\x00' * 12, mac_len=16)
示例27
def file_cipher_generator(secret, salt):
key = hkdf(16, secret, salt=salt, info=b'Content-Encoding: aes128gcm\0')
nonce_base = hkdf(12, secret, salt=salt, info=b'Content-Encoding: nonce\0')
seq = 0
while True:
if seq >= (1 << 32):
raise ValueError("Tried to encrypt too many chunks!")
tail, = struct.unpack('>I', nonce_base[-4:])
tail ^= seq
nonce = nonce_base[:-4] + struct.pack('>I', tail)
yield AES.new(key, AES.MODE_GCM, nonce, mac_len=16)
seq += 1
示例28
def __initializeCipher(self, key, iv, adata, tagsize):
from pbincli.utils import json_encode
cipher = AES.new(key, AES.MODE_GCM, nonce=iv, mac_len=tagsize)
cipher.update(json_encode(adata))
return cipher
示例29
def legacy_decrypt(self, plain_sk, enc_secret):
try:
b64_iv = 16
b64_mac = 24
iv = base64.b64decode(enc_secret[:b64_iv])
mac = base64.b64decode(enc_secret[b64_iv:(b64_iv + b64_mac)])
enc_secret = base64.b64decode(enc_secret[(b64_iv + b64_mac):])
cipher = AES.new(plain_sk, AES.MODE_GCM, iv)
dec_secret = cipher.decrypt(enc_secret)
#cipher.verify(mac)
return base64.b64encode(dec_secret)
except Exception as e:
raise Exception("ERROR: Legacy Decryption of the secret failed!", e)
示例30
def dec_secret(enc_secret):
obj = AES.new(SK.value, AES.MODE_GCM, base64.b64decode(enc_secret[:16]))
print "Decrypted Secret : " + obj.decrypt(base64.b64decode(enc_secret[40:]))
