mirror of
https://github.com/certbot/certbot.git
synced 2026-07-30 16:14:44 +02:00
Allow using cryptography certs and keys in the standalone plugin (#10133)
Co-authored-by: Brad Warren <bmw@users.noreply.github.com>
This commit is contained in:
co-authored by
Brad Warren
parent
ed972a130f
commit
e050fe91a3
@@ -11,6 +11,8 @@ from unittest import mock
|
||||
import josepy as jose
|
||||
import pytest
|
||||
import requests
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
from acme import challenges
|
||||
from acme import crypto_util
|
||||
@@ -116,13 +118,13 @@ class TLSALPN01ServerTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.certs = {b'localhost': (
|
||||
test_util.load_pyopenssl_private_key('rsa2048_key.pem'),
|
||||
test_util.load_cert('rsa2048_cert.pem'),
|
||||
serialization.load_pem_private_key(test_util.load_vector('rsa2048_key.pem'), password=None),
|
||||
x509.load_pem_x509_certificate(test_util.load_vector('rsa2048_cert.pem')),
|
||||
)}
|
||||
# Use different certificate for challenge.
|
||||
self.challenge_certs = {b'localhost': (
|
||||
test_util.load_pyopenssl_private_key('rsa4096_key.pem'),
|
||||
test_util.load_cert('rsa4096_cert.pem'),
|
||||
serialization.load_pem_private_key(test_util.load_vector('rsa4096_key.pem'), password=None),
|
||||
x509.load_pem_x509_certificate(test_util.load_vector('rsa4096_cert.pem')),
|
||||
)}
|
||||
from acme.standalone import TLSALPN01Server
|
||||
self.server = TLSALPN01Server(("localhost", 0), certs=self.certs,
|
||||
@@ -151,8 +153,7 @@ class TLSALPN01ServerTest(unittest.TestCase):
|
||||
b'localhost', host=host, port=port, timeout=1,
|
||||
alpn_protocols=[b"acme-tls/1"])
|
||||
# Expect challenge cert when connecting with ALPN.
|
||||
assert jose.ComparableX509(cert) == \
|
||||
jose.ComparableX509(self.challenge_certs[b'localhost'][1])
|
||||
assert cert.to_cryptography() == self.challenge_certs[b'localhost'][1]
|
||||
|
||||
def test_bad_alpn(self):
|
||||
host, port = self.server.socket.getsockname()[:2]
|
||||
|
||||
+23
-13
@@ -20,7 +20,7 @@ from typing import Union
|
||||
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import dsa, rsa, ec, ed25519, ed448
|
||||
from cryptography.hazmat.primitives.asymmetric import dsa, rsa, ec, ed25519, ed448, types
|
||||
import josepy as jose
|
||||
from OpenSSL import crypto
|
||||
from OpenSSL import SSL
|
||||
@@ -58,17 +58,22 @@ class Format(enum.IntEnum):
|
||||
return serialization.Encoding.PEM
|
||||
|
||||
|
||||
_KeyAndCert = Union[
|
||||
Tuple[crypto.PKey, crypto.X509],
|
||||
Tuple[types.CertificateIssuerPrivateKeyTypes, x509.Certificate],
|
||||
]
|
||||
|
||||
|
||||
class _DefaultCertSelection:
|
||||
def __init__(self, certs: Mapping[bytes, Tuple[crypto.PKey, crypto.X509]]):
|
||||
def __init__(self, certs: Mapping[bytes, _KeyAndCert]):
|
||||
self.certs = certs
|
||||
|
||||
def __call__(self, connection: SSL.Connection) -> Optional[Tuple[crypto.PKey, crypto.X509]]:
|
||||
def __call__(self, connection: SSL.Connection) -> Optional[_KeyAndCert]:
|
||||
server_name = connection.get_servername()
|
||||
if server_name:
|
||||
return self.certs.get(server_name, None)
|
||||
return None # pragma: no cover
|
||||
|
||||
|
||||
class SSLSocket: # pylint: disable=too-few-public-methods
|
||||
"""SSL wrapper for sockets.
|
||||
|
||||
@@ -82,14 +87,19 @@ class SSLSocket: # pylint: disable=too-few-public-methods
|
||||
`certs` parameter would be ignored, and therefore must be empty.
|
||||
|
||||
"""
|
||||
def __init__(self, sock: socket.socket,
|
||||
certs: Optional[Mapping[bytes, Tuple[crypto.PKey, crypto.X509]]] = None,
|
||||
method: int = _DEFAULT_SSL_METHOD,
|
||||
alpn_selection: Optional[Callable[[SSL.Connection, List[bytes]], bytes]] = None,
|
||||
cert_selection: Optional[Callable[[SSL.Connection],
|
||||
Optional[Tuple[crypto.PKey,
|
||||
crypto.X509]]]] = None
|
||||
) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
sock: socket.socket,
|
||||
certs: Optional[Mapping[bytes, _KeyAndCert]] = None,
|
||||
method: int = _DEFAULT_SSL_METHOD,
|
||||
alpn_selection: Optional[Callable[[SSL.Connection, List[bytes]], bytes]] = None,
|
||||
cert_selection: Optional[
|
||||
Callable[
|
||||
[SSL.Connection],
|
||||
Optional[_KeyAndCert],
|
||||
]
|
||||
] = None,
|
||||
) -> None:
|
||||
self.sock = sock
|
||||
self.alpn_selection = alpn_selection
|
||||
self.method = method
|
||||
@@ -231,7 +241,7 @@ def probe_sni(name: bytes, host: bytes, port: int = 443, timeout: int = 300, #
|
||||
client_ssl.set_connect_state()
|
||||
client_ssl.set_tlsext_host_name(name) # pyOpenSSL>=0.13
|
||||
if alpn_protocols is not None:
|
||||
client_ssl.set_alpn_protos(alpn_protocols)
|
||||
client_ssl.set_alpn_protos(list(alpn_protocols))
|
||||
try:
|
||||
client_ssl.do_handshake()
|
||||
client_ssl.shutdown()
|
||||
|
||||
@@ -16,7 +16,6 @@ from typing import Set
|
||||
from typing import Tuple
|
||||
from typing import Type
|
||||
|
||||
from OpenSSL import crypto
|
||||
from OpenSSL import SSL
|
||||
|
||||
from acme import challenges
|
||||
@@ -46,7 +45,7 @@ class TLSServer(socketserver.TCPServer):
|
||||
method=self.method))
|
||||
|
||||
def _cert_selection(self, connection: SSL.Connection
|
||||
) -> Optional[Tuple[crypto.PKey, crypto.X509]]: # pragma: no cover
|
||||
) -> Optional[crypto_util._KeyAndCert]: # pragma: no cover
|
||||
"""Callback selecting certificate for connection."""
|
||||
server_name = connection.get_servername()
|
||||
if server_name:
|
||||
@@ -152,8 +151,8 @@ class TLSALPN01Server(TLSServer, ACMEServerMixin):
|
||||
ACME_TLS_1_PROTOCOL = b"acme-tls/1"
|
||||
|
||||
def __init__(self, server_address: Tuple[str, int],
|
||||
certs: List[Tuple[crypto.PKey, crypto.X509]],
|
||||
challenge_certs: Mapping[bytes, Tuple[crypto.PKey, crypto.X509]],
|
||||
certs: List[crypto_util._KeyAndCert],
|
||||
challenge_certs: Mapping[bytes, crypto_util._KeyAndCert],
|
||||
ipv6: bool = False) -> None:
|
||||
# We don't need to implement a request handler here because the work
|
||||
# (including logging) is being done by wrapped socket set up in the
|
||||
@@ -163,8 +162,7 @@ class TLSALPN01Server(TLSServer, ACMEServerMixin):
|
||||
ipv6=ipv6)
|
||||
self.challenge_certs = challenge_certs
|
||||
|
||||
def _cert_selection(self, connection: SSL.Connection) -> Optional[Tuple[crypto.PKey,
|
||||
crypto.X509]]:
|
||||
def _cert_selection(self, connection: SSL.Connection) -> Optional[crypto_util._KeyAndCert]:
|
||||
# TODO: We would like to serve challenge cert only if asked for it via
|
||||
# ALPN. To do this, we need to retrieve the list of protos from client
|
||||
# hello, but this is currently impossible with openssl [0], and ALPN
|
||||
|
||||
Reference in New Issue
Block a user