| Server IP : 172.67.137.82 / Your IP : 104.23.197.222 Web Server : Apache/2.4.51 (Unix) OpenSSL/1.1.1n System : Linux ip-172-26-8-243 4.19.0-27-cloud-amd64 #1 SMP Debian 4.19.316-1 (2024-06-25) x86_64 User : daemon ( 1) PHP Version : 7.4.24 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /usr/lib/python3/dist-packages/acme/ |
Upload File : |
"""Crypto utilities."""
import binascii
import contextlib
import logging
import os
import re
import socket
from OpenSSL import crypto
from OpenSSL import SSL # type: ignore # https://github.com/python/typeshed/issues/2052
import josepy as jose
from acme import errors
# pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Callable, Union, Tuple, Optional
# pylint: enable=unused-import, no-name-in-module
logger = logging.getLogger(__name__)
# TLSSNI01 certificate serving and probing is not affected by SSL
# vulnerabilities: prober needs to check certificate for expected
# contents anyway. Working SNI is the only thing that's necessary for
# the challenge and thus scoping down SSL/TLS method (version) would
# cause interoperability issues: TLSv1_METHOD is only compatible with
# TLSv1_METHOD, while SSLv23_METHOD is compatible with all other
# methods, including TLSv2_METHOD (read more at
# https://www.openssl.org/docs/ssl/SSLv23_method.html). _serve_sni
# should be changed to use "set_options" to disable SSLv2 and SSLv3,
# in case it's used for things other than probing/serving!
_DEFAULT_TLSSNI01_SSL_METHOD = SSL.SSLv23_METHOD # type: ignore
class SSLSocket(object): # pylint: disable=too-few-public-methods
"""SSL wrapper for sockets.
:ivar socket sock: Original wrapped socket.
:ivar dict certs: Mapping from domain names (`bytes`) to
`OpenSSL.crypto.X509`.
:ivar method: See `OpenSSL.SSL.Context` for allowed values.
"""
def __init__(self, sock, certs, method=_DEFAULT_TLSSNI01_SSL_METHOD):
self.sock = sock
self.certs = certs
self.method = method
def __getattr__(self, name):
return getattr(self.sock, name)
def _pick_certificate_cb(self, connection):
"""SNI certificate callback.
This method will set a new OpenSSL context object for this
connection when an incoming connection provides an SNI name
(in order to serve the appropriate certificate, if any).
:param connection: The TLS connection object on which the SNI
extension was received.
:type connection: :class:`OpenSSL.Connection`
"""
server_name = connection.get_servername()
try:
key, cert = self.certs[server_name]
except KeyError:
logger.debug("Server name (%s) not recognized, dropping SSL",
server_name)
return
new_context = SSL.Context(self.method)
new_context.set_options(SSL.OP_NO_SSLv2)
new_context.set_options(SSL.OP_NO_SSLv3)
new_context.use_privatekey(key)
new_context.use_certificate(cert)
connection.set_context(new_context)
class FakeConnection(object):
"""Fake OpenSSL.SSL.Connection."""
# pylint: disable=too-few-public-methods,missing-docstring
def __init__(self, connection):
self._wrapped = connection
def __getattr__(self, name):
return getattr(self._wrapped, name)
def shutdown(self, *unused_args):
# OpenSSL.SSL.Connection.shutdown doesn't accept any args
return self._wrapped.shutdown()
def accept(self): # pylint: disable=missing-docstring
sock, addr = self.sock.accept()
context = SSL.Context(self.method)
context.set_options(SSL.OP_NO_SSLv2)
context.set_options(SSL.OP_NO_SSLv3)
context.set_tlsext_servername_callback(self._pick_certificate_cb)
ssl_sock = self.FakeConnection(SSL.Connection(context, sock))
ssl_sock.set_accept_state()
logger.debug("Performing handshake with %s", addr)
try:
ssl_sock.do_handshake()
except SSL.Error as error:
# _pick_certificate_cb might have returned without
# creating SSL context (wrong server name)
raise socket.error(error)
return ssl_sock, addr
def probe_sni(name, host, port=443, timeout=300,
method=_DEFAULT_TLSSNI01_SSL_METHOD, source_address=('', 0)):
"""Probe SNI server for SSL certificate.
:param bytes name: Byte string to send as the server name in the
client hello message.
:param bytes host: Host to connect to.
:param int port: Port to connect to.
:param int timeout: Timeout in seconds.
:param method: See `OpenSSL.SSL.Context` for allowed values.
:param tuple source_address: Enables multi-path probing (selection
of source interface). See `socket.creation_connection` for more
info. Available only in Python 2.7+.
:raises acme.errors.Error: In case of any problems.
:returns: SSL certificate presented by the server.
:rtype: OpenSSL.crypto.X509
"""
context = SSL.Context(method)
context.set_timeout(timeout)
socket_kwargs = {'source_address': source_address}
try:
# pylint: disable=star-args
logger.debug(
"Attempting to connect to %s:%d%s.", host, port,
" from {0}:{1}".format(
source_address[0],
source_address[1]
) if socket_kwargs else ""
)
socket_tuple = (host, port) # type: Tuple[str, int]
sock = socket.create_connection(socket_tuple, **socket_kwargs) # type: ignore
except socket.error as error:
raise errors.Error(error)
with contextlib.closing(sock) as client:
client_ssl = SSL.Connection(context, client)
client_ssl.set_connect_state()
client_ssl.set_tlsext_host_name(name) # pyOpenSSL>=0.13
try:
client_ssl.do_handshake()
client_ssl.shutdown()
except SSL.Error as error:
raise errors.Error(error)
return client_ssl.get_peer_certificate()
def make_csr(private_key_pem, domains, must_staple=False):
"""Generate a CSR containing a list of domains as subjectAltNames.
:param buffer private_key_pem: Private key, in PEM PKCS#8 format.
:param list domains: List of DNS names to include in subjectAltNames of CSR.
:param bool must_staple: Whether to include the TLS Feature extension (aka
OCSP Must Staple: https://tools.ietf.org/html/rfc7633).
:returns: buffer PEM-encoded Certificate Signing Request.
"""
private_key = crypto.load_privatekey(
crypto.FILETYPE_PEM, private_key_pem)
csr = crypto.X509Req()
extensions = [
crypto.X509Extension(
b'subjectAltName',
critical=False,
value=', '.join('DNS:' + d for d in domains).encode('ascii')
),
]
if must_staple:
extensions.append(crypto.X509Extension(
b"1.3.6.1.5.5.7.1.24",
critical=False,
value=b"DER:30:03:02:01:05"))
csr.add_extensions(extensions)
csr.set_pubkey(private_key)
csr.set_version(2)
csr.sign(private_key, 'sha256')
return crypto.dump_certificate_request(
crypto.FILETYPE_PEM, csr)
def _pyopenssl_cert_or_req_all_names(loaded_cert_or_req):
common_name = loaded_cert_or_req.get_subject().CN
sans = _pyopenssl_cert_or_req_san(loaded_cert_or_req)
if common_name is None:
return sans
else:
return [common_name] + [d for d in sans if d != common_name]
def _pyopenssl_cert_or_req_san(cert_or_req):
"""Get Subject Alternative Names from certificate or CSR using pyOpenSSL.
.. todo:: Implement directly in PyOpenSSL!
.. note:: Although this is `acme` internal API, it is used by
`letsencrypt`.
:param cert_or_req: Certificate or CSR.
:type cert_or_req: `OpenSSL.crypto.X509` or `OpenSSL.crypto.X509Req`.
:returns: A list of Subject Alternative Names.
:rtype: `list` of `unicode`
"""
# This function finds SANs by dumping the certificate/CSR to text and
# searching for "X509v3 Subject Alternative Name" in the text. This method
# is used to support PyOpenSSL version 0.13 where the
# `_subjectAltNameString` and `get_extensions` methods are not available
# for CSRs.
# constants based on PyOpenSSL certificate/CSR text dump
part_separator = ":"
parts_separator = ", "
prefix = "DNS" + part_separator
if isinstance(cert_or_req, crypto.X509):
# pylint: disable=line-too-long
func = crypto.dump_certificate # type: Union[Callable[[int, crypto.X509Req], bytes], Callable[[int, crypto.X509], bytes]]
else:
func = crypto.dump_certificate_request
text = func(crypto.FILETYPE_TEXT, cert_or_req).decode("utf-8")
# WARNING: this function does not support multiple SANs extensions.
# Multiple X509v3 extensions of the same type is disallowed by RFC 5280.
match = re.search(r"X509v3 Subject Alternative Name:(?: critical)?\s*(.*)", text)
# WARNING: this function assumes that no SAN can include
# parts_separator, hence the split!
sans_parts = [] if match is None else match.group(1).split(parts_separator)
return [part.split(part_separator)[1]
for part in sans_parts if part.startswith(prefix)]
def gen_ss_cert(key, domains, not_before=None,
validity=(7 * 24 * 60 * 60), force_san=True):
"""Generate new self-signed certificate.
:type domains: `list` of `unicode`
:param OpenSSL.crypto.PKey key:
:param bool force_san:
If more than one domain is provided, all of the domains are put into
``subjectAltName`` X.509 extension and first domain is set as the
subject CN. If only one domain is provided no ``subjectAltName``
extension is used, unless `force_san` is ``True``.
"""
assert domains, "Must provide one or more hostnames for the cert."
cert = crypto.X509()
cert.set_serial_number(int(binascii.hexlify(os.urandom(16)), 16))
cert.set_version(2)
extensions = [
crypto.X509Extension(
b"basicConstraints", True, b"CA:TRUE, pathlen:0"),
]
cert.get_subject().CN = domains[0]
# TODO: what to put into cert.get_subject()?
cert.set_issuer(cert.get_subject())
if force_san or len(domains) > 1:
extensions.append(crypto.X509Extension(
b"subjectAltName",
critical=False,
value=b", ".join(b"DNS:" + d.encode() for d in domains)
))
cert.add_extensions(extensions)
cert.gmtime_adj_notBefore(0 if not_before is None else not_before)
cert.gmtime_adj_notAfter(validity)
cert.set_pubkey(key)
cert.sign(key, "sha256")
return cert
def dump_pyopenssl_chain(chain, filetype=crypto.FILETYPE_PEM):
"""Dump certificate chain into a bundle.
:param list chain: List of `OpenSSL.crypto.X509` (or wrapped in
:class:`josepy.util.ComparableX509`).
:returns: certificate chain bundle
:rtype: bytes
"""
# XXX: returns empty string when no chain is available, which
# shuts up RenewableCert, but might not be the best solution...
def _dump_cert(cert):
if isinstance(cert, jose.ComparableX509):
# pylint: disable=protected-access
cert = cert.wrapped
return crypto.dump_certificate(filetype, cert)
# assumes that OpenSSL.crypto.dump_certificate includes ending
# newline character
return b"".join(_dump_cert(cert) for cert in chain)
| .. | |||||
| __pycache__ | |||||
| __init__.py | |||||
| challenges.py | |||||
| challenges_test.py | |||||
| client.py | |||||
| client_test.py | |||||
| crypto_util.py | |||||
| crypto_util_test.py | |||||
| errors.py | |||||
| errors_test.py | |||||
| fields.py | |||||
| fields_test.py | |||||
| jose_test.py | |||||
| jws.py | |||||
| jws_test.py | |||||
| magic_typing.py | |||||
| magic_typing_test.py | |||||
| messages.py | |||||
| messages_test.py | |||||
| standalone.py | |||||
| standalone_test.py | |||||
| test_util.py | |||||
| util.py | |||||
| util_test.py |