Merge pull request #2108 from mail-in-a-box/master
support othername in general names
diff --git a/docs/x509.rst b/docs/x509.rst
index 2dac33b..bcb6ee6 100644
--- a/docs/x509.rst
+++ b/docs/x509.rst
@@ -328,6 +328,8 @@
.. method:: public_bytes(encoding)
+ .. versionadded:: 1.0
+
:param encoding: The
:class:`~cryptography.hazmat.primitives.serialization.Encoding`
that will be used to serialize the certificate.
@@ -435,6 +437,8 @@
.. method:: public_bytes(encoding)
+ .. versionadded:: 1.0
+
:param encoding: The
:class:`~cryptography.hazmat.primitives.serialization.Encoding`
that will be used to serialize the certificate request.
diff --git a/src/_cffi_src/openssl/ssl.py b/src/_cffi_src/openssl/ssl.py
index fa0aefc..5841ee2 100644
--- a/src/_cffi_src/openssl/ssl.py
+++ b/src/_cffi_src/openssl/ssl.py
@@ -182,6 +182,7 @@
int SSL_pending(const SSL *);
int SSL_write(SSL *, const void *, int);
int SSL_read(SSL *, void *, int);
+int SSL_peek(SSL *, void *, int);
X509 *SSL_get_peer_certificate(const SSL *);
int SSL_get_ex_data_X509_STORE_CTX_idx(void);
diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py
index 73a5863..d649377 100644
--- a/src/cryptography/hazmat/backends/openssl/backend.py
+++ b/src/cryptography/hazmat/backends/openssl/backend.py
@@ -8,6 +8,8 @@
import itertools
from contextlib import contextmanager
+import idna
+
import six
from cryptography import utils, x509
@@ -136,6 +138,56 @@
return pp, r
+def _encode_subject_alt_name(backend, san):
+ general_names = backend._lib.GENERAL_NAMES_new()
+ assert general_names != backend._ffi.NULL
+ general_names = backend._ffi.gc(
+ general_names, backend._lib.GENERAL_NAMES_free
+ )
+
+ for alt_name in san:
+ if isinstance(alt_name, x509.DNSName):
+ gn = backend._lib.GENERAL_NAME_new()
+ assert gn != backend._ffi.NULL
+ gn.type = backend._lib.GEN_DNS
+
+ ia5 = backend._lib.ASN1_IA5STRING_new()
+ assert ia5 != backend._ffi.NULL
+
+ if alt_name.value.startswith(u"*."):
+ value = b"*." + idna.encode(alt_name.value[2:])
+ else:
+ value = idna.encode(alt_name.value)
+
+ res = backend._lib.ASN1_STRING_set(ia5, value, len(value))
+ assert res == 1
+ gn.d.dNSName = ia5
+ elif isinstance(alt_name, x509.RegisteredID):
+ gn = backend._lib.GENERAL_NAME_new()
+ assert gn != backend._ffi.NULL
+ gn.type = backend._lib.GEN_RID
+ obj = backend._lib.OBJ_txt2obj(
+ alt_name.value.dotted_string.encode('ascii'), 1
+ )
+ assert obj != backend._ffi.NULL
+ gn.d.registeredID = obj
+ else:
+ raise NotImplementedError(
+ "Only DNSName and RegisteredID supported right now"
+ )
+
+ res = backend._lib.sk_GENERAL_NAME_push(general_names, gn)
+ assert res != 0
+
+ pp = backend._ffi.new("unsigned char **")
+ r = backend._lib.i2d_GENERAL_NAMES(general_names, pp)
+ assert r > 0
+ pp = backend._ffi.gc(
+ pp, lambda pointer: backend._lib.OPENSSL_free(pointer[0])
+ )
+ return pp, r
+
+
@utils.register_interface(CipherBackend)
@utils.register_interface(CMACBackend)
@utils.register_interface(DERSerializationBackend)
@@ -841,12 +893,14 @@
self._lib.sk_X509_EXTENSION_free,
)
for extension in builder._extensions:
- obj = _txt2obj(self, extension.oid.dotted_string)
if isinstance(extension.value, x509.BasicConstraints):
pp, r = _encode_basic_constraints(self, extension.value)
+ elif isinstance(extension.value, x509.SubjectAlternativeName):
+ pp, r = _encode_subject_alt_name(self, extension.value)
else:
raise NotImplementedError('Extension not yet supported.')
+ obj = _txt2obj(self, extension.oid.dotted_string)
extension = self._lib.X509_EXTENSION_create_by_OBJ(
self._ffi.NULL,
obj,
diff --git a/src/cryptography/hazmat/backends/openssl/x509.py b/src/cryptography/hazmat/backends/openssl/x509.py
index e720bfd..399e6a6 100644
--- a/src/cryptography/hazmat/backends/openssl/x509.py
+++ b/src/cryptography/hazmat/backends/openssl/x509.py
@@ -36,6 +36,14 @@
return backend._bn_to_int(bn)
+def _asn1_string_to_bytes(backend, asn1_string):
+ return backend._ffi.buffer(asn1_string.data, asn1_string.length)[:]
+
+
+def _asn1_string_to_ascii(backend, asn1_string):
+ return _asn1_string_to_bytes(backend, asn1_string).decode("ascii")
+
+
def _asn1_string_to_utf8(backend, asn1_string):
buf = backend._ffi.new("unsigned char **")
res = backend._lib.ASN1_STRING_to_UTF8(buf, asn1_string)
@@ -92,7 +100,7 @@
def _decode_general_name(backend, gn):
if gn.type == backend._lib.GEN_DNS:
- data = backend._ffi.buffer(gn.d.dNSName.data, gn.d.dNSName.length)[:]
+ data = _asn1_string_to_bytes(backend, gn.d.dNSName)
if data.startswith(b"*."):
# This is a wildcard name. We need to remove the leading wildcard,
# IDNA decode, then re-add the wildcard. Wildcard characters should
@@ -109,10 +117,7 @@
return x509.DNSName(decoded)
elif gn.type == backend._lib.GEN_URI:
- data = backend._ffi.buffer(
- gn.d.uniformResourceIdentifier.data,
- gn.d.uniformResourceIdentifier.length
- )[:].decode("ascii")
+ data = _asn1_string_to_ascii(backend, gn.d.uniformResourceIdentifier)
parsed = urllib_parse.urlparse(data)
hostname = idna.decode(parsed.hostname)
if parsed.port:
@@ -138,9 +143,7 @@
elif gn.type == backend._lib.GEN_IPADD:
return x509.IPAddress(
ipaddress.ip_address(
- backend._ffi.buffer(
- gn.d.iPAddress.data, gn.d.iPAddress.length
- )[:]
+ _asn1_string_to_bytes(backend, gn.d.iPAddress)
)
)
elif gn.type == backend._lib.GEN_DIRNAME:
@@ -148,9 +151,7 @@
_decode_x509_name(backend, gn.d.directoryName)
)
elif gn.type == backend._lib.GEN_EMAIL:
- data = backend._ffi.buffer(
- gn.d.rfc822Name.data, gn.d.rfc822Name.length
- )[:].decode("ascii")
+ data = _asn1_string_to_ascii(backend, gn.d.rfc822Name)
name, address = parseaddr(data)
parts = address.split(u"@")
if name or len(parts) > 2 or not address:
@@ -240,15 +241,12 @@
def __ne__(self, other):
return not self == other
+ def __hash__(self):
+ return hash(self.public_bytes(serialization.Encoding.DER))
+
def fingerprint(self, algorithm):
h = hashes.Hash(algorithm, self._backend)
- bio = self._backend._create_mem_bio()
- res = self._backend._lib.i2d_X509_bio(
- bio, self._x509
- )
- assert res == 1
- der = self._backend._read_mem_bio(bio)
- h.update(der)
+ h.update(self.public_bytes(serialization.Encoding.DER))
return h.finalize()
@property
@@ -295,11 +293,10 @@
generalized_time = self._backend._ffi.gc(
generalized_time, self._backend._lib.ASN1_GENERALIZEDTIME_free
)
- time = self._backend._ffi.string(
- self._backend._lib.ASN1_STRING_data(
- self._backend._ffi.cast("ASN1_STRING *", generalized_time)
- )
- ).decode("ascii")
+ time = _asn1_string_to_ascii(
+ self._backend,
+ self._backend._ffi.cast("ASN1_STRING *", generalized_time)
+ )
return datetime.datetime.strptime(time, "%Y%m%d%H%M%SZ")
@property
@@ -716,6 +713,17 @@
self._backend = backend
self._x509_req = x509_req
+ def __eq__(self, other):
+ if not isinstance(other, _CertificateSigningRequest):
+ return NotImplemented
+
+ self_bytes = self.public_bytes(serialization.Encoding.DER)
+ other_bytes = other.public_bytes(serialization.Encoding.DER)
+ return self_bytes == other_bytes
+
+ def __ne__(self, other):
+ return not self == other
+
def public_key(self):
pkey = self._backend._lib.X509_REQ_get_pubkey(self._x509_req)
assert pkey != self._backend._ffi.NULL
diff --git a/src/cryptography/x509.py b/src/cryptography/x509.py
index c9d0c26..d9d6db4 100644
--- a/src/cryptography/x509.py
+++ b/src/cryptography/x509.py
@@ -1358,6 +1358,12 @@
"""
@abc.abstractmethod
+ def __hash__(self):
+ """
+ Computes a hash.
+ """
+
+ @abc.abstractmethod
def public_bytes(self, encoding):
"""
Serializes the certificate to PEM or DER format.
@@ -1426,6 +1432,18 @@
@six.add_metaclass(abc.ABCMeta)
class CertificateSigningRequest(object):
@abc.abstractmethod
+ def __eq__(self, other):
+ """
+ Checks equality.
+ """
+
+ @abc.abstractmethod
+ def __ne__(self, other):
+ """
+ Checks not equal.
+ """
+
+ @abc.abstractmethod
def public_key(self):
"""
Returns the public key
diff --git a/tests/test_x509.py b/tests/test_x509.py
index 1e0c9cd..ccb24d7 100644
--- a/tests/test_x509.py
+++ b/tests/test_x509.py
@@ -347,6 +347,29 @@
assert cert != cert2
assert cert != object()
+ def test_hash(self, backend):
+ cert1 = _load_cert(
+ os.path.join("x509", "custom", "post2000utctime.pem"),
+ x509.load_pem_x509_certificate,
+ backend
+ )
+ cert2 = _load_cert(
+ os.path.join("x509", "custom", "post2000utctime.pem"),
+ x509.load_pem_x509_certificate,
+ backend
+ )
+ cert3 = _load_cert(
+ os.path.join(
+ "x509", "PKITS_data", "certs",
+ "ValidGeneralizedTimenotAfterDateTest8EE.crt"
+ ),
+ x509.load_der_x509_certificate,
+ backend
+ )
+
+ assert hash(cert1) == hash(cert2)
+ assert hash(cert1) != hash(cert3)
+
def test_version_1_cert(self, backend):
cert = _load_cert(
os.path.join("x509", "v1_cert.pem"),
@@ -694,6 +717,35 @@
serialized = request.public_bytes(encoding)
assert serialized == request_bytes
+ def test_eq(self, backend):
+ request1 = _load_cert(
+ os.path.join("x509", "requests", "rsa_sha1.pem"),
+ x509.load_pem_x509_csr,
+ backend
+ )
+ request2 = _load_cert(
+ os.path.join("x509", "requests", "rsa_sha1.pem"),
+ x509.load_pem_x509_csr,
+ backend
+ )
+
+ assert request1 == request2
+
+ def test_ne(self, backend):
+ request1 = _load_cert(
+ os.path.join("x509", "requests", "rsa_sha1.pem"),
+ x509.load_pem_x509_csr,
+ backend
+ )
+ request2 = _load_cert(
+ os.path.join("x509", "requests", "san_rsa_sha1.pem"),
+ x509.load_pem_x509_csr,
+ backend
+ )
+
+ assert request1 != request2
+ assert request1 != object()
+
@pytest.mark.requires_backend_interface(interface=X509Backend)
class TestCertificateSigningRequestBuilder(object):
@@ -911,6 +963,51 @@
])
)
+ def test_subject_alt_names(self, backend):
+ private_key = RSA_KEY_2048.private_key(backend)
+
+ csr = x509.CertificateSigningRequestBuilder().subject_name(
+ x509.Name([
+ x509.NameAttribute(x509.OID_COMMON_NAME, u"SAN"),
+ ])
+ ).add_extension(
+ x509.SubjectAlternativeName([
+ x509.DNSName(u"example.com"),
+ x509.DNSName(u"*.example.com"),
+ x509.RegisteredID(x509.ObjectIdentifier("1.2.3.4.5.6.7")),
+ ]),
+ critical=False,
+ ).sign(private_key, hashes.SHA256(), backend)
+
+ assert len(csr.extensions) == 1
+ ext = csr.extensions.get_extension_for_oid(
+ x509.OID_SUBJECT_ALTERNATIVE_NAME
+ )
+ assert not ext.critical
+ assert ext.oid == x509.OID_SUBJECT_ALTERNATIVE_NAME
+ assert list(ext.value) == [
+ x509.DNSName(u"example.com"),
+ x509.DNSName(u"*.example.com"),
+ x509.RegisteredID(x509.ObjectIdentifier("1.2.3.4.5.6.7")),
+ ]
+
+ def test_subject_alt_name_unsupported_general_name(self, backend):
+ private_key = RSA_KEY_2048.private_key(backend)
+
+ builder = x509.CertificateSigningRequestBuilder().subject_name(
+ x509.Name([
+ x509.NameAttribute(x509.OID_COMMON_NAME, u"SAN"),
+ ])
+ ).add_extension(
+ x509.SubjectAlternativeName([
+ x509.RFC822Name(u"test@example.com"),
+ ]),
+ critical=False,
+ )
+
+ with pytest.raises(NotImplementedError):
+ builder.sign(private_key, hashes.SHA256(), backend)
+
@pytest.mark.requires_backend_interface(interface=DSABackend)
@pytest.mark.requires_backend_interface(interface=X509Backend)