mirror of
https://github.com/certbot/certbot.git
synced 2026-08-03 03:41:53 +02:00
Expose wrapped, not dump
This commit is contained in:
@@ -422,7 +422,8 @@ class ProofOfPossessionHintsTest(unittest.TestCase):
|
||||
self.jmsg_to = {
|
||||
'jwk': jwk,
|
||||
'certFingerprints': cert_fingerprints,
|
||||
'certs': (jose.encode_b64jose(CERT.dump()),),
|
||||
'certs': (jose.encode_b64jose(OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_ASN1, CERT.wrapped)),),
|
||||
'subjectKeyIdentifiers': subject_key_identifiers,
|
||||
'serialNumbers': serial_numbers,
|
||||
'issuers': issuers,
|
||||
|
||||
@@ -7,7 +7,6 @@ import sys
|
||||
import OpenSSL
|
||||
|
||||
from acme import errors
|
||||
from acme import jose
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -167,7 +166,12 @@ def _pyopenssl_cert_or_req_san(cert_or_req):
|
||||
prefix = label + part_separator
|
||||
title = "X509v3 Subject Alternative Name:"
|
||||
|
||||
text = jose.ComparableX509(cert_or_req).dump(OpenSSL.crypto.FILETYPE_TEXT)
|
||||
if isinstance(cert_or_req, OpenSSL.crypto.X509):
|
||||
func = OpenSSL.crypto.dump_certificate
|
||||
else:
|
||||
func = OpenSSL.crypto.dump_certificate_request
|
||||
text = func(OpenSSL.crypto.FILETYPE_TEXT, cert_or_req)
|
||||
|
||||
lines = iter(text.decode("utf-8").splitlines())
|
||||
sans = [next(lines).split(parts_separator)
|
||||
for line in lines if title in line]
|
||||
|
||||
@@ -18,7 +18,7 @@ class SSLSocketAndProbeSNITest(unittest.TestCase):
|
||||
self.cert = test_util.load_comparable_cert('cert.pem')
|
||||
key = test_util.load_pyopenssl_private_key('rsa512_key.pem')
|
||||
# pylint: disable=protected-access
|
||||
certs = {b'foo': (key, self.cert._wrapped)}
|
||||
certs = {b'foo': (key, self.cert.wrapped)}
|
||||
|
||||
from acme.crypto_util import SSLSocket
|
||||
|
||||
|
||||
@@ -372,7 +372,8 @@ def encode_cert(cert):
|
||||
:rtype: unicode
|
||||
|
||||
"""
|
||||
return encode_b64jose(cert.dump())
|
||||
return encode_b64jose(OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_ASN1, cert.wrapped))
|
||||
|
||||
|
||||
def decode_cert(b64der):
|
||||
@@ -396,7 +397,8 @@ def encode_csr(csr):
|
||||
:rtype: unicode
|
||||
|
||||
"""
|
||||
return encode_b64jose(csr.dump())
|
||||
return encode_b64jose(OpenSSL.crypto.dump_certificate_request(
|
||||
OpenSSL.crypto.FILETYPE_ASN1, csr.wrapped))
|
||||
|
||||
|
||||
def decode_csr(b64der):
|
||||
|
||||
@@ -123,7 +123,8 @@ class Header(json_util.JSONObjectWithFields):
|
||||
|
||||
@x5c.encoder
|
||||
def x5c(value): # pylint: disable=missing-docstring,no-self-argument
|
||||
return [base64.b64encode(cert.dump()) for cert in value]
|
||||
return [base64.b64encode(OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_ASN1, cert.wrapped)) for cert in value]
|
||||
|
||||
@x5c.decoder
|
||||
def x5c(value): # pylint: disable=missing-docstring,no-self-argument
|
||||
|
||||
@@ -3,6 +3,7 @@ import base64
|
||||
import unittest
|
||||
|
||||
import mock
|
||||
import OpenSSL
|
||||
|
||||
from acme import test_util
|
||||
|
||||
@@ -67,10 +68,12 @@ class HeaderTest(unittest.TestCase):
|
||||
from acme.jose.jws import Header
|
||||
header = Header(x5c=(CERT, CERT))
|
||||
jobj = header.to_partial_json()
|
||||
cert_b64 = base64.b64encode(CERT.dump())
|
||||
cert_asn1 = OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_ASN1, CERT.wrapped)
|
||||
cert_b64 = base64.b64encode(cert_asn1)
|
||||
self.assertEqual(jobj, {'x5c': [cert_b64, cert_b64]})
|
||||
self.assertEqual(header, Header.from_json(jobj))
|
||||
jobj['x5c'][0] = base64.b64encode(b'xxx' + CERT.dump())
|
||||
jobj['x5c'][0] = base64.b64encode(b'xxx' + cert_asn1)
|
||||
self.assertRaises(errors.DeserializationError, Header.from_json, jobj)
|
||||
|
||||
def test_find_key(self):
|
||||
|
||||
+14
-14
@@ -29,50 +29,50 @@ class abstractclassmethod(classmethod):
|
||||
class ComparableX509(object): # pylint: disable=too-few-public-methods
|
||||
"""Wrapper for OpenSSL.crypto.X509** objects that supports __eq__.
|
||||
|
||||
Wraps around:
|
||||
|
||||
- :class:`OpenSSL.crypto.X509`
|
||||
- :class:`OpenSSL.crypto.X509Req`
|
||||
:ivar wrapped: Wrapped certificate or certificate request.
|
||||
:type wrapped: `OpenSSL.crypto.X509` or `OpenSSL.crypto.X509Req`.
|
||||
|
||||
"""
|
||||
def __init__(self, wrapped):
|
||||
assert isinstance(wrapped, OpenSSL.crypto.X509) or isinstance(
|
||||
wrapped, OpenSSL.crypto.X509Req)
|
||||
self._wrapped = wrapped
|
||||
self.wrapped = wrapped
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(self._wrapped, name)
|
||||
return getattr(self.wrapped, name)
|
||||
|
||||
def dump(self, filetype=OpenSSL.crypto.FILETYPE_ASN1):
|
||||
def _dump(self, filetype=OpenSSL.crypto.FILETYPE_ASN1):
|
||||
"""Dumps the object into a buffer with the specified encoding.
|
||||
|
||||
:param int filetype: The desired encoding. Should be one of
|
||||
OpenSSL.crypto.FILETYPE_ASN1, OpenSSL.crypto.FILETYPE_PEM,
|
||||
or OpenSSL.crypto.FILETYPE_TEXT.
|
||||
`OpenSSL.crypto.FILETYPE_ASN1`,
|
||||
`OpenSSL.crypto.FILETYPE_PEM`, or
|
||||
`OpenSSL.crypto.FILETYPE_TEXT`.
|
||||
|
||||
:returns: Encoded X509 object.
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
if isinstance(self._wrapped, OpenSSL.crypto.X509):
|
||||
if isinstance(self.wrapped, OpenSSL.crypto.X509):
|
||||
func = OpenSSL.crypto.dump_certificate
|
||||
else: # assert in __init__ makes sure this is X509Req
|
||||
func = OpenSSL.crypto.dump_certificate_request
|
||||
return func(filetype, self._wrapped)
|
||||
return func(filetype, self.wrapped)
|
||||
|
||||
def __eq__(self, other):
|
||||
if not isinstance(other, self.__class__):
|
||||
return NotImplemented
|
||||
return self.dump() == other.dump()
|
||||
# pylint: disable=protected-access
|
||||
return self._dump() == other._dump()
|
||||
|
||||
def __hash__(self):
|
||||
return hash((self.__class__, self.dump()))
|
||||
return hash((self.__class__, self._dump()))
|
||||
|
||||
def __ne__(self, other):
|
||||
return not self == other
|
||||
|
||||
def __repr__(self):
|
||||
return '<{0}({1!r})>'.format(self.__class__.__name__, self._wrapped)
|
||||
return '<{0}({1!r})>'.format(self.__class__.__name__, self.wrapped)
|
||||
|
||||
|
||||
class ComparableKey(object): # pylint: disable=too-few-public-methods
|
||||
|
||||
+3
-3
@@ -391,9 +391,9 @@ def _auth_from_domains(le_client, config, domains):
|
||||
new_certr, new_chain, new_key, _ = le_client.obtain_certificate(domains)
|
||||
# TODO: Check whether it worked! <- or make sure errors are thrown (jdk)
|
||||
lineage.save_successor(
|
||||
lineage.latest_common_version(),
|
||||
new_certr.body.dump(OpenSSL.crypto.FILETYPE_PEM), new_key.pem,
|
||||
crypto_util.dump_pyopenssl_chain(new_chain))
|
||||
lineage.latest_common_version(), OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM, new_certr.body.wrapped),
|
||||
new_key.pem, crypto_util.dump_pyopenssl_chain(new_chain))
|
||||
|
||||
lineage.update_all_links_to(lineage.latest_common_version())
|
||||
# TODO: Check return value of save_successor
|
||||
|
||||
@@ -299,7 +299,8 @@ class Client(object):
|
||||
"by your operating system package manager")
|
||||
|
||||
lineage = storage.RenewableCert.new_lineage(
|
||||
domains[0], certr.body.dump(OpenSSL.crypto.FILETYPE_PEM),
|
||||
domains[0], OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM, certr.body.wrapped),
|
||||
key.pem, crypto_util.dump_pyopenssl_chain(chain),
|
||||
params, config, cli_config)
|
||||
return lineage
|
||||
@@ -328,7 +329,8 @@ class Client(object):
|
||||
os.path.dirname(path), 0o755, os.geteuid(),
|
||||
self.config.strict_permissions)
|
||||
|
||||
cert_pem = certr.body.dump(OpenSSL.crypto.FILETYPE_PEM)
|
||||
cert_pem = OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM, certr.body.wrapped)
|
||||
cert_file, act_cert_path = le_util.unique_file(cert_path, 0o644)
|
||||
try:
|
||||
cert_file.write(cert_pem)
|
||||
|
||||
@@ -271,7 +271,7 @@ def dump_pyopenssl_chain(chain, filetype=OpenSSL.crypto.FILETYPE_PEM):
|
||||
def _dump_cert(cert):
|
||||
if isinstance(cert, jose.ComparableX509):
|
||||
# pylint: disable=protected-access
|
||||
cert = cert._wrapped
|
||||
cert = cert.wrapped
|
||||
return OpenSSL.crypto.dump_certificate(filetype, cert)
|
||||
|
||||
# assumes that OpenSSL.crypto.dump_certificate includes ending
|
||||
|
||||
@@ -102,7 +102,8 @@ def renew(cert, old_version):
|
||||
# new_key if the old key is to be used (since save_successor
|
||||
# already understands this distinction!)
|
||||
return cert.save_successor(
|
||||
old_version, new_certr.body.dump(OpenSSL.crypto.FILETYPE_PEM),
|
||||
old_version, OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM, new_certr.body.wrapped),
|
||||
new_key.pem, crypto_util.dump_pyopenssl_chain(new_chain))
|
||||
# TODO: Notify results
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user