support certificate policies in the openssl backend

2 files changed, 165 insertions, 0 deletions

diff --git a/src/cryptography/hazmat/backends/openssl/x509.py b/src/cryptography/hazmat/backends/openssl/x509.py
index 8b5f3920..9d70d72f 100644
--- a/src/cryptography/hazmat/backends/openssl/x509.py
+++ b/src/cryptography/hazmat/backends/openssl/x509.py
@@ -290,6 +290,8 @@ class _Certificate(object):
                 value = self._build_authority_key_identifier(ext)
             elif oid == x509.OID_AUTHORITY_INFORMATION_ACCESS:
                 value = self._build_authority_information_access(ext)
+            elif oid == x509.OID_CERTIFICATE_POLICIES:
+                value = self._build_certificate_policies(ext)
             elif critical:
                 raise x509.UnsupportedExtension(
                     "{0} is not currently supported".format(oid), oid
@@ -304,6 +306,75 @@ class _Certificate(object):
 
         return x509.Extensions(extensions)
 
+    def _build_certificate_policies(self, ext):
+        cp = self._backend._ffi.cast(
+            "Cryptography_STACK_OF_POLICYINFO *",
+            self._backend._lib.X509V3_EXT_d2i(ext)
+        )
+        assert cp != self._backend._ffi.NULL
+        cp = self._backend._ffi.gc(cp, self._backend._lib.sk_POLICYINFO_free)
+        num = self._backend._lib.sk_POLICYINFO_num(cp)
+        certificate_policies = []
+        for i in range(num):
+            qualifiers = None
+            pi = self._backend._lib.sk_POLICYINFO_value(cp, i)
+            oid = x509.ObjectIdentifier(_obj2txt(self._backend, pi.policyid))
+            if pi.qualifiers != self._backend._ffi.NULL:
+                qnum = self._backend._lib.sk_POLICYQUALINFO_num(pi.qualifiers)
+                qualifiers = []
+                for j in range(qnum):
+                    pqi = self._backend._lib.sk_POLICYQUALINFO_value(
+                        pi.qualifiers, j
+                    )
+                    pqualid = x509.ObjectIdentifier(
+                        _obj2txt(self._backend, pqi.pqualid)
+                    )
+                    if pqualid == x509.OID_CPS_QUALIFIER:
+                        cpsuri = self._backend._ffi.buffer(
+                            pqi.d.cpsuri.data, pqi.d.cpsuri.length
+                        )[:].decode('ascii')
+                        qualifiers.append(cpsuri)
+                    elif pqualid == x509.OID_CPS_USER_NOTICE:
+                        user_notice = self._build_user_notice(pqi.d.usernotice)
+                        qualifiers.append(user_notice)
+
+            certificate_policies.append(
+                x509.PolicyInformation(oid, qualifiers)
+            )
+
+        return x509.CertificatePolicies(certificate_policies)
+
+    def _build_user_notice(self, un):
+        explicit_text = None
+        notice_reference = None
+
+        if un.exptext != self._backend._ffi.NULL:
+            explicit_text = _asn1_string_to_utf8(self._backend, un.exptext)
+
+        if un.noticeref != self._backend._ffi.NULL:
+            organization = _asn1_string_to_utf8(
+                self._backend, un.noticeref.organization
+            )
+
+            num = self._backend._lib.sk_ASN1_INTEGER_num(
+                un.noticeref.noticenos
+            )
+            notice_numbers = []
+            for i in range(num):
+                asn1_int = self._backend._lib.sk_ASN1_INTEGER_value(
+                    un.noticeref.noticenos, i
+                )
+                notice_num = _asn1_integer_to_int(
+                    self._backend, asn1_int
+                )
+                notice_numbers.append(notice_num)
+
+            notice_reference = x509.NoticeReference(
+                organization, notice_numbers
+            )
+
+        return x509.UserNotice(notice_reference, explicit_text)
+
     def _build_basic_constraints(self, ext):
         bc_st = self._backend._lib.X509V3_EXT_d2i(ext)
         assert bc_st != self._backend._ffi.NULL
diff --git a/tests/test_x509_ext.py b/tests/test_x509_ext.py
index 701ea167..2852776b 100644
--- a/tests/test_x509_ext.py
+++ b/tests/test_x509_ext.py
@@ -235,6 +235,100 @@ class TestCertificatePolicies(object):
         assert cp != object()
 
 
+@pytest.mark.requires_backend_interface(interface=RSABackend)
+@pytest.mark.requires_backend_interface(interface=X509Backend)
+class TestCertificatePoliciesExtension(object):
+    def test_cps_uri_policy_qualifier(self, backend):
+        cert = _load_cert(
+            os.path.join("x509", "custom", "cp_cps_uri.pem"),
+            x509.load_pem_x509_certificate,
+            backend
+        )
+
+        cp = cert.extensions.get_extension_for_oid(
+            x509.OID_CERTIFICATE_POLICIES
+        ).value
+
+        assert cp == x509.CertificatePolicies([
+            x509.PolicyInformation(
+                x509.ObjectIdentifier("2.16.840.1.12345.1.2.3.4.1"),
+                [u"http://other.com/cps"]
+            )
+        ])
+
+    def test_user_notice_with_notice_reference(self, backend):
+        cert = _load_cert(
+            os.path.join(
+                "x509", "custom", "cp_user_notice_with_notice_reference.pem"
+            ),
+            x509.load_pem_x509_certificate,
+            backend
+        )
+
+        cp = cert.extensions.get_extension_for_oid(
+            x509.OID_CERTIFICATE_POLICIES
+        ).value
+
+        assert cp == x509.CertificatePolicies([
+            x509.PolicyInformation(
+                x509.ObjectIdentifier("2.16.840.1.12345.1.2.3.4.1"),
+                [
+                    u"http://example.com/cps",
+                    u"http://other.com/cps",
+                    x509.UserNotice(
+                        x509.NoticeReference(u"my org", [1, 2, 3, 4]),
+                        u"thing"
+                    )
+                ]
+            )
+        ])
+
+    def test_user_notice_with_explicit_text(self, backend):
+        cert = _load_cert(
+            os.path.join(
+                "x509", "custom", "cp_user_notice_with_explicit_text.pem"
+            ),
+            x509.load_pem_x509_certificate,
+            backend
+        )
+
+        cp = cert.extensions.get_extension_for_oid(
+            x509.OID_CERTIFICATE_POLICIES
+        ).value
+
+        assert cp == x509.CertificatePolicies([
+            x509.PolicyInformation(
+                x509.ObjectIdentifier("2.16.840.1.12345.1.2.3.4.1"),
+                [x509.UserNotice(None, u"thing")]
+            )
+        ])
+
+    def test_user_notice_no_explicit_text(self, backend):
+        cert = _load_cert(
+            os.path.join(
+                "x509", "custom", "cp_user_notice_no_explicit_text.pem"
+            ),
+            x509.load_pem_x509_certificate,
+            backend
+        )
+
+        cp = cert.extensions.get_extension_for_oid(
+            x509.OID_CERTIFICATE_POLICIES
+        ).value
+
+        assert cp == x509.CertificatePolicies([
+            x509.PolicyInformation(
+                x509.ObjectIdentifier("2.16.840.1.12345.1.2.3.4.1"),
+                [
+                    x509.UserNotice(
+                        x509.NoticeReference(u"my org", [1, 2, 3, 4]),
+                        None
+                    )
+                ]
+            )
+        ])
+
+
 class TestKeyUsage(object):
     def test_key_agreement_false_encipher_decipher_true(self):
         with pytest.raises(ValueError):