starred/certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-28 16:54:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

Check_untyped_defs in mypy with clean output for acme (#5874)

Browse Source 
* check_untyped_defs in mypy with clean output for acme

* test entire acme module

* Add typing as a dependency because it's only in the stdlib for 3.5+

* Add str_utils, modified for python2.7 compatibility

* make mypy happy in acme

* typing is needed in prod

* we actually only need typing in acme so far

* add tests and more docs for str_utils

* pragma no cover

* add magic_typing

* s/from typing/from magic_typing/g

* move typing to dev_extras

* correctly set up imports

* remove str_utils

* only type: ignore for OpenSSL.SSL, not crypto

* Since we only run mypy with python3 anyway and we're fine importing it when it's not actually there, there's no actual need for typing to be present as a dependency

* comment magic_typing.py

* disable wildcard-import im magic_typing

* disable pylint errors

* add magic_typing_test

* make magic_typing tests work alongside other tests

* make sure temp_typing is set

* add typing as a dev dependency for python3.4

* run mypy with python3.4 on travis to get a little more testing with different environments

* don't stick typing into sys.modules

* reorder imports
This commit is contained in:
ohemorange
2018-05-03 13:10:33 -07:00
committed by Brad Warren
parent 32e85e9a23
commit 3eaf35f1e2
14 changed files with 121 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+2
View File
@@ -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"
acme/acme/challenges.py
+1 -1
View File
@@ -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)
acme/acme/client.py
+5 -4
View File
@@ -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
acme/acme/client_test.py
+3 -1
View File
@@ -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')
acme/acme/crypto_util.py
+33 -30
View File
@@ -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
acme/acme/crypto_util_test.py
+2 -1
View File
@@ -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)
acme/acme/magic_typing.py
+13
View File
@@ -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()
acme/acme/magic_typing_test.py
+41
View File
@@ -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
acme/acme/messages_test.py
+2 -1
View File
@@ -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')
acme/acme/standalone.py
+5 -4
View File
@@ -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()
acme/acme/standalone_test.py
+3 -2
View File
@@ -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)
acme/acme/test_util.py
+7 -7
View File
@@ -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
mypy.ini
+3
View File
@@ -1,3 +1,6 @@
[mypy]
python_version = 2.7
ignore_missing_imports = True
[mypy-acme.*]
check_untyped_defs = True
setup.py
+1
View File
@@ -67,6 +67,7 @@ dev_extras = [
dev3_extras = [
'mypy',
'typing', # for python3.4
]
docs_extras = [