Add rfc based validators

Add a validator which collects various standard format/behaviour tests.
These are not user-configurable and any valid request failing them is a
bug in Anchor.

All checks reference the document where they're defined.

Closes-bug: 1476877
Partial-bug: 1476875
Change-Id: I208685d8d7cde40ed5294e7235d64ca17617c094

anchor/validators_standards.py

@@ -0,0 +1,107 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+Standards based validator.
+
+This module provides validators which should be included in all deployments and
+which are based directly on the standards documents. All exceptions must have a
+comment referencing the document / section they're based on.
+
+All the rules are pulled into a single validator: ``standards_compliance``.
+"""
+
+from __future__ import absolute_import
+
+import re
+
+from anchor import validators
+from anchor.X509 import extension
+
+
+def standards_compliance(csr=None, **kwargs):
+    """Collection of separate cases of standards validation."""
+    _no_extension_duplicates(csr)
+    _critical_flags(csr)
+    _valid_domains(csr)
+    # TODO(stan): validate srv/uri, distinct DNs, email format, identity keys
+
+
+def _no_extension_duplicates(csr):
+    """Only one extension with a given oid is allowed.
+
+    See RFC5280 section 4.2
+    """
+    seen_oids = set()
+    for ext in csr.get_extensions():
+        oid = ext.get_oid()
+        if oid in seen_oids:
+            raise validators.ValidationError(
+                "Duplicate extension with oid %s (RFC5280/4.2)" % oid)
+        seen_oids.add(oid)
+
+
+def _critical_flags(csr):
+    """Various rules define whether critical flag is required."""
+    for ext in csr.get_extensions():
+        if isinstance(ext, extension.X509ExtensionSubjectAltName):
+            if len(csr.get_subject()) == 0 and not ext.get_critical():
+                raise validators.ValidationError(
+                    "SAN must be critical if subject is empty "
+                    "(RFC5280/4.1.2.6)")
+        if isinstance(ext, extension.X509ExtensionBasicConstraints):
+            if not ext.get_critical():
+                raise validators.ValidationError(
+                    "Basic constraints has to be marked critical "
+                    "(RFC5280/4.1.2.9)")
+
+
+# RFC1034 allows a simple " " too, but it's not allowed in certificates, so it
+# will not match
+RE_DOMAIN_LABEL = re.compile("^[a-z](?:[-a-z0-9]*[a-z0-9])?$", re.IGNORECASE)
+
+
+def _valid_domains(csr):
+    """Format of the domin names
+
+    See RFC5280 section 4.2.1.6 / RFC6125 / RFC1034
+    """
+    def verify_domain(domain):
+        labels = domain.split('.')
+        if labels[-1] == "":
+            # single trailing . is ok, ignore
+            labels.pop(-1)
+        for i, label in enumerate(labels):
+            if len(label) > 63:
+                raise validators.ValidationError(
+                    "SAN entry <%s> it too long (RFC5280/4.2.1.6)" % (domain,))
+
+            # check for wildcard labels, ignore partial-wildcard labels
+            if '*' == label:
+                if i != 0:
+                    raise validators.ValidationError(
+                        "SAN entry <%s> has wildcard that's not in the "
+                        "left-most label (RFC6125/6.4.3)" % (domain,))
+            else:
+                if RE_DOMAIN_LABEL.match(label) is None:
+                    raise validators.ValidationError(
+                        "SAN entry <%s> contains invalid characters "
+                        "(RFC1034/3.5)" % (domain,))
+
+    sans = csr.get_extensions(extension.X509ExtensionSubjectAltName)
+    if not sans:
+        return
+
+    ext = sans[0]
+    for domain in ext.get_dns_ids():
+        verify_domain(domain)

config.json

@@ -21,6 +21,7 @@
       "authentication": "method_1",
       "signing_ca": "local",
       "validators": {
+        "standards_compliance": {},
         "ca_status": {
           "ca_requested": false
         },

doc/source/validators.rst

@@ -11,6 +11,16 @@ Included validators
 
 The following validators are implemented at the moment:
 
+``standards_compliance``
+    Verifies: CSR.
+
+    Ensures that the CSR does not break any rules defined in the standards
+    documents (mostly RFC5280). Specific checks may be added over time in new
+    versions of Anchor. This validator should be only skipped if there's a
+    known compatibility issue. Otherwise it should be used in all environments.
+    Any requests produced using standard tooling that fail this check should be
+    reported as Anchor issues.
+
 ``common_name``
     Verifies: CSR. Parameters: ``allowed_domains``, ``allowed_networks``.
 

setup.cfg

@@ -43,6 +43,7 @@ anchor.validators =
     key_usage = anchor.validators:key_usage
     ca_status = anchor.validators:ca_status
     source_cidrs = anchor.validators:source_cidrs
+    standards_compliance = anchor.validators_standards:standards_compliance
 
 anchor.authentication =
     keystone = anchor.auth.keystone:login

tests/validators/test_standards_validator.py

@@ -0,0 +1,174 @@
+# -*- coding:utf-8 -*-
+#
+# Copyright 2015 Hewlett-Packard Development Company, L.P.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import textwrap
+import unittest
+
+from pyasn1.codec.der import encoder
+from pyasn1_modules import rfc2459
+
+from anchor import validators
+from anchor import validators_standards
+from anchor.X509 import extension
+from anchor.X509 import name
+from anchor.X509 import signing_request
+
+
+class TestStandardsValidator(unittest.TestCase):
+    csr_data = textwrap.dedent(u"""
+        -----BEGIN CERTIFICATE REQUEST-----
+        MIIB1TCCAT4CAQAwgZQxCzAJBgNVBAYTAlVLMQ8wDQYDVQQIDAZOYXJuaWExEjAQ
+        BgNVBAcMCUZ1bmt5dG93bjEXMBUGA1UECgwOQW5jaG9yIFRlc3RpbmcxEDAOBgNV
+        BAsMB3Rlc3RpbmcxFDASBgNVBAMMC2FuY2hvci50ZXN0MR8wHQYJKoZIhvcNAQkB
+        FhB0ZXN0QGFuY2hvci50ZXN0MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCe
+        eqg1Qeccv8hqj1BP9KEJX5QsFCxR62M8plPb5t4sLo8UYfZd6kFLcOP8xzwwvx/e
+        FY6Sux52enQ197o8aMwyP77hMhZqtd8NCgLJMVlUbRhwLti0SkHFPic0wAg+esfX
+        a6yhd5TxC+bti7MgV/ljA80XQxHH8xOjdOoGN0DHfQIDAQABoAAwDQYJKoZIhvcN
+        AQELBQADgYEAI4eMihRKSeNLt1DLg6l+WYU4ssRTEHpxwBRo0lh5IGEBjtL+NrPY
+        /A9AKfbkyW7BnKd9IT5wvenZajl5UzCveTCkqVDbSEOwLpUY3GeHf0jujml8gKFb
+        AFrlaOkOuDai+an0EdbeLef1kYh8CWd573MPvKTwOsiaGP/EACrlIEM=
+        -----END CERTIFICATE REQUEST-----""")
+
+    def test_passing(self):
+        csr = signing_request.X509Csr.from_buffer(self.csr_data)
+        validators_standards.standards_compliance(csr=csr)
+
+
+class TestExtensionDuplicates(unittest.TestCase):
+    def test_no_extensions(self):
+        csr = signing_request.X509Csr()
+        validators_standards._no_extension_duplicates(csr)
+
+    def test_no_duplicates(self):
+        csr = signing_request.X509Csr()
+        ext = extension.X509ExtensionSubjectAltName()
+        csr.add_extension(ext)
+        validators_standards._no_extension_duplicates(csr)
+
+    def test_with_duplicates(self):
+        csr = signing_request.X509Csr()
+        ext = extension.X509ExtensionSubjectAltName()
+        ext.add_dns_id('example.com')
+        exts = rfc2459.Extensions()
+        exts[0] = ext._ext
+        exts[1] = ext._ext
+        # Anchor doesn't allow this normally, so tests need to cheat
+        attrs = csr.get_attributes()
+        attrs[0] = None
+        attrs[0]['type'] = signing_request.OID_extensionRequest
+        attrs[0]['vals'] = None
+        attrs[0]['vals'][0] = encoder.encode(exts)
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._no_extension_duplicates(csr)
+
+
+class TestExtensionCriticalFlags(unittest.TestCase):
+    def test_no_subject_san_not_critical(self):
+        csr = signing_request.X509Csr()
+        ext = extension.X509ExtensionSubjectAltName()
+        ext.set_critical(False)
+        ext.add_dns_id('example.com')
+        csr.add_extension(ext)
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._critical_flags(csr)
+
+    def test_no_subject_san_critical(self):
+        csr = signing_request.X509Csr()
+        ext = extension.X509ExtensionSubjectAltName()
+        ext.set_critical(True)
+        ext.add_dns_id('example.com')
+        csr.add_extension(ext)
+        validators_standards._critical_flags(csr)
+
+    def test_with_subject_san_not_critical(self):
+        csr = signing_request.X509Csr()
+        subject = name.X509Name()
+        subject.add_name_entry(name.OID_commonName, "example.com")
+        csr.set_subject(subject)
+        ext = extension.X509ExtensionSubjectAltName()
+        ext.set_critical(False)
+        ext.add_dns_id('example.com')
+        csr.add_extension(ext)
+        validators_standards._critical_flags(csr)
+
+    def test_basic_constraints_not_critical(self):
+        csr = signing_request.X509Csr()
+        ext = extension.X509ExtensionBasicConstraints()
+        ext.set_critical(False)
+        csr.add_extension(ext)
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._critical_flags(csr)
+
+    def test_basic_constraints_critical(self):
+        csr = signing_request.X509Csr()
+        ext = extension.X509ExtensionBasicConstraints()
+        ext.set_critical(True)
+        csr.add_extension(ext)
+        validators_standards._critical_flags(csr)
+
+
+class TestValidDomains(unittest.TestCase):
+    def _create_csr_with_domain_san(self, domain):
+        csr = signing_request.X509Csr()
+        ext = extension.X509ExtensionSubjectAltName()
+        ext.add_dns_id(domain)
+        csr.add_extension(ext)
+        return csr
+
+    def test_all_valid(self):
+        csr = self._create_csr_with_domain_san('a-123.example.com')
+        validators_standards._valid_domains(csr)
+
+    def test_all_valid_trailing_dot(self):
+        csr = self._create_csr_with_domain_san('a-123.example.com.')
+        validators_standards._valid_domains(csr)
+
+    def test_too_long(self):
+        csr = self._create_csr_with_domain_san(
+            'very-long-label-over-63-characters-'
+            'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.example.com')
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._valid_domains(csr)
+
+    def test_beginning_hyphen(self):
+        csr = self._create_csr_with_domain_san('-label.example.com.')
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._valid_domains(csr)
+
+    def test_trailing_hyphen(self):
+        csr = self._create_csr_with_domain_san('label-.example.com.')
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._valid_domains(csr)
+
+    def test_san_space(self):
+        # valid domain, but not in CSRs
+        csr = self._create_csr_with_domain_san(' ')
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._valid_domains(csr)
+
+    def test_wildcard(self):
+        csr = self._create_csr_with_domain_san('*.example.com')
+        validators_standards._valid_domains(csr)
+
+    def test_wildcard_middle(self):
+        csr = self._create_csr_with_domain_san('foo.*.example.com')
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._valid_domains(csr)
+
+    def test_wildcard_partial(self):
+        csr = self._create_csr_with_domain_san('foo*.example.com')
+        with self.assertRaises(validators.ValidationError):
+            validators_standards._valid_domains(csr)