/snap/core18/2979/usr/lib/python3/dist-packages/jwt
Edit: /snap/core18/2979/usr/lib/python3/dist-packages/jwt/utils.py (4121B)
import base64
import binascii
import re
import struct
from .compat import binary_type, bytes_from_int, text_type
try:
from cryptography.hazmat.primitives.asymmetric.utils import (
decode_dss_signature, encode_dss_signature
)
except ImportError:
pass
def force_unicode(value):
if isinstance(value, binary_type):
return value.decode('utf-8')
elif isinstance(value, text_type):
return value
else:
raise TypeError('Expected a string value')
def force_bytes(value):
if isinstance(value, text_type):
return value.encode('utf-8')
elif isinstance(value, binary_type):
return value
else:
raise TypeError('Expected a string value')
def base64url_decode(input):
if isinstance(input, text_type):
input = input.encode('ascii')
rem = len(input) % 4
if rem > 0:
input += b'=' * (4 - rem)
return base64.urlsafe_b64decode(input)
def base64url_encode(input):
return base64.urlsafe_b64encode(input).replace(b'=', b'')
def to_base64url_uint(val):
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):
if isinstance(val, text_type):
val = val.encode('ascii')
data = base64url_decode(val)
buf = struct.unpack('%sB' % len(data), data)
return int(''.join(["%02x" % byte for byte in buf]), 16)
def merge_dict(original, updates):
if not updates:
return original
try:
merged_options = original.copy()
merged_options.update(updates)
except (AttributeError, ValueError) as e:
raise TypeError('original and updates must be a dictionary: %s' % e)
return merged_options
def number_to_bytes(num, num_bytes):
padded_hex = '%0*x' % (2 * num_bytes, num)
big_endian = binascii.a2b_hex(padded_hex.encode('ascii'))
return big_endian
def bytes_to_number(string):
return int(binascii.b2a_hex(string), 16)
def der_to_raw_signature(der_sig, curve):
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, curve):
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)
# 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):
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):
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