mirror of
https://github.com/Tautulli/Tautulli.git
synced 2025-07-06 21:21:15 -07:00
Bump pyjwt from 2.3.0 to 2.4.0 (#1743)
* Bump pyjwt from 2.3.0 to 2.4.0 Bumps [pyjwt](https://github.com/jpadilla/pyjwt) from 2.3.0 to 2.4.0. - [Release notes](https://github.com/jpadilla/pyjwt/releases) - [Changelog](https://github.com/jpadilla/pyjwt/blob/master/CHANGELOG.rst) - [Commits](https://github.com/jpadilla/pyjwt/compare/2.3.0...2.4.0) --- updated-dependencies: - dependency-name: pyjwt dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> * Update pyjwt==2.4.0 Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: JonnyWong16 <9099342+JonnyWong16@users.noreply.github.com> [skip ci]
This commit is contained in:
dependabot[bot]
GitHub
parent
d17015de44
commit
b3aeaafd00
9 changed files with 171 additions and 61 deletions
|
|
@ -25,19 +25,19 @@ from .exceptions import (
|
|||
)
|
||||
from .jwks_client import PyJWKClient
|
||||
|
||||
__version__ = "2.3.0"
|
||||
__version__ = "2.4.0"
|
||||
|
||||
__title__ = "PyJWT"
|
||||
__description__ = "JSON Web Token implementation in Python"
|
||||
__url__ = "https://pyjwt.readthedocs.io"
|
||||
__uri__ = __url__
|
||||
__doc__ = __description__ + " <" + __uri__ + ">"
|
||||
__doc__ = f"{__description__} <{__uri__}>"
|
||||
|
||||
__author__ = "José Padilla"
|
||||
__email__ = "hello@jpadilla.com"
|
||||
|
||||
__license__ = "MIT"
|
||||
__copyright__ = "Copyright 2015-2020 José Padilla"
|
||||
__copyright__ = "Copyright 2015-2022 José Padilla"
|
||||
|
||||
|
||||
__all__ = [
|
||||
|
|
|
|||
|
|
@ -9,6 +9,8 @@ from .utils import (
|
|||
der_to_raw_signature,
|
||||
force_bytes,
|
||||
from_base64url_uint,
|
||||
is_pem_format,
|
||||
is_ssh_key,
|
||||
raw_to_der_signature,
|
||||
to_base64url_uint,
|
||||
)
|
||||
|
|
@ -183,14 +185,7 @@ class HMACAlgorithm(Algorithm):
|
|||
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):
|
||||
if is_pem_format(key) or is_ssh_key(key):
|
||||
raise InvalidKeyError(
|
||||
"The specified key is an asymmetric key or x509 certificate and"
|
||||
" should not be used as an HMAC secret."
|
||||
|
|
@ -417,6 +412,12 @@ if has_crypto:
|
|||
except ValueError:
|
||||
key = load_pem_private_key(key, password=None)
|
||||
|
||||
# Explicit check the key to prevent confusing errors from cryptography
|
||||
if not isinstance(key, (EllipticCurvePrivateKey, EllipticCurvePublicKey)):
|
||||
raise InvalidKeyError(
|
||||
"Expecting a EllipticCurvePrivateKey/EllipticCurvePublicKey. Wrong key provided for ECDSA algorithms"
|
||||
)
|
||||
|
||||
return key
|
||||
|
||||
def sign(self, msg, key):
|
||||
|
|
@ -545,26 +546,28 @@ if has_crypto:
|
|||
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)
|
||||
key = load_pem_public_key(key)
|
||||
elif "-----BEGIN PRIVATE" in str_key:
|
||||
key = load_pem_private_key(key, password=None)
|
||||
elif str_key[0:4] == "ssh-":
|
||||
key = load_ssh_public_key(key)
|
||||
|
||||
raise TypeError("Expecting a PEM-formatted or OpenSSH key.")
|
||||
# Explicit check the key to prevent confusing errors from cryptography
|
||||
if not isinstance(
|
||||
key,
|
||||
(Ed25519PrivateKey, Ed25519PublicKey, Ed448PrivateKey, Ed448PublicKey),
|
||||
):
|
||||
raise InvalidKeyError(
|
||||
"Expecting a EllipticCurvePrivateKey/EllipticCurvePublicKey. Wrong key provided for EdDSA algorithms"
|
||||
)
|
||||
|
||||
return key
|
||||
|
||||
def sign(self, msg, key):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ class PyJWK:
|
|||
|
||||
kty = self._jwk_data.get("kty", None)
|
||||
if not kty:
|
||||
raise InvalidKeyError("kty is not found: %s" % self._jwk_data)
|
||||
raise InvalidKeyError(f"kty is not found: {self._jwk_data}")
|
||||
|
||||
if not algorithm and isinstance(self._jwk_data, dict):
|
||||
algorithm = self._jwk_data.get("alg", None)
|
||||
|
|
@ -29,25 +29,25 @@ class PyJWK:
|
|||
elif crv == "secp256k1":
|
||||
algorithm = "ES256K"
|
||||
else:
|
||||
raise InvalidKeyError("Unsupported crv: %s" % crv)
|
||||
raise InvalidKeyError(f"Unsupported crv: {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)
|
||||
raise InvalidKeyError(f"crv is not found: {self._jwk_data}")
|
||||
if crv == "Ed25519":
|
||||
algorithm = "EdDSA"
|
||||
else:
|
||||
raise InvalidKeyError("Unsupported crv: %s" % crv)
|
||||
raise InvalidKeyError(f"Unsupported crv: {crv}")
|
||||
else:
|
||||
raise InvalidKeyError("Unsupported kty: %s" % kty)
|
||||
raise InvalidKeyError(f"Unsupported kty: {kty}")
|
||||
|
||||
self.Algorithm = self._algorithms.get(algorithm)
|
||||
|
||||
if not self.Algorithm:
|
||||
raise PyJWKError("Unable to find a algorithm for key: %s" % self._jwk_data)
|
||||
raise PyJWKError(f"Unable to find a algorithm for key: {self._jwk_data}")
|
||||
|
||||
self.key = self.Algorithm.from_jwk(self._jwk_data)
|
||||
|
||||
|
|
@ -95,3 +95,9 @@ class PyJWKSet:
|
|||
def from_json(data):
|
||||
obj = json.loads(data)
|
||||
return PyJWKSet.from_dict(obj)
|
||||
|
||||
def __getitem__(self, kid):
|
||||
for key in self.keys:
|
||||
if key.key_id == kid:
|
||||
return key
|
||||
raise KeyError(f"keyset has no key for kid: {kid}")
|
||||
|
|
|
|||
|
|
@ -80,34 +80,54 @@ class PyJWS:
|
|||
algorithm: Optional[str] = "HS256",
|
||||
headers: Optional[Dict] = None,
|
||||
json_encoder: Optional[Type[json.JSONEncoder]] = None,
|
||||
is_payload_detached: bool = False,
|
||||
) -> 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"]:
|
||||
# Prefer headers values if present to function parameters.
|
||||
if headers:
|
||||
headers_alg = headers.get("alg")
|
||||
if headers_alg:
|
||||
algorithm = headers["alg"]
|
||||
|
||||
headers_b64 = headers.get("b64")
|
||||
if headers_b64 is False:
|
||||
is_payload_detached = True
|
||||
|
||||
# Header
|
||||
header = {"typ": self.header_typ, "alg": algorithm}
|
||||
header = {"typ": self.header_typ, "alg": algorithm} # type: Dict[str, Any]
|
||||
|
||||
if headers:
|
||||
self._validate_headers(headers)
|
||||
header.update(headers)
|
||||
|
||||
if not header["typ"]:
|
||||
del header["typ"]
|
||||
|
||||
if is_payload_detached:
|
||||
header["b64"] = False
|
||||
elif "b64" in header:
|
||||
# True is the standard value for b64, so no need for it
|
||||
del header["b64"]
|
||||
|
||||
json_header = json.dumps(
|
||||
header, separators=(",", ":"), cls=json_encoder
|
||||
).encode()
|
||||
|
||||
segments.append(base64url_encode(json_header))
|
||||
segments.append(base64url_encode(payload))
|
||||
|
||||
if is_payload_detached:
|
||||
msg_payload = payload
|
||||
else:
|
||||
msg_payload = base64url_encode(payload)
|
||||
segments.append(msg_payload)
|
||||
|
||||
# Segments
|
||||
signing_input = b".".join(segments)
|
||||
|
||||
try:
|
||||
alg_obj = self._algorithms[algorithm]
|
||||
key = alg_obj.prepare_key(key)
|
||||
|
|
@ -116,14 +136,15 @@ class PyJWS:
|
|||
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
|
||||
f"Algorithm '{algorithm}' could not be found. Do you have cryptography installed?"
|
||||
) from e
|
||||
else:
|
||||
raise NotImplementedError("Algorithm not supported") from e
|
||||
|
||||
segments.append(base64url_encode(signature))
|
||||
|
||||
# Don't put the payload content inside the encoded token when detached
|
||||
if is_payload_detached:
|
||||
segments[1] = b""
|
||||
encoded_string = b".".join(segments)
|
||||
|
||||
return encoded_string.decode("utf-8")
|
||||
|
|
@ -132,8 +153,9 @@ class PyJWS:
|
|||
self,
|
||||
jwt: str,
|
||||
key: str = "",
|
||||
algorithms: List[str] = None,
|
||||
options: Dict = None,
|
||||
algorithms: Optional[List[str]] = None,
|
||||
options: Optional[Dict] = None,
|
||||
detached_payload: Optional[bytes] = None,
|
||||
**kwargs,
|
||||
) -> Dict[str, Any]:
|
||||
if options is None:
|
||||
|
|
@ -148,6 +170,14 @@ class PyJWS:
|
|||
|
||||
payload, signing_input, header, signature = self._load(jwt)
|
||||
|
||||
if header.get("b64", True) is False:
|
||||
if detached_payload is None:
|
||||
raise DecodeError(
|
||||
'It is required that you pass in a value for the "detached_payload" argument to decode a message having the b64 header set to false.'
|
||||
)
|
||||
payload = detached_payload
|
||||
signing_input = b".".join([signing_input.rsplit(b".", 1)[0], payload])
|
||||
|
||||
if verify_signature:
|
||||
self._verify_signature(signing_input, header, signature, key, algorithms)
|
||||
|
||||
|
|
@ -161,8 +191,8 @@ class PyJWS:
|
|||
self,
|
||||
jwt: str,
|
||||
key: str = "",
|
||||
algorithms: List[str] = None,
|
||||
options: Dict = None,
|
||||
algorithms: Optional[List[str]] = None,
|
||||
options: Optional[Dict] = None,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
decoded = self.decode_complete(jwt, key, algorithms, options, **kwargs)
|
||||
|
|
@ -200,7 +230,7 @@ class PyJWS:
|
|||
try:
|
||||
header = json.loads(header_data)
|
||||
except ValueError as e:
|
||||
raise DecodeError("Invalid header string: %s" % e) from e
|
||||
raise DecodeError(f"Invalid header string: {e}") from e
|
||||
|
||||
if not isinstance(header, Mapping):
|
||||
raise DecodeError("Invalid header string: must be a json object")
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import json
|
||||
import warnings
|
||||
from calendar import timegm
|
||||
from collections.abc import Iterable, Mapping
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
|
@ -66,15 +67,24 @@ class PyJWT:
|
|||
self,
|
||||
jwt: str,
|
||||
key: str = "",
|
||||
algorithms: List[str] = None,
|
||||
options: Dict = None,
|
||||
algorithms: Optional[List[str]] = None,
|
||||
options: Optional[Dict] = None,
|
||||
**kwargs,
|
||||
) -> Dict[str, Any]:
|
||||
if options is None:
|
||||
options = {"verify_signature": True}
|
||||
else:
|
||||
options = dict(options or {}) # shallow-copy or initialize an empty dict
|
||||
options.setdefault("verify_signature", True)
|
||||
|
||||
# If the user has set the legacy `verify` argument, and it doesn't match
|
||||
# what the relevant `options` entry for the argument is, inform the user
|
||||
# that they're likely making a mistake.
|
||||
if "verify" in kwargs and kwargs["verify"] != options["verify_signature"]:
|
||||
warnings.warn(
|
||||
"The `verify` argument to `decode` does nothing in PyJWT 2.0 and newer. "
|
||||
"The equivalent is setting `verify_signature` to False in the `options` dictionary. "
|
||||
"This invocation has a mismatch between the kwarg and the option entry.",
|
||||
category=DeprecationWarning,
|
||||
)
|
||||
|
||||
if not options["verify_signature"]:
|
||||
options.setdefault("verify_exp", False)
|
||||
options.setdefault("verify_nbf", False)
|
||||
|
|
@ -98,7 +108,7 @@ class PyJWT:
|
|||
try:
|
||||
payload = json.loads(decoded["payload"])
|
||||
except ValueError as e:
|
||||
raise DecodeError("Invalid payload string: %s" % e)
|
||||
raise DecodeError(f"Invalid payload string: {e}")
|
||||
if not isinstance(payload, dict):
|
||||
raise DecodeError("Invalid payload string: must be a json object")
|
||||
|
||||
|
|
@ -112,8 +122,8 @@ class PyJWT:
|
|||
self,
|
||||
jwt: str,
|
||||
key: str = "",
|
||||
algorithms: List[str] = None,
|
||||
options: Dict = None,
|
||||
algorithms: Optional[List[str]] = None,
|
||||
options: Optional[Dict] = None,
|
||||
**kwargs,
|
||||
) -> Dict[str, Any]:
|
||||
decoded = self.decode_complete(jwt, key, algorithms, options, **kwargs)
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ class MissingRequiredClaimError(InvalidTokenError):
|
|||
self.claim = claim
|
||||
|
||||
def __str__(self):
|
||||
return 'Token is missing the "%s" claim' % self.claim
|
||||
return f'Token is missing the "{self.claim}" claim'
|
||||
|
||||
|
||||
class PyJWKError(PyJWTError):
|
||||
|
|
|
|||
|
|
@ -28,10 +28,10 @@ def info():
|
|||
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,
|
||||
implementation_version = (
|
||||
f"{sys.pypy_version_info.major}."
|
||||
f"{sys.pypy_version_info.minor}."
|
||||
f"{sys.pypy_version_info.micro}"
|
||||
)
|
||||
if sys.pypy_version_info.releaselevel != "final":
|
||||
implementation_version = "".join(
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import base64
|
||||
import binascii
|
||||
import re
|
||||
from typing import Any, Union
|
||||
|
||||
try:
|
||||
|
|
@ -97,3 +98,63 @@ def raw_to_der_signature(raw_sig: bytes, curve: EllipticCurve) -> bytes:
|
|||
s = bytes_to_number(raw_sig[num_bytes:])
|
||||
|
||||
return encode_dss_signature(r, s)
|
||||
|
||||
|
||||
# Based on https://github.com/hynek/pem/blob/7ad94db26b0bc21d10953f5dbad3acfdfacf57aa/src/pem/_core.py#L224-L252
|
||||
_PEMS = {
|
||||
b"CERTIFICATE",
|
||||
b"TRUSTED CERTIFICATE",
|
||||
b"PRIVATE KEY",
|
||||
b"PUBLIC KEY",
|
||||
b"ENCRYPTED PRIVATE KEY",
|
||||
b"OPENSSH PRIVATE KEY",
|
||||
b"DSA PRIVATE KEY",
|
||||
b"RSA PRIVATE KEY",
|
||||
b"RSA PUBLIC KEY",
|
||||
b"EC PRIVATE KEY",
|
||||
b"DH PARAMETERS",
|
||||
b"NEW CERTIFICATE REQUEST",
|
||||
b"CERTIFICATE REQUEST",
|
||||
b"SSH2 PUBLIC KEY",
|
||||
b"SSH2 ENCRYPTED PRIVATE KEY",
|
||||
b"X509 CRL",
|
||||
}
|
||||
|
||||
_PEM_RE = re.compile(
|
||||
b"----[- ]BEGIN ("
|
||||
+ b"|".join(_PEMS)
|
||||
+ b""")[- ]----\r?
|
||||
.+?\r?
|
||||
----[- ]END \\1[- ]----\r?\n?""",
|
||||
re.DOTALL,
|
||||
)
|
||||
|
||||
|
||||
def is_pem_format(key: bytes) -> bool:
|
||||
return bool(_PEM_RE.search(key))
|
||||
|
||||
|
||||
# Based on https://github.com/pyca/cryptography/blob/bcb70852d577b3f490f015378c75cba74986297b/src/cryptography/hazmat/primitives/serialization/ssh.py#L40-L46
|
||||
_CERT_SUFFIX = b"-cert-v01@openssh.com"
|
||||
_SSH_PUBKEY_RC = re.compile(br"\A(\S+)[ \t]+(\S+)")
|
||||
_SSH_KEY_FORMATS = [
|
||||
b"ssh-ed25519",
|
||||
b"ssh-rsa",
|
||||
b"ssh-dss",
|
||||
b"ecdsa-sha2-nistp256",
|
||||
b"ecdsa-sha2-nistp384",
|
||||
b"ecdsa-sha2-nistp521",
|
||||
]
|
||||
|
||||
|
||||
def is_ssh_key(key: bytes) -> bool:
|
||||
if any(string_value in key for string_value in _SSH_KEY_FORMATS):
|
||||
return True
|
||||
|
||||
ssh_pubkey_match = _SSH_PUBKEY_RC.match(key)
|
||||
if ssh_pubkey_match:
|
||||
key_type = ssh_pubkey_match.group(1)
|
||||
if _CERT_SUFFIX == key_type[-len(_CERT_SUFFIX) :]:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ paho-mqtt==1.6.1
|
|||
plexapi==4.9.2
|
||||
portend==3.1.0
|
||||
profilehooks==1.12.0
|
||||
PyJWT==2.3.0
|
||||
PyJWT==2.4.0
|
||||
pyparsing==3.0.9
|
||||
python-dateutil==2.8.2
|
||||
python-twitter==3.5
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue