Merge branch 'master' into kubernetes-improvment

lemur/common/defaults.py

@@ -7,18 +7,21 @@ from lemur.extensions import sentry
 from lemur.constants import SAN_NAMING_TEMPLATE, DEFAULT_NAMING_TEMPLATE
 
 
-def text_to_slug(value):
-    """Normalize a string to a "slug" value, stripping character accents and removing non-alphanum characters."""
+def text_to_slug(value, joiner='-'):
+    """
+    Normalize a string to a "slug" value, stripping character accents and removing non-alphanum characters.
+    A series of non-alphanumeric characters is replaced with the joiner character.
+    """
 
     # Strip all character accents: decompose Unicode characters and then drop combining chars.
     value = ''.join(c for c in unicodedata.normalize('NFKD', value) if not unicodedata.combining(c))
 
-    # Replace all remaining non-alphanumeric characters with '-'. Multiple characters get collapsed into a single dash.
-    # Except, keep 'xn--' used in IDNA domain names as is.
-    value = re.sub(r'[^A-Za-z0-9.]+(?<!xn--)', '-', value)
+    # Replace all remaining non-alphanumeric characters with joiner string. Multiple characters get collapsed into a
+    # single joiner. Except, keep 'xn--' used in IDNA domain names as is.
+    value = re.sub(r'[^A-Za-z0-9.]+(?<!xn--)', joiner, value)
 
     # '-' in the beginning or end of string looks ugly.
-    return value.strip('-')
+    return value.strip(joiner)
 
 
 def certificate_name(common_name, issuer, not_before, not_after, san):
@@ -224,25 +227,20 @@ def bitstrength(cert):
 
 def issuer(cert):
     """
-    Gets a sane issuer name from a given certificate.
+    Gets a sane issuer slug from a given certificate, stripping non-alphanumeric characters.
 
     :param cert:
-    :return: Issuer
+    :return: Issuer slug
     """
-    delchars = ''.join(c for c in map(chr, range(256)) if not c.isalnum())
-    try:
-        # Try organization name or fall back to CN
-        issuer = (cert.issuer.get_attributes_for_oid(x509.OID_COMMON_NAME) or
-                  cert.issuer.get_attributes_for_oid(x509.OID_ORGANIZATION_NAME))
-        issuer = str(issuer[0].value)
-        for c in delchars:
-            issuer = issuer.replace(c, "")
-        return issuer
-    except Exception as e:
-        sentry.captureException()
-        current_app.logger.error("Unable to get issuer! {0}".format(e))
+    # Try Common Name or fall back to Organization name
+    attrs = (cert.issuer.get_attributes_for_oid(x509.OID_COMMON_NAME) or
+             cert.issuer.get_attributes_for_oid(x509.OID_ORGANIZATION_NAME))
+    if not attrs:
+        current_app.logger.error("Unable to get issuer! Cert serial {:x}".format(cert.serial_number))
         return "Unknown"
 
+    return text_to_slug(attrs[0].value, '')
+
 
 def not_before(cert):
     """

lemur/plugins/lemur_kubernetes/plugin.py

@@ -73,7 +73,6 @@ def _resolve_uri(k8s_base_uri, namespace, kind, name=None, api_ver=DEFAULT_API_V
 def base64encode(string):
     return base64.b64encode(string.encode()).decode()
 
-
 def build_secret(secret_format, secret_name, body, private_key, cert_chain):
     secret = {
         'apiVersion': 'v1',
@@ -102,7 +101,6 @@ def build_secret(secret_format, secret_name, body, private_key, cert_chain):
         }
     return secret
 
-
 class KubernetesDestinationPlugin(DestinationPlugin):
     title = 'Kubernetes'
     slug = 'kubernetes-destination'

lemur/tests/conftest.py

@@ -11,7 +11,7 @@ from flask_principal import identity_changed, Identity
 from lemur import create_app
 from lemur.database import db as _db
 from lemur.auth.service import create_token
-from lemur.tests.vectors import SAN_CERT_KEY
+from lemur.tests.vectors import SAN_CERT_KEY, INTERMEDIATE_KEY
 
 from .factories import ApiKeyFactory, AuthorityFactory, NotificationFactory, DestinationFactory, \
     CertificateFactory, UserFactory, RoleFactory, SourceFactory, EndpointFactory, \
@@ -231,6 +231,11 @@ def private_key():
     return load_pem_private_key(SAN_CERT_KEY.encode(), password=None, backend=default_backend())
 
 
+@pytest.fixture
+def issuer_private_key():
+    return load_pem_private_key(INTERMEDIATE_KEY.encode(), password=None, backend=default_backend())
+
+
 @pytest.fixture
 def cert_builder(private_key):
     return (x509.CertificateBuilder()

lemur/tests/test_defaults.py

@@ -1,3 +1,7 @@
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes
+
 from .vectors import SAN_CERT, WILDCARD_CERT, INTERMEDIATE_CERT
 
 
@@ -41,12 +45,14 @@ def test_cert_issuer(client):
 def test_text_to_slug(client):
     from lemur.common.defaults import text_to_slug
     assert text_to_slug('test - string') == 'test-string'
+    assert text_to_slug('test - string', '') == 'teststring'
     # Accented characters are decomposed
     assert text_to_slug('föö bär') == 'foo-bar'
     # Melt away the Unicode Snowman
     assert text_to_slug('\u2603') == ''
     assert text_to_slug('\u2603test\u2603') == 'test'
     assert text_to_slug('snow\u2603man') == 'snow-man'
+    assert text_to_slug('snow\u2603man', '') == 'snowman'
     # IDNA-encoded domain names should be kept as-is
     assert text_to_slug('xn--i1b6eqas.xn--xmpl-loa9b3671b.com') == 'xn--i1b6eqas.xn--xmpl-loa9b3671b.com'
 
@@ -75,3 +81,29 @@ def test_create_name(client):
         datetime(2015, 5, 12, 0, 0, 0),
         False
     ) == 'xn--mnchen-3ya.de-VertrauenswurdigAutoritat-20150507-20150512'
+
+
+def test_issuer(client, cert_builder, issuer_private_key):
+    from lemur.common.defaults import issuer
+
+    assert issuer(INTERMEDIATE_CERT) == 'LemurTrustUnittestsRootCA2018'
+
+    # We need to override builder's issuer name
+    cert_builder._issuer_name = None
+    # Unicode issuer name
+    cert = (cert_builder
+            .issuer_name(x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, 'Vertrauenswürdig Autorität')]))
+            .sign(issuer_private_key, hashes.SHA256(), default_backend()))
+    assert issuer(cert) == 'VertrauenswurdigAutoritat'
+
+    # Fallback to 'Organization' field when issuer CN is missing
+    cert = (cert_builder
+            .issuer_name(x509.Name([x509.NameAttribute(x509.NameOID.ORGANIZATION_NAME, 'No Such Organization')]))
+            .sign(issuer_private_key, hashes.SHA256(), default_backend()))
+    assert issuer(cert) == 'NoSuchOrganization'
+
+    # Missing issuer name
+    cert = (cert_builder
+            .issuer_name(x509.Name([]))
+            .sign(issuer_private_key, hashes.SHA256(), default_backend()))
+    assert issuer(cert) == 'Unknown'