starred/certbot-certbot-1
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-08-03 20:12:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/certbot/certbot

Browse Source
This commit is contained in:
Filip Lajszczak
2020-02-06 15:14:17 +00:00
parent b27e5804b9 7da5196206
commit 2b051dd197
980 changed files with 20261 additions and 16621 deletions
Download Patch File Download Diff File Expand all files Collapse all files
acme/MANIFEST.in
+3 -1
View File
@@ -3,4 +3,6 @@ include README.rst
include pytest.ini
recursive-include docs *
recursive-include examples *
recursive-include acme/testdata *
recursive-include tests *
global-exclude __pycache__
global-exclude *.py[cod]
acme/acme/__init__.py
+1 -1
View File
@@ -6,13 +6,13 @@ This module is an implementation of the `ACME protocol`_.
"""
import sys
import warnings
# This code exists to keep backwards compatibility with people using acme.jose
# before it became the standalone josepy package.
#
# It is based on
# https://github.com/requests/requests/blob/1278ecdf71a312dc2268f3bfc0aabfab3c006dcf/requests/packages.py
import josepy as jose
for mod in list(sys.modules):
acme/acme/challenges.py
+16 -169
View File
@@ -3,27 +3,19 @@ import abc
import functools
import hashlib
import logging
import socket
import warnings
from cryptography.hazmat.primitives import hashes # type: ignore
import josepy as jose
import OpenSSL
import requests
import six
from acme import errors
from acme import crypto_util
from acme import fields
logger = logging.getLogger(__name__)
# pylint: disable=too-few-public-methods
class Challenge(jose.TypedJSONObjectWithFields):
# _fields_to_partial_json | pylint: disable=abstract-method
# _fields_to_partial_json
"""ACME challenge."""
TYPES = {} # type: dict
@@ -37,7 +29,7 @@ class Challenge(jose.TypedJSONObjectWithFields):
class ChallengeResponse(jose.TypedJSONObjectWithFields):
# _fields_to_partial_json | pylint: disable=abstract-method
# _fields_to_partial_json
"""ACME challenge response."""
TYPES = {} # type: dict
resource_type = 'challenge'
@@ -62,8 +54,7 @@ class UnrecognizedChallenge(Challenge):
object.__setattr__(self, "jobj", jobj)
def to_partial_json(self):
# pylint: disable=no-member
return self.jobj
return self.jobj # pylint: disable=no-member
@classmethod
def from_json(cls, jobj):
@@ -96,6 +87,7 @@ class _TokenChallenge(Challenge):
"""
# TODO: check that path combined with uri does not go above
# URI_ROOT_PATH!
# pylint: disable=unsupported-membership-test
return b'..' not in self.token and b'/' not in self.token
@@ -120,7 +112,7 @@ class KeyAuthorizationChallengeResponse(ChallengeResponse):
:rtype: bool
"""
parts = self.key_authorization.split('.') # pylint: disable=no-member
parts = self.key_authorization.split('.')
if len(parts) != 2:
logger.debug("Key authorization (%r) is not well formed",
self.key_authorization)
@@ -140,10 +132,14 @@ class KeyAuthorizationChallengeResponse(ChallengeResponse):
return True
def to_partial_json(self):
jobj = super(KeyAuthorizationChallengeResponse, self).to_partial_json()
jobj.pop('keyAuthorization', None)
return jobj
@six.add_metaclass(abc.ABCMeta)
class KeyAuthorizationChallenge(_TokenChallenge):
# pylint: disable=abstract-class-little-used,too-many-ancestors
"""Challenge based on Key Authorization.
:param response_cls: Subclass of `KeyAuthorizationChallengeResponse`
@@ -175,7 +171,7 @@ class KeyAuthorizationChallenge(_TokenChallenge):
:rtype: KeyAuthorizationChallengeResponse
"""
return self.response_cls(
return self.response_cls( # pylint: disable=not-callable
key_authorization=self.key_authorization(account_key))
@abc.abstractmethod
@@ -212,7 +208,7 @@ class DNS01Response(KeyAuthorizationChallengeResponse):
"""ACME dns-01 challenge response."""
typ = "dns-01"
def simple_verify(self, chall, domain, account_public_key):
def simple_verify(self, chall, domain, account_public_key): # pylint: disable=unused-argument
"""Simple verify.
This method no longer checks DNS records and is a simple wrapper
@@ -228,14 +224,13 @@ class DNS01Response(KeyAuthorizationChallengeResponse):
:rtype: bool
"""
# pylint: disable=unused-argument
verified = self.verify(chall, account_public_key)
if not verified:
logger.debug("Verification of key authorization in response failed")
return verified
@Challenge.register # pylint: disable=too-many-ancestors
@Challenge.register
class DNS01(KeyAuthorizationChallenge):
"""ACME dns-01 challenge."""
response_cls = DNS01Response
@@ -325,7 +320,7 @@ class HTTP01Response(KeyAuthorizationChallengeResponse):
return True
@Challenge.register # pylint: disable=too-many-ancestors
@Challenge.register
class HTTP01(KeyAuthorizationChallenge):
"""ACME http-01 challenge."""
response_cls = HTTP01Response
@@ -365,154 +360,6 @@ class HTTP01(KeyAuthorizationChallenge):
return self.key_authorization(account_key)
@ChallengeResponse.register
class TLSSNI01Response(KeyAuthorizationChallengeResponse):
"""ACME tls-sni-01 challenge response."""
typ = "tls-sni-01"
DOMAIN_SUFFIX = b".acme.invalid"
"""Domain name suffix."""
PORT = 443
"""Verification port as defined by the protocol.
You can override it (e.g. for testing) by passing ``port`` to
`simple_verify`.
"""
@property
def z(self): # pylint: disable=invalid-name
"""``z`` value used for verification.
:rtype bytes:
"""
return hashlib.sha256(
self.key_authorization.encode("utf-8")).hexdigest().lower().encode()
@property
def z_domain(self):
"""Domain name used for verification, generated from `z`.
:rtype bytes:
"""
return self.z[:32] + b'.' + self.z[32:] + self.DOMAIN_SUFFIX
def gen_cert(self, key=None, bits=2048):
"""Generate tls-sni-01 certificate.
:param OpenSSL.crypto.PKey key: Optional private key used in
certificate generation. If not provided (``None``), then
fresh key will be generated.
:param int bits: Number of bits for newly generated key.
:rtype: `tuple` of `OpenSSL.crypto.X509` and `OpenSSL.crypto.PKey`
"""
if key is None:
key = OpenSSL.crypto.PKey()
key.generate_key(OpenSSL.crypto.TYPE_RSA, bits)
return crypto_util.gen_ss_cert(key, [
# z_domain is too big to fit into CN, hence first dummy domain
'dummy', self.z_domain.decode()], force_san=True), key
def probe_cert(self, domain, **kwargs):
"""Probe tls-sni-01 challenge certificate.
:param unicode domain:
"""
# TODO: domain is not necessary if host is provided
if "host" not in kwargs:
host = socket.gethostbyname(domain)
logger.debug('%s resolved to %s', domain, host)
kwargs["host"] = host
kwargs.setdefault("port", self.PORT)
kwargs["name"] = self.z_domain
# TODO: try different methods?
# pylint: disable=protected-access
return crypto_util.probe_sni(**kwargs)
def verify_cert(self, cert):
"""Verify tls-sni-01 challenge certificate.
:param OpensSSL.crypto.X509 cert: Challenge certificate.
:returns: Whether the certificate was successfully verified.
:rtype: bool
"""
# pylint: disable=protected-access
sans = crypto_util._pyopenssl_cert_or_req_san(cert)
logger.debug('Certificate %s. SANs: %s', cert.digest('sha256'), sans)
return self.z_domain.decode() in sans
def simple_verify(self, chall, domain, account_public_key,
cert=None, **kwargs):
"""Simple verify.
Verify ``validation`` using ``account_public_key``, optionally
probe tls-sni-01 certificate and check using `verify_cert`.
:param .challenges.TLSSNI01 chall: Corresponding challenge.
:param str domain: Domain name being validated.
:param JWK account_public_key:
:param OpenSSL.crypto.X509 cert: Optional certificate. If not
provided (``None``) certificate will be retrieved using
`probe_cert`.
:param int port: Port used to probe the certificate.
:returns: ``True`` iff client's control of the domain has been
verified.
:rtype: bool
"""
if not self.verify(chall, account_public_key):
logger.debug("Verification of key authorization in response failed")
return False
if cert is None:
try:
cert = self.probe_cert(domain=domain, **kwargs)
except errors.Error as error:
logger.debug(str(error), exc_info=True)
return False
return self.verify_cert(cert)
@Challenge.register # pylint: disable=too-many-ancestors
class TLSSNI01(KeyAuthorizationChallenge):
"""ACME tls-sni-01 challenge."""
response_cls = TLSSNI01Response
typ = response_cls.typ
# boulder#962, ietf-wg-acme#22
#n = jose.Field("n", encoder=int, decoder=int)
def __init__(self, *args, **kwargs):
warnings.warn("TLS-SNI-01 is deprecated, and will stop working soon.",
DeprecationWarning, stacklevel=2)
super(TLSSNI01, self).__init__(*args, **kwargs)
def validation(self, account_key, **kwargs):
"""Generate validation.
:param JWK account_key:
:param OpenSSL.crypto.PKey cert_key: Optional private key used
in certificate generation. If not provided (``None``), then
fresh key will be generated.
:rtype: `tuple` of `OpenSSL.crypto.X509` and `OpenSSL.crypto.PKey`
"""
return self.response(account_key).gen_cert(key=kwargs.get('cert_key'))
@ChallengeResponse.register
class TLSALPN01Response(KeyAuthorizationChallengeResponse):
"""ACME TLS-ALPN-01 challenge response.
@@ -524,7 +371,7 @@ class TLSALPN01Response(KeyAuthorizationChallengeResponse):
typ = "tls-alpn-01"
@Challenge.register # pylint: disable=too-many-ancestors
@Challenge.register
class TLSALPN01(KeyAuthorizationChallenge):
"""ACME tls-alpn-01 challenge.
@@ -540,7 +387,7 @@ class TLSALPN01(KeyAuthorizationChallenge):
raise NotImplementedError()
@Challenge.register # pylint: disable=too-many-ancestors
@Challenge.register
class DNS(_TokenChallenge):
"""ACME "dns" challenge."""
typ = "dns"
acme/acme/client.py
+57 -75
View File
@@ -5,25 +5,26 @@ import datetime
from email.utils import parsedate_tz
import heapq
import logging
import re
import sys
import time
import six
from six.moves import http_client # pylint: disable=import-error
import josepy as jose
import OpenSSL
import re
from requests_toolbelt.adapters.source import SourceAddressAdapter
import requests
from requests.adapters import HTTPAdapter
import sys
from requests_toolbelt.adapters.source import SourceAddressAdapter
import six
from six.moves import http_client # pylint: disable=import-error
from acme import crypto_util
from acme import errors
from acme import jws
from acme import messages
# pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Dict, List, Set, Text
from acme.magic_typing import Dict # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Set # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Text # pylint: disable=unused-import, no-name-in-module
logger = logging.getLogger(__name__)
@@ -33,7 +34,6 @@ logger = logging.getLogger(__name__)
# https://urllib3.readthedocs.org/en/latest/security.html#insecureplatformwarning
if sys.version_info < (2, 7, 9): # pragma: no cover
try:
# pylint: disable=no-member
requests.packages.urllib3.contrib.pyopenssl.inject_into_urllib3() # type: ignore
except AttributeError:
import urllib3.contrib.pyopenssl # pylint: disable=import-error
@@ -44,7 +44,7 @@ DEFAULT_NETWORK_TIMEOUT = 45
DER_CONTENT_TYPE = 'application/pkix-cert'
class ClientBase(object): # pylint: disable=too-many-instance-attributes
class ClientBase(object):
"""ACME client base object.
:ivar messages.Directory directory:
@@ -123,14 +123,21 @@ class ClientBase(object): # pylint: disable=too-many-instance-attributes
"""
return self.update_registration(regr, update={'status': 'deactivated'})
def query_registration(self, regr):
"""Query server about registration.
def deactivate_authorization(self, authzr):
# type: (messages.AuthorizationResource) -> messages.AuthorizationResource
"""Deactivate authorization.
:param messages.RegistrationResource: Existing Registration
Resource.
:param messages.AuthorizationResource authzr: The Authorization resource
to be deactivated.
:returns: The Authorization resource that was deactivated.
:rtype: `.AuthorizationResource`
"""
return self._send_recv_regr(regr, messages.UpdateRegistration())
body = messages.UpdateAuthorization(status='deactivated')
response = self._post(authzr.uri, body)
return self._authzr_from_response(response,
authzr.body.identifier, authzr.uri)
def _authzr_from_response(self, response, identifier=None, uri=None):
authzr = messages.AuthorizationResource(
@@ -247,7 +254,6 @@ class Client(ClientBase):
URI from which the resource will be downloaded.
"""
# pylint: disable=too-many-arguments
self.key = key
if net is None:
net = ClientNetwork(key, alg=alg, verify_ssl=verify_ssl)
@@ -273,9 +279,17 @@ class Client(ClientBase):
assert response.status_code == http_client.CREATED
# "Instance of 'Field' has no key/contact member" bug:
# pylint: disable=no-member
return self._regr_from_response(response)
def query_registration(self, regr):
"""Query server about registration.
:param messages.RegistrationResource: Existing Registration
Resource.
"""
return self._send_recv_regr(regr, messages.UpdateRegistration())
def agree_to_tos(self, regr):
"""Agree to the terms-of-service.
@@ -419,7 +433,6 @@ class Client(ClientBase):
was marked by the CA as invalid
"""
# pylint: disable=too-many-locals
assert max_attempts > 0
attempts = collections.defaultdict(int) # type: Dict[messages.AuthorizationResource, int]
exhausted = set()
@@ -450,7 +463,6 @@ class Client(ClientBase):
updated[authzr] = updated_authzr
attempts[authzr] += 1
# pylint: disable=no-member
if updated_authzr.body.status not in (
messages.STATUS_VALID, messages.STATUS_INVALID):
if attempts[authzr] < max_attempts:
@@ -591,7 +603,6 @@ class ClientV2(ClientBase):
if response.status_code == 200 and 'Location' in response.headers:
raise errors.ConflictError(response.headers.get('Location'))
# "Instance of 'Field' has no key/contact member" bug:
# pylint: disable=no-member
regr = self._regr_from_response(response)
self.net.account = regr
return regr
@@ -603,10 +614,13 @@ class ClientV2(ClientBase):
Resource.
"""
self.net.account = regr
updated_regr = super(ClientV2, self).query_registration(regr)
self.net.account = updated_regr
return updated_regr
self.net.account = regr # See certbot/certbot#6258
# ACME v2 requires to use a POST-as-GET request (POST an empty JWS) here.
# This is done by passing None instead of an empty UpdateRegistration to _post().
response = self._post(regr.uri, None)
self.net.account = self._regr_from_response(response, uri=regr.uri,
terms_of_service=regr.terms_of_service)
return self.net.account
def update_registration(self, regr, update=None):
"""Update registration.
@@ -652,7 +666,7 @@ class ClientV2(ClientBase):
response = self._post(self.directory['newOrder'], order)
body = messages.Order.from_json(response.json())
authorizations = []
for url in body.authorizations:
for url in body.authorizations: # pylint: disable=not-an-iterable
authorizations.append(self._authzr_from_response(self._post_as_get(url), uri=url))
return messages.OrderResource(
body=body,
@@ -712,9 +726,9 @@ class ClientV2(ClientBase):
for authzr in responses:
if authzr.body.status != messages.STATUS_VALID:
for chall in authzr.body.challenges:
if chall.error != None:
if chall.error is not None:
failed.append(authzr)
if len(failed) > 0:
if failed:
raise errors.ValidationError(failed)
return orderr.update(authorizations=responses)
@@ -739,8 +753,7 @@ class ClientV2(ClientBase):
if body.error is not None:
raise errors.IssuanceError(body.error)
if body.certificate is not None:
certificate_response = self._post_as_get(body.certificate,
content_type=DER_CONTENT_TYPE).text
certificate_response = self._post_as_get(body.certificate).text
return orderr.update(body=body, fullchain_pem=certificate_response)
raise errors.TimeoutError()
@@ -759,36 +772,17 @@ class ClientV2(ClientBase):
def external_account_required(self):
"""Checks if ACME server requires External Account Binding authentication."""
if hasattr(self.directory, 'meta') and self.directory.meta.external_account_required:
return True
else:
return False
return hasattr(self.directory, 'meta') and self.directory.meta.external_account_required
def _post_as_get(self, *args, **kwargs):
"""
Send GET request using the POST-as-GET protocol if needed.
The request will be first issued using POST-as-GET for ACME v2. If the ACME CA servers do
not support this yet and return an error, request will be retried using GET.
For ACME v1, only GET request will be tried, as POST-as-GET is not supported.
Send GET request using the POST-as-GET protocol.
:param args:
:param kwargs:
:return:
"""
if self.acme_version >= 2:
# We add an empty payload for POST-as-GET requests
new_args = args[:1] + (None,) + args[1:]
try:
return self._post(*new_args, **kwargs) # pylint: disable=star-args
except messages.Error as error:
if error.code == 'malformed':
logger.debug('Error during a POST-as-GET request, '
'your ACME CA may not support it:\n%s', error)
logger.debug('Retrying request with GET.')
else: # pragma: no cover
raise
# If POST-as-GET is not supported yet, we use a GET instead.
return self.net.get(*args, **kwargs)
new_args = args[:1] + (None,) + args[1:]
return self._post(*new_args, **kwargs)
class BackwardsCompatibleClientV2(object):
@@ -866,8 +860,7 @@ class BackwardsCompatibleClientV2(object):
for domain in dnsNames:
authorizations.append(self.client.request_domain_challenges(domain))
return messages.OrderResource(authorizations=authorizations, csr_pem=csr_pem)
else:
return self.client.new_order(csr_pem)
return self.client.new_order(csr_pem)
def finalize_order(self, orderr, deadline):
"""Finalize an order and obtain a certificate.
@@ -904,8 +897,7 @@ class BackwardsCompatibleClientV2(object):
chain = crypto_util.dump_pyopenssl_chain(chain).decode()
return orderr.update(fullchain_pem=(cert + chain))
else:
return self.client.finalize_order(orderr, deadline)
return self.client.finalize_order(orderr, deadline)
def revoke(self, cert, rsn):
"""Revoke certificate.
@@ -923,8 +915,7 @@ class BackwardsCompatibleClientV2(object):
def _acme_version_from_directory(self, directory):
if hasattr(directory, 'newNonce'):
return 2
else:
return 1
return 1
def external_account_required(self):
"""Checks if the server requires an external account for ACMEv2 servers.
@@ -932,11 +923,10 @@ class BackwardsCompatibleClientV2(object):
Always return False for ACMEv1 servers, as it doesn't use External Account Binding."""
if self.acme_version == 1:
return False
else:
return self.client.external_account_required()
return self.client.external_account_required()
class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
class ClientNetwork(object):
"""Wrapper around requests that signs POSTs for authentication.
Also adds user agent, and handles Content-Type.
@@ -952,7 +942,7 @@ class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
:param messages.RegistrationResource account: Account object. Required if you are
planning to use .post() with acme_version=2 for anything other than
creating a new account; may be set later after registering.
:param josepy.JWASignature alg: Algoritm to use in signing JWS.
:param josepy.JWASignature alg: Algorithm to use in signing JWS.
:param bool verify_ssl: Whether to verify certificates on SSL connections.
:param str user_agent: String to send as User-Agent header.
:param float timeout: Timeout for requests.
@@ -962,7 +952,6 @@ class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
def __init__(self, key, account=None, alg=jose.RS256, verify_ssl=True,
user_agent='acme-python', timeout=DEFAULT_NETWORK_TIMEOUT,
source_address=None):
# pylint: disable=too-many-arguments
self.key = key
self.account = account
self.alg = alg
@@ -1011,7 +1000,6 @@ class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
if self.account is not None:
kwargs["kid"] = self.account["uri"]
kwargs["key"] = self.key
# pylint: disable=star-args
return jws.JWS.sign(jobj, **kwargs).json_dumps(indent=2)
@classmethod
@@ -1071,7 +1059,6 @@ class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
return response
def _send_request(self, method, url, *args, **kwargs):
# pylint: disable=too-many-locals
"""Send HTTP request.
Makes sure that `verify_ssl` is respected. Logs request and
@@ -1118,10 +1105,9 @@ class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
err_regex = r".*host='(\S*)'.*Max retries exceeded with url\: (\/\w*).*(\[Errno \d+\])([A-Za-z ]*)"
m = re.match(err_regex, str(e))
if m is None:
raise # pragma: no cover
else:
host, path, _err_no, err_msg = m.groups()
raise ValueError("Requesting {0}{1}:{2}".format(host, path, err_msg))
raise # pragma: no cover
host, path, _err_no, err_msg = m.groups()
raise ValueError("Requesting {0}{1}:{2}".format(host, path, err_msg))
# If content is DER, log the base64 of it instead of raw bytes, to keep
# binary data out of the logs.
@@ -1187,15 +1173,11 @@ class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
if error.code == 'badNonce':
logger.debug('Retrying request after error:\n%s', error)
return self._post_once(*args, **kwargs)
else:
raise
raise
def _post_once(self, url, obj, content_type=JOSE_CONTENT_TYPE,
acme_version=1, **kwargs):
try:
new_nonce_url = kwargs.pop('new_nonce_url')
except KeyError:
new_nonce_url = None
new_nonce_url = kwargs.pop('new_nonce_url', None)
data = self._wrap_in_jws(obj, self._get_nonce(url, new_nonce_url), url, acme_version)
kwargs.setdefault('headers', {'Content-Type': content_type})
response = self._send_request('POST', url, data=data, **kwargs)
acme/acme/crypto_util.py
+14 -19
View File
@@ -6,32 +6,29 @@ 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 OpenSSL import crypto
from OpenSSL import SSL # type: ignore # https://github.com/python/typeshed/issues/2052
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
from acme.magic_typing import Callable # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Optional # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Tuple # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Union # pylint: disable=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
# Default SSL method selected here is the most compatible, while secure
# SSL method: 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
_DEFAULT_SSL_METHOD = SSL.SSLv23_METHOD # type: ignore
class SSLSocket(object): # pylint: disable=too-few-public-methods
class SSLSocket(object):
"""SSL wrapper for sockets.
:ivar socket sock: Original wrapped socket.
@@ -40,7 +37,7 @@ class SSLSocket(object): # pylint: disable=too-few-public-methods
:ivar method: See `OpenSSL.SSL.Context` for allowed values.
"""
def __init__(self, sock, certs, method=_DEFAULT_TLSSNI01_SSL_METHOD):
def __init__(self, sock, certs, method=_DEFAULT_SSL_METHOD):
self.sock = sock
self.certs = certs
self.method = method
@@ -77,7 +74,7 @@ class SSLSocket(object): # pylint: disable=too-few-public-methods
class FakeConnection(object):
"""Fake OpenSSL.SSL.Connection."""
# pylint: disable=too-few-public-methods,missing-docstring
# pylint: disable=missing-docstring
def __init__(self, connection):
self._wrapped = connection
@@ -112,7 +109,7 @@ class SSLSocket(object): # pylint: disable=too-few-public-methods
def probe_sni(name, host, port=443, timeout=300,
method=_DEFAULT_TLSSNI01_SSL_METHOD, source_address=('', 0)):
method=_DEFAULT_SSL_METHOD, source_address=('', 0)):
"""Probe SNI server for SSL certificate.
:param bytes name: Byte string to send as the server name in the
@@ -137,7 +134,6 @@ def probe_sni(name, host, port=443, timeout=300,
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(
@@ -198,8 +194,7 @@ def _pyopenssl_cert_or_req_all_names(loaded_cert_or_req):
if common_name is None:
return sans
else:
return [common_name] + [d for d in sans if d != common_name]
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.
acme/acme/errors.py
+13 -3
View File
@@ -29,7 +29,12 @@ class NonceError(ClientError):
class BadNonce(NonceError):
"""Bad nonce error."""
def __init__(self, nonce, error, *args, **kwargs):
super(BadNonce, self).__init__(*args, **kwargs)
# MyPy complains here that there is too many arguments for BaseException constructor.
# This is an error fixed in typeshed, see https://github.com/python/mypy/issues/4183
# The fix is included in MyPy>=0.740, but upgrading it would bring dozen of errors due to
# new types definitions. So we ignore the error until the code base is fixed to match
# with MyPy>=0.740 referential.
super(BadNonce, self).__init__(*args, **kwargs) # type: ignore
self.nonce = nonce
self.error = error
@@ -48,7 +53,8 @@ class MissingNonce(NonceError):
"""
def __init__(self, response, *args, **kwargs):
super(MissingNonce, self).__init__(*args, **kwargs)
# See comment in BadNonce constructor above for an explanation of type: ignore here.
super(MissingNonce, self).__init__(*args, **kwargs) # type: ignore
self.response = response
def __str__(self):
@@ -83,6 +89,7 @@ class PollError(ClientError):
return '{0}(exhausted={1!r}, updated={2!r})'.format(
self.__class__.__name__, self.exhausted, self.updated)
class ValidationError(Error):
"""Error for authorization failures. Contains a list of authorization
resources, each of which is invalid and should have an error field.
@@ -91,9 +98,11 @@ class ValidationError(Error):
self.failed_authzrs = failed_authzrs
super(ValidationError, self).__init__()
class TimeoutError(Error):
class TimeoutError(Error): # pylint: disable=redefined-builtin
"""Error for when polling an authorization or an order times out."""
class IssuanceError(Error):
"""Error sent by the server after requesting issuance of a certificate."""
@@ -105,6 +114,7 @@ class IssuanceError(Error):
self.error = error
super(IssuanceError, self).__init__()
class ConflictError(ClientError):
"""Error for when the server returns a 409 (Conflict) HTTP status.
acme/acme/fields.py
-1
View File
@@ -4,7 +4,6 @@ import logging
import josepy as jose
import pyrfc3339
logger = logging.getLogger(__name__)
acme/acme/jws.py
+2 -2
View File
@@ -40,10 +40,10 @@ class Signature(jose.Signature):
class JWS(jose.JWS):
"""ACME-specific JWS. Includes none, url, and kid in protected header."""
signature_cls = Signature
__slots__ = jose.JWS._orig_slots # pylint: disable=no-member
__slots__ = jose.JWS._orig_slots
@classmethod
# pylint: disable=arguments-differ,too-many-arguments
# pylint: disable=arguments-differ
def sign(cls, payload, key, alg, nonce, url=None, kid=None):
# Per ACME spec, jwk and kid are mutually exclusive, so only include a
# jwk field if kid is not provided.
acme/acme/magic_typing.py
+1
View File
@@ -1,6 +1,7 @@
"""Shim class to not have to depend on typing module in prod."""
import sys
class TypingClass(object):
"""Ignore import errors by getting anything"""
def __getattr__(self, name):
acme/acme/messages.py
+43 -19
View File
@@ -1,37 +1,55 @@
"""ACME protocol messages."""
import six
import json
try:
from collections.abc import Hashable # pylint: disable=no-name-in-module
except ImportError:
from collections import Hashable
import josepy as jose
import six
from acme import challenges
from acme import errors
from acme import fields
from acme import util
from acme import jws
from acme import util
try:
from collections.abc import Hashable # pylint: disable=no-name-in-module
except ImportError: # pragma: no cover
from collections import Hashable
OLD_ERROR_PREFIX = "urn:acme:error:"
ERROR_PREFIX = "urn:ietf:params:acme:error:"
ERROR_CODES = {
'accountDoesNotExist': 'The request specified an account that does not exist',
'alreadyRevoked': 'The request specified a certificate to be revoked that has' \
' already been revoked',
'badCSR': 'The CSR is unacceptable (e.g., due to a short key)',
'badNonce': 'The client sent an unacceptable anti-replay nonce',
'badPublicKey': 'The JWS was signed by a public key the server does not support',
'badRevocationReason': 'The revocation reason provided is not allowed by the server',
'badSignatureAlgorithm': 'The JWS was signed with an algorithm the server does not support',
'caa': 'Certification Authority Authorization (CAA) records forbid the CA from issuing' \
' a certificate',
'compound': 'Specific error conditions are indicated in the "subproblems" array',
'connection': ('The server could not connect to the client to verify the'
' domain'),
'dns': 'There was a problem with a DNS query during identifier validation',
'dnssec': 'The server could not validate a DNSSEC signed domain',
'incorrectResponse': 'Response received didn\'t match the challenge\'s requirements',
# deprecate invalidEmail
'invalidEmail': 'The provided email for a registration was invalid',
'invalidContact': 'The provided contact URI was invalid',
'malformed': 'The request message was malformed',
'rejectedIdentifier': 'The server will not issue certificates for the identifier',
'orderNotReady': 'The request attempted to finalize an order that is not ready to be finalized',
'rateLimited': 'There were too many requests of a given type',
'serverInternal': 'The server experienced an internal error',
'tls': 'The server experienced a TLS error during domain verification',
'unauthorized': 'The client lacks sufficient authorization',
'unsupportedContact': 'A contact URL for an account used an unsupported protocol scheme',
'unknownHost': 'The server could not resolve a domain name',
'unsupportedIdentifier': 'An identifier is of an unsupported type',
'externalAccountRequired': 'The server requires external account binding',
}
@@ -46,8 +64,7 @@ def is_acme_error(err):
"""Check if argument is an ACME error."""
if isinstance(err, Error) and (err.typ is not None):
return (ERROR_PREFIX in err.typ) or (OLD_ERROR_PREFIX in err.typ)
else:
return False
return False
@six.python_2_unicode_compatible
@@ -102,6 +119,7 @@ class Error(jose.JSONObjectWithFields, errors.Error):
code = str(self.typ).split(':')[-1]
if code in ERROR_CODES:
return code
return None
def __str__(self):
return b' :: '.join(
@@ -116,18 +134,19 @@ class _Constant(jose.JSONDeSerializable, Hashable): # type: ignore
POSSIBLE_NAMES = NotImplemented
def __init__(self, name):
self.POSSIBLE_NAMES[name] = self
super(_Constant, self).__init__()
self.POSSIBLE_NAMES[name] = self # pylint: disable=unsupported-assignment-operation
self.name = name
def to_partial_json(self):
return self.name
@classmethod
def from_json(cls, value):
if value not in cls.POSSIBLE_NAMES:
def from_json(cls, jobj):
if jobj not in cls.POSSIBLE_NAMES: # pylint: disable=unsupported-membership-test
raise jose.DeserializationError(
'{0} not recognized'.format(cls.__name__))
return cls.POSSIBLE_NAMES[value]
return cls.POSSIBLE_NAMES[jobj]
def __repr__(self):
return '{0}({1})'.format(self.__class__.__name__, self.name)
@@ -152,6 +171,7 @@ STATUS_VALID = Status('valid')
STATUS_INVALID = Status('invalid')
STATUS_REVOKED = Status('revoked')
STATUS_READY = Status('ready')
STATUS_DEACTIVATED = Status('deactivated')
class IdentifierType(_Constant):
@@ -186,7 +206,6 @@ class Directory(jose.JSONDeSerializable):
def __init__(self, **kwargs):
kwargs = dict((self._internal_name(k), v) for k, v in kwargs.items())
# pylint: disable=star-args
super(Directory.Meta, self).__init__(**kwargs)
@property
@@ -226,13 +245,13 @@ class Directory(jose.JSONDeSerializable):
try:
return self[name.replace('_', '-')]
except KeyError as error:
raise AttributeError(str(error) + ': ' + name)
raise AttributeError(str(error))
def __getitem__(self, name):
try:
return self._jobj[self._canon_key(name)]
except KeyError:
raise KeyError('Directory field not found')
raise KeyError('Directory field "' + self._canon_key(name) + '" not found')
def to_partial_json(self):
return self._jobj
@@ -322,7 +341,7 @@ class Registration(ResourceBody):
def _filter_contact(self, prefix):
return tuple(
detail[len(prefix):] for detail in self.contact
detail[len(prefix):] for detail in self.contact # pylint: disable=not-an-iterable
if detail.startswith(prefix))
@property
@@ -394,7 +413,6 @@ class ChallengeBody(ResourceBody):
def __init__(self, **kwargs):
kwargs = dict((self._internal_name(k), v) for k, v in kwargs.items())
# pylint: disable=star-args
super(ChallengeBody, self).__init__(**kwargs)
def encode(self, name):
@@ -457,7 +475,7 @@ class Authorization(ResourceBody):
:ivar datetime.datetime expires:
"""
identifier = jose.Field('identifier', decoder=Identifier.from_json)
identifier = jose.Field('identifier', decoder=Identifier.from_json, omitempty=True)
challenges = jose.Field('challenges', omitempty=True)
combinations = jose.Field('combinations', omitempty=True)
@@ -477,7 +495,7 @@ class Authorization(ResourceBody):
def resolved_combinations(self):
"""Combinations with challenges instead of indices."""
return tuple(tuple(self.challenges[idx] for idx in combo)
for combo in self.combinations)
for combo in self.combinations) # pylint: disable=not-an-iterable
@Directory.register
@@ -487,6 +505,12 @@ class NewAuthorization(Authorization):
resource = fields.Resource(resource_type)
class UpdateAuthorization(Authorization):
"""Update authorization."""
resource_type = 'authz'
resource = fields.Resource(resource_type)
class AuthorizationResource(ResourceWithURI):
"""Authorization Resource.
acme/acme/standalone.py
+8 -80
View File
@@ -1,28 +1,22 @@
"""Support for standalone client challenge solvers. """
import argparse
import collections
import functools
import logging
import os
import socket
import sys
import threading
from six.moves import BaseHTTPServer # type: ignore # pylint: disable=import-error
from six.moves import http_client # pylint: disable=import-error
from six.moves import socketserver # type: ignore # pylint: disable=import-error
import OpenSSL
from acme import challenges
from acme import crypto_util
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
logger = logging.getLogger(__name__)
# six.moves.* | pylint: disable=no-member,attribute-defined-outside-init
# pylint: disable=too-few-public-methods,no-init
# pylint: disable=no-init
class TLSServer(socketserver.TCPServer):
@@ -37,7 +31,7 @@ class TLSServer(socketserver.TCPServer):
self.certs = kwargs.pop("certs", {})
self.method = kwargs.pop(
# pylint: disable=protected-access
"method", crypto_util._DEFAULT_TLSSNI01_SSL_METHOD)
"method", crypto_util._DEFAULT_SSL_METHOD)
self.allow_reuse_address = kwargs.pop("allow_reuse_address", True)
socketserver.TCPServer.__init__(self, *args, **kwargs)
@@ -50,7 +44,7 @@ class TLSServer(socketserver.TCPServer):
return socketserver.TCPServer.server_bind(self)
class ACMEServerMixin: # pylint: disable=old-style-class
class ACMEServerMixin:
"""ACME server common settings mixin."""
# TODO: c.f. #858
server_version = "ACME client standalone challenge solver"
@@ -82,7 +76,7 @@ class BaseDualNetworkedServers(object):
kwargs["ipv6"] = ip_version
new_address = (server_address[0],) + (port,) + server_address[2:]
new_args = (new_address,) + remaining_args
server = ServerClass(*new_args, **kwargs) # pylint: disable=star-args
server = ServerClass(*new_args, **kwargs)
logger.debug(
"Successfully bound to %s:%s using %s", new_address[0],
new_address[1], "IPv6" if ip_version else "IPv4")
@@ -90,8 +84,8 @@ class BaseDualNetworkedServers(object):
if self.servers:
# Already bound using IPv6.
logger.debug(
"Certbot wasn't able to bind to %s:%s using %s, this " +
"is often expected due to the dual stack nature of " +
"Certbot wasn't able to bind to %s:%s using %s, this "
"is often expected due to the dual stack nature of "
"IPv6 socket implementations.",
new_address[0], new_address[1],
"IPv6" if ip_version else "IPv4")
@@ -104,14 +98,13 @@ class BaseDualNetworkedServers(object):
# If two servers are set up and port 0 was passed in, ensure we always
# bind to the same port for both servers.
port = server.socket.getsockname()[1]
if len(self.servers) == 0:
if not self.servers:
raise socket.error("Could not bind to IPv4 or IPv6.")
def serve_forever(self):
"""Wraps socketserver.TCPServer.serve_forever"""
for server in self.servers:
thread = threading.Thread(
# pylint: disable=no-member
target=server.serve_forever)
thread.start()
self.threads.append(thread)
@@ -131,35 +124,6 @@ class BaseDualNetworkedServers(object):
self.threads = []
class TLSSNI01Server(TLSServer, ACMEServerMixin):
"""TLSSNI01 Server."""
def __init__(self, server_address, certs, ipv6=False):
TLSServer.__init__(
self, server_address, BaseRequestHandlerWithLogging, certs=certs, ipv6=ipv6)
class TLSSNI01DualNetworkedServers(BaseDualNetworkedServers):
"""TLSSNI01Server Wrapper. Tries everything for both. Failures for one don't
affect the other."""
def __init__(self, *args, **kwargs):
BaseDualNetworkedServers.__init__(self, TLSSNI01Server, *args, **kwargs)
class BaseRequestHandlerWithLogging(socketserver.BaseRequestHandler):
"""BaseRequestHandler with logging."""
def log_message(self, format, *args): # pylint: disable=redefined-builtin
"""Log arbitrary message."""
logger.debug("%s - - %s", self.client_address[0], format % args)
def handle(self):
"""Handle request."""
self.log_message("Incoming request")
socketserver.BaseRequestHandler.handle(self)
class HTTPServer(BaseHTTPServer.HTTPServer):
"""Generic HTTP Server."""
@@ -262,39 +226,3 @@ class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
"""
return functools.partial(
cls, simple_http_resources=simple_http_resources)
def simple_tls_sni_01_server(cli_args, forever=True):
"""Run simple standalone TLSSNI01 server."""
logging.basicConfig(level=logging.DEBUG)
parser = argparse.ArgumentParser()
parser.add_argument(
"-p", "--port", default=0, help="Port to serve at. By default "
"picks random free port.")
args = parser.parse_args(cli_args[1:])
certs = {}
_, hosts, _ = next(os.walk('.')) # type: ignore # https://github.com/python/mypy/issues/465
for host in hosts:
with open(os.path.join(host, "cert.pem")) as cert_file:
cert_contents = cert_file.read()
with open(os.path.join(host, "key.pem")) as key_file:
key_contents = key_file.read()
certs[host.encode()] = (
OpenSSL.crypto.load_privatekey(
OpenSSL.crypto.FILETYPE_PEM, key_contents),
OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, cert_contents))
server = TLSSNI01Server(('', int(args.port)), certs=certs)
logger.info("Serving at https://%s:%s...", *server.socket.getsockname()[:2])
if forever: # pragma: no cover
server.serve_forever()
else:
server.handle_request()
if __name__ == "__main__":
sys.exit(simple_tls_sni_01_server(sys.argv)) # pragma: no cover
acme/docs/conf.py
+2 -3
View File
@@ -12,10 +12,9 @@
# All configuration values have a default; values that are commented out
# serve to show the default.
import sys
import os
import shlex
import sys
here = os.path.abspath(os.path.dirname(__file__))
@@ -42,7 +41,7 @@ extensions = [
]
autodoc_member_order = 'bysource'
autodoc_default_flags = ['show-inheritance', 'private-members']
autodoc_default_flags = ['show-inheritance']
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
acme/examples/http01_example.py
+241
View File
@@ -0,0 +1,241 @@
"""Example ACME-V2 API for HTTP-01 challenge.
Brief:
This a complete usage example of the python-acme API.
Limitations of this example:
- Works for only one Domain name
- Performs only HTTP-01 challenge
- Uses ACME-v2
Workflow:
(Account creation)
- Create account key
- Register account and accept TOS
(Certificate actions)
- Select HTTP-01 within offered challenges by the CA server
- Set up http challenge resource
- Set up standalone web server
- Create domain private key and CSR
- Issue certificate
- Renew certificate
- Revoke certificate
(Account update actions)
- Change contact information
- Deactivate Account
"""
from contextlib import contextmanager
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
import josepy as jose
import OpenSSL
from acme import challenges
from acme import client
from acme import crypto_util
from acme import errors
from acme import messages
from acme import standalone
# Constants:
# This is the staging point for ACME-V2 within Let's Encrypt.
DIRECTORY_URL = 'https://acme-staging-v02.api.letsencrypt.org/directory'
USER_AGENT = 'python-acme-example'
# Account key size
ACC_KEY_BITS = 2048
# Certificate private key size
CERT_PKEY_BITS = 2048
# Domain name for the certificate.
DOMAIN = 'client.example.com'
# If you are running Boulder locally, it is possible to configure any port
# number to execute the challenge, but real CA servers will always use port
# 80, as described in the ACME specification.
PORT = 80
# Useful methods and classes:
def new_csr_comp(domain_name, pkey_pem=None):
"""Create certificate signing request."""
if pkey_pem is None:
# Create private key.
pkey = OpenSSL.crypto.PKey()
pkey.generate_key(OpenSSL.crypto.TYPE_RSA, CERT_PKEY_BITS)
pkey_pem = OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM,
pkey)
csr_pem = crypto_util.make_csr(pkey_pem, [domain_name])
return pkey_pem, csr_pem
def select_http01_chall(orderr):
"""Extract authorization resource from within order resource."""
# Authorization Resource: authz.
# This object holds the offered challenges by the server and their status.
authz_list = orderr.authorizations
for authz in authz_list:
# Choosing challenge.
# authz.body.challenges is a set of ChallengeBody objects.
for i in authz.body.challenges:
# Find the supported challenge.
if isinstance(i.chall, challenges.HTTP01):
return i
raise Exception('HTTP-01 challenge was not offered by the CA server.')
@contextmanager
def challenge_server(http_01_resources):
"""Manage standalone server set up and shutdown."""
# Setting up a fake server that binds at PORT and any address.
address = ('', PORT)
try:
servers = standalone.HTTP01DualNetworkedServers(address,
http_01_resources)
# Start client standalone web server.
servers.serve_forever()
yield servers
finally:
# Shutdown client web server and unbind from PORT
servers.shutdown_and_server_close()
def perform_http01(client_acme, challb, orderr):
"""Set up standalone webserver and perform HTTP-01 challenge."""
response, validation = challb.response_and_validation(client_acme.net.key)
resource = standalone.HTTP01RequestHandler.HTTP01Resource(
chall=challb.chall, response=response, validation=validation)
with challenge_server({resource}):
# Let the CA server know that we are ready for the challenge.
client_acme.answer_challenge(challb, response)
# Wait for challenge status and then issue a certificate.
# It is possible to set a deadline time.
finalized_orderr = client_acme.poll_and_finalize(orderr)
return finalized_orderr.fullchain_pem
# Main examples:
def example_http():
"""This example executes the whole process of fulfilling a HTTP-01
challenge for one specific domain.
The workflow consists of:
(Account creation)
- Create account key
- Register account and accept TOS
(Certificate actions)
- Select HTTP-01 within offered challenges by the CA server
- Set up http challenge resource
- Set up standalone web server
- Create domain private key and CSR
- Issue certificate
- Renew certificate
- Revoke certificate
(Account update actions)
- Change contact information
- Deactivate Account
"""
# Create account key
acc_key = jose.JWKRSA(
key=rsa.generate_private_key(public_exponent=65537,
key_size=ACC_KEY_BITS,
backend=default_backend()))
# Register account and accept TOS
net = client.ClientNetwork(acc_key, user_agent=USER_AGENT)
directory = messages.Directory.from_json(net.get(DIRECTORY_URL).json())
client_acme = client.ClientV2(directory, net=net)
# Terms of Service URL is in client_acme.directory.meta.terms_of_service
# Registration Resource: regr
# Creates account with contact information.
email = ('fake@example.com')
regr = client_acme.new_account(
messages.NewRegistration.from_data(
email=email, terms_of_service_agreed=True))
# Create domain private key and CSR
pkey_pem, csr_pem = new_csr_comp(DOMAIN)
# Issue certificate
orderr = client_acme.new_order(csr_pem)
# Select HTTP-01 within offered challenges by the CA server
challb = select_http01_chall(orderr)
# The certificate is ready to be used in the variable "fullchain_pem".
fullchain_pem = perform_http01(client_acme, challb, orderr)
# Renew certificate
_, csr_pem = new_csr_comp(DOMAIN, pkey_pem)
orderr = client_acme.new_order(csr_pem)
challb = select_http01_chall(orderr)
# Performing challenge
fullchain_pem = perform_http01(client_acme, challb, orderr)
# Revoke certificate
fullchain_com = jose.ComparableX509(
OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, fullchain_pem))
try:
client_acme.revoke(fullchain_com, 0) # revocation reason = 0
except errors.ConflictError:
# Certificate already revoked.
pass
# Query registration status.
client_acme.net.account = regr
try:
regr = client_acme.query_registration(regr)
except errors.Error as err:
if err.typ == messages.OLD_ERROR_PREFIX + 'unauthorized' \
or err.typ == messages.ERROR_PREFIX + 'unauthorized':
# Status is deactivated.
pass
raise
# Change contact information
email = 'newfake@example.com'
regr = client_acme.update_registration(
regr.update(
body=regr.body.update(
contact=('mailto:' + email,)
)
)
)
# Deactivate account/registration
regr = client_acme.deactivate_registration(regr)
if __name__ == "__main__":
example_http()
acme/readthedocs.org.requirements.txt
+3 -3
View File
@@ -1,10 +1,10 @@
# readthedocs.org gives no way to change the install command to "pip
# install -e .[docs]" (that would in turn install documentation
# install -e acme[docs]" (that would in turn install documentation
# dependencies), but it allows to specify a requirements.txt file at
# https://readthedocs.org/dashboard/letsencrypt/advanced/ (c.f. #259)
# Although ReadTheDocs certainly doesn't need to install the project
# in --editable mode (-e), just "pip install .[docs]" does not work as
# expected and "pip install -e .[docs]" must be used instead
# in --editable mode (-e), just "pip install acme[docs]" does not work as
# expected and "pip install -e acme[docs]" must be used instead
-e acme[docs]
acme/setup.py
+14 -9
View File
@@ -1,9 +1,10 @@
from setuptools import setup
from setuptools import find_packages
from setuptools.command.test import test as TestCommand
import sys
version = '0.31.0.dev0'
from setuptools import find_packages
from setuptools import setup
from setuptools.command.test import test as TestCommand
version = '1.3.0.dev0'
# Please update tox.ini when modifying dependency version requirements
install_requires = [
@@ -11,9 +12,11 @@ install_requires = [
# rsa_recover_prime_factors (>=0.8)
'cryptography>=1.2.3',
# formerly known as acme.jose:
'josepy>=1.0.0',
# Connection.set_tlsext_host_name (>=0.13)
# 1.1.0+ is required to avoid the warnings described at
# https://github.com/certbot/josepy/issues/13.
'josepy>=1.1.0',
'mock',
# Connection.set_tlsext_host_name (>=0.13)
'PyOpenSSL>=0.13.1',
'pyrfc3339',
'pytz',
@@ -34,6 +37,7 @@ docs_extras = [
'sphinx_rtd_theme',
]
class PyTest(TestCommand):
user_options = []
@@ -48,6 +52,7 @@ class PyTest(TestCommand):
errno = pytest.main(shlex.split(self.pytest_args))
sys.exit(errno)
setup(
name='acme',
version=version,
@@ -56,7 +61,7 @@ setup(
author="Certbot Project",
author_email='client-dev@letsencrypt.org',
license='Apache License 2.0',
python_requires='>=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*',
python_requires='>=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*',
classifiers=[
'Development Status :: 5 - Production/Stable',
'Intended Audience :: Developers',
@@ -65,10 +70,10 @@ setup(
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Topic :: Internet :: WWW/HTTP',
'Topic :: Security',
],
@@ -80,7 +85,7 @@ setup(
'dev': dev_extras,
'docs': docs_extras,
},
tests_require=["pytest"],
test_suite='acme',
tests_require=["pytest"],
cmdclass={"test": PyTest},
)
acme/acme/challenges_test.py → acme/tests/challenges_test.py
+8 -157
View File
@@ -1,16 +1,12 @@
"""Tests for acme.challenges."""
import unittest
import warnings
import josepy as jose
import mock
import OpenSSL
import requests
from six.moves.urllib import parse as urllib_parse
from six.moves.urllib import parse as urllib_parse # pylint: disable=import-error
from acme import errors
from acme import test_util
import test_util
CERT = test_util.load_comparable_cert('cert.pem')
KEY = jose.JWKRSA(key=test_util.load_rsa_private_key('rsa512_key.pem'))
@@ -22,7 +18,6 @@ class ChallengeTest(unittest.TestCase):
from acme.challenges import Challenge
from acme.challenges import UnrecognizedChallenge
chall = UnrecognizedChallenge({"type": "foo"})
# pylint: disable=no-member
self.assertEqual(chall, Challenge.from_json(chall.jobj))
@@ -78,7 +73,6 @@ class KeyAuthorizationChallengeResponseTest(unittest.TestCase):
class DNS01ResponseTest(unittest.TestCase):
# pylint: disable=too-many-instance-attributes
def setUp(self):
from acme.challenges import DNS01Response
@@ -94,7 +88,8 @@ class DNS01ResponseTest(unittest.TestCase):
self.response = self.chall.response(KEY)
def test_to_partial_json(self):
self.assertEqual(self.jmsg, self.msg.to_partial_json())
self.assertEqual({k: v for k, v in self.jmsg.items() if k != 'keyAuthorization'},
self.msg.to_partial_json())
def test_from_json(self):
from acme.challenges import DNS01Response
@@ -149,7 +144,6 @@ class DNS01Test(unittest.TestCase):
class HTTP01ResponseTest(unittest.TestCase):
# pylint: disable=too-many-instance-attributes
def setUp(self):
from acme.challenges import HTTP01Response
@@ -165,7 +159,8 @@ class HTTP01ResponseTest(unittest.TestCase):
self.response = self.chall.response(KEY)
def test_to_partial_json(self):
self.assertEqual(self.jmsg, self.msg.to_partial_json())
self.assertEqual({k: v for k, v in self.jmsg.items() if k != 'keyAuthorization'},
self.msg.to_partial_json())
def test_from_json(self):
from acme.challenges import HTTP01Response
@@ -258,152 +253,7 @@ class HTTP01Test(unittest.TestCase):
self.msg.update(token=b'..').good_token)
class TLSSNI01ResponseTest(unittest.TestCase):
# pylint: disable=too-many-instance-attributes
def setUp(self):
from acme.challenges import TLSSNI01
self.chall = TLSSNI01(
token=jose.b64decode(b'a82d5ff8ef740d12881f6d3c2277ab2e'))
self.response = self.chall.response(KEY)
self.jmsg = {
'resource': 'challenge',
'type': 'tls-sni-01',
'keyAuthorization': self.response.key_authorization,
}
# pylint: disable=invalid-name
label1 = b'dc38d9c3fa1a4fdcc3a5501f2d38583f'
label2 = b'b7793728f084394f2a1afd459556bb5c'
self.z = label1 + label2
self.z_domain = label1 + b'.' + label2 + b'.acme.invalid'
self.domain = 'foo.com'
def test_z_and_domain(self):
self.assertEqual(self.z, self.response.z)
self.assertEqual(self.z_domain, self.response.z_domain)
def test_to_partial_json(self):
self.assertEqual(self.jmsg, self.response.to_partial_json())
def test_from_json(self):
from acme.challenges import TLSSNI01Response
self.assertEqual(self.response, TLSSNI01Response.from_json(self.jmsg))
def test_from_json_hashable(self):
from acme.challenges import TLSSNI01Response
hash(TLSSNI01Response.from_json(self.jmsg))
@mock.patch('acme.challenges.socket.gethostbyname')
@mock.patch('acme.challenges.crypto_util.probe_sni')
def test_probe_cert(self, mock_probe_sni, mock_gethostbyname):
mock_gethostbyname.return_value = '127.0.0.1'
self.response.probe_cert('foo.com')
mock_gethostbyname.assert_called_once_with('foo.com')
mock_probe_sni.assert_called_once_with(
host='127.0.0.1', port=self.response.PORT,
name=self.z_domain)
self.response.probe_cert('foo.com', host='8.8.8.8')
mock_probe_sni.assert_called_with(
host='8.8.8.8', port=mock.ANY, name=mock.ANY)
self.response.probe_cert('foo.com', port=1234)
mock_probe_sni.assert_called_with(
host=mock.ANY, port=1234, name=mock.ANY)
self.response.probe_cert('foo.com', bar='baz')
mock_probe_sni.assert_called_with(
host=mock.ANY, port=mock.ANY, name=mock.ANY, bar='baz')
self.response.probe_cert('foo.com', name=b'xxx')
mock_probe_sni.assert_called_with(
host=mock.ANY, port=mock.ANY,
name=self.z_domain)
def test_gen_verify_cert(self):
key1 = test_util.load_pyopenssl_private_key('rsa512_key.pem')
cert, key2 = self.response.gen_cert(key1)
self.assertEqual(key1, key2)
self.assertTrue(self.response.verify_cert(cert))
def test_gen_verify_cert_gen_key(self):
cert, key = self.response.gen_cert()
self.assertTrue(isinstance(key, OpenSSL.crypto.PKey))
self.assertTrue(self.response.verify_cert(cert))
def test_verify_bad_cert(self):
self.assertFalse(self.response.verify_cert(
test_util.load_cert('cert.pem')))
def test_simple_verify_bad_key_authorization(self):
key2 = jose.JWKRSA.load(test_util.load_vector('rsa256_key.pem'))
self.response.simple_verify(self.chall, "local", key2.public_key())
@mock.patch('acme.challenges.TLSSNI01Response.verify_cert', autospec=True)
def test_simple_verify(self, mock_verify_cert):
mock_verify_cert.return_value = mock.sentinel.verification
self.assertEqual(
mock.sentinel.verification, self.response.simple_verify(
self.chall, self.domain, KEY.public_key(),
cert=mock.sentinel.cert))
mock_verify_cert.assert_called_once_with(
self.response, mock.sentinel.cert)
@mock.patch('acme.challenges.TLSSNI01Response.probe_cert')
def test_simple_verify_false_on_probe_error(self, mock_probe_cert):
mock_probe_cert.side_effect = errors.Error
self.assertFalse(self.response.simple_verify(
self.chall, self.domain, KEY.public_key()))
class TLSSNI01Test(unittest.TestCase):
def setUp(self):
self.jmsg = {
'type': 'tls-sni-01',
'token': 'a82d5ff8ef740d12881f6d3c2277ab2e',
}
def _msg(self):
from acme.challenges import TLSSNI01
with warnings.catch_warnings(record=True) as warn:
warnings.simplefilter("always")
msg = TLSSNI01(
token=jose.b64decode('a82d5ff8ef740d12881f6d3c2277ab2e'))
assert warn is not None # using a raw assert for mypy
self.assertTrue(len(warn) == 1)
self.assertTrue(issubclass(warn[-1].category, DeprecationWarning))
self.assertTrue('deprecated' in str(warn[-1].message))
return msg
def test_to_partial_json(self):
self.assertEqual(self.jmsg, self._msg().to_partial_json())
def test_from_json(self):
from acme.challenges import TLSSNI01
self.assertEqual(self._msg(), TLSSNI01.from_json(self.jmsg))
def test_from_json_hashable(self):
from acme.challenges import TLSSNI01
hash(TLSSNI01.from_json(self.jmsg))
def test_from_json_invalid_token_length(self):
from acme.challenges import TLSSNI01
self.jmsg['token'] = jose.encode_b64jose(b'abcd')
self.assertRaises(
jose.DeserializationError, TLSSNI01.from_json, self.jmsg)
@mock.patch('acme.challenges.TLSSNI01Response.gen_cert')
def test_validation(self, mock_gen_cert):
mock_gen_cert.return_value = ('cert', 'key')
self.assertEqual(('cert', 'key'), self._msg().validation(
KEY, cert_key=mock.sentinel.cert_key))
mock_gen_cert.assert_called_once_with(key=mock.sentinel.cert_key)
class TLSALPN01ResponseTest(unittest.TestCase):
# pylint: disable=too-many-instance-attributes
def setUp(self):
from acme.challenges import TLSALPN01Response
@@ -419,7 +269,8 @@ class TLSALPN01ResponseTest(unittest.TestCase):
self.response = self.chall.response(KEY)
def test_to_partial_json(self):
self.assertEqual(self.jmsg, self.msg.to_partial_json())
self.assertEqual({k: v for k, v in self.jmsg.items() if k != 'keyAuthorization'},
self.msg.to_partial_json())
def test_from_json(self):
from acme.challenges import TLSALPN01Response
acme/acme/client_test.py → acme/tests/client_test.py
+21 -34
View File
@@ -5,21 +5,19 @@ import datetime
import json
import unittest
from six.moves import http_client # pylint: disable=import-error
import josepy as jose
import mock
import OpenSSL
import requests
from six.moves import http_client # pylint: disable=import-error
from acme import challenges
from acme import errors
from acme import jws as acme_jws
from acme import messages
from acme import messages_test
from acme import test_util
from acme.magic_typing import Dict # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Dict # pylint: disable=unused-import, no-name-in-module
import messages_test
import test_util
CERT_DER = test_util.load_vector('cert.der')
CERT_SAN_PEM = test_util.load_vector('cert-san.pem')
@@ -63,8 +61,8 @@ class ClientTestBase(unittest.TestCase):
self.contact = ('mailto:cert-admin@example.com', 'tel:+12025551212')
reg = messages.Registration(
contact=self.contact, key=KEY.public_key())
the_arg = dict(reg) # type: Dict
self.new_reg = messages.NewRegistration(**the_arg) # pylint: disable=star-args
the_arg = dict(reg) # type: Dict
self.new_reg = messages.NewRegistration(**the_arg)
self.regr = messages.RegistrationResource(
body=reg, uri='https://www.letsencrypt-demo.org/acme/reg/1')
@@ -318,7 +316,6 @@ class BackwardsCompatibleClientV2Test(ClientTestBase):
class ClientTest(ClientTestBase):
"""Tests for acme.client.Client."""
# pylint: disable=too-many-instance-attributes,too-many-public-methods
def setUp(self):
super(ClientTest, self).setUp()
@@ -358,7 +355,6 @@ class ClientTest(ClientTestBase):
def test_register(self):
# "Instance of 'Field' has no to_json/update member" bug:
# pylint: disable=no-member
self.response.status_code = http_client.CREATED
self.response.json.return_value = self.regr.body.to_json()
self.response.headers['Location'] = self.regr.uri
@@ -371,7 +367,6 @@ class ClientTest(ClientTestBase):
def test_update_registration(self):
# "Instance of 'Field' has no to_json/update member" bug:
# pylint: disable=no-member
self.response.headers['Location'] = self.regr.uri
self.response.json.return_value = self.regr.body.to_json()
self.assertEqual(self.regr, self.client.update_registration(self.regr))
@@ -639,6 +634,14 @@ class ClientTest(ClientTestBase):
errors.PollError, self.client.poll_and_request_issuance,
csr, authzrs, mintime=mintime, max_attempts=2)
def test_deactivate_authorization(self):
authzb = self.authzr.body.update(status=messages.STATUS_DEACTIVATED)
self.response.json.return_value = authzb.to_json()
authzr = self.client.deactivate_authorization(self.authzr)
self.assertEqual(authzb, authzr.body)
self.assertEqual(self.client.net.post.call_count, 1)
self.assertTrue(self.authzr.uri in self.net.post.call_args_list[0][0])
def test_check_cert(self):
self.response.headers['Location'] = self.certr.uri
self.response.content = CERT_DER
@@ -844,7 +847,6 @@ class ClientV2Test(ClientTestBase):
def test_update_registration(self):
# "Instance of 'Field' has no to_json/update member" bug:
# pylint: disable=no-member
self.response.headers['Location'] = self.regr.uri
self.response.json.return_value = self.regr.body.to_json()
self.assertEqual(self.regr, self.client.update_registration(self.regr))
@@ -883,19 +885,6 @@ class ClientV2Test(ClientTestBase):
new_nonce_url='https://www.letsencrypt-demo.org/acme/new-nonce')
self.client.net.get.assert_not_called()
class FakeError(messages.Error): # pylint: disable=too-many-ancestors
"""Fake error to reproduce a malformed request ACME error"""
def __init__(self): # pylint: disable=super-init-not-called
pass
@property
def code(self):
return 'malformed'
self.client.net.post.side_effect = FakeError()
self.client.poll(self.authzr2) # pylint: disable=protected-access
self.client.net.get.assert_called_once_with(self.authzr2.uri)
class MockJSONDeSerializable(jose.JSONDeSerializable):
# pylint: disable=missing-docstring
@@ -906,13 +895,12 @@ class MockJSONDeSerializable(jose.JSONDeSerializable):
return {'foo': self.value}
@classmethod
def from_json(cls, value):
def from_json(cls, jobj):
pass # pragma: no cover
class ClientNetworkTest(unittest.TestCase):
"""Tests for acme.client.ClientNetwork."""
# pylint: disable=too-many-public-methods
def setUp(self):
self.verify_ssl = mock.MagicMock()
@@ -962,8 +950,8 @@ class ClientNetworkTest(unittest.TestCase):
def test_check_response_not_ok_jobj_error(self):
self.response.ok = False
self.response.json.return_value = messages.Error(
detail='foo', typ='serverInternal', title='some title').to_json()
self.response.json.return_value = messages.Error.with_code(
'serverInternal', detail='foo', title='some title').to_json()
# pylint: disable=protected-access
self.assertRaises(
messages.Error, self.net._check_response, self.response)
@@ -988,7 +976,7 @@ class ClientNetworkTest(unittest.TestCase):
self.response.json.side_effect = ValueError
for response_ct in [self.net.JSON_CONTENT_TYPE, 'foo']:
self.response.headers['Content-Type'] = response_ct
# pylint: disable=protected-access,no-value-for-parameter
# pylint: disable=protected-access
self.assertEqual(
self.response, self.net._check_response(self.response))
@@ -1002,7 +990,7 @@ class ClientNetworkTest(unittest.TestCase):
self.response.json.return_value = {}
for response_ct in [self.net.JSON_CONTENT_TYPE, 'foo']:
self.response.headers['Content-Type'] = response_ct
# pylint: disable=protected-access,no-value-for-parameter
# pylint: disable=protected-access
self.assertEqual(
self.response, self.net._check_response(self.response))
@@ -1118,7 +1106,6 @@ class ClientNetworkTest(unittest.TestCase):
class ClientNetworkWithMockedResponseTest(unittest.TestCase):
"""Tests for acme.client.ClientNetwork which mock out response."""
# pylint: disable=too-many-instance-attributes
def setUp(self):
from acme.client import ClientNetwork
@@ -1128,8 +1115,8 @@ class ClientNetworkWithMockedResponseTest(unittest.TestCase):
self.response.headers = {}
self.response.links = {}
self.response.checked = False
self.acmev1_nonce_response = mock.MagicMock(ok=False,
status_code=http_client.METHOD_NOT_ALLOWED)
self.acmev1_nonce_response = mock.MagicMock(
ok=False, status_code=http_client.METHOD_NOT_ALLOWED)
self.acmev1_nonce_response.headers = {}
self.obj = mock.MagicMock()
self.wrapped_obj = mock.MagicMock()
acme/acme/crypto_util_test.py → acme/tests/crypto_util_test.py
+5 -8
View File
@@ -5,15 +5,14 @@ import threading
import time
import unittest
import six
from six.moves import socketserver #type: ignore # pylint: disable=import-error
import josepy as jose
import OpenSSL
import six
from six.moves import socketserver # type: ignore # pylint: disable=import-error
from acme import errors
from acme import test_util
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
import test_util
class SSLSocketAndProbeSNITest(unittest.TestCase):
@@ -30,7 +29,6 @@ class SSLSocketAndProbeSNITest(unittest.TestCase):
class _TestServer(socketserver.TCPServer):
# pylint: disable=too-few-public-methods
# six.moves.* | pylint: disable=attribute-defined-outside-init,no-init
def server_bind(self): # pylint: disable=missing-docstring
@@ -40,7 +38,6 @@ class SSLSocketAndProbeSNITest(unittest.TestCase):
self.server = _TestServer(('', 0), socketserver.BaseRequestHandler)
self.port = self.server.socket.getsockname()[1]
self.server_thread = threading.Thread(
# pylint: disable=no-member
target=self.server.handle_request)
def tearDown(self):
@@ -67,7 +64,7 @@ class SSLSocketAndProbeSNITest(unittest.TestCase):
def test_probe_connection_error(self):
# pylint has a hard time with six
self.server.server_close() # pylint: disable=no-member
self.server.server_close()
original_timeout = socket.getdefaulttimeout()
try:
socket.setdefaulttimeout(1)
acme/acme/errors_test.py → acme/tests/errors_test.py
View File
acme/acme/fields_test.py → acme/tests/fields_test.py
View File
acme/acme/jose_test.py → acme/tests/jose_test.py
+5 -5
View File
@@ -2,6 +2,7 @@
import importlib
import unittest
class JoseTest(unittest.TestCase):
"""Tests for acme.jose shim."""
@@ -20,11 +21,10 @@ class JoseTest(unittest.TestCase):
# We use the imports below with eval, but pylint doesn't
# understand that.
# pylint: disable=eval-used,unused-variable
import acme
import josepy
acme_jose_mod = eval(acme_jose_path)
josepy_mod = eval(josepy_path)
import acme # pylint: disable=unused-import
import josepy # pylint: disable=unused-import
acme_jose_mod = eval(acme_jose_path) # pylint: disable=eval-used
josepy_mod = eval(josepy_path) # pylint: disable=eval-used
self.assertIs(acme_jose_mod, josepy_mod)
self.assertIs(getattr(acme_jose_mod, attribute), getattr(josepy_mod, attribute))
acme/acme/jws_test.py → acme/tests/jws_test.py
+1 -2
View File
@@ -3,8 +3,7 @@ import unittest
import josepy as jose
from acme import test_util
import test_util
KEY = jose.JWKRSA.load(test_util.load_vector('rsa512_key.pem'))
acme/acme/magic_typing_test.py → acme/tests/magic_typing_test.py
View File
acme/acme/messages_test.py → acme/tests/messages_test.py
+10 -15
View File
@@ -5,9 +5,8 @@ import josepy as jose
import mock
from acme import challenges
from acme import test_util
from acme.magic_typing import Dict # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Dict # pylint: disable=unused-import, no-name-in-module
import test_util
CERT = test_util.load_comparable_cert('cert.der')
CSR = test_util.load_comparable_csr('csr.der')
@@ -19,8 +18,7 @@ class ErrorTest(unittest.TestCase):
def setUp(self):
from acme.messages import Error, ERROR_PREFIX
self.error = Error(
detail='foo', typ=ERROR_PREFIX + 'malformed', title='title')
self.error = Error.with_code('malformed', detail='foo', title='title')
self.jobj = {
'detail': 'foo',
'title': 'some title',
@@ -28,7 +26,6 @@ class ErrorTest(unittest.TestCase):
}
self.error_custom = Error(typ='custom', detail='bar')
self.empty_error = Error()
self.jobj_custom = {'type': 'custom', 'detail': 'bar'}
def test_default_typ(self):
from acme.messages import Error
@@ -43,8 +40,7 @@ class ErrorTest(unittest.TestCase):
hash(Error.from_json(self.error.to_json()))
def test_description(self):
self.assertEqual(
'The request message was malformed', self.error.description)
self.assertEqual('The request message was malformed', self.error.description)
self.assertTrue(self.error_custom.description is None)
def test_code(self):
@@ -54,17 +50,17 @@ class ErrorTest(unittest.TestCase):
self.assertEqual(None, Error().code)
def test_is_acme_error(self):
from acme.messages import is_acme_error
from acme.messages import is_acme_error, Error
self.assertTrue(is_acme_error(self.error))
self.assertFalse(is_acme_error(self.error_custom))
self.assertFalse(is_acme_error(Error()))
self.assertFalse(is_acme_error(self.empty_error))
self.assertFalse(is_acme_error("must pet all the {dogs|rabbits}"))
def test_unicode_error(self):
from acme.messages import Error, ERROR_PREFIX, is_acme_error
arabic_error = Error(
detail=u'\u0639\u062f\u0627\u0644\u0629', typ=ERROR_PREFIX + 'malformed',
title='title')
from acme.messages import Error, is_acme_error
arabic_error = Error.with_code(
'malformed', detail=u'\u0639\u062f\u0627\u0644\u0629', title='title')
self.assertTrue(is_acme_error(arabic_error))
def test_with_code(self):
@@ -305,8 +301,7 @@ class ChallengeBodyTest(unittest.TestCase):
from acme.messages import Error
from acme.messages import STATUS_INVALID
self.status = STATUS_INVALID
error = Error(typ='urn:ietf:params:acme:error:serverInternal',
detail='Unable to communicate with DNS server')
error = Error.with_code('serverInternal', detail='Unable to communicate with DNS server')
self.challb = ChallengeBody(
uri='http://challb', chall=self.chall, status=self.status,
error=error)
acme/acme/standalone_test.py → acme/tests/standalone_test.py
+7 -116
View File
@@ -1,23 +1,17 @@
"""Tests for acme.standalone."""
import os
import shutil
import socket
import threading
import tempfile
import unittest
from six.moves import http_client # pylint: disable=import-error
from six.moves import queue # pylint: disable=import-error
from six.moves import socketserver # type: ignore # pylint: disable=import-error
import josepy as jose
import mock
import requests
from six.moves import http_client # pylint: disable=import-error
from six.moves import socketserver # type: ignore # pylint: disable=import-error
from acme import challenges
from acme import crypto_util
from acme import test_util
from acme.magic_typing import Set # pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Set # pylint: disable=unused-import, no-name-in-module
import test_util
class TLSServerTest(unittest.TestCase):
@@ -28,41 +22,14 @@ class TLSServerTest(unittest.TestCase):
from acme.standalone import TLSServer
server = TLSServer(
('', 0), socketserver.BaseRequestHandler, bind_and_activate=True)
server.server_close() # pylint: disable=no-member
server.server_close()
def test_ipv6(self):
if socket.has_ipv6:
from acme.standalone import TLSServer
server = TLSServer(
('', 0), socketserver.BaseRequestHandler, bind_and_activate=True, ipv6=True)
server.server_close() # pylint: disable=no-member
class TLSSNI01ServerTest(unittest.TestCase):
"""Test for acme.standalone.TLSSNI01Server."""
def setUp(self):
self.certs = {b'localhost': (
test_util.load_pyopenssl_private_key('rsa2048_key.pem'),
test_util.load_cert('rsa2048_cert.pem'),
)}
from acme.standalone import TLSSNI01Server
self.server = TLSSNI01Server(('localhost', 0), certs=self.certs)
# pylint: disable=no-member
self.thread = threading.Thread(target=self.server.serve_forever)
self.thread.start()
def tearDown(self):
self.server.shutdown() # pylint: disable=no-member
self.thread.join()
def test_it(self):
host, port = self.server.socket.getsockname()[:2]
cert = crypto_util.probe_sni(
b'localhost', host=host, port=port, timeout=1)
self.assertEqual(jose.ComparableX509(cert),
jose.ComparableX509(self.certs[b'localhost'][1]))
server.server_close()
class HTTP01ServerTest(unittest.TestCase):
@@ -77,13 +44,12 @@ class HTTP01ServerTest(unittest.TestCase):
from acme.standalone import HTTP01Server
self.server = HTTP01Server(('', 0), resources=self.resources)
# pylint: disable=no-member
self.port = self.server.socket.getsockname()[1]
self.thread = threading.Thread(target=self.server.serve_forever)
self.thread.start()
def tearDown(self):
self.server.shutdown() # pylint: disable=no-member
self.server.shutdown()
self.thread.join()
def test_index(self):
@@ -136,7 +102,6 @@ class BaseDualNetworkedServersTest(unittest.TestCase):
# NB: On Windows, socket.IPPROTO_IPV6 constant may be missing.
# We use the corresponding value (41) instead.
level = getattr(socket, "IPPROTO_IPV6", 41)
# pylint: disable=no-member
self.socket.setsockopt(level, socket.IPV6_V6ONLY, 1)
try:
self.server_bind()
@@ -170,33 +135,6 @@ class BaseDualNetworkedServersTest(unittest.TestCase):
prev_port = port
class TLSSNI01DualNetworkedServersTest(unittest.TestCase):
"""Test for acme.standalone.TLSSNI01DualNetworkedServers."""
def setUp(self):
self.certs = {b'localhost': (
test_util.load_pyopenssl_private_key('rsa2048_key.pem'),
test_util.load_cert('rsa2048_cert.pem'),
)}
from acme.standalone import TLSSNI01DualNetworkedServers
self.servers = TLSSNI01DualNetworkedServers(('localhost', 0), certs=self.certs)
self.servers.serve_forever()
def tearDown(self):
self.servers.shutdown_and_server_close()
def test_connect(self):
socknames = self.servers.getsocknames()
# connect to all addresses
for sockname in socknames:
host, port = sockname[:2]
cert = crypto_util.probe_sni(
b'localhost', host=host, port=port, timeout=1)
self.assertEqual(jose.ComparableX509(cert),
jose.ComparableX509(self.certs[b'localhost'][1]))
class HTTP01DualNetworkedServersTest(unittest.TestCase):
"""Tests for acme.standalone.HTTP01DualNetworkedServers."""
@@ -209,7 +147,6 @@ class HTTP01DualNetworkedServersTest(unittest.TestCase):
from acme.standalone import HTTP01DualNetworkedServers
self.servers = HTTP01DualNetworkedServers(('', 0), resources=self.resources)
# pylint: disable=no-member
self.port = self.servers.getsocknames()[0][1]
self.servers.serve_forever()
@@ -248,51 +185,5 @@ class HTTP01DualNetworkedServersTest(unittest.TestCase):
self.assertFalse(self._test_http01(add=False))
@test_util.broken_on_windows
class TestSimpleTLSSNI01Server(unittest.TestCase):
"""Tests for acme.standalone.simple_tls_sni_01_server."""
def setUp(self):
# mirror ../examples/standalone
self.test_cwd = tempfile.mkdtemp()
localhost_dir = os.path.join(self.test_cwd, 'localhost')
os.makedirs(localhost_dir)
shutil.copy(test_util.vector_path('rsa2048_cert.pem'),
os.path.join(localhost_dir, 'cert.pem'))
shutil.copy(test_util.vector_path('rsa2048_key.pem'),
os.path.join(localhost_dir, 'key.pem'))
from acme.standalone import simple_tls_sni_01_server
self.thread = threading.Thread(
target=simple_tls_sni_01_server, kwargs={
'cli_args': ('filename',),
'forever': False,
},
)
self.old_cwd = os.getcwd()
os.chdir(self.test_cwd)
def tearDown(self):
os.chdir(self.old_cwd)
self.thread.join()
shutil.rmtree(self.test_cwd)
@mock.patch('acme.standalone.logger')
def test_it(self, mock_logger):
# Use a Queue because mock objects aren't thread safe.
q = queue.Queue() # type: queue.Queue[int]
# Add port number to the queue.
mock_logger.info.side_effect = lambda *args: q.put(args[-1])
self.thread.start()
# After the timeout, an exception is raised if the queue is empty.
port = q.get(timeout=5)
cert = crypto_util.probe_sni(b'localhost', b'0.0.0.0', port)
self.assertEqual(jose.ComparableX509(cert),
test_util.load_comparable_cert(
'rsa2048_cert.pem'))
if __name__ == "__main__":
unittest.main() # pragma: no cover
acme/acme/test_util.py → acme/tests/test_util.py
+2 -40
View File
@@ -4,20 +4,12 @@
"""
import os
import sys
import pkg_resources
import unittest
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
import josepy as jose
from OpenSSL import crypto
def vector_path(*names):
"""Path to a test vector."""
return pkg_resources.resource_filename(
__name__, os.path.join('testdata', *names))
import pkg_resources
def load_vector(*names):
@@ -33,8 +25,7 @@ def _guess_loader(filename, loader_pem, loader_der):
return loader_pem
elif ext.lower() == '.der':
return loader_der
else: # pragma: no cover
raise ValueError("Loader could not be recognized based on extension")
raise ValueError("Loader could not be recognized based on extension") # pragma: no cover
def load_cert(*names):
@@ -74,32 +65,3 @@ def load_pyopenssl_private_key(*names):
loader = _guess_loader(
names[-1], crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1)
return crypto.load_privatekey(loader, load_vector(*names))
def skip_unless(condition, reason): # pragma: no cover
"""Skip tests unless a condition holds.
This implements the basic functionality of unittest.skipUnless
which is only available on Python 2.7+.
:param bool condition: If ``False``, the test will be skipped
:param str reason: the reason for skipping the test
:rtype: callable
:returns: decorator that hides tests unless condition is ``True``
"""
if hasattr(unittest, "skipUnless"):
return unittest.skipUnless(condition, reason)
elif condition:
return lambda cls: cls
else:
return lambda cls: None
def broken_on_windows(function):
"""Decorator to skip temporarily a broken test on Windows."""
reason = 'Test is broken and ignored on windows but should be fixed.'
return unittest.skipIf(
sys.platform == 'win32'
and os.environ.get('SKIP_BROKEN_TESTS_ON_WINDOWS', 'true') == 'true',
reason)(function)
acme/acme/testdata/README → acme/tests/testdata/README
Vendored
View File
acme/acme/testdata/cert-100sans.pem → acme/tests/testdata/cert-100sans.pem
Vendored
View File
acme/acme/testdata/cert-idnsans.pem → acme/tests/testdata/cert-idnsans.pem
Vendored
View File
acme/acme/testdata/cert-nocn.der → acme/tests/testdata/cert-nocn.der
Vendored
View File
acme/acme/testdata/cert-san.pem → acme/tests/testdata/cert-san.pem
Vendored
View File
acme/acme/testdata/cert.der → acme/tests/testdata/cert.der
Vendored
View File
acme/acme/testdata/cert.pem → acme/tests/testdata/cert.pem
Vendored
View File
acme/acme/testdata/critical-san.pem → acme/tests/testdata/critical-san.pem
Vendored
View File
acme/acme/testdata/csr-100sans.pem → acme/tests/testdata/csr-100sans.pem
Vendored
View File
acme/acme/testdata/csr-6sans.pem → acme/tests/testdata/csr-6sans.pem
Vendored
View File
acme/acme/testdata/csr-idnsans.pem → acme/tests/testdata/csr-idnsans.pem
Vendored
View File
acme/acme/testdata/csr-nosans.pem → acme/tests/testdata/csr-nosans.pem
Vendored
View File
acme/acme/testdata/csr-san.pem → acme/tests/testdata/csr-san.pem
Vendored
View File
acme/acme/testdata/csr.der → acme/tests/testdata/csr.der
Vendored
View File
acme/acme/testdata/csr.pem → acme/tests/testdata/csr.pem
Vendored
View File
acme/acme/testdata/dsa512_key.pem → acme/tests/testdata/dsa512_key.pem
Vendored
View File
acme/acme/testdata/rsa1024_key.pem → acme/tests/testdata/rsa1024_key.pem
Vendored
View File
acme/acme/testdata/rsa2048_cert.pem → acme/tests/testdata/rsa2048_cert.pem
Vendored
View File
acme/acme/testdata/rsa2048_key.pem → acme/tests/testdata/rsa2048_key.pem
Vendored
View File
acme/acme/testdata/rsa256_key.pem → acme/tests/testdata/rsa256_key.pem
Vendored
View File
acme/acme/testdata/rsa512_key.pem → acme/tests/testdata/rsa512_key.pem
Vendored
View File
acme/acme/util_test.py → acme/tests/util_test.py
View File