diff options
author | alexv-smirnov <alex@ydb.tech> | 2023-12-01 12:02:50 +0300 |
---|---|---|
committer | alexv-smirnov <alex@ydb.tech> | 2023-12-01 13:28:10 +0300 |
commit | 0e578a4c44d4abd539d9838347b9ebafaca41dfb (patch) | |
tree | a0c1969c37f818c830ebeff9c077eacf30be6ef8 /contrib/python/PyJWT/py3 | |
parent | 84f2d3d4cc985e63217cff149bd2e6d67ae6fe22 (diff) | |
download | ydb-0e578a4c44d4abd539d9838347b9ebafaca41dfb.tar.gz |
Change "ya.make"
Diffstat (limited to 'contrib/python/PyJWT/py3')
-rw-r--r-- | contrib/python/PyJWT/py3/.dist-info/METADATA | 107 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/.dist-info/top_level.txt | 1 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/AUTHORS.rst | 7 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/LICENSE | 21 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/README.rst | 62 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/__init__.py | 70 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/algorithms.py | 674 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/api_jwk.py | 97 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/api_jws.py | 259 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/api_jwt.py | 221 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/exceptions.py | 66 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/help.py | 60 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/jwks_client.py | 59 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/py.typed | 0 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/jwt/utils.py | 99 | ||||
-rw-r--r-- | contrib/python/PyJWT/py3/ya.make | 39 |
16 files changed, 1842 insertions, 0 deletions
diff --git a/contrib/python/PyJWT/py3/.dist-info/METADATA b/contrib/python/PyJWT/py3/.dist-info/METADATA new file mode 100644 index 0000000000..f5fbdf64a5 --- /dev/null +++ b/contrib/python/PyJWT/py3/.dist-info/METADATA @@ -0,0 +1,107 @@ +Metadata-Version: 2.1 +Name: PyJWT +Version: 2.3.0 +Summary: JSON Web Token implementation in Python +Home-page: https://github.com/jpadilla/pyjwt +Author: Jose Padilla +Author-email: hello@jpadilla.com +License: MIT +Keywords: json,jwt,security,signing,token,web +Platform: UNKNOWN +Classifier: Development Status :: 5 - Production/Stable +Classifier: Intended Audience :: Developers +Classifier: Natural Language :: English +Classifier: License :: OSI Approved :: MIT License +Classifier: Programming Language :: Python +Classifier: Programming Language :: Python :: 3 +Classifier: Programming Language :: Python :: 3 :: Only +Classifier: Programming Language :: Python :: 3.6 +Classifier: Programming Language :: Python :: 3.7 +Classifier: Programming Language :: Python :: 3.8 +Classifier: Programming Language :: Python :: 3.9 +Classifier: Topic :: Utilities +Requires-Python: >=3.6 +Description-Content-Type: text/x-rst +Provides-Extra: crypto +Requires-Dist: cryptography (>=3.3.1) ; extra == 'crypto' +Provides-Extra: dev +Requires-Dist: sphinx ; extra == 'dev' +Requires-Dist: sphinx-rtd-theme ; extra == 'dev' +Requires-Dist: zope.interface ; extra == 'dev' +Requires-Dist: cryptography (>=3.3.1) ; extra == 'dev' +Requires-Dist: pytest (<7.0.0,>=6.0.0) ; extra == 'dev' +Requires-Dist: coverage[toml] (==5.0.4) ; extra == 'dev' +Requires-Dist: mypy ; extra == 'dev' +Requires-Dist: pre-commit ; extra == 'dev' +Provides-Extra: docs +Requires-Dist: sphinx ; extra == 'docs' +Requires-Dist: sphinx-rtd-theme ; extra == 'docs' +Requires-Dist: zope.interface ; extra == 'docs' +Provides-Extra: tests +Requires-Dist: pytest (<7.0.0,>=6.0.0) ; extra == 'tests' +Requires-Dist: coverage[toml] (==5.0.4) ; extra == 'tests' + +PyJWT +===== + +.. image:: https://github.com/jpadilla/pyjwt/workflows/CI/badge.svg + :target: https://github.com/jpadilla/pyjwt/actions?query=workflow%3ACI + +.. image:: https://img.shields.io/pypi/v/pyjwt.svg + :target: https://pypi.python.org/pypi/pyjwt + +.. image:: https://codecov.io/gh/jpadilla/pyjwt/branch/master/graph/badge.svg + :target: https://codecov.io/gh/jpadilla/pyjwt + +.. image:: https://readthedocs.org/projects/pyjwt/badge/?version=stable + :target: https://pyjwt.readthedocs.io/en/stable/ + +A Python implementation of `RFC 7519 <https://tools.ietf.org/html/rfc7519>`_. Original implementation was written by `@progrium <https://github.com/progrium>`_. + +Sponsor +------- + ++--------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| |auth0-logo| | If you want to quickly add secure token-based authentication to Python projects, feel free to check Auth0's Python SDK and free plan at `auth0.com/developers <https://auth0.com/developers?utm_source=GHsponsor&utm_medium=GHsponsor&utm_campaign=pyjwt&utm_content=auth>`_. | ++--------------+-----------------------------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + +.. |auth0-logo| image:: https://user-images.githubusercontent.com/83319/31722733-de95bbde-b3ea-11e7-96bf-4f4e8f915588.png + +Installing +---------- + +Install with **pip**: + +.. code-block:: console + + $ pip install PyJWT + + +Usage +----- + +.. code-block:: pycon + + >>> import jwt + >>> encoded = jwt.encode({"some": "payload"}, "secret", algorithm="HS256") + >>> print(encoded) + eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzb21lIjoicGF5bG9hZCJ9.Joh1R2dYzkRvDkqv3sygm5YyK8Gi4ShZqbhK2gxcs2U + >>> jwt.decode(encoded, "secret", algorithms=["HS256"]) + {'some': 'payload'} + +Documentation +------------- + +View the full docs online at https://pyjwt.readthedocs.io/en/stable/ + + +Tests +----- + +You can run tests from the project root after cloning with: + +.. code-block:: console + + $ tox + + diff --git a/contrib/python/PyJWT/py3/.dist-info/top_level.txt b/contrib/python/PyJWT/py3/.dist-info/top_level.txt new file mode 100644 index 0000000000..27ccc9bc3a --- /dev/null +++ b/contrib/python/PyJWT/py3/.dist-info/top_level.txt @@ -0,0 +1 @@ +jwt diff --git a/contrib/python/PyJWT/py3/AUTHORS.rst b/contrib/python/PyJWT/py3/AUTHORS.rst new file mode 100644 index 0000000000..88e2b6ad75 --- /dev/null +++ b/contrib/python/PyJWT/py3/AUTHORS.rst @@ -0,0 +1,7 @@ +Authors +======= + +``pyjwt`` is currently written and maintained by `Jose Padilla <https://github.com/jpadilla>`_. +Originally written and maintained by `Jeff Lindsay <https://github.com/progrium>`_. + +A full list of contributors can be found on GitHub’s `overview <https://github.com/jpadilla/pyjwt/graphs/contributors>`_. diff --git a/contrib/python/PyJWT/py3/LICENSE b/contrib/python/PyJWT/py3/LICENSE new file mode 100644 index 0000000000..bdc7819ea1 --- /dev/null +++ b/contrib/python/PyJWT/py3/LICENSE @@ -0,0 +1,21 @@ +The MIT License (MIT) + +Copyright (c) 2015 José Padilla + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/contrib/python/PyJWT/py3/README.rst b/contrib/python/PyJWT/py3/README.rst new file mode 100644 index 0000000000..49aa77a8e7 --- /dev/null +++ b/contrib/python/PyJWT/py3/README.rst @@ -0,0 +1,62 @@ +PyJWT +===== + +.. image:: https://github.com/jpadilla/pyjwt/workflows/CI/badge.svg + :target: https://github.com/jpadilla/pyjwt/actions?query=workflow%3ACI + +.. image:: https://img.shields.io/pypi/v/pyjwt.svg + :target: https://pypi.python.org/pypi/pyjwt + +.. image:: https://codecov.io/gh/jpadilla/pyjwt/branch/master/graph/badge.svg + :target: https://codecov.io/gh/jpadilla/pyjwt + +.. image:: https://readthedocs.org/projects/pyjwt/badge/?version=stable + :target: https://pyjwt.readthedocs.io/en/stable/ + +A Python implementation of `RFC 7519 <https://tools.ietf.org/html/rfc7519>`_. Original implementation was written by `@progrium <https://github.com/progrium>`_. + +Sponsor +------- + ++--------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| |auth0-logo| | If you want to quickly add secure token-based authentication to Python projects, feel free to check Auth0's Python SDK and free plan at `auth0.com/developers <https://auth0.com/developers?utm_source=GHsponsor&utm_medium=GHsponsor&utm_campaign=pyjwt&utm_content=auth>`_. | ++--------------+-----------------------------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + +.. |auth0-logo| image:: https://user-images.githubusercontent.com/83319/31722733-de95bbde-b3ea-11e7-96bf-4f4e8f915588.png + +Installing +---------- + +Install with **pip**: + +.. code-block:: console + + $ pip install PyJWT + + +Usage +----- + +.. code-block:: pycon + + >>> import jwt + >>> encoded = jwt.encode({"some": "payload"}, "secret", algorithm="HS256") + >>> print(encoded) + eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzb21lIjoicGF5bG9hZCJ9.Joh1R2dYzkRvDkqv3sygm5YyK8Gi4ShZqbhK2gxcs2U + >>> jwt.decode(encoded, "secret", algorithms=["HS256"]) + {'some': 'payload'} + +Documentation +------------- + +View the full docs online at https://pyjwt.readthedocs.io/en/stable/ + + +Tests +----- + +You can run tests from the project root after cloning with: + +.. code-block:: console + + $ tox diff --git a/contrib/python/PyJWT/py3/jwt/__init__.py b/contrib/python/PyJWT/py3/jwt/__init__.py new file mode 100644 index 0000000000..3208c39f39 --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/__init__.py @@ -0,0 +1,70 @@ +from .api_jwk import PyJWK, PyJWKSet +from .api_jws import ( + PyJWS, + get_unverified_header, + register_algorithm, + unregister_algorithm, +) +from .api_jwt import PyJWT, decode, encode +from .exceptions import ( + DecodeError, + ExpiredSignatureError, + ImmatureSignatureError, + InvalidAlgorithmError, + InvalidAudienceError, + InvalidIssuedAtError, + InvalidIssuerError, + InvalidKeyError, + InvalidSignatureError, + InvalidTokenError, + MissingRequiredClaimError, + PyJWKClientError, + PyJWKError, + PyJWKSetError, + PyJWTError, +) +from .jwks_client import PyJWKClient + +__version__ = "2.3.0" + +__title__ = "PyJWT" +__description__ = "JSON Web Token implementation in Python" +__url__ = "https://pyjwt.readthedocs.io" +__uri__ = __url__ +__doc__ = __description__ + " <" + __uri__ + ">" + +__author__ = "José Padilla" +__email__ = "hello@jpadilla.com" + +__license__ = "MIT" +__copyright__ = "Copyright 2015-2020 José Padilla" + + +__all__ = [ + "PyJWS", + "PyJWT", + "PyJWKClient", + "PyJWK", + "PyJWKSet", + "decode", + "encode", + "get_unverified_header", + "register_algorithm", + "unregister_algorithm", + # Exceptions + "DecodeError", + "ExpiredSignatureError", + "ImmatureSignatureError", + "InvalidAlgorithmError", + "InvalidAudienceError", + "InvalidIssuedAtError", + "InvalidIssuerError", + "InvalidKeyError", + "InvalidSignatureError", + "InvalidTokenError", + "MissingRequiredClaimError", + "PyJWKClientError", + "PyJWKError", + "PyJWKSetError", + "PyJWTError", +] diff --git a/contrib/python/PyJWT/py3/jwt/algorithms.py b/contrib/python/PyJWT/py3/jwt/algorithms.py new file mode 100644 index 0000000000..1f8865afbd --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/algorithms.py @@ -0,0 +1,674 @@ +import hashlib +import hmac +import json + +from .exceptions import InvalidKeyError +from .utils import ( + base64url_decode, + base64url_encode, + der_to_raw_signature, + force_bytes, + from_base64url_uint, + raw_to_der_signature, + to_base64url_uint, +) + +try: + import cryptography.exceptions + from cryptography.exceptions import InvalidSignature + from cryptography.hazmat.primitives import hashes + from cryptography.hazmat.primitives.asymmetric import ec, padding + from cryptography.hazmat.primitives.asymmetric.ec import ( + EllipticCurvePrivateKey, + EllipticCurvePublicKey, + ) + from cryptography.hazmat.primitives.asymmetric.ed448 import ( + Ed448PrivateKey, + Ed448PublicKey, + ) + from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, + ) + from cryptography.hazmat.primitives.asymmetric.rsa import ( + RSAPrivateKey, + RSAPrivateNumbers, + RSAPublicKey, + RSAPublicNumbers, + rsa_crt_dmp1, + rsa_crt_dmq1, + rsa_crt_iqmp, + rsa_recover_prime_factors, + ) + from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, + PublicFormat, + load_pem_private_key, + load_pem_public_key, + load_ssh_public_key, + ) + + has_crypto = True +except ModuleNotFoundError: + has_crypto = False + +requires_cryptography = { + "RS256", + "RS384", + "RS512", + "ES256", + "ES256K", + "ES384", + "ES521", + "ES512", + "PS256", + "PS384", + "PS512", + "EdDSA", +} + + +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), + "ES256K": ECAlgorithm(ECAlgorithm.SHA256), + "ES384": ECAlgorithm(ECAlgorithm.SHA384), + "ES521": ECAlgorithm(ECAlgorithm.SHA512), + "ES512": ECAlgorithm( + ECAlgorithm.SHA512 + ), # Backward compat for #219 fix + "PS256": RSAPSSAlgorithm(RSAPSSAlgorithm.SHA256), + "PS384": RSAPSSAlgorithm(RSAPSSAlgorithm.SHA384), + "PS512": RSAPSSAlgorithm(RSAPSSAlgorithm.SHA512), + "EdDSA": OKPAlgorithm(), + } + ) + + return default_algorithms + + +class Algorithm: + """ + The interface for an algorithm used to sign and verify tokens. + """ + + 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): + """ + Placeholder for use when no signing or verification + operations are required. + """ + + 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): + """ + Performs signing and verification operations using HMAC + and the specified hash function. + """ + + 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": base64url_encode(force_bytes(key_obj)).decode(), + "kty": "oct", + } + ) + + @staticmethod + def from_jwk(jwk): + try: + if isinstance(jwk, str): + obj = json.loads(jwk) + elif isinstance(jwk, dict): + obj = jwk + else: + raise ValueError + except ValueError: + raise InvalidKeyError("Key is not valid JSON") + + 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 hmac.compare_digest(sig, self.sign(msg, key)) + + +if has_crypto: + + class RSAAlgorithm(Algorithm): + """ + Performs signing and verification operations using + RSASSA-PKCS-v1_5 and the specified hash function. + """ + + 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, RSAPublicKey)): + return key + + if not isinstance(key, (bytes, str)): + raise TypeError("Expecting a PEM-formatted key.") + + key = force_bytes(key) + + try: + if key.startswith(b"ssh-rsa"): + key = load_ssh_public_key(key) + else: + key = load_pem_private_key(key, password=None) + except ValueError: + key = load_pem_public_key(key) + return key + + @staticmethod + def to_jwk(key_obj): + obj = None + + if getattr(key_obj, "private_numbers", None): + # Private key + numbers = key_obj.private_numbers() + + obj = { + "kty": "RSA", + "key_ops": ["sign"], + "n": to_base64url_uint(numbers.public_numbers.n).decode(), + "e": to_base64url_uint(numbers.public_numbers.e).decode(), + "d": to_base64url_uint(numbers.d).decode(), + "p": to_base64url_uint(numbers.p).decode(), + "q": to_base64url_uint(numbers.q).decode(), + "dp": to_base64url_uint(numbers.dmp1).decode(), + "dq": to_base64url_uint(numbers.dmq1).decode(), + "qi": to_base64url_uint(numbers.iqmp).decode(), + } + + elif getattr(key_obj, "verify", None): + # Public key + numbers = key_obj.public_numbers() + + obj = { + "kty": "RSA", + "key_ops": ["verify"], + "n": to_base64url_uint(numbers.n).decode(), + "e": to_base64url_uint(numbers.e).decode(), + } + else: + raise InvalidKeyError("Not a public or private key") + + return json.dumps(obj) + + @staticmethod + def from_jwk(jwk): + try: + if isinstance(jwk, str): + obj = json.loads(jwk) + elif isinstance(jwk, dict): + obj = jwk + else: + raise ValueError + except ValueError: + raise InvalidKeyError("Key is not valid JSON") + + if obj.get("kty") != "RSA": + raise InvalidKeyError("Not an RSA key") + + if "d" in obj and "e" in obj and "n" in obj: + # Private key + if "oth" in obj: + raise InvalidKeyError( + "Unsupported RSA private key: > 2 primes not supported" + ) + + 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 and 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() + elif "n" in obj and "e" in obj: + # Public key + numbers = RSAPublicNumbers( + from_base64url_uint(obj["e"]), + from_base64url_uint(obj["n"]), + ) + + return numbers.public_key() + else: + 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): + """ + Performs signing and verification operations using + ECDSA and the specified hash function + """ + + 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, EllipticCurvePublicKey)): + return key + + if not isinstance(key, (bytes, str)): + raise TypeError("Expecting a PEM-formatted key.") + + key = force_bytes(key) + + # Attempt to load key. We don't know if it's + # a Signing Key or a Verifying Key, so we try + # the Verifying Key first. + try: + if key.startswith(b"ecdsa-sha2-"): + key = load_ssh_public_key(key) + else: + key = load_pem_public_key(key) + except ValueError: + key = load_pem_private_key(key, password=None) + + 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 + + try: + if isinstance(key, EllipticCurvePrivateKey): + key = key.public_key() + key.verify(der_sig, msg, ec.ECDSA(self.hash_alg())) + return True + except InvalidSignature: + return False + + @staticmethod + def from_jwk(jwk): + try: + if isinstance(jwk, str): + obj = json.loads(jwk) + elif isinstance(jwk, dict): + obj = jwk + else: + raise ValueError + except ValueError: + raise InvalidKeyError("Key is not valid JSON") + + if obj.get("kty") != "EC": + raise InvalidKeyError("Not an Elliptic curve key") + + if "x" not in obj or "y" not in obj: + raise InvalidKeyError("Not an Elliptic curve key") + + x = base64url_decode(obj.get("x")) + y = base64url_decode(obj.get("y")) + + curve = obj.get("crv") + if curve == "P-256": + if len(x) == len(y) == 32: + curve_obj = ec.SECP256R1() + else: + raise InvalidKeyError("Coords should be 32 bytes for curve P-256") + elif curve == "P-384": + if len(x) == len(y) == 48: + curve_obj = ec.SECP384R1() + else: + raise InvalidKeyError("Coords should be 48 bytes for curve P-384") + elif curve == "P-521": + if len(x) == len(y) == 66: + curve_obj = ec.SECP521R1() + else: + raise InvalidKeyError("Coords should be 66 bytes for curve P-521") + elif curve == "secp256k1": + if len(x) == len(y) == 32: + curve_obj = ec.SECP256K1() + else: + raise InvalidKeyError( + "Coords should be 32 bytes for curve secp256k1" + ) + else: + raise InvalidKeyError(f"Invalid curve: {curve}") + + public_numbers = ec.EllipticCurvePublicNumbers( + x=int.from_bytes(x, byteorder="big"), + y=int.from_bytes(y, byteorder="big"), + curve=curve_obj, + ) + + if "d" not in obj: + return public_numbers.public_key() + + d = base64url_decode(obj.get("d")) + if len(d) != len(x): + raise InvalidKeyError( + "D should be {} bytes for curve {}", len(x), curve + ) + + return ec.EllipticCurvePrivateNumbers( + int.from_bytes(d, byteorder="big"), public_numbers + ).private_key() + + class RSAPSSAlgorithm(RSAAlgorithm): + """ + Performs a signature using RSASSA-PSS with MGF1 + """ + + 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 + + class OKPAlgorithm(Algorithm): + """ + Performs signing and verification operations using EdDSA + + This class requires ``cryptography>=2.6`` to be installed. + """ + + def __init__(self, **kwargs): + pass + + def prepare_key(self, key): + + if isinstance( + key, + (Ed25519PrivateKey, Ed25519PublicKey, Ed448PrivateKey, Ed448PublicKey), + ): + return key + + if isinstance(key, (bytes, str)): + if isinstance(key, str): + key = key.encode("utf-8") + str_key = key.decode("utf-8") + + if "-----BEGIN PUBLIC" in str_key: + return load_pem_public_key(key) + if "-----BEGIN PRIVATE" in str_key: + return load_pem_private_key(key, password=None) + if str_key[0:4] == "ssh-": + return load_ssh_public_key(key) + + raise TypeError("Expecting a PEM-formatted or OpenSSH key.") + + def sign(self, msg, key): + """ + Sign a message ``msg`` using the EdDSA private key ``key`` + :param str|bytes msg: Message to sign + :param Ed25519PrivateKey}Ed448PrivateKey key: A :class:`.Ed25519PrivateKey` + or :class:`.Ed448PrivateKey` iinstance + :return bytes signature: The signature, as bytes + """ + msg = bytes(msg, "utf-8") if type(msg) is not bytes else msg + return key.sign(msg) + + def verify(self, msg, key, sig): + """ + Verify a given ``msg`` against a signature ``sig`` using the EdDSA key ``key`` + + :param str|bytes sig: EdDSA signature to check ``msg`` against + :param str|bytes msg: Message to sign + :param Ed25519PrivateKey|Ed25519PublicKey|Ed448PrivateKey|Ed448PublicKey key: + A private or public EdDSA key instance + :return bool verified: True if signature is valid, False if not. + """ + try: + msg = bytes(msg, "utf-8") if type(msg) is not bytes else msg + sig = bytes(sig, "utf-8") if type(sig) is not bytes else sig + + if isinstance(key, (Ed25519PrivateKey, Ed448PrivateKey)): + key = key.public_key() + key.verify(sig, msg) + return True # If no exception was raised, the signature is valid. + except cryptography.exceptions.InvalidSignature: + return False + + @staticmethod + def to_jwk(key): + if isinstance(key, (Ed25519PublicKey, Ed448PublicKey)): + x = key.public_bytes( + encoding=Encoding.Raw, + format=PublicFormat.Raw, + ) + crv = "Ed25519" if isinstance(key, Ed25519PublicKey) else "Ed448" + return json.dumps( + { + "x": base64url_encode(force_bytes(x)).decode(), + "kty": "OKP", + "crv": crv, + } + ) + + if isinstance(key, (Ed25519PrivateKey, Ed448PrivateKey)): + d = key.private_bytes( + encoding=Encoding.Raw, + format=PrivateFormat.Raw, + encryption_algorithm=NoEncryption(), + ) + + x = key.public_key().public_bytes( + encoding=Encoding.Raw, + format=PublicFormat.Raw, + ) + + crv = "Ed25519" if isinstance(key, Ed25519PrivateKey) else "Ed448" + return json.dumps( + { + "x": base64url_encode(force_bytes(x)).decode(), + "d": base64url_encode(force_bytes(d)).decode(), + "kty": "OKP", + "crv": crv, + } + ) + + raise InvalidKeyError("Not a public or private key") + + @staticmethod + def from_jwk(jwk): + try: + if isinstance(jwk, str): + obj = json.loads(jwk) + elif isinstance(jwk, dict): + obj = jwk + else: + raise ValueError + except ValueError: + raise InvalidKeyError("Key is not valid JSON") + + if obj.get("kty") != "OKP": + raise InvalidKeyError("Not an Octet Key Pair") + + curve = obj.get("crv") + if curve != "Ed25519" and curve != "Ed448": + raise InvalidKeyError(f"Invalid curve: {curve}") + + if "x" not in obj: + raise InvalidKeyError('OKP should have "x" parameter') + x = base64url_decode(obj.get("x")) + + try: + if "d" not in obj: + if curve == "Ed25519": + return Ed25519PublicKey.from_public_bytes(x) + return Ed448PublicKey.from_public_bytes(x) + d = base64url_decode(obj.get("d")) + if curve == "Ed25519": + return Ed25519PrivateKey.from_private_bytes(d) + return Ed448PrivateKey.from_private_bytes(d) + except ValueError as err: + raise InvalidKeyError("Invalid key parameter") from err diff --git a/contrib/python/PyJWT/py3/jwt/api_jwk.py b/contrib/python/PyJWT/py3/jwt/api_jwk.py new file mode 100644 index 0000000000..a0f6364da0 --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/api_jwk.py @@ -0,0 +1,97 @@ +import json + +from .algorithms import get_default_algorithms +from .exceptions import InvalidKeyError, PyJWKError, PyJWKSetError + + +class PyJWK: + def __init__(self, jwk_data, algorithm=None): + self._algorithms = get_default_algorithms() + self._jwk_data = jwk_data + + kty = self._jwk_data.get("kty", None) + if not kty: + raise InvalidKeyError("kty is not found: %s" % self._jwk_data) + + if not algorithm and isinstance(self._jwk_data, dict): + algorithm = self._jwk_data.get("alg", None) + + if not algorithm: + # Determine alg with kty (and crv). + crv = self._jwk_data.get("crv", None) + if kty == "EC": + if crv == "P-256" or not crv: + algorithm = "ES256" + elif crv == "P-384": + algorithm = "ES384" + elif crv == "P-521": + algorithm = "ES512" + elif crv == "secp256k1": + algorithm = "ES256K" + else: + raise InvalidKeyError("Unsupported crv: %s" % crv) + elif kty == "RSA": + algorithm = "RS256" + elif kty == "oct": + algorithm = "HS256" + elif kty == "OKP": + if not crv: + raise InvalidKeyError("crv is not found: %s" % self._jwk_data) + if crv == "Ed25519": + algorithm = "EdDSA" + else: + raise InvalidKeyError("Unsupported crv: %s" % crv) + else: + raise InvalidKeyError("Unsupported kty: %s" % kty) + + self.Algorithm = self._algorithms.get(algorithm) + + if not self.Algorithm: + raise PyJWKError("Unable to find a algorithm for key: %s" % self._jwk_data) + + self.key = self.Algorithm.from_jwk(self._jwk_data) + + @staticmethod + def from_dict(obj, algorithm=None): + return PyJWK(obj, algorithm) + + @staticmethod + def from_json(data, algorithm=None): + obj = json.loads(data) + return PyJWK.from_dict(obj, algorithm) + + @property + def key_type(self): + return self._jwk_data.get("kty", None) + + @property + def key_id(self): + return self._jwk_data.get("kid", None) + + @property + def public_key_use(self): + return self._jwk_data.get("use", None) + + +class PyJWKSet: + def __init__(self, keys): + self.keys = [] + + if not keys or not isinstance(keys, list): + raise PyJWKSetError("Invalid JWK Set value") + + if len(keys) == 0: + raise PyJWKSetError("The JWK Set did not contain any keys") + + for key in keys: + self.keys.append(PyJWK(key)) + + @staticmethod + def from_dict(obj): + keys = obj.get("keys", []) + return PyJWKSet(keys) + + @staticmethod + def from_json(data): + obj = json.loads(data) + return PyJWKSet.from_dict(obj) diff --git a/contrib/python/PyJWT/py3/jwt/api_jws.py b/contrib/python/PyJWT/py3/jwt/api_jws.py new file mode 100644 index 0000000000..f85072e05e --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/api_jws.py @@ -0,0 +1,259 @@ +import binascii +import json +from collections.abc import Mapping +from typing import Any, Dict, List, Optional, Type + +from .algorithms import ( + Algorithm, + get_default_algorithms, + has_crypto, + requires_cryptography, +) +from .exceptions import ( + DecodeError, + InvalidAlgorithmError, + InvalidSignatureError, + InvalidTokenError, +) +from .utils import base64url_decode, base64url_encode + + +class PyJWS: + header_typ = "JWT" + + def __init__(self, algorithms=None, options=None): + self._algorithms = get_default_algorithms() + self._valid_algs = ( + set(algorithms) if algorithms is not None else set(self._algorithms) + ) + + # Remove algorithms that aren't on the whitelist + for key in list(self._algorithms.keys()): + if key not in self._valid_algs: + del self._algorithms[key] + + if options is None: + options = {} + self.options = {**self._get_default_options(), **options} + + @staticmethod + def _get_default_options(): + return {"verify_signature": True} + + def register_algorithm(self, alg_id, alg_obj): + """ + Registers a new Algorithm for use when creating and verifying tokens. + """ + if alg_id in self._algorithms: + raise ValueError("Algorithm already has a handler.") + + if not isinstance(alg_obj, Algorithm): + raise TypeError("Object is not of type `Algorithm`") + + self._algorithms[alg_id] = alg_obj + self._valid_algs.add(alg_id) + + def unregister_algorithm(self, alg_id): + """ + Unregisters an Algorithm for use when creating and verifying tokens + Throws KeyError if algorithm is not registered. + """ + if alg_id not in self._algorithms: + raise KeyError( + "The specified algorithm could not be removed" + " because it is not registered." + ) + + del self._algorithms[alg_id] + self._valid_algs.remove(alg_id) + + def get_algorithms(self): + """ + Returns a list of supported values for the 'alg' parameter. + """ + return list(self._valid_algs) + + def encode( + self, + payload: bytes, + key: str, + algorithm: Optional[str] = "HS256", + headers: Optional[Dict] = None, + json_encoder: Optional[Type[json.JSONEncoder]] = None, + ) -> str: + segments = [] + + if algorithm is None: + algorithm = "none" + + # Prefer headers["alg"] if present to algorithm parameter. + if headers and "alg" in headers and headers["alg"]: + algorithm = headers["alg"] + + # Header + header = {"typ": self.header_typ, "alg": algorithm} + + if headers: + self._validate_headers(headers) + header.update(headers) + if not header["typ"]: + del header["typ"] + + json_header = json.dumps( + header, separators=(",", ":"), cls=json_encoder + ).encode() + + segments.append(base64url_encode(json_header)) + segments.append(base64url_encode(payload)) + + # Segments + signing_input = b".".join(segments) + try: + alg_obj = self._algorithms[algorithm] + key = alg_obj.prepare_key(key) + signature = alg_obj.sign(signing_input, key) + + except KeyError as e: + if not has_crypto and algorithm in requires_cryptography: + raise NotImplementedError( + "Algorithm '%s' could not be found. Do you have cryptography " + "installed?" % algorithm + ) from e + else: + raise NotImplementedError("Algorithm not supported") from e + + segments.append(base64url_encode(signature)) + + encoded_string = b".".join(segments) + + return encoded_string.decode("utf-8") + + def decode_complete( + self, + jwt: str, + key: str = "", + algorithms: List[str] = None, + options: Dict = None, + **kwargs, + ) -> Dict[str, Any]: + if options is None: + options = {} + merged_options = {**self.options, **options} + verify_signature = merged_options["verify_signature"] + + if verify_signature and not algorithms: + raise DecodeError( + 'It is required that you pass in a value for the "algorithms" argument when calling decode().' + ) + + payload, signing_input, header, signature = self._load(jwt) + + if verify_signature: + self._verify_signature(signing_input, header, signature, key, algorithms) + + return { + "payload": payload, + "header": header, + "signature": signature, + } + + def decode( + self, + jwt: str, + key: str = "", + algorithms: List[str] = None, + options: Dict = None, + **kwargs, + ) -> str: + decoded = self.decode_complete(jwt, key, algorithms, options, **kwargs) + return decoded["payload"] + + def get_unverified_header(self, jwt): + """Returns back the JWT header parameters as a dict() + + Note: The signature is not verified so the header parameters + should not be fully trusted until signature verification is complete + """ + headers = self._load(jwt)[2] + self._validate_headers(headers) + + return headers + + def _load(self, jwt): + if isinstance(jwt, str): + jwt = jwt.encode("utf-8") + + if not isinstance(jwt, bytes): + raise DecodeError(f"Invalid token type. Token must be a {bytes}") + + try: + signing_input, crypto_segment = jwt.rsplit(b".", 1) + header_segment, payload_segment = signing_input.split(b".", 1) + except ValueError as err: + raise DecodeError("Not enough segments") from err + + try: + header_data = base64url_decode(header_segment) + except (TypeError, binascii.Error) as err: + raise DecodeError("Invalid header padding") from err + + try: + header = json.loads(header_data) + except ValueError as e: + raise DecodeError("Invalid header string: %s" % e) from e + + if not isinstance(header, Mapping): + raise DecodeError("Invalid header string: must be a json object") + + try: + payload = base64url_decode(payload_segment) + except (TypeError, binascii.Error) as err: + raise DecodeError("Invalid payload padding") from err + + try: + signature = base64url_decode(crypto_segment) + except (TypeError, binascii.Error) as err: + raise DecodeError("Invalid crypto padding") from err + + return (payload, signing_input, header, signature) + + def _verify_signature( + self, + signing_input, + header, + signature, + key="", + algorithms=None, + ): + + alg = header.get("alg") + + if algorithms is not None and alg not in algorithms: + raise InvalidAlgorithmError("The specified alg value is not allowed") + + try: + alg_obj = self._algorithms[alg] + key = alg_obj.prepare_key(key) + + if not alg_obj.verify(signing_input, key, signature): + raise InvalidSignatureError("Signature verification failed") + + except KeyError as e: + raise InvalidAlgorithmError("Algorithm not supported") from e + + def _validate_headers(self, headers): + if "kid" in headers: + self._validate_kid(headers["kid"]) + + def _validate_kid(self, kid): + if not isinstance(kid, str): + raise InvalidTokenError("Key ID header parameter must be a string") + + +_jws_global_obj = PyJWS() +encode = _jws_global_obj.encode +decode_complete = _jws_global_obj.decode_complete +decode = _jws_global_obj.decode +register_algorithm = _jws_global_obj.register_algorithm +unregister_algorithm = _jws_global_obj.unregister_algorithm +get_unverified_header = _jws_global_obj.get_unverified_header diff --git a/contrib/python/PyJWT/py3/jwt/api_jwt.py b/contrib/python/PyJWT/py3/jwt/api_jwt.py new file mode 100644 index 0000000000..f3b55d360e --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/api_jwt.py @@ -0,0 +1,221 @@ +import json +from calendar import timegm +from collections.abc import Iterable, Mapping +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional, Type, Union + +from . import api_jws +from .exceptions import ( + DecodeError, + ExpiredSignatureError, + ImmatureSignatureError, + InvalidAudienceError, + InvalidIssuedAtError, + InvalidIssuerError, + MissingRequiredClaimError, +) + + +class PyJWT: + def __init__(self, options=None): + if options is None: + options = {} + self.options = {**self._get_default_options(), **options} + + @staticmethod + def _get_default_options() -> Dict[str, Union[bool, List[str]]]: + return { + "verify_signature": True, + "verify_exp": True, + "verify_nbf": True, + "verify_iat": True, + "verify_aud": True, + "verify_iss": True, + "require": [], + } + + def encode( + self, + payload: Dict[str, Any], + key: str, + algorithm: Optional[str] = "HS256", + headers: Optional[Dict] = None, + json_encoder: Optional[Type[json.JSONEncoder]] = None, + ) -> str: + # Check that we get a mapping + if not isinstance(payload, Mapping): + raise TypeError( + "Expecting a mapping object, as JWT only supports " + "JSON objects as payloads." + ) + + # Payload + payload = payload.copy() + for time_claim in ["exp", "iat", "nbf"]: + # Convert datetime to a intDate value in known time-format claims + if isinstance(payload.get(time_claim), datetime): + payload[time_claim] = timegm(payload[time_claim].utctimetuple()) + + json_payload = json.dumps( + payload, separators=(",", ":"), cls=json_encoder + ).encode("utf-8") + + return api_jws.encode(json_payload, key, algorithm, headers, json_encoder) + + def decode_complete( + self, + jwt: str, + key: str = "", + algorithms: List[str] = None, + options: Dict = None, + **kwargs, + ) -> Dict[str, Any]: + if options is None: + options = {"verify_signature": True} + else: + options.setdefault("verify_signature", True) + + if not options["verify_signature"]: + options.setdefault("verify_exp", False) + options.setdefault("verify_nbf", False) + options.setdefault("verify_iat", False) + options.setdefault("verify_aud", False) + options.setdefault("verify_iss", False) + + if options["verify_signature"] and not algorithms: + raise DecodeError( + 'It is required that you pass in a value for the "algorithms" argument when calling decode().' + ) + + decoded = api_jws.decode_complete( + jwt, + key=key, + algorithms=algorithms, + options=options, + **kwargs, + ) + + try: + payload = json.loads(decoded["payload"]) + except ValueError as e: + raise DecodeError("Invalid payload string: %s" % e) + if not isinstance(payload, dict): + raise DecodeError("Invalid payload string: must be a json object") + + merged_options = {**self.options, **options} + self._validate_claims(payload, merged_options, **kwargs) + + decoded["payload"] = payload + return decoded + + def decode( + self, + jwt: str, + key: str = "", + algorithms: List[str] = None, + options: Dict = None, + **kwargs, + ) -> Dict[str, Any]: + decoded = self.decode_complete(jwt, key, algorithms, options, **kwargs) + return decoded["payload"] + + def _validate_claims( + self, payload, options, audience=None, issuer=None, leeway=0, **kwargs + ): + if isinstance(leeway, timedelta): + leeway = leeway.total_seconds() + + if not isinstance(audience, (bytes, str, type(None), Iterable)): + raise TypeError("audience must be a string, iterable, or None") + + self._validate_required_claims(payload, options) + + now = timegm(datetime.now(tz=timezone.utc).utctimetuple()) + + if "iat" in payload and options["verify_iat"]: + self._validate_iat(payload, now, leeway) + + if "nbf" in payload and options["verify_nbf"]: + self._validate_nbf(payload, now, leeway) + + if "exp" in payload and options["verify_exp"]: + self._validate_exp(payload, now, leeway) + + if options["verify_iss"]: + self._validate_iss(payload, issuer) + + if options["verify_aud"]: + self._validate_aud(payload, audience) + + def _validate_required_claims(self, payload, options): + for claim in options["require"]: + if payload.get(claim) is None: + raise MissingRequiredClaimError(claim) + + def _validate_iat(self, payload, now, leeway): + try: + int(payload["iat"]) + except ValueError: + raise InvalidIssuedAtError("Issued At claim (iat) must be an integer.") + + def _validate_nbf(self, payload, now, leeway): + try: + nbf = int(payload["nbf"]) + except ValueError: + raise DecodeError("Not Before claim (nbf) must be an integer.") + + if nbf > (now + leeway): + raise ImmatureSignatureError("The token is not yet valid (nbf)") + + def _validate_exp(self, payload, now, leeway): + try: + exp = int(payload["exp"]) + except ValueError: + raise DecodeError("Expiration Time claim (exp) must be an" " integer.") + + if exp < (now - leeway): + raise ExpiredSignatureError("Signature has expired") + + def _validate_aud(self, payload, audience): + if audience is None: + if "aud" not in payload or not payload["aud"]: + return + # Application did not specify an audience, but + # the token has the 'aud' claim + raise InvalidAudienceError("Invalid audience") + + if "aud" not in payload or not payload["aud"]: + # Application specified an audience, but it could not be + # verified since the token does not contain a claim. + raise MissingRequiredClaimError("aud") + + audience_claims = payload["aud"] + + if isinstance(audience_claims, str): + audience_claims = [audience_claims] + if not isinstance(audience_claims, list): + raise InvalidAudienceError("Invalid claim format in token") + if any(not isinstance(c, str) for c in audience_claims): + raise InvalidAudienceError("Invalid claim format in token") + + if isinstance(audience, str): + audience = [audience] + + if all(aud not in audience_claims for aud in audience): + raise InvalidAudienceError("Invalid audience") + + def _validate_iss(self, payload, issuer): + if issuer is None: + return + + if "iss" not in payload: + raise MissingRequiredClaimError("iss") + + if payload["iss"] != issuer: + raise InvalidIssuerError("Invalid issuer") + + +_jwt_global_obj = PyJWT() +encode = _jwt_global_obj.encode +decode_complete = _jwt_global_obj.decode_complete +decode = _jwt_global_obj.decode diff --git a/contrib/python/PyJWT/py3/jwt/exceptions.py b/contrib/python/PyJWT/py3/jwt/exceptions.py new file mode 100644 index 0000000000..308899aa6a --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/exceptions.py @@ -0,0 +1,66 @@ +class PyJWTError(Exception): + """ + Base class for all exceptions + """ + + pass + + +class InvalidTokenError(PyJWTError): + pass + + +class DecodeError(InvalidTokenError): + pass + + +class InvalidSignatureError(DecodeError): + pass + + +class ExpiredSignatureError(InvalidTokenError): + pass + + +class InvalidAudienceError(InvalidTokenError): + pass + + +class InvalidIssuerError(InvalidTokenError): + pass + + +class InvalidIssuedAtError(InvalidTokenError): + pass + + +class ImmatureSignatureError(InvalidTokenError): + pass + + +class InvalidKeyError(PyJWTError): + pass + + +class InvalidAlgorithmError(InvalidTokenError): + pass + + +class MissingRequiredClaimError(InvalidTokenError): + def __init__(self, claim): + self.claim = claim + + def __str__(self): + return 'Token is missing the "%s" claim' % self.claim + + +class PyJWKError(PyJWTError): + pass + + +class PyJWKSetError(PyJWTError): + pass + + +class PyJWKClientError(PyJWTError): + pass diff --git a/contrib/python/PyJWT/py3/jwt/help.py b/contrib/python/PyJWT/py3/jwt/help.py new file mode 100644 index 0000000000..d8f2302421 --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/help.py @@ -0,0 +1,60 @@ +import json +import platform +import sys + +from . import __version__ as pyjwt_version + +try: + import cryptography +except ModuleNotFoundError: + cryptography = None # type: ignore + + +def info(): + """ + Generate information for a bug report. + Based on the requests package help utility module. + """ + try: + platform_info = { + "system": platform.system(), + "release": platform.release(), + } + except OSError: + platform_info = {"system": "Unknown", "release": "Unknown"} + + implementation = platform.python_implementation() + + if implementation == "CPython": + implementation_version = platform.python_version() + elif implementation == "PyPy": + implementation_version = "{}.{}.{}".format( + sys.pypy_version_info.major, + sys.pypy_version_info.minor, + sys.pypy_version_info.micro, + ) + if sys.pypy_version_info.releaselevel != "final": + implementation_version = "".join( + [implementation_version, sys.pypy_version_info.releaselevel] + ) + else: + implementation_version = "Unknown" + + return { + "platform": platform_info, + "implementation": { + "name": implementation, + "version": implementation_version, + }, + "cryptography": {"version": getattr(cryptography, "__version__", "")}, + "pyjwt": {"version": pyjwt_version}, + } + + +def main(): + """Pretty-print the bug information as JSON.""" + print(json.dumps(info(), sort_keys=True, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/contrib/python/PyJWT/py3/jwt/jwks_client.py b/contrib/python/PyJWT/py3/jwt/jwks_client.py new file mode 100644 index 0000000000..767b7179db --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/jwks_client.py @@ -0,0 +1,59 @@ +import json +import urllib.request +from functools import lru_cache +from typing import Any, List + +from .api_jwk import PyJWK, PyJWKSet +from .api_jwt import decode_complete as decode_token +from .exceptions import PyJWKClientError + + +class PyJWKClient: + def __init__(self, uri: str, cache_keys: bool = True, max_cached_keys: int = 16): + self.uri = uri + if cache_keys: + # Cache signing keys + # Ignore mypy (https://github.com/python/mypy/issues/2427) + self.get_signing_key = lru_cache(maxsize=max_cached_keys)(self.get_signing_key) # type: ignore + + def fetch_data(self) -> Any: + with urllib.request.urlopen(self.uri) as response: + return json.load(response) + + def get_jwk_set(self) -> PyJWKSet: + data = self.fetch_data() + return PyJWKSet.from_dict(data) + + def get_signing_keys(self) -> List[PyJWK]: + jwk_set = self.get_jwk_set() + signing_keys = [ + jwk_set_key + for jwk_set_key in jwk_set.keys + if jwk_set_key.public_key_use in ["sig", None] and jwk_set_key.key_id + ] + + if not signing_keys: + raise PyJWKClientError("The JWKS endpoint did not contain any signing keys") + + return signing_keys + + def get_signing_key(self, kid: str) -> PyJWK: + signing_keys = self.get_signing_keys() + signing_key = None + + for key in signing_keys: + if key.key_id == kid: + signing_key = key + break + + if not signing_key: + raise PyJWKClientError( + f'Unable to find a signing key that matches: "{kid}"' + ) + + return signing_key + + def get_signing_key_from_jwt(self, token: str) -> PyJWK: + unverified = decode_token(token, options={"verify_signature": False}) + header = unverified["header"] + return self.get_signing_key(header.get("kid")) diff --git a/contrib/python/PyJWT/py3/jwt/py.typed b/contrib/python/PyJWT/py3/jwt/py.typed new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/py.typed diff --git a/contrib/python/PyJWT/py3/jwt/utils.py b/contrib/python/PyJWT/py3/jwt/utils.py new file mode 100644 index 0000000000..9dde10cf8e --- /dev/null +++ b/contrib/python/PyJWT/py3/jwt/utils.py @@ -0,0 +1,99 @@ +import base64 +import binascii +from typing import Any, Union + +try: + from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurve + from cryptography.hazmat.primitives.asymmetric.utils import ( + decode_dss_signature, + encode_dss_signature, + ) +except ModuleNotFoundError: + EllipticCurve = Any # type: ignore + + +def force_bytes(value: Union[str, bytes]) -> bytes: + if isinstance(value, str): + return value.encode("utf-8") + elif isinstance(value, bytes): + return value + else: + raise TypeError("Expected a string value") + + +def base64url_decode(input: Union[str, bytes]) -> bytes: + if isinstance(input, str): + input = input.encode("ascii") + + rem = len(input) % 4 + + if rem > 0: + input += b"=" * (4 - rem) + + return base64.urlsafe_b64decode(input) + + +def base64url_encode(input: bytes) -> bytes: + return base64.urlsafe_b64encode(input).replace(b"=", b"") + + +def to_base64url_uint(val: int) -> bytes: + if val < 0: + raise ValueError("Must be a positive integer") + + int_bytes = bytes_from_int(val) + + if len(int_bytes) == 0: + int_bytes = b"\x00" + + return base64url_encode(int_bytes) + + +def from_base64url_uint(val: Union[str, bytes]) -> int: + if isinstance(val, str): + val = val.encode("ascii") + + data = base64url_decode(val) + return int.from_bytes(data, byteorder="big") + + +def number_to_bytes(num: int, num_bytes: int) -> bytes: + padded_hex = "%0*x" % (2 * num_bytes, num) + return binascii.a2b_hex(padded_hex.encode("ascii")) + + +def bytes_to_number(string: bytes) -> int: + return int(binascii.b2a_hex(string), 16) + + +def bytes_from_int(val: int) -> bytes: + remaining = val + byte_length = 0 + + while remaining != 0: + remaining >>= 8 + byte_length += 1 + + return val.to_bytes(byte_length, "big", signed=False) + + +def der_to_raw_signature(der_sig: bytes, curve: EllipticCurve) -> bytes: + num_bits = curve.key_size + num_bytes = (num_bits + 7) // 8 + + r, s = decode_dss_signature(der_sig) + + return number_to_bytes(r, num_bytes) + number_to_bytes(s, num_bytes) + + +def raw_to_der_signature(raw_sig: bytes, curve: EllipticCurve) -> bytes: + num_bits = curve.key_size + num_bytes = (num_bits + 7) // 8 + + if len(raw_sig) != 2 * num_bytes: + raise ValueError("Invalid signature") + + r = bytes_to_number(raw_sig[:num_bytes]) + s = bytes_to_number(raw_sig[num_bytes:]) + + return encode_dss_signature(r, s) diff --git a/contrib/python/PyJWT/py3/ya.make b/contrib/python/PyJWT/py3/ya.make new file mode 100644 index 0000000000..0cbee2bb2e --- /dev/null +++ b/contrib/python/PyJWT/py3/ya.make @@ -0,0 +1,39 @@ +# Generated by devtools/yamaker (pypi). + +PY3_LIBRARY() + +VERSION(2.3.0) + +LICENSE(MIT) + +PEERDIR( + contrib/python/cryptography +) + +NO_LINT() + +NO_CHECK_IMPORTS( + jwt.contrib.* +) + +PY_SRCS( + TOP_LEVEL + jwt/__init__.py + jwt/algorithms.py + jwt/api_jwk.py + jwt/api_jws.py + jwt/api_jwt.py + jwt/exceptions.py + jwt/help.py + jwt/jwks_client.py + jwt/utils.py +) + +RESOURCE_FILES( + PREFIX contrib/python/PyJWT/py3/ + .dist-info/METADATA + .dist-info/top_level.txt + jwt/py.typed +) + +END() |