Integrated my history into data
This commit is contained in:
Karthik Pullela
@@ -0,0 +1,29 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# flake8: noqa
|
||||
|
||||
"""
|
||||
JSON Web Token implementation
|
||||
|
||||
Minimum implementation based on this spec:
|
||||
http://self-issued.info/docs/draft-jones-json-web-token-01.html
|
||||
"""
|
||||
|
||||
|
||||
__title__ = 'pyjwt'
|
||||
__version__ = '1.5.3'
|
||||
__author__ = 'José Padilla'
|
||||
__license__ = 'MIT'
|
||||
__copyright__ = 'Copyright 2015 José Padilla'
|
||||
|
||||
|
||||
from .api_jwt import (
|
||||
encode, decode, register_algorithm, unregister_algorithm,
|
||||
get_unverified_header, PyJWT
|
||||
)
|
||||
from .api_jws import PyJWS
|
||||
from .exceptions import (
|
||||
InvalidTokenError, DecodeError, InvalidAlgorithmError,
|
||||
InvalidAudienceError, ExpiredSignatureError, ImmatureSignatureError,
|
||||
InvalidIssuedAtError, InvalidIssuerError, ExpiredSignature,
|
||||
InvalidAudience, InvalidIssuer, MissingRequiredClaimError
|
||||
)
|
||||
@@ -0,0 +1,162 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
from __future__ import absolute_import, print_function
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
|
||||
from . import DecodeError, __version__, decode, encode
|
||||
|
||||
|
||||
def encode_payload(args):
|
||||
# Try to encode
|
||||
if args.key is None:
|
||||
raise ValueError('Key is required when encoding. See --help for usage.')
|
||||
|
||||
# Build payload object to encode
|
||||
payload = {}
|
||||
|
||||
for arg in args.payload:
|
||||
k, v = arg.split('=', 1)
|
||||
|
||||
# exp +offset special case?
|
||||
if k == 'exp' and v[0] == '+' and len(v) > 1:
|
||||
v = str(int(time.time()+int(v[1:])))
|
||||
|
||||
# Cast to integer?
|
||||
if v.isdigit():
|
||||
v = int(v)
|
||||
else:
|
||||
# Cast to float?
|
||||
try:
|
||||
v = float(v)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Cast to true, false, or null?
|
||||
constants = {'true': True, 'false': False, 'null': None}
|
||||
|
||||
if v in constants:
|
||||
v = constants[v]
|
||||
|
||||
payload[k] = v
|
||||
|
||||
token = encode(
|
||||
payload,
|
||||
key=args.key,
|
||||
algorithm=args.algorithm
|
||||
)
|
||||
|
||||
return token.decode('utf-8')
|
||||
|
||||
|
||||
def decode_payload(args):
|
||||
try:
|
||||
if sys.stdin.isatty():
|
||||
token = sys.stdin.read()
|
||||
else:
|
||||
token = args.token
|
||||
|
||||
token = token.encode('utf-8')
|
||||
data = decode(token, key=args.key, verify=args.verify)
|
||||
|
||||
return json.dumps(data)
|
||||
|
||||
except DecodeError as e:
|
||||
raise DecodeError('There was an error decoding the token: %s' % e)
|
||||
|
||||
|
||||
def build_argparser():
|
||||
|
||||
usage = '''
|
||||
Encodes or decodes JSON Web Tokens based on input.
|
||||
|
||||
%(prog)s [options] <command> [options] input
|
||||
|
||||
Decoding examples:
|
||||
|
||||
%(prog)s --key=secret decode json.web.token
|
||||
%(prog)s decode --no-verify json.web.token
|
||||
|
||||
Encoding requires the key option and takes space separated key/value pairs
|
||||
separated by equals (=) as input. Examples:
|
||||
|
||||
%(prog)s --key=secret encode iss=me exp=1302049071
|
||||
%(prog)s --key=secret encode foo=bar exp=+10
|
||||
|
||||
The exp key is special and can take an offset to current Unix time.
|
||||
'''
|
||||
|
||||
arg_parser = argparse.ArgumentParser(
|
||||
prog='pyjwt',
|
||||
usage=usage
|
||||
)
|
||||
|
||||
arg_parser.add_argument(
|
||||
'-v', '--version',
|
||||
action='version',
|
||||
version='%(prog)s ' + __version__
|
||||
)
|
||||
|
||||
arg_parser.add_argument(
|
||||
'--key',
|
||||
dest='key',
|
||||
metavar='KEY',
|
||||
default=None,
|
||||
help='set the secret key to sign with'
|
||||
)
|
||||
|
||||
arg_parser.add_argument(
|
||||
'--alg',
|
||||
dest='algorithm',
|
||||
metavar='ALG',
|
||||
default='HS256',
|
||||
help='set crypto algorithm to sign with. default=HS256'
|
||||
)
|
||||
|
||||
subparsers = arg_parser.add_subparsers(
|
||||
title='PyJWT subcommands',
|
||||
description='valid subcommands',
|
||||
help='additional help'
|
||||
)
|
||||
|
||||
# Encode subcommand
|
||||
encode_parser = subparsers.add_parser('encode', help='use to encode a supplied payload')
|
||||
|
||||
payload_help = """Payload to encode. Must be a space separated list of key/value
|
||||
pairs separated by equals (=) sign."""
|
||||
|
||||
encode_parser.add_argument('payload', nargs='+', help=payload_help)
|
||||
encode_parser.set_defaults(func=encode_payload)
|
||||
|
||||
# Decode subcommand
|
||||
decode_parser = subparsers.add_parser('decode', help='use to decode a supplied JSON web token')
|
||||
decode_parser.add_argument('token', help='JSON web token to decode.')
|
||||
|
||||
decode_parser.add_argument(
|
||||
'-n', '--no-verify',
|
||||
action='store_false',
|
||||
dest='verify',
|
||||
default=True,
|
||||
help='ignore signature and claims verification on decode'
|
||||
)
|
||||
|
||||
decode_parser.set_defaults(func=decode_payload)
|
||||
|
||||
return arg_parser
|
||||
|
||||
|
||||
def main():
|
||||
arg_parser = build_argparser()
|
||||
|
||||
try:
|
||||
arguments = arg_parser.parse_args(sys.argv[1:])
|
||||
|
||||
output = arguments.func(arguments)
|
||||
|
||||
print(output)
|
||||
except Exception as e:
|
||||
print('There was an unforseen error: ', e)
|
||||
arg_parser.print_help()
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,403 @@
|
||||
import hashlib
|
||||
import hmac
|
||||
import 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), # Backward compat for #219 fix
|
||||
'PS256': RSAPSSAlgorithm(RSAPSSAlgorithm.SHA256),
|
||||
'PS384': RSAPSSAlgorithm(RSAPSSAlgorithm.SHA384),
|
||||
'PS512': RSAPSSAlgorithm(RSAPSSAlgorithm.SHA512)
|
||||
})
|
||||
|
||||
return default_algorithms
|
||||
|
||||
|
||||
class Algorithm(object):
|
||||
"""
|
||||
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': 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):
|
||||
"""
|
||||
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) or \
|
||||
isinstance(key, RSAPublicKey):
|
||||
return key
|
||||
|
||||
if 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):
|
||||
# Private key
|
||||
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))
|
||||
}
|
||||
|
||||
elif getattr(key_obj, 'verify', None):
|
||||
# Public key
|
||||
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')
|
||||
|
||||
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(default_backend())
|
||||
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(default_backend())
|
||||
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) or \
|
||||
isinstance(key, EllipticCurvePublicKey):
|
||||
return key
|
||||
|
||||
if isinstance(key, string_types):
|
||||
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, 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
|
||||
|
||||
try:
|
||||
key.verify(der_sig, msg, ec.ECDSA(self.hash_alg()))
|
||||
return True
|
||||
except InvalidSignature:
|
||||
return False
|
||||
|
||||
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
|
||||
@@ -0,0 +1,226 @@
|
||||
import binascii
|
||||
import json
|
||||
import warnings
|
||||
|
||||
from collections import Mapping
|
||||
|
||||
from .algorithms import (
|
||||
Algorithm, get_default_algorithms, has_crypto, requires_cryptography # NOQA
|
||||
)
|
||||
from .compat import binary_type, string_types, text_type
|
||||
from .exceptions import DecodeError, InvalidAlgorithmError, InvalidTokenError
|
||||
from .utils import base64url_decode, base64url_encode, force_bytes, merge_dict
|
||||
|
||||
|
||||
class PyJWS(object):
|
||||
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 not options:
|
||||
options = {}
|
||||
|
||||
self.options = merge_dict(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, key, algorithm='HS256', headers=None,
|
||||
json_encoder=None):
|
||||
segments = []
|
||||
|
||||
if algorithm is None:
|
||||
algorithm = 'none'
|
||||
|
||||
if algorithm not in self._valid_algs:
|
||||
pass
|
||||
|
||||
# Header
|
||||
header = {'typ': self.header_typ, 'alg': algorithm}
|
||||
|
||||
if headers:
|
||||
self._validate_headers(headers)
|
||||
header.update(headers)
|
||||
|
||||
json_header = force_bytes(
|
||||
json.dumps(
|
||||
header,
|
||||
separators=(',', ':'),
|
||||
cls=json_encoder
|
||||
)
|
||||
)
|
||||
|
||||
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:
|
||||
if not has_crypto and algorithm in requires_cryptography:
|
||||
raise NotImplementedError(
|
||||
"Algorithm '%s' could not be found. Do you have cryptography "
|
||||
"installed?" % algorithm
|
||||
)
|
||||
else:
|
||||
raise NotImplementedError('Algorithm not supported')
|
||||
|
||||
segments.append(base64url_encode(signature))
|
||||
|
||||
return b'.'.join(segments)
|
||||
|
||||
def decode(self, jws, key='', verify=True, algorithms=None, options=None,
|
||||
**kwargs):
|
||||
|
||||
merged_options = merge_dict(self.options, options)
|
||||
verify_signature = merged_options['verify_signature']
|
||||
|
||||
if verify_signature and not algorithms:
|
||||
warnings.warn(
|
||||
'It is strongly recommended that you pass in a ' +
|
||||
'value for the "algorithms" argument when calling decode(). ' +
|
||||
'This argument will be mandatory in a future version.',
|
||||
DeprecationWarning
|
||||
)
|
||||
|
||||
payload, signing_input, header, signature = self._load(jws)
|
||||
|
||||
if not verify:
|
||||
warnings.warn('The verify parameter is deprecated. '
|
||||
'Please use verify_signature in options instead.',
|
||||
DeprecationWarning, stacklevel=2)
|
||||
elif verify_signature:
|
||||
self._verify_signature(payload, signing_input, header, signature,
|
||||
key, algorithms)
|
||||
|
||||
return 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, text_type):
|
||||
jwt = jwt.encode('utf-8')
|
||||
|
||||
if not issubclass(type(jwt), binary_type):
|
||||
raise DecodeError("Invalid token type. Token must be a {0}".format(
|
||||
binary_type))
|
||||
|
||||
try:
|
||||
signing_input, crypto_segment = jwt.rsplit(b'.', 1)
|
||||
header_segment, payload_segment = signing_input.split(b'.', 1)
|
||||
except ValueError:
|
||||
raise DecodeError('Not enough segments')
|
||||
|
||||
try:
|
||||
header_data = base64url_decode(header_segment)
|
||||
except (TypeError, binascii.Error):
|
||||
raise DecodeError('Invalid header padding')
|
||||
|
||||
try:
|
||||
header = json.loads(header_data.decode('utf-8'))
|
||||
except ValueError as e:
|
||||
raise DecodeError('Invalid header string: %s' % 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):
|
||||
raise DecodeError('Invalid payload padding')
|
||||
|
||||
try:
|
||||
signature = base64url_decode(crypto_segment)
|
||||
except (TypeError, binascii.Error):
|
||||
raise DecodeError('Invalid crypto padding')
|
||||
|
||||
return (payload, signing_input, header, signature)
|
||||
|
||||
def _verify_signature(self, payload, 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 DecodeError('Signature verification failed')
|
||||
|
||||
except KeyError:
|
||||
raise InvalidAlgorithmError('Algorithm not supported')
|
||||
|
||||
def _validate_headers(self, headers):
|
||||
if 'kid' in headers:
|
||||
self._validate_kid(headers['kid'])
|
||||
|
||||
def _validate_kid(self, kid):
|
||||
if not isinstance(kid, string_types):
|
||||
raise InvalidTokenError('Key ID header parameter must be a string')
|
||||
|
||||
|
||||
_jws_global_obj = PyJWS()
|
||||
encode = _jws_global_obj.encode
|
||||
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
|
||||
@@ -0,0 +1,199 @@
|
||||
import json
|
||||
import warnings
|
||||
|
||||
from calendar import timegm
|
||||
from collections import Mapping
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from .api_jws import PyJWS
|
||||
from .algorithms import Algorithm, get_default_algorithms # NOQA
|
||||
from .compat import string_types, timedelta_total_seconds
|
||||
from .exceptions import (
|
||||
DecodeError, ExpiredSignatureError, ImmatureSignatureError,
|
||||
InvalidAudienceError, InvalidIssuedAtError,
|
||||
InvalidIssuerError, MissingRequiredClaimError
|
||||
)
|
||||
from .utils import merge_dict
|
||||
|
||||
|
||||
class PyJWT(PyJWS):
|
||||
header_type = 'JWT'
|
||||
|
||||
@staticmethod
|
||||
def _get_default_options():
|
||||
return {
|
||||
'verify_signature': True,
|
||||
'verify_exp': True,
|
||||
'verify_nbf': True,
|
||||
'verify_iat': True,
|
||||
'verify_aud': True,
|
||||
'verify_iss': True,
|
||||
'require_exp': False,
|
||||
'require_iat': False,
|
||||
'require_nbf': False
|
||||
}
|
||||
|
||||
def encode(self, payload, key, algorithm='HS256', headers=None,
|
||||
json_encoder=None):
|
||||
# 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
|
||||
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 super(PyJWT, self).encode(
|
||||
json_payload, key, algorithm, headers, json_encoder
|
||||
)
|
||||
|
||||
def decode(self, jwt, key='', verify=True, algorithms=None, options=None,
|
||||
**kwargs):
|
||||
|
||||
if verify and not algorithms:
|
||||
warnings.warn(
|
||||
'It is strongly recommended that you pass in a ' +
|
||||
'value for the "algorithms" argument when calling decode(). ' +
|
||||
'This argument will be mandatory in a future version.',
|
||||
DeprecationWarning
|
||||
)
|
||||
|
||||
payload, signing_input, header, signature = self._load(jwt)
|
||||
|
||||
if options is None:
|
||||
options = {'verify_signature': verify}
|
||||
else:
|
||||
options.setdefault('verify_signature', verify)
|
||||
|
||||
decoded = super(PyJWT, self).decode(
|
||||
jwt, key=key, algorithms=algorithms, options=options, **kwargs
|
||||
)
|
||||
|
||||
try:
|
||||
payload = json.loads(decoded.decode('utf-8'))
|
||||
except ValueError as e:
|
||||
raise DecodeError('Invalid payload string: %s' % e)
|
||||
if not isinstance(payload, Mapping):
|
||||
raise DecodeError('Invalid payload string: must be a json object')
|
||||
|
||||
if verify:
|
||||
merged_options = merge_dict(self.options, options)
|
||||
self._validate_claims(payload, merged_options, **kwargs)
|
||||
|
||||
return payload
|
||||
|
||||
def _validate_claims(self, payload, options, audience=None, issuer=None,
|
||||
leeway=0, **kwargs):
|
||||
|
||||
if 'verify_expiration' in kwargs:
|
||||
options['verify_exp'] = kwargs.get('verify_expiration', True)
|
||||
warnings.warn('The verify_expiration parameter is deprecated. '
|
||||
'Please use verify_exp in options instead.',
|
||||
DeprecationWarning)
|
||||
|
||||
if isinstance(leeway, timedelta):
|
||||
leeway = timedelta_total_seconds(leeway)
|
||||
|
||||
if not isinstance(audience, (string_types, type(None))):
|
||||
raise TypeError('audience must be a string or None')
|
||||
|
||||
self._validate_required_claims(payload, options)
|
||||
|
||||
now = timegm(datetime.utcnow().utctimetuple())
|
||||
|
||||
if 'iat' in payload and options.get('verify_iat'):
|
||||
self._validate_iat(payload, now, leeway)
|
||||
|
||||
if 'nbf' in payload and options.get('verify_nbf'):
|
||||
self._validate_nbf(payload, now, leeway)
|
||||
|
||||
if 'exp' in payload and options.get('verify_exp'):
|
||||
self._validate_exp(payload, now, leeway)
|
||||
|
||||
if options.get('verify_iss'):
|
||||
self._validate_iss(payload, issuer)
|
||||
|
||||
if options.get('verify_aud'):
|
||||
self._validate_aud(payload, audience)
|
||||
|
||||
def _validate_required_claims(self, payload, options):
|
||||
if options.get('require_exp') and payload.get('exp') is None:
|
||||
raise MissingRequiredClaimError('exp')
|
||||
|
||||
if options.get('require_iat') and payload.get('iat') is None:
|
||||
raise MissingRequiredClaimError('iat')
|
||||
|
||||
if options.get('require_nbf') and payload.get('nbf') is None:
|
||||
raise MissingRequiredClaimError('nbf')
|
||||
|
||||
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 and 'aud' not in payload:
|
||||
return
|
||||
|
||||
if audience is not None and 'aud' not in payload:
|
||||
# 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, string_types):
|
||||
audience_claims = [audience_claims]
|
||||
if not isinstance(audience_claims, list):
|
||||
raise InvalidAudienceError('Invalid claim format in token')
|
||||
if any(not isinstance(c, string_types) for c in audience_claims):
|
||||
raise InvalidAudienceError('Invalid claim format in token')
|
||||
if audience not in audience_claims:
|
||||
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 = _jwt_global_obj.decode
|
||||
register_algorithm = _jwt_global_obj.register_algorithm
|
||||
unregister_algorithm = _jwt_global_obj.unregister_algorithm
|
||||
get_unverified_header = _jwt_global_obj.get_unverified_header
|
||||
@@ -0,0 +1,76 @@
|
||||
"""
|
||||
The `compat` module provides support for backwards compatibility with older
|
||||
versions of python, and compatibility wrappers around optional packages.
|
||||
"""
|
||||
# flake8: noqa
|
||||
import hmac
|
||||
import struct
|
||||
import sys
|
||||
|
||||
|
||||
PY3 = sys.version_info[0] == 3
|
||||
|
||||
|
||||
if PY3:
|
||||
text_type = str
|
||||
binary_type = bytes
|
||||
else:
|
||||
text_type = unicode
|
||||
binary_type = str
|
||||
|
||||
string_types = (text_type, binary_type)
|
||||
|
||||
|
||||
def timedelta_total_seconds(delta):
|
||||
try:
|
||||
delta.total_seconds
|
||||
except AttributeError:
|
||||
# On Python 2.6, timedelta instances do not have
|
||||
# a .total_seconds() method.
|
||||
total_seconds = delta.days * 24 * 60 * 60 + delta.seconds
|
||||
else:
|
||||
total_seconds = delta.total_seconds()
|
||||
|
||||
return total_seconds
|
||||
|
||||
|
||||
try:
|
||||
constant_time_compare = hmac.compare_digest
|
||||
except AttributeError:
|
||||
# Fallback for Python < 2.7
|
||||
def constant_time_compare(val1, val2):
|
||||
"""
|
||||
Returns True if the two strings are equal, False otherwise.
|
||||
|
||||
The time taken is independent of the number of characters that match.
|
||||
"""
|
||||
if len(val1) != len(val2):
|
||||
return False
|
||||
|
||||
result = 0
|
||||
|
||||
for x, y in zip(val1, val2):
|
||||
result |= ord(x) ^ ord(y)
|
||||
|
||||
return result == 0
|
||||
|
||||
# Use int.to_bytes if it exists (Python 3)
|
||||
if getattr(int, 'to_bytes', None):
|
||||
def bytes_from_int(val):
|
||||
remaining = val
|
||||
byte_length = 0
|
||||
|
||||
while remaining != 0:
|
||||
remaining = remaining >> 8
|
||||
byte_length += 1
|
||||
|
||||
return val.to_bytes(byte_length, 'big', signed=False)
|
||||
else:
|
||||
def bytes_from_int(val):
|
||||
buf = []
|
||||
while val:
|
||||
val, remainder = divmod(val, 256)
|
||||
buf.append(remainder)
|
||||
|
||||
buf.reverse()
|
||||
return struct.pack('%sB' % len(buf), *buf)
|
||||
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
@@ -0,0 +1,60 @@
|
||||
# Note: This file is named py_ecdsa.py because import behavior in Python 2
|
||||
# would cause ecdsa.py to squash the ecdsa library that it depends upon.
|
||||
|
||||
import hashlib
|
||||
|
||||
import ecdsa
|
||||
|
||||
from jwt.algorithms import Algorithm
|
||||
from jwt.compat import string_types, text_type
|
||||
|
||||
|
||||
class ECAlgorithm(Algorithm):
|
||||
"""
|
||||
Performs signing and verification operations using
|
||||
ECDSA and the specified hash function
|
||||
|
||||
This class requires the ecdsa package to be installed.
|
||||
|
||||
This is based off of the implementation in PyJWT 0.3.2
|
||||
"""
|
||||
SHA256 = hashlib.sha256
|
||||
SHA384 = hashlib.sha384
|
||||
SHA512 = hashlib.sha512
|
||||
|
||||
def __init__(self, hash_alg):
|
||||
self.hash_alg = hash_alg
|
||||
|
||||
def prepare_key(self, key):
|
||||
|
||||
if isinstance(key, ecdsa.SigningKey) or \
|
||||
isinstance(key, ecdsa.VerifyingKey):
|
||||
return key
|
||||
|
||||
if isinstance(key, string_types):
|
||||
if isinstance(key, text_type):
|
||||
key = key.encode('utf-8')
|
||||
|
||||
# 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:
|
||||
key = ecdsa.VerifyingKey.from_pem(key)
|
||||
except ecdsa.der.UnexpectedDER:
|
||||
key = ecdsa.SigningKey.from_pem(key)
|
||||
|
||||
else:
|
||||
raise TypeError('Expecting a PEM-formatted key.')
|
||||
|
||||
return key
|
||||
|
||||
def sign(self, msg, key):
|
||||
return key.sign(msg, hashfunc=self.hash_alg,
|
||||
sigencode=ecdsa.util.sigencode_string)
|
||||
|
||||
def verify(self, msg, key, sig):
|
||||
try:
|
||||
return key.verify(sig, msg, hashfunc=self.hash_alg,
|
||||
sigdecode=ecdsa.util.sigdecode_string)
|
||||
except AssertionError:
|
||||
return False
|
||||
@@ -0,0 +1,47 @@
|
||||
import Crypto.Hash.SHA256
|
||||
import Crypto.Hash.SHA384
|
||||
import Crypto.Hash.SHA512
|
||||
|
||||
from Crypto.PublicKey import RSA
|
||||
from Crypto.Signature import PKCS1_v1_5
|
||||
|
||||
from jwt.algorithms import Algorithm
|
||||
from jwt.compat import string_types, text_type
|
||||
|
||||
|
||||
class RSAAlgorithm(Algorithm):
|
||||
"""
|
||||
Performs signing and verification operations using
|
||||
RSASSA-PKCS-v1_5 and the specified hash function.
|
||||
|
||||
This class requires PyCrypto package to be installed.
|
||||
|
||||
This is based off of the implementation in PyJWT 0.3.2
|
||||
"""
|
||||
SHA256 = Crypto.Hash.SHA256
|
||||
SHA384 = Crypto.Hash.SHA384
|
||||
SHA512 = Crypto.Hash.SHA512
|
||||
|
||||
def __init__(self, hash_alg):
|
||||
self.hash_alg = hash_alg
|
||||
|
||||
def prepare_key(self, key):
|
||||
|
||||
if isinstance(key, RSA._RSAobj):
|
||||
return key
|
||||
|
||||
if isinstance(key, string_types):
|
||||
if isinstance(key, text_type):
|
||||
key = key.encode('utf-8')
|
||||
|
||||
key = RSA.importKey(key)
|
||||
else:
|
||||
raise TypeError('Expecting a PEM- or RSA-formatted key.')
|
||||
|
||||
return key
|
||||
|
||||
def sign(self, msg, key):
|
||||
return PKCS1_v1_5.new(key).sign(self.hash_alg.new(msg))
|
||||
|
||||
def verify(self, msg, key, sig):
|
||||
return PKCS1_v1_5.new(key).verify(self.hash_alg.new(msg), sig)
|
||||
@@ -0,0 +1,48 @@
|
||||
class InvalidTokenError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class DecodeError(InvalidTokenError):
|
||||
pass
|
||||
|
||||
|
||||
class ExpiredSignatureError(InvalidTokenError):
|
||||
pass
|
||||
|
||||
|
||||
class InvalidAudienceError(InvalidTokenError):
|
||||
pass
|
||||
|
||||
|
||||
class InvalidIssuerError(InvalidTokenError):
|
||||
pass
|
||||
|
||||
|
||||
class InvalidIssuedAtError(InvalidTokenError):
|
||||
pass
|
||||
|
||||
|
||||
class ImmatureSignatureError(InvalidTokenError):
|
||||
pass
|
||||
|
||||
|
||||
class InvalidKeyError(Exception):
|
||||
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
|
||||
|
||||
|
||||
# Compatibility aliases (deprecated)
|
||||
ExpiredSignature = ExpiredSignatureError
|
||||
InvalidAudience = InvalidAudienceError
|
||||
InvalidIssuer = InvalidIssuerError
|
||||
@@ -0,0 +1,113 @@
|
||||
import base64
|
||||
import binascii
|
||||
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)
|
||||
Reference in New Issue
Block a user