Removing python2 compatibility. (#518)
This commit is contained in:
parent
6eca2eb147
commit
dd6d332166
|
@ -8,7 +8,6 @@
|
|||
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
|
||||
|
||||
"""
|
||||
import sys
|
||||
import jwt
|
||||
import json
|
||||
import binascii
|
||||
|
@ -40,12 +39,8 @@ def get_rsa_public_key(n, e):
|
|||
:param e:
|
||||
:return: a RSA Public Key in PEM format
|
||||
"""
|
||||
if sys.version_info >= (3, 0):
|
||||
n = int(binascii.hexlify(jwt.utils.base64url_decode(bytes(n, 'utf-8'))), 16)
|
||||
e = int(binascii.hexlify(jwt.utils.base64url_decode(bytes(e, 'utf-8'))), 16)
|
||||
else:
|
||||
n = int(binascii.hexlify(jwt.utils.base64url_decode(str(n))), 16)
|
||||
e = int(binascii.hexlify(jwt.utils.base64url_decode(str(e))), 16)
|
||||
|
||||
pub = RSAPublicNumbers(e, n).public_key(default_backend())
|
||||
return pub.public_bytes(
|
||||
|
@ -128,10 +123,7 @@ def fetch_token_header(token):
|
|||
raise jwt.DecodeError('Not enough segments')
|
||||
|
||||
try:
|
||||
if sys.version_info >= (3, 0):
|
||||
return json.loads(jwt.utils.base64url_decode(header_segment).decode('utf-8'))
|
||||
else:
|
||||
return json.loads(jwt.utils.base64url_decode(header_segment))
|
||||
except TypeError as e:
|
||||
current_app.logger.exception(e)
|
||||
raise jwt.DecodeError('Invalid header padding')
|
||||
|
|
|
@ -5,7 +5,6 @@
|
|||
:license: Apache, see LICENSE for more details.
|
||||
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
|
||||
"""
|
||||
import sys
|
||||
import jwt
|
||||
import base64
|
||||
import requests
|
||||
|
@ -143,12 +142,8 @@ class Ping(Resource):
|
|||
# the secret and cliendId will be given to you when you signup for the provider
|
||||
token = '{0}:{1}'.format(args['clientId'], current_app.config.get("PING_SECRET"))
|
||||
|
||||
if sys.version_info >= (3, 0):
|
||||
basic = base64.b64encode(bytes(token, 'utf-8'))
|
||||
headers = {'authorization': 'basic {0}'.format(basic.decode('utf-8'))}
|
||||
else:
|
||||
basic = base64.b64encode(token, 'utf-8')
|
||||
headers = {'authorization': 'basic {0}'.format(basic)}
|
||||
|
||||
# exchange authorization code for access token.
|
||||
|
||||
|
@ -172,10 +167,7 @@ class Ping(Resource):
|
|||
|
||||
# validate your token based on the key it was signed with
|
||||
try:
|
||||
if sys.version_info >= (3, 0):
|
||||
jwt.decode(id_token, secret.decode('utf-8'), algorithms=[algo], audience=args['clientId'])
|
||||
else:
|
||||
jwt.decode(id_token, secret, algorithms=[algo], audience=args['clientId'])
|
||||
except jwt.DecodeError:
|
||||
return dict(message='Token is invalid'), 403
|
||||
except jwt.ExpiredSignatureError:
|
||||
|
|
|
@ -6,8 +6,6 @@
|
|||
|
||||
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
|
||||
"""
|
||||
import six
|
||||
import sys
|
||||
import string
|
||||
import random
|
||||
|
||||
|
@ -37,10 +35,7 @@ def get_psuedo_random_string():
|
|||
|
||||
|
||||
def parse_certificate(body):
|
||||
if sys.version_info[0] <= 2:
|
||||
return x509.load_pem_x509_certificate(bytes(body), default_backend())
|
||||
|
||||
if isinstance(body, six.string_types):
|
||||
if isinstance(body, str):
|
||||
body = body.encode('utf-8')
|
||||
|
||||
return x509.load_pem_x509_certificate(body, default_backend())
|
||||
|
|
|
@ -477,14 +477,6 @@ def unlock(path=None):
|
|||
sys.stdout.write("[+] Keys have been unencrypted!\n")
|
||||
|
||||
|
||||
def unicode_(data):
|
||||
import sys
|
||||
|
||||
if sys.version_info.major < 3:
|
||||
return data.decode('UTF-8')
|
||||
return data
|
||||
|
||||
|
||||
def print_certificate_details(details):
|
||||
"""
|
||||
Print the certificate details with formatting.
|
||||
|
|
|
@ -1,5 +1,4 @@
|
|||
import pytest
|
||||
import six
|
||||
|
||||
from lemur.tests.vectors import INTERNAL_CERTIFICATE_A_STR, INTERNAL_PRIVATE_KEY_A_STR
|
||||
|
||||
|
@ -24,7 +23,7 @@ def test_export_truststore_default_password(app):
|
|||
actual = p.export(INTERNAL_CERTIFICATE_A_STR, "", "", options)
|
||||
|
||||
assert actual[0] == 'jks'
|
||||
assert isinstance(actual[1], six.string_types)
|
||||
assert isinstance(actual[1], str)
|
||||
assert isinstance(actual[2], bytes)
|
||||
|
||||
|
||||
|
@ -56,5 +55,5 @@ def test_export_keystore_default_password(app):
|
|||
actual = p.export(INTERNAL_CERTIFICATE_A_STR, "", INTERNAL_PRIVATE_KEY_A_STR, options)
|
||||
|
||||
assert actual[0] == 'jks'
|
||||
assert isinstance(actual[1], six.string_types)
|
||||
assert isinstance(actual[1], str)
|
||||
assert isinstance(actual[2], bytes)
|
||||
|
|
|
@ -6,8 +6,6 @@
|
|||
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import six
|
||||
from flask import current_app
|
||||
from cryptography.fernet import Fernet, MultiFernet
|
||||
import sqlalchemy.types as types
|
||||
|
@ -97,11 +95,8 @@ class Vault(types.TypeDecorator):
|
|||
if not value:
|
||||
return
|
||||
|
||||
if sys.version_info[0] <= 2:
|
||||
return MultiFernet(self.keys).encrypt(bytes(value))
|
||||
|
||||
# ensure bytes for fernet
|
||||
if isinstance(value, six.string_types):
|
||||
if isinstance(value, str):
|
||||
value = value.encode('utf-8')
|
||||
|
||||
return MultiFernet(self.keys).encrypt(value)
|
||||
|
@ -122,6 +117,4 @@ class Vault(types.TypeDecorator):
|
|||
if not value:
|
||||
return
|
||||
|
||||
if sys.version_info[0] <= 2:
|
||||
return MultiFernet(self.keys).decrypt(value)
|
||||
return MultiFernet(self.keys).decrypt(value).decode('utf8')
|
||||
|
|
Loading…
Reference in New Issue