Add EKU extension validator

Add a validator for the extended key usage, matching the existing key
usage one.

Change-Id: I10efc01cabf3a63adce95f2a3a2b615070d84e2b

anchor/X509/extension.py

@@ -44,6 +44,20 @@ EXT_KEY_USAGE_NAMES = {
     id_kp_OCSPSigning: "OCSP Signing",
     anyExtendedKeyUsage: "Any Extended Key Usage",
 }
+EXT_KEY_USAGE_NAMES_INV = dict((v, k) for k, v in EXT_KEY_USAGE_NAMES.items())
+
+
+EXT_KEY_USAGE_SHORT_NAMES = {
+    rfc2459.id_kp_serverAuth: "serverAuth",
+    rfc2459.id_kp_clientAuth: "clientAuth",
+    rfc2459.id_kp_codeSigning: "codeSigning",
+    rfc2459.id_kp_emailProtection: "emailProtection",
+    rfc2459.id_kp_timeStamping: "timeStamping",
+    id_kp_OCSPSigning: "ocspSigning",
+    anyExtendedKeyUsage: "anyExtendedKeyUsage",
+}
+EXT_KEY_USAGE_SHORT_NAMES_INV = dict((v, k) for k, v in
+                                     EXT_KEY_USAGE_SHORT_NAMES.items())
 
 
 EXTENSION_NAMES = {

anchor/validators.py

@@ -16,6 +16,7 @@ from __future__ import absolute_import
 import logging
 
 import netaddr
+from pyasn1.type import univ as pyasn1_univ
 
 from anchor.X509 import errors
 from anchor.X509 import extension
@@ -204,10 +205,39 @@ def key_usage(csr=None, allowed_usage=None, **kwargs):
             usages = set(ext.get_all_usages())
             denied = denied | (usages - allowed)
     if denied:
-        raise ValidationError("Found some not allowed key usages: %s"
+        raise ValidationError("Found some prohibited key usages: %s"
                               % ', '.join(denied))
 
 
+def ext_key_usage(csr=None, allowed_usage=None, **kwargs):
+    """Ensure only accepted extended key usages are specified."""
+
+    # transform all possible names into oids we actually check
+    for i, usage in enumerate(allowed_usage):
+        if usage in extension.EXT_KEY_USAGE_NAMES_INV:
+            allowed_usage[i] = extension.EXT_KEY_USAGE_NAMES_INV[usage]
+        elif usage in extension.EXT_KEY_USAGE_SHORT_NAMES_INV:
+            allowed_usage[i] = extension.EXT_KEY_USAGE_SHORT_NAMES_INV[usage]
+        else:
+            try:
+                oid = pyasn1_univ.ObjectIdentifier(usage)
+                allowed_usage[i] = oid
+            except Exception:
+                raise ValidationError("Unknown usage: %s" % (usage,))
+
+    allowed = set(allowed_usage)
+    denied = set()
+
+    for ext in csr.get_extensions(extension.X509ExtensionExtendedKeyUsage):
+        usages = set(ext.get_all_usages())
+        denied = denied | (usages - allowed)
+    if denied:
+        text_denied = [extension.EXT_KEY_USAGE_SHORT_NAMES.get(x)
+                       for x in denied]
+        raise ValidationError("Found some prohibited key usages: %s"
+                              % ', '.join(text_denied))
+
+
 def ca_status(csr=None, ca_requested=False, **kwargs):
     """Ensure the request has/hasn't got the CA flag."""
     request_ca_flags = False

doc/source/validators.rst

@@ -81,6 +81,22 @@ The following validators are implemented at the moment:
     digitalSignature, nonRepudiation, keyEncipherment, dataEncipherment,
     keyAgreement, keyCertSign, cRLSign, encipherOnly, decipherOnly
 
+``ext_key_usage``
+    Verifies: CSR. Parameters: ``allowed_usage``.
+
+    Ensures only ``allowed_usage`` is requested for the certificate. The names
+    recognised by Anchor are:
+
+    TLS Web Server Authentication, TLS Web Client Authentication, Code Signing,
+    E-mail Protection, Time Stamping, OCSP Signing, Any Extended Key Usage
+
+    as well as short versions:
+
+    serverAuth, clientAuth, codeSigning, emailProtection, timeStamping,
+    ocspSigning, anyExtendedKeyUsage
+
+    or text representation of custom OIDs.
+
 ``ca_status``
     Verifies: CSR. Parameters: ``ca_requested``.
 

setup.cfg

@@ -41,6 +41,7 @@ anchor.validators =
     server_group = anchor.validators:server_group
     extensions = anchor.validators:extensions
     key_usage = anchor.validators:key_usage
+    ext_key_usage = anchor.validators:ext_key_usage
     ca_status = anchor.validators:ca_status
     source_cidrs = anchor.validators:source_cidrs
 

tests/validators/test_callable_validators.py

@@ -323,7 +323,7 @@ class TestValidators(unittest.TestCase):
             validators.key_usage(
                 csr=csr,
                 allowed_usage=allowed_usage)
-        self.assertEqual("Found some not allowed key usages: "
+        self.assertEqual("Found some prohibited key usages: "
                          "keyCertSign", str(e.exception))
 
     def test_key_usage_good(self):
@@ -345,6 +345,69 @@ class TestValidators(unittest.TestCase):
             )
         )
 
+    def test_ext_key_usage_good_short(self):
+        allowed_usage = ['serverAuth']
+
+        csr = x509_csr.X509Csr()
+        ext = x509_ext.X509ExtensionExtendedKeyUsage()
+        ext.set_usage(rfc2459.id_kp_serverAuth, True)
+        csr.add_extension(ext)
+
+        self.assertEqual(
+            None,
+            validators.ext_key_usage(
+                csr=csr,
+                allowed_usage=allowed_usage
+            )
+        )
+
+    def test_ext_key_usage_good_long(self):
+        allowed_usage = ['TLS Web Server Authentication']
+
+        csr = x509_csr.X509Csr()
+        ext = x509_ext.X509ExtensionExtendedKeyUsage()
+        ext.set_usage(rfc2459.id_kp_serverAuth, True)
+        csr.add_extension(ext)
+
+        self.assertEqual(
+            None,
+            validators.ext_key_usage(
+                csr=csr,
+                allowed_usage=allowed_usage
+            )
+        )
+
+    def test_ext_key_usage_good_oid(self):
+        allowed_usage = ["1.3.6.1.5.5.7.3.1"]
+
+        csr = x509_csr.X509Csr()
+        ext = x509_ext.X509ExtensionExtendedKeyUsage()
+        ext.set_usage(rfc2459.id_kp_serverAuth, True)
+        csr.add_extension(ext)
+
+        self.assertEqual(
+            None,
+            validators.ext_key_usage(
+                csr=csr,
+                allowed_usage=allowed_usage
+            )
+        )
+
+    def test_ext_key_usage_bad(self):
+        allowed_usage = ['serverAuth']
+
+        csr = x509_csr.X509Csr()
+        ext = x509_ext.X509ExtensionExtendedKeyUsage()
+        ext.set_usage(rfc2459.id_kp_clientAuth, True)
+        csr.add_extension(ext)
+
+        with self.assertRaises(validators.ValidationError) as e:
+            validators.ext_key_usage(
+                csr=csr,
+                allowed_usage=allowed_usage)
+        self.assertEqual("Found some prohibited key usages: "
+                         "clientAuth", str(e.exception))
+
     def test_ca_status_good1(self):
         csr = x509_csr.X509Csr()
         ext = x509_ext.X509ExtensionBasicConstraints()