317 lines
13 KiB
Python
317 lines
13 KiB
Python
# uncompyle6 version 3.9.2
|
|
# Python bytecode version base 3.7.0 (3394)
|
|
# Decompiled from: Python 3.8.19 (default, Mar 20 2024, 15:27:52)
|
|
# [Clang 14.0.6 ]
|
|
# Embedded file name: /var/user/app/device_supervisorbak/device_supervisor/lib/jwt/algorithms.py
|
|
# Compiled at: 2024-04-18 03:12:55
|
|
# Size of source mod 2**32: 13739 bytes
|
|
import hashlib, hmac, json
|
|
from .compat import constant_time_compare, string_types
|
|
from .exceptions import InvalidKeyError
|
|
from .utils import base64url_decode, base64url_encode, der_to_raw_signature, force_bytes, force_unicode, from_base64url_uint, raw_to_der_signature, to_base64url_uint
|
|
try:
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.serialization import load_pem_private_key, load_pem_public_key, load_ssh_public_key
|
|
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, RSAPublicKey, RSAPrivateNumbers, RSAPublicNumbers, rsa_recover_prime_factors, rsa_crt_dmp1, rsa_crt_dmq1, rsa_crt_iqmp
|
|
from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey, EllipticCurvePublicKey
|
|
from cryptography.hazmat.primitives.asymmetric import ec, padding
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.exceptions import InvalidSignature
|
|
has_crypto = True
|
|
except ImportError:
|
|
has_crypto = False
|
|
|
|
requires_cryptography = set(['RS256', 'RS384', 'RS512', 'ES256', 'ES384',
|
|
'ES521', 'ES512', 'PS256',
|
|
'PS384', 'PS512'])
|
|
|
|
def get_default_algorithms():
|
|
"""
|
|
Returns the algorithms that are implemented by the library.
|
|
"""
|
|
default_algorithms = {'none':NoneAlgorithm(),
|
|
'HS256':HMACAlgorithm(HMACAlgorithm.SHA256),
|
|
'HS384':HMACAlgorithm(HMACAlgorithm.SHA384),
|
|
'HS512':HMACAlgorithm(HMACAlgorithm.SHA512)}
|
|
if has_crypto:
|
|
default_algorithms.update({'RS256':RSAAlgorithm(RSAAlgorithm.SHA256),
|
|
'RS384':RSAAlgorithm(RSAAlgorithm.SHA384),
|
|
'RS512':RSAAlgorithm(RSAAlgorithm.SHA512),
|
|
'ES256':ECAlgorithm(ECAlgorithm.SHA256),
|
|
'ES384':ECAlgorithm(ECAlgorithm.SHA384),
|
|
'ES521':ECAlgorithm(ECAlgorithm.SHA512),
|
|
'ES512':ECAlgorithm(ECAlgorithm.SHA512),
|
|
'PS256':RSAPSSAlgorithm(RSAPSSAlgorithm.SHA256),
|
|
'PS384':RSAPSSAlgorithm(RSAPSSAlgorithm.SHA384),
|
|
'PS512':RSAPSSAlgorithm(RSAPSSAlgorithm.SHA512)})
|
|
return default_algorithms
|
|
|
|
|
|
class Algorithm(object):
|
|
__doc__ = "\n The interface for an algorithm used to sign and verify tokens.\n "
|
|
|
|
def prepare_key(self, key):
|
|
"""
|
|
Performs necessary validation and conversions on the key and returns
|
|
the key value in the proper format for sign() and verify().
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def sign(self, msg, key):
|
|
"""
|
|
Returns a digital signature for the specified message
|
|
using the specified key value.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
def verify(self, msg, key, sig):
|
|
"""
|
|
Verifies that the specified digital signature is valid
|
|
for the specified message and key values.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@staticmethod
|
|
def to_jwk(key_obj):
|
|
"""
|
|
Serializes a given RSA key into a JWK
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@staticmethod
|
|
def from_jwk(jwk):
|
|
"""
|
|
Deserializes a given RSA key from JWK back into a PublicKey or PrivateKey object
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
|
|
class NoneAlgorithm(Algorithm):
|
|
__doc__ = "\n Placeholder for use when no signing or verification\n operations are required.\n "
|
|
|
|
def prepare_key(self, key):
|
|
if key == "":
|
|
key = None
|
|
if key is not None:
|
|
raise InvalidKeyError('When alg = "none", key value must be None.')
|
|
return key
|
|
|
|
def sign(self, msg, key):
|
|
return b''
|
|
|
|
def verify(self, msg, key, sig):
|
|
return False
|
|
|
|
|
|
class HMACAlgorithm(Algorithm):
|
|
__doc__ = "\n Performs signing and verification operations using HMAC\n and the specified hash function.\n "
|
|
SHA256 = hashlib.sha256
|
|
SHA384 = hashlib.sha384
|
|
SHA512 = hashlib.sha512
|
|
|
|
def __init__(self, hash_alg):
|
|
self.hash_alg = hash_alg
|
|
|
|
def prepare_key(self, key):
|
|
key = force_bytes(key)
|
|
invalid_strings = [
|
|
b'-----BEGIN PUBLIC KEY-----',
|
|
b'-----BEGIN CERTIFICATE-----',
|
|
b'-----BEGIN RSA PUBLIC KEY-----',
|
|
b'ssh-rsa']
|
|
if any([string_value in key for string_value in invalid_strings]):
|
|
raise InvalidKeyError("The specified key is an asymmetric key or x509 certificate and should not be used as an HMAC secret.")
|
|
return key
|
|
|
|
@staticmethod
|
|
def to_jwk(key_obj):
|
|
return json.dumps({'k':force_unicode(base64url_encode(force_bytes(key_obj))),
|
|
'kty':"oct"})
|
|
|
|
@staticmethod
|
|
def from_jwk(jwk):
|
|
obj = json.loads(jwk)
|
|
if obj.get("kty") != "oct":
|
|
raise InvalidKeyError("Not an HMAC key")
|
|
return base64url_decode(obj["k"])
|
|
|
|
def sign(self, msg, key):
|
|
return hmac.new(key, msg, self.hash_alg).digest()
|
|
|
|
def verify(self, msg, key, sig):
|
|
return constant_time_compare(sig, self.sign(msg, key))
|
|
|
|
|
|
if has_crypto:
|
|
|
|
class RSAAlgorithm(Algorithm):
|
|
__doc__ = "\n Performs signing and verification operations using\n RSASSA-PKCS-v1_5 and the specified hash function.\n "
|
|
SHA256 = hashes.SHA256
|
|
SHA384 = hashes.SHA384
|
|
SHA512 = hashes.SHA512
|
|
|
|
def __init__(self, hash_alg):
|
|
self.hash_alg = hash_alg
|
|
|
|
def prepare_key(self, key):
|
|
if isinstance(key, RSAPrivateKey) or isinstance(key, RSAPublicKey):
|
|
return key
|
|
elif isinstance(key, string_types):
|
|
key = force_bytes(key)
|
|
try:
|
|
if key.startswith(b'ssh-rsa'):
|
|
key = load_ssh_public_key(key, backend=(default_backend()))
|
|
else:
|
|
key = load_pem_private_key(key, password=None, backend=(default_backend()))
|
|
except ValueError:
|
|
key = load_pem_public_key(key, backend=(default_backend()))
|
|
|
|
else:
|
|
raise TypeError("Expecting a PEM-formatted key.")
|
|
return key
|
|
|
|
@staticmethod
|
|
def to_jwk(key_obj):
|
|
obj = None
|
|
if getattr(key_obj, "private_numbers", None):
|
|
numbers = key_obj.private_numbers()
|
|
obj = {'kty':"RSA",
|
|
'key_ops':[
|
|
"sign"],
|
|
'n':force_unicode(to_base64url_uint(numbers.public_numbers.n)),
|
|
'e':force_unicode(to_base64url_uint(numbers.public_numbers.e)),
|
|
'd':force_unicode(to_base64url_uint(numbers.d)),
|
|
'p':force_unicode(to_base64url_uint(numbers.p)),
|
|
'q':force_unicode(to_base64url_uint(numbers.q)),
|
|
'dp':force_unicode(to_base64url_uint(numbers.dmp1)),
|
|
'dq':force_unicode(to_base64url_uint(numbers.dmq1)),
|
|
'qi':force_unicode(to_base64url_uint(numbers.iqmp))}
|
|
else:
|
|
if getattr(key_obj, "verify", None):
|
|
numbers = key_obj.public_numbers()
|
|
obj = {'kty':"RSA",
|
|
'key_ops':[
|
|
"verify"],
|
|
'n':force_unicode(to_base64url_uint(numbers.n)),
|
|
'e':force_unicode(to_base64url_uint(numbers.e))}
|
|
else:
|
|
raise InvalidKeyError("Not a public or private key")
|
|
return json.dumps(obj)
|
|
|
|
@staticmethod
|
|
def from_jwk(jwk):
|
|
try:
|
|
obj = json.loads(jwk)
|
|
except ValueError:
|
|
raise InvalidKeyError("Key is not valid JSON")
|
|
|
|
if obj.get("kty") != "RSA":
|
|
raise InvalidKeyError("Not an RSA key")
|
|
elif "d" in obj:
|
|
if "e" in obj and "n" in obj:
|
|
if "oth" in obj:
|
|
raise InvalidKeyError("Unsupported RSA private key: > 2 primes not supported")
|
|
else:
|
|
other_props = [
|
|
'p', 'q',
|
|
'dp', 'dq', 'qi']
|
|
props_found = [prop in obj for prop in other_props]
|
|
any_props_found = any(props_found)
|
|
if any_props_found:
|
|
if not all(props_found):
|
|
raise InvalidKeyError("RSA key must include all parameters if any are present besides d")
|
|
public_numbers = RSAPublicNumbers(from_base64url_uint(obj["e"]), from_base64url_uint(obj["n"]))
|
|
if any_props_found:
|
|
numbers = RSAPrivateNumbers(d=(from_base64url_uint(obj["d"])),
|
|
p=(from_base64url_uint(obj["p"])),
|
|
q=(from_base64url_uint(obj["q"])),
|
|
dmp1=(from_base64url_uint(obj["dp"])),
|
|
dmq1=(from_base64url_uint(obj["dq"])),
|
|
iqmp=(from_base64url_uint(obj["qi"])),
|
|
public_numbers=public_numbers)
|
|
else:
|
|
d = from_base64url_uint(obj["d"])
|
|
p, q = rsa_recover_prime_factors(public_numbers.n, d, public_numbers.e)
|
|
numbers = RSAPrivateNumbers(d=d,
|
|
p=p,
|
|
q=q,
|
|
dmp1=(rsa_crt_dmp1(d, p)),
|
|
dmq1=(rsa_crt_dmq1(d, q)),
|
|
iqmp=(rsa_crt_iqmp(p, q)),
|
|
public_numbers=public_numbers)
|
|
return numbers.private_key(default_backend())
|
|
if "n" in obj:
|
|
if "e" in obj:
|
|
numbers = RSAPublicNumbers(from_base64url_uint(obj["e"]), from_base64url_uint(obj["n"]))
|
|
return numbers.public_key(default_backend())
|
|
raise InvalidKeyError("Not a public or private key")
|
|
|
|
def sign(self, msg, key):
|
|
return key.sign(msg, padding.PKCS1v15(), self.hash_alg())
|
|
|
|
def verify(self, msg, key, sig):
|
|
try:
|
|
key.verify(sig, msg, padding.PKCS1v15(), self.hash_alg())
|
|
return True
|
|
except InvalidSignature:
|
|
return False
|
|
|
|
|
|
class ECAlgorithm(Algorithm):
|
|
__doc__ = "\n Performs signing and verification operations using\n ECDSA and the specified hash function\n "
|
|
SHA256 = hashes.SHA256
|
|
SHA384 = hashes.SHA384
|
|
SHA512 = hashes.SHA512
|
|
|
|
def __init__(self, hash_alg):
|
|
self.hash_alg = hash_alg
|
|
|
|
def prepare_key(self, key):
|
|
if isinstance(key, EllipticCurvePrivateKey) or isinstance(key, EllipticCurvePublicKey):
|
|
return key
|
|
elif isinstance(key, string_types):
|
|
key = force_bytes(key)
|
|
try:
|
|
if key.startswith(b'ecdsa-sha2-'):
|
|
key = load_ssh_public_key(key, backend=(default_backend()))
|
|
else:
|
|
key = load_pem_public_key(key, backend=(default_backend()))
|
|
except ValueError:
|
|
key = load_pem_private_key(key, password=None, backend=(default_backend()))
|
|
|
|
else:
|
|
raise TypeError("Expecting a PEM-formatted key.")
|
|
return key
|
|
|
|
def sign(self, msg, key):
|
|
der_sig = key.sign(msg, ec.ECDSA(self.hash_alg()))
|
|
return der_to_raw_signature(der_sig, key.curve)
|
|
|
|
def verify(self, msg, key, sig):
|
|
try:
|
|
der_sig = raw_to_der_signature(sig, key.curve)
|
|
except ValueError:
|
|
return False
|
|
else:
|
|
try:
|
|
key.verify(der_sig, msg, ec.ECDSA(self.hash_alg()))
|
|
return True
|
|
except InvalidSignature:
|
|
return False
|
|
|
|
|
|
class RSAPSSAlgorithm(RSAAlgorithm):
|
|
__doc__ = "\n Performs a signature using RSASSA-PSS with MGF1\n "
|
|
|
|
def sign(self, msg, key):
|
|
return key.sign(msg, padding.PSS(mgf=(padding.MGF1(self.hash_alg())),
|
|
salt_length=(self.hash_alg.digest_size)), self.hash_alg())
|
|
|
|
def verify(self, msg, key, sig):
|
|
try:
|
|
key.verify(sig, msg, padding.PSS(mgf=(padding.MGF1(self.hash_alg())),
|
|
salt_length=(self.hash_alg.digest_size)), self.hash_alg())
|
|
return True
|
|
except InvalidSignature:
|
|
return False
|