628 lines
17 KiB
Python
628 lines
17 KiB
Python
"""
|
|
Java crypto standard library APIs implemented in Python.
|
|
javax.crypto package - Cipher, KeyGenerator, Mac, SecretKey, etc.
|
|
"""
|
|
|
|
from .._core import Stub, AndroidOnlyError
|
|
from .lang import RuntimeException
|
|
import os as _os
|
|
|
|
# Try to import cryptography library
|
|
try:
|
|
from cryptography.hazmat.primitives.ciphers import Cipher as _Cipher
|
|
from cryptography.hazmat.primitives.ciphers import algorithms as _algorithms
|
|
from cryptography.hazmat.primitives.ciphers import modes as _modes
|
|
from cryptography.hazmat.primitives import hashes as _hashes
|
|
from cryptography.hazmat.primitives.asymmetric import rsa as _rsa
|
|
from cryptography.hazmat.primitives.asymmetric import padding as _padding
|
|
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC as _PBKDF2
|
|
from cryptography.hazmat.backends import default_backend as _crypto_backend
|
|
HAS_CRYPTOGRAPHY = True
|
|
except ImportError:
|
|
HAS_CRYPTOGRAPHY = False
|
|
|
|
from .security import (
|
|
GeneralSecurityException, InvalidKeyException,
|
|
NoSuchAlgorithmException, NoSuchProviderException,
|
|
Key, SecretKeySpec, AlgorithmParameters, AlgorithmParametersSpi,
|
|
AlgorithmParameterGenerator, AlgorithmParameterGeneratorSpi,
|
|
Provider, SecureRandom,
|
|
)
|
|
|
|
IllegalArgumentException = type("IllegalArgumentException", (RuntimeException,), {})
|
|
AEADBadTagException = type("AEADBadTagException", (GeneralSecurityException,), {})
|
|
BadPaddingException = type("BadPaddingException", (GeneralSecurityException,), {})
|
|
IllegalBlockSizeException = type("IllegalBlockSizeException", (GeneralSecurityException,), {})
|
|
NoSuchPaddingException = type("NoSuchPaddingException", (GeneralSecurityException,), {})
|
|
ShortBufferException = type("ShortBufferException", (GeneralSecurityException,), {})
|
|
ExemptionMechanismException = type("ExemptionMechanismException", (GeneralSecurityException,), {})
|
|
|
|
__all__ = [
|
|
"Cipher", "CipherSpi",
|
|
"KeyGenerator", "KeyGeneratorSpi",
|
|
"SecretKey", "SecretKeyFactory", "SecretKeyFactorySpi",
|
|
"Mac", "MacSpi",
|
|
"IvParameterSpec", "PBEParameterSpec", "GCMParameterSpec",
|
|
"OAEPParameterSpec", "PSource", "MGF1ParameterSpec",
|
|
"DHParameterSpec", "DSAParameterSpec",
|
|
"ECParameterSpec", "ECGenParameterSpec", "ECPoint",
|
|
"ECField", "ECFieldF2m", "ECFieldFp", "EllipticCurve",
|
|
"AlgorithmParameters", "AlgorithmParameterGenerator",
|
|
"AlgorithmParametersSpi",
|
|
"AEADBadTagException", "BadPaddingException",
|
|
"IllegalBlockSizeException", "NoSuchPaddingException",
|
|
"ShortBufferException",
|
|
"ExemptionMechanism", "ExemptionMechanismException",
|
|
"ExemptionMechanismSpi",
|
|
]
|
|
|
|
|
|
# ============================================================
|
|
# SecretKey
|
|
# ============================================================
|
|
|
|
class SecretKey(Key):
|
|
def __init__(self, key_bytes=None, algorithm=""):
|
|
super().__init__()
|
|
self._key = key_bytes or b""
|
|
self._algorithm = algorithm
|
|
|
|
def getAlgorithm(self):
|
|
return self._algorithm
|
|
|
|
def getFormat(self):
|
|
return "RAW"
|
|
|
|
def getEncoded(self):
|
|
return self._key
|
|
|
|
|
|
# ============================================================
|
|
# Cipher
|
|
# ============================================================
|
|
|
|
class Cipher(Stub):
|
|
ENCRYPT_MODE = 1
|
|
DECRYPT_MODE = 2
|
|
WRAP_MODE = 3
|
|
UNWRAP_MODE = 4
|
|
PUBLIC_KEY = 1
|
|
PRIVATE_KEY = 2
|
|
SECRET_KEY = 3
|
|
|
|
def __init__(self, transformation):
|
|
super().__init__()
|
|
self._transformation = transformation
|
|
self._opmode = 0
|
|
self._key = None
|
|
self._iv = None
|
|
self._provider = Provider("AndroidOpenSSL", 1.0, "Android OpenSSL provider")
|
|
|
|
@staticmethod
|
|
def getInstance(transformation, provider=None):
|
|
return Cipher(transformation)
|
|
|
|
def init(self, opmode, key_or_cert, params=None, random=None):
|
|
self._opmode = opmode
|
|
self._key = key_or_cert
|
|
if isinstance(params, IvParameterSpec):
|
|
self._iv = params.getIV()
|
|
elif isinstance(params, GCMParameterSpec):
|
|
self._iv = params.getIV()
|
|
|
|
def doFinal(self, input_data=None):
|
|
if input_data is None:
|
|
return b""
|
|
if HAS_CRYPTOGRAPHY and self._key is not None:
|
|
try:
|
|
algo = self._transformation.split("/")[0].upper()
|
|
key_bytes = self._key.getEncoded() if hasattr(self._key, "getEncoded") else bytes(self._key)
|
|
iv = self._iv or b"\x00" * 16
|
|
if algo == "AES":
|
|
if len(iv) == 12:
|
|
mode = _modes.GCM(iv)
|
|
else:
|
|
mode = _modes.CBC(iv)
|
|
cipher = _Cipher(_algorithms.AES(key_bytes), mode, backend=_crypto_backend())
|
|
elif algo == "DES":
|
|
cipher = _Cipher(_algorithms.TripleDES(key_bytes), _modes.CBC(iv), backend=_crypto_backend())
|
|
else:
|
|
return self._basic_cipher(input_data)
|
|
if self._opmode == Cipher.ENCRYPT_MODE:
|
|
encryptor = cipher.encryptor()
|
|
result = encryptor.update(input_data) + encryptor.finalize()
|
|
return result
|
|
else:
|
|
decryptor = cipher.decryptor()
|
|
result = decryptor.update(input_data) + decryptor.finalize()
|
|
return result
|
|
except Exception:
|
|
return self._basic_cipher(input_data)
|
|
else:
|
|
return self._basic_cipher(input_data)
|
|
|
|
def _basic_cipher(self, data):
|
|
if self._opmode == Cipher.ENCRYPT_MODE:
|
|
if self._key is not None:
|
|
key_bytes = self._key.getEncoded() if hasattr(self._key, "getEncoded") else bytes(self._key)
|
|
return bytes([(b ^ key_bytes[i % len(key_bytes)]) & 0xFF for i, b in enumerate(data)])
|
|
return data
|
|
else:
|
|
if self._key is not None:
|
|
key_bytes = self._key.getEncoded() if hasattr(self._key, "getEncoded") else bytes(self._key)
|
|
return bytes([(b ^ key_bytes[i % len(key_bytes)]) & 0xFF for i, b in enumerate(data)])
|
|
return data
|
|
|
|
def update(self, input_data):
|
|
return self._basic_cipher(input_data)
|
|
|
|
def getBlockSize(self):
|
|
algo = self._transformation.split("/")[0].upper()
|
|
if algo == "AES":
|
|
return 16
|
|
if algo == "DES" or algo == "DESede":
|
|
return 8
|
|
return 1
|
|
|
|
def getOutputSize(self, input_len):
|
|
bs = self.getBlockSize()
|
|
if bs > 1:
|
|
return input_len + bs - (input_len % bs)
|
|
return input_len
|
|
|
|
def getIV(self):
|
|
return self._iv
|
|
|
|
def getParameters(self):
|
|
if self._iv:
|
|
return AlgorithmParameters(self._transformation.split("/")[0])
|
|
return None
|
|
|
|
def getAlgorithm(self):
|
|
return self._transformation
|
|
|
|
def getProvider(self):
|
|
return self._provider
|
|
|
|
def getExemptionMechanism(self):
|
|
return None
|
|
|
|
def unwrap(self, wrapped_key, wrapped_key_algorithm, wrapped_key_type):
|
|
return SecretKey(wrapped_key, wrapped_key_algorithm)
|
|
|
|
def wrap(self, key):
|
|
key_bytes = key.getEncoded() if hasattr(key, "getEncoded") else b""
|
|
return self.doFinal(key_bytes)
|
|
|
|
@staticmethod
|
|
def maxAllowedKeyLength(transformation):
|
|
return 2147483647
|
|
|
|
|
|
class CipherSpi(Stub):
|
|
pass
|
|
|
|
|
|
# ============================================================
|
|
# KeyGenerator
|
|
# ============================================================
|
|
|
|
class KeyGenerator(Stub):
|
|
def __init__(self, algorithm):
|
|
super().__init__()
|
|
self._algorithm = algorithm
|
|
self._key_size = 128
|
|
self._secure_random = None
|
|
|
|
@staticmethod
|
|
def getInstance(algorithm, provider=None):
|
|
return KeyGenerator(algorithm)
|
|
|
|
def init(self, size_or_random, random=None):
|
|
if isinstance(size_or_random, int):
|
|
self._key_size = size_or_random
|
|
self._secure_random = random
|
|
else:
|
|
self._secure_random = size_or_random
|
|
|
|
def generateKey(self):
|
|
key_bytes = _os.urandom(self._key_size // 8)
|
|
return SecretKey(key_bytes, self._algorithm)
|
|
|
|
def getAlgorithm(self):
|
|
return self._algorithm
|
|
|
|
def getProvider(self):
|
|
return Provider("AndroidOpenSSL", 1.0, "Android OpenSSL provider")
|
|
|
|
|
|
class KeyGeneratorSpi(Stub):
|
|
pass
|
|
|
|
|
|
# ============================================================
|
|
# SecretKeyFactory
|
|
# ============================================================
|
|
|
|
class SecretKeyFactory(Stub):
|
|
def __init__(self, algorithm):
|
|
super().__init__()
|
|
self._algorithm = algorithm
|
|
|
|
@staticmethod
|
|
def getInstance(algorithm, provider=None):
|
|
return SecretKeyFactory(algorithm)
|
|
|
|
def generateSecret(self, key_spec):
|
|
if isinstance(key_spec, SecretKeySpec):
|
|
return SecretKey(key_spec.getEncoded(), self._algorithm)
|
|
return SecretKey(b"", self._algorithm)
|
|
|
|
def getKeySpec(self, secret_key, key_spec_cls):
|
|
return SecretKeySpec(secret_key.getEncoded(), self._algorithm)
|
|
|
|
def translateKey(self, key):
|
|
return key
|
|
|
|
def getAlgorithm(self):
|
|
return self._algorithm
|
|
|
|
|
|
class SecretKeyFactorySpi(Stub):
|
|
pass
|
|
|
|
|
|
# ============================================================
|
|
# Mac
|
|
# ============================================================
|
|
|
|
class Mac(Stub):
|
|
def __init__(self, algorithm):
|
|
super().__init__()
|
|
self._algorithm = algorithm
|
|
self._key = None
|
|
self._state = "UNINITIALIZED"
|
|
self._data = b""
|
|
|
|
@staticmethod
|
|
def getInstance(algorithm, provider=None):
|
|
return Mac(algorithm)
|
|
|
|
def init(self, key, params=None):
|
|
self._key = key
|
|
self._state = "INITIALIZED"
|
|
self._data = b""
|
|
|
|
def update(self, input_data, offset=0, length=-1):
|
|
if length < 0:
|
|
self._data += bytes(input_data)
|
|
else:
|
|
self._data += bytes(input_data[offset:offset+length])
|
|
|
|
def doFinal(self, input_data=None):
|
|
if input_data is not None:
|
|
self._data += bytes(input_data)
|
|
if HAS_CRYPTOGRAPHY and self._key is not None:
|
|
try:
|
|
key_bytes = self._key.getEncoded() if hasattr(self._key, "getEncoded") else bytes(self._key)
|
|
algo = self._algorithm.lower().replace("-", "")
|
|
if "sha256" in algo or "sha-256" in self._algorithm:
|
|
h = _hashlib.sha256(key_bytes + self._data)
|
|
elif "sha1" in algo or "sha-1" in self._algorithm:
|
|
h = _hashlib.sha1(key_bytes + self._data)
|
|
elif "md5" in algo:
|
|
h = _hashlib.md5(key_bytes + self._data)
|
|
else:
|
|
h = _hashlib.sha256(key_bytes + self._data)
|
|
result = h.digest()
|
|
self._data = b""
|
|
return result
|
|
except Exception:
|
|
pass
|
|
import hashlib as _hashlib
|
|
key_bytes = self._key.getEncoded() if self._key and hasattr(self._key, "getEncoded") else b"key"
|
|
h = _hashlib.sha256(key_bytes + self._data)
|
|
result = h.digest()
|
|
self._data = b""
|
|
return result
|
|
|
|
def reset(self):
|
|
self._data = b""
|
|
|
|
def getAlgorithm(self):
|
|
return self._algorithm
|
|
|
|
def getMacLength(self):
|
|
return 32
|
|
|
|
def getProvider(self):
|
|
return Provider("AndroidOpenSSL", 1.0, "Android OpenSSL provider")
|
|
|
|
|
|
class MacSpi(Stub):
|
|
pass
|
|
|
|
|
|
# ============================================================
|
|
# Parameter specs
|
|
# ============================================================
|
|
|
|
class IvParameterSpec(Stub):
|
|
def __init__(self, iv):
|
|
super().__init__()
|
|
self._iv = bytes(iv)
|
|
|
|
def getIV(self):
|
|
return self._iv
|
|
|
|
|
|
class PBEParameterSpec(Stub):
|
|
def __init__(self, salt, iteration_count):
|
|
super().__init__()
|
|
self._salt = bytes(salt)
|
|
self._iteration_count = iteration_count
|
|
|
|
def getSalt(self):
|
|
return self._salt
|
|
|
|
def getIterationCount(self):
|
|
return self._iteration_count
|
|
|
|
|
|
class GCMParameterSpec(Stub):
|
|
def __init__(self, t_len, src):
|
|
super().__init__()
|
|
self._t_len = t_len
|
|
self._iv = bytes(src)
|
|
|
|
def getTLen(self):
|
|
return self._t_len
|
|
|
|
def getIV(self):
|
|
return self._iv
|
|
|
|
|
|
class OAEPParameterSpec(Stub):
|
|
def __init__(self, md_name, mgf_name, mgf_spec, p_src):
|
|
super().__init__()
|
|
self._md_name = md_name
|
|
self._mgf_name = mgf_name
|
|
self._mgf_spec = mgf_spec
|
|
self._p_src = p_src
|
|
|
|
def getDigestAlgorithm(self):
|
|
return self._md_name
|
|
|
|
def getMGFAlgorithm(self):
|
|
return self._mgf_name
|
|
|
|
def getMGFParameters(self):
|
|
return self._mgf_spec
|
|
|
|
def getPSource(self):
|
|
return self._p_src
|
|
|
|
|
|
class PSource(Stub):
|
|
class PSpecified(Stub):
|
|
DEFAULT = None
|
|
|
|
def __init__(self, p):
|
|
super().__init__()
|
|
self._p = p
|
|
|
|
def getValue(self):
|
|
return self._p
|
|
|
|
def __init__(self, p_src_id):
|
|
super().__init__()
|
|
self._p_src_id = p_src_id
|
|
|
|
def getAlgorithm(self):
|
|
return self._p_src_id
|
|
|
|
|
|
PSource.PSpecified.DEFAULT = PSource.PSpecified(b"")
|
|
|
|
|
|
class MGF1ParameterSpec(Stub):
|
|
SHA1 = None
|
|
SHA256 = None
|
|
SHA384 = None
|
|
SHA512 = None
|
|
|
|
def __init__(self, md_name):
|
|
super().__init__()
|
|
self._md_name = md_name
|
|
|
|
def getDigestAlgorithm(self):
|
|
return self._md_name
|
|
|
|
def toString(self):
|
|
return f"MGF1 {self._md_name}"
|
|
|
|
|
|
MGF1ParameterSpec.SHA1 = MGF1ParameterSpec("SHA-1")
|
|
MGF1ParameterSpec.SHA256 = MGF1ParameterSpec("SHA-256")
|
|
MGF1ParameterSpec.SHA384 = MGF1ParameterSpec("SHA-384")
|
|
MGF1ParameterSpec.SHA512 = MGF1ParameterSpec("SHA-512")
|
|
|
|
|
|
# ============================================================
|
|
# DH / DSA / EC parameter specs (stubs)
|
|
# ============================================================
|
|
|
|
class DHParameterSpec(Stub):
|
|
def __init__(self, p, g, l=0):
|
|
super().__init__()
|
|
self._p = p
|
|
self._g = g
|
|
self._l = l
|
|
|
|
def getP(self):
|
|
return self._p
|
|
|
|
def getG(self):
|
|
return self._g
|
|
|
|
def getL(self):
|
|
return self._l
|
|
|
|
|
|
class DSAParameterSpec(Stub):
|
|
def __init__(self, p, q, g):
|
|
super().__init__()
|
|
self._p = p
|
|
self._q = q
|
|
self._g = g
|
|
|
|
def getP(self):
|
|
return self._p
|
|
|
|
def getQ(self):
|
|
return self._q
|
|
|
|
def getG(self):
|
|
return self._g
|
|
|
|
|
|
class ECPoint(Stub):
|
|
def __init__(self, affine_x, affine_y):
|
|
super().__init__()
|
|
self._affine_x = affine_x
|
|
self._affine_y = affine_y
|
|
|
|
def getAffineX(self):
|
|
return self._affine_x
|
|
|
|
def getAffineY(self):
|
|
return self._affine_y
|
|
|
|
def equals(self, other):
|
|
if isinstance(other, ECPoint):
|
|
return self._affine_x == other._affine_x and self._affine_y == other._affine_y
|
|
return False
|
|
|
|
def hashCode(self):
|
|
return hash((self._affine_x, self._affine_y))
|
|
|
|
|
|
class ECField(Stub):
|
|
def getFieldSize(self):
|
|
raise NotImplementedError
|
|
|
|
|
|
class ECFieldFp(ECField):
|
|
def __init__(self, p):
|
|
super().__init__()
|
|
self._p = p
|
|
|
|
def getP(self):
|
|
return self._p
|
|
|
|
def getFieldSize(self):
|
|
return self._p.bit_length()
|
|
|
|
|
|
class ECFieldF2m(ECField):
|
|
def __init__(self, m, ks=None):
|
|
super().__init__()
|
|
self._m = m
|
|
self._ks = ks
|
|
|
|
def getM(self):
|
|
return self._m
|
|
|
|
def getMidTermsOfReductionPolynomial(self):
|
|
return self._ks
|
|
|
|
def getFieldSize(self):
|
|
return self._m
|
|
|
|
|
|
class EllipticCurve(Stub):
|
|
def __init__(self, field, a, b, seed=None):
|
|
super().__init__()
|
|
self._field = field
|
|
self._a = a
|
|
self._b = b
|
|
self._seed = seed
|
|
|
|
def getField(self):
|
|
return self._field
|
|
|
|
def getA(self):
|
|
return self._a
|
|
|
|
def getB(self):
|
|
return self._b
|
|
|
|
def getSeed(self):
|
|
return self._seed
|
|
|
|
|
|
class ECParameterSpec(Stub):
|
|
def __init__(self, curve, generator, order, cofactor, seed=None):
|
|
super().__init__()
|
|
self._curve = curve
|
|
self._generator = generator
|
|
self._order = order
|
|
self._cofactor = cofactor
|
|
|
|
def getCurve(self):
|
|
return self._curve
|
|
|
|
def getGenerator(self):
|
|
return self._generator
|
|
|
|
def getOrder(self):
|
|
return self._order
|
|
|
|
def getCofactor(self):
|
|
return self._cofactor
|
|
|
|
|
|
class ECGenParameterSpec(Stub):
|
|
def __init__(self, name):
|
|
super().__init__()
|
|
self._name = name
|
|
|
|
def getName(self):
|
|
return self._name
|
|
|
|
|
|
# ============================================================
|
|
# ExemptionMechanism
|
|
# ============================================================
|
|
|
|
class ExemptionMechanism(Stub):
|
|
def __init__(self, name):
|
|
super().__init__()
|
|
self._name = name
|
|
self._state = False
|
|
|
|
@staticmethod
|
|
def getInstance(name, provider=None):
|
|
return ExemptionMechanism(name)
|
|
|
|
def init(self, key):
|
|
self._state = False
|
|
|
|
def isCryptoAllowed(self, key):
|
|
return True
|
|
|
|
def genExemptionBlob(self, data=None):
|
|
return b""
|
|
|
|
def getName(self):
|
|
return self._name
|
|
|
|
def getOutputSize(self, input_len):
|
|
return input_len
|
|
|
|
def storeMechanism(self, mechanism):
|
|
pass
|
|
|
|
|
|
class ExemptionMechanismSpi(Stub):
|
|
pass |