From e24a94d798bd69a0110b1e5ddf532192621ca754 Mon Sep 17 00:00:00 2001 From: Marti Raudsepp Date: Wed, 26 Dec 2018 19:49:56 +0200 Subject: [PATCH] Enforce that PEM strings (certs, keys, CSR) are internally passed as str, not bytes This was already true in most places but not 100%, leading to lots of redundant checks and conversions. --- lemur/certificates/service.py | 10 +--------- lemur/common/utils.py | 17 +++++++---------- lemur/plugins/lemur_aws/iam.py | 3 +-- lemur/plugins/lemur_cryptography/plugin.py | 11 ++++------- lemur/plugins/lemur_csr/plugin.py | 11 +++-------- lemur/plugins/lemur_java/plugin.py | 18 +++++------------- lemur/plugins/lemur_openssl/plugin.py | 11 +++-------- lemur/tests/conftest.py | 7 +++---- 8 files changed, 27 insertions(+), 61 deletions(-) diff --git a/lemur/certificates/service.py b/lemur/certificates/service.py index 1b203260..0f37d70e 100644 --- a/lemur/certificates/service.py +++ b/lemur/certificates/service.py @@ -221,11 +221,6 @@ def upload(**kwargs): else: kwargs['roles'] = roles - if kwargs.get('private_key'): - private_key = kwargs['private_key'] - if not isinstance(private_key, bytes): - kwargs['private_key'] = private_key.encode('utf-8') - cert = Certificate(**kwargs) cert.authority = kwargs.get('authority') cert = database.create(cert) @@ -432,10 +427,7 @@ def create_csr(**csr_config): encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, # would like to use PKCS8 but AWS ELBs don't like it encryption_algorithm=serialization.NoEncryption() - ) - - if isinstance(private_key, bytes): - private_key = private_key.decode('utf-8') + ).decode('utf-8') csr = request.public_bytes( encoding=serialization.Encoding.PEM diff --git a/lemur/common/utils.py b/lemur/common/utils.py index 0504c958..32271e89 100644 --- a/lemur/common/utils.py +++ b/lemur/common/utils.py @@ -48,24 +48,22 @@ def parse_certificate(body): :param body: :return: """ - if isinstance(body, str): - body = body.encode('utf-8') + assert isinstance(body, str) - return x509.load_pem_x509_certificate(body, default_backend()) + return x509.load_pem_x509_certificate(body.encode('utf-8'), default_backend()) def parse_private_key(private_key): """ Parses a PEM-format private key (RSA, DSA, ECDSA or any other supported algorithm). - Raises ValueError for an invalid string. + Raises ValueError for an invalid string. Raises AssertionError when passed value is not str-type. :param private_key: String containing PEM private key """ - if isinstance(private_key, str): - private_key = private_key.encode('utf8') + assert isinstance(private_key, str) - return load_pem_private_key(private_key, password=None, backend=default_backend()) + return load_pem_private_key(private_key.encode('utf8'), password=None, backend=default_backend()) def parse_csr(csr): @@ -75,10 +73,9 @@ def parse_csr(csr): :param csr: :return: """ - if isinstance(csr, str): - csr = csr.encode('utf-8') + assert isinstance(csr, str) - return x509.load_pem_x509_csr(csr, default_backend()) + return x509.load_pem_x509_csr(csr.encode('utf-8'), default_backend()) def get_authority_key(body): diff --git a/lemur/plugins/lemur_aws/iam.py b/lemur/plugins/lemur_aws/iam.py index 7010c909..49816c2b 100644 --- a/lemur/plugins/lemur_aws/iam.py +++ b/lemur/plugins/lemur_aws/iam.py @@ -64,6 +64,7 @@ def upload_cert(name, body, private_key, path, cert_chain=None, **kwargs): :param path: :return: """ + assert isinstance(private_key, str) client = kwargs.pop('client') if not path or path == '/': @@ -72,8 +73,6 @@ def upload_cert(name, body, private_key, path, cert_chain=None, **kwargs): name = name + '-' + path.strip('/') try: - if isinstance(private_key, bytes): - private_key = private_key.decode("utf-8") if cert_chain: return client.upload_server_certificate( Path=path, diff --git a/lemur/plugins/lemur_cryptography/plugin.py b/lemur/plugins/lemur_cryptography/plugin.py index fe9d7bb3..97060391 100644 --- a/lemur/plugins/lemur_cryptography/plugin.py +++ b/lemur/plugins/lemur_cryptography/plugin.py @@ -14,6 +14,7 @@ from cryptography import x509 from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes, serialization +from lemur.common.utils import parse_private_key from lemur.plugins.bases import IssuerPlugin from lemur.plugins import lemur_cryptography as cryptography_issuer @@ -40,7 +41,8 @@ def issue_certificate(csr, options, private_key=None): if options.get("authority"): # Issue certificate signed by an existing lemur_certificates authority issuer_subject = options['authority'].authority_certificate.subject - issuer_private_key = options['authority'].authority_certificate.private_key + assert private_key is None, "Private would be ignored, authority key used instead" + private_key = options['authority'].authority_certificate.private_key chain_cert_pem = options['authority'].authority_certificate.body authority_key_identifier_public = options['authority'].authority_certificate.public_key authority_key_identifier_subject = x509.SubjectKeyIdentifier.from_public_key(authority_key_identifier_public) @@ -52,7 +54,6 @@ def issue_certificate(csr, options, private_key=None): else: # Issue certificate that is self-signed (new lemur_certificates root authority) issuer_subject = csr.subject - issuer_private_key = private_key chain_cert_pem = "" authority_key_identifier_public = csr.public_key() authority_key_identifier_subject = None @@ -112,11 +113,7 @@ def issue_certificate(csr, options, private_key=None): # FIXME: Not implemented in lemur/schemas.py yet https://github.com/Netflix/lemur/issues/662 pass - private_key = serialization.load_pem_private_key( - bytes(str(issuer_private_key).encode('utf-8')), - password=None, - backend=default_backend() - ) + private_key = parse_private_key(private_key) cert = builder.sign(private_key, hashes.SHA256(), default_backend()) cert_pem = cert.public_bytes( diff --git a/lemur/plugins/lemur_csr/plugin.py b/lemur/plugins/lemur_csr/plugin.py index e06035d1..13f42084 100644 --- a/lemur/plugins/lemur_csr/plugin.py +++ b/lemur/plugins/lemur_csr/plugin.py @@ -38,14 +38,9 @@ def create_csr(cert, chain, csr_tmp, key): :param csr_tmp: :param key: """ - if isinstance(cert, bytes): - cert = cert.decode('utf-8') - - if isinstance(chain, bytes): - chain = chain.decode('utf-8') - - if isinstance(key, bytes): - key = key.decode('utf-8') + assert isinstance(cert, str) + assert isinstance(chain, str) + assert isinstance(key, str) with mktempfile() as key_tmp: with open(key_tmp, 'w') as f: diff --git a/lemur/plugins/lemur_java/plugin.py b/lemur/plugins/lemur_java/plugin.py index 151794da..5aab5342 100644 --- a/lemur/plugins/lemur_java/plugin.py +++ b/lemur/plugins/lemur_java/plugin.py @@ -59,11 +59,8 @@ def split_chain(chain): def create_truststore(cert, chain, jks_tmp, alias, passphrase): - if isinstance(cert, bytes): - cert = cert.decode('utf-8') - - if isinstance(chain, bytes): - chain = chain.decode('utf-8') + assert isinstance(cert, str) + assert isinstance(chain, str) with mktempfile() as cert_tmp: with open(cert_tmp, 'w') as f: @@ -98,14 +95,9 @@ def create_truststore(cert, chain, jks_tmp, alias, passphrase): def create_keystore(cert, chain, jks_tmp, key, alias, passphrase): - if isinstance(cert, bytes): - cert = cert.decode('utf-8') - - if isinstance(chain, bytes): - chain = chain.decode('utf-8') - - if isinstance(key, bytes): - key = key.decode('utf-8') + assert isinstance(cert, str) + assert isinstance(chain, str) + assert isinstance(key, str) # Create PKCS12 keystore from private key and public certificate with mktempfile() as cert_tmp: diff --git a/lemur/plugins/lemur_openssl/plugin.py b/lemur/plugins/lemur_openssl/plugin.py index d50b4e43..9ddce925 100644 --- a/lemur/plugins/lemur_openssl/plugin.py +++ b/lemur/plugins/lemur_openssl/plugin.py @@ -44,14 +44,9 @@ def create_pkcs12(cert, chain, p12_tmp, key, alias, passphrase): :param alias: :param passphrase: """ - if isinstance(cert, bytes): - cert = cert.decode('utf-8') - - if isinstance(chain, bytes): - chain = chain.decode('utf-8') - - if isinstance(key, bytes): - key = key.decode('utf-8') + assert isinstance(cert, str) + assert isinstance(chain, str) + assert isinstance(key, str) with mktempfile() as key_tmp: with open(key_tmp, 'w') as f: diff --git a/lemur/tests/conftest.py b/lemur/tests/conftest.py index 9a48eb94..3790358e 100644 --- a/lemur/tests/conftest.py +++ b/lemur/tests/conftest.py @@ -3,12 +3,11 @@ import os import datetime import pytest from cryptography import x509 -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives.serialization import load_pem_private_key from flask import current_app from flask_principal import identity_changed, Identity from lemur import create_app +from lemur.common.utils import parse_private_key from lemur.database import db as _db from lemur.auth.service import create_token from lemur.tests.vectors import SAN_CERT_KEY, INTERMEDIATE_KEY @@ -235,12 +234,12 @@ def logged_in_admin(session, app): @pytest.fixture def private_key(): - return load_pem_private_key(SAN_CERT_KEY.encode(), password=None, backend=default_backend()) + return parse_private_key(SAN_CERT_KEY) @pytest.fixture def issuer_private_key(): - return load_pem_private_key(INTERMEDIATE_KEY.encode(), password=None, backend=default_backend()) + return parse_private_key(INTERMEDIATE_KEY) @pytest.fixture