add load_der_x509_csr to support loading DER encoded CSRs
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 2d705b1..b520e48 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -23,7 +23,8 @@
and
:class:`~cryptography.hazmat.primitives.asymmetric.ec.EllipticCurvePrivateKeyWithSerialization`.
* Add support for parsing X.509 certificate signing requests (CSRs) with
- :func:`~cryptography.x509.load_pem_x509_csr`.
+ :func:`~cryptography.x509.load_pem_x509_csr` and
+ :func:`~cryptography.x509.load_der_x509_csr`.
0.8.1 - 2015-03-20
~~~~~~~~~~~~~~~~~~
diff --git a/docs/x509.rst b/docs/x509.rst
index 22b282c..dbfb441 100644
--- a/docs/x509.rst
+++ b/docs/x509.rst
@@ -118,6 +118,22 @@
:returns: An instance of
:class:`~cryptography.x509.CertificateSigningRequest`.
+.. function:: load_der_x509_csr(data, backend)
+
+ .. versionadded:: 0.9
+
+ Deserialize a certificate signing request (CSR) from DER encoded data. DER
+ is a binary format and is not commonly used with CSRs.
+
+ :param bytes data: The DER encoded request data.
+
+ :param backend: A backend supporting the
+ :class:`~cryptography.hazmat.backends.interfaces.X509Backend`
+ interface.
+
+ :returns: An instance of
+ :class:`~cryptography.x509.CertificateSigningRequest`.
+
.. doctest::
>>> from cryptography import x509
diff --git a/src/cryptography/hazmat/backends/interfaces.py b/src/cryptography/hazmat/backends/interfaces.py
index 354b514..5224f5c 100644
--- a/src/cryptography/hazmat/backends/interfaces.py
+++ b/src/cryptography/hazmat/backends/interfaces.py
@@ -263,6 +263,12 @@
"""
@abc.abstractmethod
+ def load_der_x509_csr(self, data):
+ """
+ Load an X.509 CSR from DER encoded data.
+ """
+
+ @abc.abstractmethod
def load_pem_x509_csr(self, data):
"""
Load an X.509 CSR from PEM encoded data.
diff --git a/src/cryptography/hazmat/backends/multibackend.py b/src/cryptography/hazmat/backends/multibackend.py
index 54b044f..784ab84 100644
--- a/src/cryptography/hazmat/backends/multibackend.py
+++ b/src/cryptography/hazmat/backends/multibackend.py
@@ -325,6 +325,15 @@
_Reasons.UNSUPPORTED_X509
)
+ def load_der_x509_csr(self, data):
+ for b in self._filtered_backends(X509Backend):
+ return b.load_der_x509_csr(data)
+
+ raise UnsupportedAlgorithm(
+ "This backend does not support X.509.",
+ _Reasons.UNSUPPORTED_X509
+ )
+
def load_pem_x509_csr(self, data):
for b in self._filtered_backends(X509Backend):
return b.load_pem_x509_csr(data)
diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py
index 854219f..665771a 100644
--- a/src/cryptography/hazmat/backends/openssl/backend.py
+++ b/src/cryptography/hazmat/backends/openssl/backend.py
@@ -834,6 +834,16 @@
x509_req = self._ffi.gc(x509_req, self._lib.X509_REQ_free)
return _CertificateSigningRequest(self, x509_req)
+ def load_der_x509_csr(self, data):
+ mem_bio = self._bytes_to_bio(data)
+ x509_req = self._lib.d2i_X509_REQ_bio(mem_bio.bio, self._ffi.NULL)
+ if x509_req == self._ffi.NULL:
+ self._consume_errors()
+ raise ValueError("Unable to load request")
+
+ x509_req = self._ffi.gc(x509_req, self._lib.X509_REQ_free)
+ return _CertificateSigningRequest(self, x509_req)
+
def _load_key(self, openssl_read_func, convert_func, data, password):
mem_bio = self._bytes_to_bio(data)
diff --git a/src/cryptography/x509.py b/src/cryptography/x509.py
index 25bbe9c..113c35e 100644
--- a/src/cryptography/x509.py
+++ b/src/cryptography/x509.py
@@ -64,6 +64,10 @@
return backend.load_pem_x509_csr(data)
+def load_der_x509_csr(data, backend):
+ return backend.load_der_x509_csr(data)
+
+
class InvalidVersion(Exception):
def __init__(self, msg, parsed_version):
super(InvalidVersion, self).__init__(msg)
diff --git a/tests/hazmat/backends/test_multibackend.py b/tests/hazmat/backends/test_multibackend.py
index 4030538..5871e6c 100644
--- a/tests/hazmat/backends/test_multibackend.py
+++ b/tests/hazmat/backends/test_multibackend.py
@@ -200,6 +200,9 @@
def load_pem_x509_csr(self, data):
pass
+ def load_der_x509_csr(self, data):
+ pass
+
class TestMultiBackend(object):
def test_ciphers(self):
@@ -476,6 +479,7 @@
backend.load_pem_x509_certificate(b"certdata")
backend.load_der_x509_certificate(b"certdata")
backend.load_pem_x509_csr(b"reqdata")
+ backend.load_der_x509_csr(b"reqdata")
backend = MultiBackend([])
with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_X509):
@@ -484,3 +488,5 @@
backend.load_der_x509_certificate(b"certdata")
with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_X509):
backend.load_pem_x509_csr(b"reqdata")
+ with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_X509):
+ backend.load_der_x509_csr(b"reqdata")
diff --git a/tests/test_x509.py b/tests/test_x509.py
index 22b93f6..dc148d9 100644
--- a/tests/test_x509.py
+++ b/tests/test_x509.py
@@ -340,12 +340,21 @@
with pytest.raises(UnsupportedAlgorithm):
cert.signature_hash_algorithm
- def test_load_rsa_certificate_request(self, backend):
- request = _load_cert(
- os.path.join("x509", "requests", "rsa_sha1.pem"),
- x509.load_pem_x509_csr,
- backend
- )
+ @pytest.mark.parametrize(
+ ("path", "loader_func"),
+ [
+ [
+ os.path.join("x509", "requests", "rsa_sha1.pem"),
+ x509.load_pem_x509_csr
+ ],
+ [
+ os.path.join("x509", "requests", "rsa_sha1.der"),
+ x509.load_der_x509_csr
+ ],
+ ]
+ )
+ def test_load_rsa_certificate_request(self, path, loader_func, backend):
+ request = _load_cert(path, loader_func, backend)
assert isinstance(request.signature_hash_algorithm, hashes.SHA1)
public_key = request.public_key()
assert isinstance(public_key, rsa.RSAPublicKey)
@@ -359,9 +368,13 @@
x509.NameAttribute(x509.OID_COMMON_NAME, 'cryptography.io'),
]
- def test_invalid_certificate_request_pem(self, backend):
+ @pytest.mark.parametrize(
+ "loader_func",
+ [x509.load_pem_x509_csr, x509.load_der_x509_csr]
+ )
+ def test_invalid_certificate_request(self, loader_func, backend):
with pytest.raises(ValueError):
- x509.load_pem_x509_csr(b"notacsr", backend)
+ loader_func(b"notacsr", backend)
def test_unsupported_signature_hash_algorithm_request(self, backend):
request = _load_cert(
@@ -424,12 +437,21 @@
"822ff5d234e073b901cf5941f58e1f538e71d40d", 16
)
- def test_load_dsa_request(self, backend):
- request = _load_cert(
- os.path.join("x509", "requests", "dsa_sha1.pem"),
- x509.load_pem_x509_csr,
- backend
- )
+ @pytest.mark.parametrize(
+ ("path", "loader_func"),
+ [
+ [
+ os.path.join("x509", "requests", "dsa_sha1.pem"),
+ x509.load_pem_x509_csr
+ ],
+ [
+ os.path.join("x509", "requests", "dsa_sha1.der"),
+ x509.load_der_x509_csr
+ ],
+ ]
+ )
+ def test_load_dsa_request(self, path, loader_func, backend):
+ request = _load_cert(path, loader_func, backend)
assert isinstance(request.signature_hash_algorithm, hashes.SHA1)
public_key = request.public_key()
assert isinstance(public_key, dsa.DSAPublicKey)
@@ -479,13 +501,22 @@
with pytest.raises(NotImplementedError):
cert.public_key()
- def test_load_ecdsa_certificate_request(self, backend):
+ @pytest.mark.parametrize(
+ ("path", "loader_func"),
+ [
+ [
+ os.path.join("x509", "requests", "ec_sha256.pem"),
+ x509.load_pem_x509_csr
+ ],
+ [
+ os.path.join("x509", "requests", "ec_sha256.der"),
+ x509.load_der_x509_csr
+ ],
+ ]
+ )
+ def test_load_ecdsa_certificate_request(self, path, loader_func, backend):
_skip_curve_unsupported(backend, ec.SECP384R1())
- request = _load_cert(
- os.path.join("x509", "requests", "ec_sha256.pem"),
- x509.load_pem_x509_csr,
- backend
- )
+ request = _load_cert(path, loader_func, backend)
assert isinstance(request.signature_hash_algorithm, hashes.SHA256)
public_key = request.public_key()
assert isinstance(public_key, ec.EllipticCurvePublicKey)