diff --git a/.travis.yml b/.travis.yml index 370137f68..111ddb3d4 100644 --- a/.travis.yml +++ b/.travis.yml @@ -29,6 +29,8 @@ matrix: addons: - python: "2.7" env: TOXENV=lint + - python: "3.4" + env: TOXENV=mypy - python: "3.5" env: TOXENV=mypy - python: "2.7" diff --git a/acme/acme/challenges.py b/acme/acme/challenges.py index 9702a7a14..b2a4882eb 100644 --- a/acme/acme/challenges.py +++ b/acme/acme/challenges.py @@ -478,7 +478,7 @@ class TLSSNI01Response(KeyAuthorizationChallengeResponse): try: cert = self.probe_cert(domain=domain, **kwargs) except errors.Error as error: - logger.debug(error, exc_info=True) + logger.debug(str(error), exc_info=True) return False return self.verify_cert(cert) diff --git a/acme/acme/client.py b/acme/acme/client.py index 19615b087..7807f0ece 100644 --- a/acme/acme/client.py +++ b/acme/acme/client.py @@ -9,7 +9,6 @@ import time import six from six.moves import http_client # pylint: disable=import-error - import josepy as jose import OpenSSL import re @@ -20,6 +19,8 @@ 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 logger = logging.getLogger(__name__) @@ -415,7 +416,7 @@ class Client(ClientBase): """ # pylint: disable=too-many-locals assert max_attempts > 0 - attempts = collections.defaultdict(int) + attempts = collections.defaultdict(int) # type: Dict[messages.AuthorizationResource, int] exhausted = set() # priority queue with datetime.datetime (based on Retry-After) as key, @@ -529,7 +530,7 @@ class Client(ClientBase): :rtype: `list` of `OpenSSL.crypto.X509` wrapped in `.ComparableX509` """ - chain = [] + chain = [] # type: List[jose.ComparableX509] uri = certr.cert_chain_uri while uri is not None and len(chain) < max_length: response, cert = self._get_cert(uri) @@ -864,7 +865,7 @@ class ClientNetwork(object): # pylint: disable=too-many-instance-attributes self.account = account self.alg = alg self.verify_ssl = verify_ssl - self._nonces = set() + self._nonces = set() # type: Set[Text] self.user_agent = user_agent self.session = requests.Session() self._default_timeout = timeout diff --git a/acme/acme/client_test.py b/acme/acme/client_test.py index be08c2919..c17b83210 100644 --- a/acme/acme/client_test.py +++ b/acme/acme/client_test.py @@ -17,6 +17,7 @@ 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 CERT_DER = test_util.load_vector('cert.der') @@ -61,7 +62,8 @@ class ClientTestBase(unittest.TestCase): self.contact = ('mailto:cert-admin@example.com', 'tel:+12025551212') reg = messages.Registration( contact=self.contact, key=KEY.public_key()) - self.new_reg = messages.NewRegistration(**dict(reg)) + the_arg = dict(reg) # type: Dict + self.new_reg = messages.NewRegistration(**the_arg) # pylint: disable=star-args self.regr = messages.RegistrationResource( body=reg, uri='https://www.letsencrypt-demo.org/acme/reg/1') diff --git a/acme/acme/crypto_util.py b/acme/acme/crypto_util.py index 2281196eb..ad914ca60 100644 --- a/acme/acme/crypto_util.py +++ b/acme/acme/crypto_util.py @@ -6,11 +6,13 @@ import os import re import socket -import OpenSSL +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, Text, Union logger = logging.getLogger(__name__) @@ -25,7 +27,7 @@ logger = logging.getLogger(__name__) # 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 = OpenSSL.SSL.SSLv23_METHOD # type: ignore +_DEFAULT_TLSSNI01_SSL_METHOD = SSL.SSLv23_METHOD # type: ignore class SSLSocket(object): # pylint: disable=too-few-public-methods @@ -64,9 +66,9 @@ class SSLSocket(object): # pylint: disable=too-few-public-methods logger.debug("Server name (%s) not recognized, dropping SSL", server_name) return - new_context = OpenSSL.SSL.Context(self.method) - new_context.set_options(OpenSSL.SSL.OP_NO_SSLv2) - new_context.set_options(OpenSSL.SSL.OP_NO_SSLv3) + 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) @@ -89,18 +91,18 @@ class SSLSocket(object): # pylint: disable=too-few-public-methods def accept(self): # pylint: disable=missing-docstring sock, addr = self.sock.accept() - context = OpenSSL.SSL.Context(self.method) - context.set_options(OpenSSL.SSL.OP_NO_SSLv2) - context.set_options(OpenSSL.SSL.OP_NO_SSLv3) + 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(OpenSSL.SSL.Connection(context, sock)) + 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 OpenSSL.SSL.Error as error: + except SSL.Error as error: # _pick_certificate_cb might have returned without # creating SSL context (wrong server name) raise socket.error(error) @@ -128,7 +130,7 @@ def probe_sni(name, host, port=443, timeout=300, :rtype: OpenSSL.crypto.X509 """ - context = OpenSSL.SSL.Context(method) + context = SSL.Context(method) context.set_timeout(timeout) socket_kwargs = {'source_address': source_address} @@ -145,13 +147,13 @@ def probe_sni(name, host, port=443, timeout=300, raise errors.Error(error) with contextlib.closing(sock) as client: - client_ssl = OpenSSL.SSL.Connection(context, 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 OpenSSL.SSL.Error as error: + except SSL.Error as error: raise errors.Error(error) return client_ssl.get_peer_certificate() @@ -164,18 +166,18 @@ def make_csr(private_key_pem, domains, must_staple=False): OCSP Must Staple: https://tools.ietf.org/html/rfc7633). :returns: buffer PEM-encoded Certificate Signing Request. """ - private_key = OpenSSL.crypto.load_privatekey( - OpenSSL.crypto.FILETYPE_PEM, private_key_pem) - csr = OpenSSL.crypto.X509Req() + private_key = crypto.load_privatekey( + crypto.FILETYPE_PEM, private_key_pem) + csr = crypto.X509Req() extensions = [ - OpenSSL.crypto.X509Extension( + crypto.X509Extension( b'subjectAltName', critical=False, value=', '.join('DNS:' + d for d in domains).encode('ascii') ), ] if must_staple: - extensions.append(OpenSSL.crypto.X509Extension( + extensions.append(crypto.X509Extension( b"1.3.6.1.5.5.7.1.24", critical=False, value=b"DER:30:03:02:01:05")) @@ -183,8 +185,8 @@ def make_csr(private_key_pem, domains, must_staple=False): csr.set_pubkey(private_key) csr.set_version(2) csr.sign(private_key, 'sha256') - return OpenSSL.crypto.dump_certificate_request( - OpenSSL.crypto.FILETYPE_PEM, csr) + 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 @@ -221,11 +223,12 @@ def _pyopenssl_cert_or_req_san(cert_or_req): parts_separator = ", " prefix = "DNS" + part_separator - if isinstance(cert_or_req, OpenSSL.crypto.X509): - func = OpenSSL.crypto.dump_certificate + 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 = OpenSSL.crypto.dump_certificate_request - text = func(OpenSSL.crypto.FILETYPE_TEXT, cert_or_req).decode("utf-8") + 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) @@ -252,12 +255,12 @@ def gen_ss_cert(key, domains, not_before=None, """ assert domains, "Must provide one or more hostnames for the cert." - cert = OpenSSL.crypto.X509() + cert = crypto.X509() cert.set_serial_number(int(binascii.hexlify(os.urandom(16)), 16)) cert.set_version(2) extensions = [ - OpenSSL.crypto.X509Extension( + crypto.X509Extension( b"basicConstraints", True, b"CA:TRUE, pathlen:0"), ] @@ -266,7 +269,7 @@ def gen_ss_cert(key, domains, not_before=None, cert.set_issuer(cert.get_subject()) if force_san or len(domains) > 1: - extensions.append(OpenSSL.crypto.X509Extension( + extensions.append(crypto.X509Extension( b"subjectAltName", critical=False, value=b", ".join(b"DNS:" + d.encode() for d in domains) @@ -281,7 +284,7 @@ def gen_ss_cert(key, domains, not_before=None, cert.sign(key, "sha256") return cert -def dump_pyopenssl_chain(chain, filetype=OpenSSL.crypto.FILETYPE_PEM): +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 @@ -298,7 +301,7 @@ def dump_pyopenssl_chain(chain, filetype=OpenSSL.crypto.FILETYPE_PEM): if isinstance(cert, jose.ComparableX509): # pylint: disable=protected-access cert = cert.wrapped - return OpenSSL.crypto.dump_certificate(filetype, cert) + return crypto.dump_certificate(filetype, cert) # assumes that OpenSSL.crypto.dump_certificate includes ending # newline character diff --git a/acme/acme/crypto_util_test.py b/acme/acme/crypto_util_test.py index 3874ba9d9..36d62b324 100644 --- a/acme/acme/crypto_util_test.py +++ b/acme/acme/crypto_util_test.py @@ -13,6 +13,7 @@ import OpenSSL from acme import errors from acme import test_util +from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module class SSLSocketAndProbeSNITest(unittest.TestCase): @@ -165,7 +166,7 @@ class RandomSnTest(unittest.TestCase): def setUp(self): self.cert_count = 5 - self.serial_num = [] + self.serial_num = [] # type: List[int] self.key = OpenSSL.crypto.PKey() self.key.generate_key(OpenSSL.crypto.TYPE_RSA, 2048) diff --git a/acme/acme/magic_typing.py b/acme/acme/magic_typing.py new file mode 100644 index 000000000..555088cf2 --- /dev/null +++ b/acme/acme/magic_typing.py @@ -0,0 +1,13 @@ +"""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): + return None + +try: + # mypy doesn't respect modifying sys.modules + from typing import * # pylint: disable=wildcard-import, unused-wildcard-import +except ImportError: + sys.modules[__name__] = TypingClass() diff --git a/acme/acme/magic_typing_test.py b/acme/acme/magic_typing_test.py new file mode 100644 index 000000000..23dfe3367 --- /dev/null +++ b/acme/acme/magic_typing_test.py @@ -0,0 +1,41 @@ +"""Tests for acme.magic_typing.""" +import sys +import unittest + +import mock + + +class MagicTypingTest(unittest.TestCase): + """Tests for acme.magic_typing.""" + def test_import_success(self): + try: + import typing as temp_typing + except ImportError: # pragma: no cover + temp_typing = None # pragma: no cover + typing_class_mock = mock.MagicMock() + text_mock = mock.MagicMock() + typing_class_mock.Text = text_mock + sys.modules['typing'] = typing_class_mock + if 'acme.magic_typing' in sys.modules: + del sys.modules['acme.magic_typing'] # pragma: no cover + from acme.magic_typing import Text # pylint: disable=no-name-in-module + self.assertEqual(Text, text_mock) + del sys.modules['acme.magic_typing'] + sys.modules['typing'] = temp_typing + + def test_import_failure(self): + try: + import typing as temp_typing + except ImportError: # pragma: no cover + temp_typing = None # pragma: no cover + sys.modules['typing'] = None + if 'acme.magic_typing' in sys.modules: + del sys.modules['acme.magic_typing'] # pragma: no cover + from acme.magic_typing import Text # pylint: disable=no-name-in-module + self.assertTrue(Text is None) + del sys.modules['acme.magic_typing'] + sys.modules['typing'] = temp_typing + + +if __name__ == '__main__': + unittest.main() # pragma: no cover diff --git a/acme/acme/messages_test.py b/acme/acme/messages_test.py index 64bc81efd..0e2d8c62d 100644 --- a/acme/acme/messages_test.py +++ b/acme/acme/messages_test.py @@ -6,6 +6,7 @@ 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 CERT = test_util.load_comparable_cert('cert.der') @@ -85,7 +86,7 @@ class ConstantTest(unittest.TestCase): from acme.messages import _Constant class MockConstant(_Constant): # pylint: disable=missing-docstring - POSSIBLE_NAMES = {} + POSSIBLE_NAMES = {} # type: Dict self.MockConstant = MockConstant # pylint: disable=invalid-name self.const_a = MockConstant('a') diff --git a/acme/acme/standalone.py b/acme/acme/standalone.py index c221f5883..a370501ee 100644 --- a/acme/acme/standalone.py +++ b/acme/acme/standalone.py @@ -16,6 +16,7 @@ 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 logger = logging.getLogger(__name__) @@ -66,8 +67,8 @@ class BaseDualNetworkedServers(object): def __init__(self, ServerClass, server_address, *remaining_args, **kwargs): port = server_address[1] - self.threads = [] - self.servers = [] + self.threads = [] # type: List[threading.Thread] + self.servers = [] # type: List[ACMEServerMixin] # Must try True first. # Ubuntu, for example, will fail to bind to IPv4 if we've already bound @@ -189,7 +190,7 @@ class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): def __init__(self, *args, **kwargs): self.simple_http_resources = kwargs.pop("simple_http_resources", set()) - socketserver.BaseRequestHandler.__init__(self, *args, **kwargs) + BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, *args, **kwargs) def log_message(self, format, *args): # pylint: disable=redefined-builtin """Log arbitrary message.""" @@ -262,7 +263,7 @@ def simple_tls_sni_01_server(cli_args, forever=True): certs = {} - _, hosts, _ = next(os.walk('.')) + _, 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() diff --git a/acme/acme/standalone_test.py b/acme/acme/standalone_test.py index 5b1b72c18..1591187e5 100644 --- a/acme/acme/standalone_test.py +++ b/acme/acme/standalone_test.py @@ -18,6 +18,7 @@ from acme import challenges from acme import crypto_util from acme import errors from acme import test_util +from acme.magic_typing import Set # pylint: disable=unused-import, no-name-in-module class TLSServerTest(unittest.TestCase): @@ -72,7 +73,7 @@ class HTTP01ServerTest(unittest.TestCase): def setUp(self): self.account_key = jose.JWK.load( test_util.load_vector('rsa1024_key.pem')) - self.resources = set() + self.resources = set() # type: Set from acme.standalone import HTTP01Server self.server = HTTP01Server(('', 0), resources=self.resources) @@ -201,7 +202,7 @@ class HTTP01DualNetworkedServersTest(unittest.TestCase): def setUp(self): self.account_key = jose.JWK.load( test_util.load_vector('rsa1024_key.pem')) - self.resources = set() + self.resources = set() # type: Set from acme.standalone import HTTP01DualNetworkedServers self.servers = HTTP01DualNetworkedServers(('', 0), resources=self.resources) diff --git a/acme/acme/test_util.py b/acme/acme/test_util.py index d5d57bd27..1a0b67056 100644 --- a/acme/acme/test_util.py +++ b/acme/acme/test_util.py @@ -10,7 +10,7 @@ import unittest from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization import josepy as jose -import OpenSSL +from OpenSSL import crypto def vector_path(*names): @@ -39,8 +39,8 @@ def _guess_loader(filename, loader_pem, loader_der): def load_cert(*names): """Load certificate.""" loader = _guess_loader( - names[-1], OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.FILETYPE_ASN1) - return OpenSSL.crypto.load_certificate(loader, load_vector(*names)) + names[-1], crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1) + return crypto.load_certificate(loader, load_vector(*names)) def load_comparable_cert(*names): @@ -51,8 +51,8 @@ def load_comparable_cert(*names): def load_csr(*names): """Load certificate request.""" loader = _guess_loader( - names[-1], OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.FILETYPE_ASN1) - return OpenSSL.crypto.load_certificate_request(loader, load_vector(*names)) + names[-1], crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1) + return crypto.load_certificate_request(loader, load_vector(*names)) def load_comparable_csr(*names): @@ -71,8 +71,8 @@ def load_rsa_private_key(*names): def load_pyopenssl_private_key(*names): """Load pyOpenSSL private key.""" loader = _guess_loader( - names[-1], OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.FILETYPE_ASN1) - return OpenSSL.crypto.load_privatekey(loader, load_vector(*names)) + names[-1], crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1) + return crypto.load_privatekey(loader, load_vector(*names)) def skip_unless(condition, reason): # pragma: no cover diff --git a/mypy.ini b/mypy.ini index aac60a2ad..2bac9249f 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1,3 +1,6 @@ [mypy] python_version = 2.7 ignore_missing_imports = True + +[mypy-acme.*] +check_untyped_defs = True diff --git a/setup.py b/setup.py index e674871a8..3760fd35b 100644 --- a/setup.py +++ b/setup.py @@ -67,6 +67,7 @@ dev_extras = [ dev3_extras = [ 'mypy', + 'typing', # for python3.4 ] docs_extras = [