Convert Python 2 type hints to Python 3 types annotations (#8640)

Fixes #8427

This PR converts the Python 2 types hints into Python 3 types annotations. I have used the project https://github.com/ilevkivskyi/com2ann which has been designed for that specific purpose and did that very well.

The only remaining things to do were to fix broken type hints that became wrong code after migration, and to fix lines too long with the new syntax.

* Raw execution of com2ann

* Fixing broken type annotations

* Cleanup imports

acme/acme/challenges.py

@@ -24,7 +24,7 @@ logger = logging.getLogger(__name__)
 class Challenge(jose.TypedJSONObjectWithFields):
     # _fields_to_partial_json
     """ACME challenge."""
-    TYPES = {}  # type: dict
+    TYPES: dict = {}
 
     @classmethod
     def from_json(cls, jobj):
@@ -38,7 +38,7 @@ class Challenge(jose.TypedJSONObjectWithFields):
 class ChallengeResponse(ResourceMixin, TypeMixin, jose.TypedJSONObjectWithFields):
     # _fields_to_partial_json
     """ACME challenge response."""
-    TYPES = {}  # type: dict
+    TYPES: dict = {}
     resource_type = 'challenge'
     resource = fields.Resource(resource_type)
 

acme/acme/client.py

@@ -112,8 +112,9 @@ class ClientBase:
         """
         return self.update_registration(regr, update={'status': 'deactivated'})
 
-    def deactivate_authorization(self, authzr):
+    def deactivate_authorization(self,
-        # type: (messages.AuthorizationResource) -> messages.AuthorizationResource
+                                 authzr: messages.AuthorizationResource
+                                 ) -> messages.AuthorizationResource:
         """Deactivate authorization.
 
         :param messages.AuthorizationResource authzr: The Authorization resource
@@ -423,7 +424,7 @@ class Client(ClientBase):
 
         """
         assert max_attempts > 0
-        attempts = collections.defaultdict(int) # type: Dict[messages.AuthorizationResource, int]
+        attempts: Dict[messages.AuthorizationResource, int] = collections.defaultdict(int)
         exhausted = set()
 
         # priority queue with datetime.datetime (based on Retry-After) as key,
@@ -536,7 +537,7 @@ class Client(ClientBase):
         :rtype: `list` of `OpenSSL.crypto.X509` wrapped in `.ComparableX509`
 
         """
-        chain = [] # type: List[jose.ComparableX509]
+        chain: List[jose.ComparableX509] = []
         uri = certr.cert_chain_uri
         while uri is not None and len(chain) < max_length:
             response, cert = self._get_cert(uri)
@@ -968,7 +969,7 @@ class ClientNetwork:
         self.account = account
         self.alg = alg
         self.verify_ssl = verify_ssl
-        self._nonces = set() # type: Set[Text]
+        self._nonces: Set[Text] = set()
         self.user_agent = user_agent
         self.session = requests.Session()
         self._default_timeout = timeout

acme/acme/crypto_util.py

@@ -168,7 +168,7 @@ def probe_sni(name, host, port=443, timeout=300, # pylint: disable=too-many-argu
                 source_address[1]
             ) if any(source_address) else ""
         )
-        socket_tuple = (host, port)  # type: Tuple[str, int]
+        socket_tuple: Tuple[str, int] = (host, port)
         sock = socket.create_connection(socket_tuple, **socket_kwargs)  # type: ignore
     except socket.error as error:
         raise errors.Error(error)
@@ -256,7 +256,7 @@ def _pyopenssl_cert_or_req_san(cert_or_req):
 
     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]]
+        func: Union[Callable[[int, crypto.X509Req], bytes], Callable[[int, crypto.X509], bytes]] = crypto.dump_certificate
     else:
         func = crypto.dump_certificate_request
     text = func(crypto.FILETYPE_TEXT, cert_or_req).decode("utf-8")

acme/acme/messages.py

@@ -153,7 +153,7 @@ class _Constant(jose.JSONDeSerializable, Hashable):  # type: ignore
 
 class Status(_Constant):
     """ACME "status" field."""
-    POSSIBLE_NAMES = {}  # type: dict
+    POSSIBLE_NAMES: dict = {}
 STATUS_UNKNOWN = Status('unknown')
 STATUS_PENDING = Status('pending')
 STATUS_PROCESSING = Status('processing')
@@ -166,7 +166,7 @@ STATUS_DEACTIVATED = Status('deactivated')
 
 class IdentifierType(_Constant):
     """ACME identifier type."""
-    POSSIBLE_NAMES = {}  # type: dict
+    POSSIBLE_NAMES: dict = {}
 IDENTIFIER_FQDN = IdentifierType('dns')  # IdentifierDNS in Boulder
 
 
@@ -184,7 +184,7 @@ class Identifier(jose.JSONObjectWithFields):
 class Directory(jose.JSONDeSerializable):
     """Directory."""
 
-    _REGISTERED_TYPES = {}  # type: dict
+    _REGISTERED_TYPES: dict = {}
 
     class Meta(jose.JSONObjectWithFields):
         """Directory Meta."""

acme/acme/standalone.py

@@ -63,8 +63,8 @@ class BaseDualNetworkedServers:
 
     def __init__(self, ServerClass, server_address, *remaining_args, **kwargs):
         port = server_address[1]
-        self.threads = [] # type: List[threading.Thread]
+        self.threads: List[threading.Thread] = []
-        self.servers = [] # type: List[ACMEServerMixin]
+        self.servers: List[ACMEServerMixin] = []
 
         # Must try True first.
         # Ubuntu, for example, will fail to bind to IPv4 if we've already bound

acme/tests/client_test.py

@@ -61,7 +61,7 @@ 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
+        the_arg: Dict = dict(reg)
         self.new_reg = messages.NewRegistration(**the_arg)
         self.regr = messages.RegistrationResource(
             body=reg, uri='https://www.letsencrypt-demo.org/acme/reg/1')

acme/tests/crypto_util_test.py

@@ -180,7 +180,7 @@ class RandomSnTest(unittest.TestCase):
 
     def setUp(self):
         self.cert_count = 5
-        self.serial_num = [] # type: List[int]
+        self.serial_num: List[int] = []
         self.key = OpenSSL.crypto.PKey()
         self.key.generate_key(OpenSSL.crypto.TYPE_RSA, 2048)
 

acme/tests/messages_test.py

@@ -1,4 +1,5 @@
 """Tests for acme.messages."""
+from typing import Dict
 import unittest
 from unittest import mock
 
@@ -81,7 +82,7 @@ class ConstantTest(unittest.TestCase):
         from acme.messages import _Constant
 
         class MockConstant(_Constant):  # pylint: disable=missing-docstring
-            POSSIBLE_NAMES = {} # type: Dict
+            POSSIBLE_NAMES: Dict = {}
 
         self.MockConstant = MockConstant  # pylint: disable=invalid-name
         self.const_a = MockConstant('a')

acme/tests/standalone_test.py

@@ -41,7 +41,7 @@ class HTTP01ServerTest(unittest.TestCase):
     def setUp(self):
         self.account_key = jose.JWK.load(
             test_util.load_vector('rsa1024_key.pem'))
-        self.resources = set() # type: Set
+        self.resources: Set = set()
 
         from acme.standalone import HTTP01Server
         self.server = HTTP01Server(('', 0), resources=self.resources)
@@ -218,7 +218,7 @@ class HTTP01DualNetworkedServersTest(unittest.TestCase):
     def setUp(self):
         self.account_key = jose.JWK.load(
             test_util.load_vector('rsa1024_key.pem'))
-        self.resources = set() # type: Set
+        self.resources: Set = set()
 
         from acme.standalone import HTTP01DualNetworkedServers
         self.servers = HTTP01DualNetworkedServers(('', 0), resources=self.resources)

certbot-apache/certbot_apache/_internal/augeasparser.py

@@ -355,7 +355,7 @@ class AugeasBlockNode(AugeasDirectiveNode):
         ownpath = self.metadata.get("augeaspath")
 
         directives = self.parser.find_dir(name, start=ownpath, exclude=exclude)
-        already_parsed = set()  # type: Set[str]
+        already_parsed: Set[str] = set()
         for directive in directives:
             # Remove the /arg part from the Augeas path
             directive = directive.partition("/arg")[0]

certbot-apache/certbot_apache/_internal/configurator.py

@@ -16,11 +16,6 @@ from typing import Union
 
 import zope.component
 import zope.interface
-try:
-    import apacheconfig
-    HAS_APACHECONFIG = True
-except ImportError:  # pragma: no cover
-    HAS_APACHECONFIG = False
 
 from acme import challenges
 from certbot import errors
@@ -41,6 +36,12 @@ from certbot_apache._internal import http_01
 from certbot_apache._internal import obj
 from certbot_apache._internal import parser
 
+try:
+    import apacheconfig
+    HAS_APACHECONFIG = True
+except ImportError:  # pragma: no cover
+    HAS_APACHECONFIG = False
+
 
 logger = logging.getLogger(__name__)
 
@@ -211,23 +212,23 @@ class ApacheConfigurator(common.Installer):
         super(ApacheConfigurator, self).__init__(*args, **kwargs)
 
         # Add name_server association dict
-        self.assoc = {}  # type: Dict[str, obj.VirtualHost]
+        self.assoc: Dict[str, obj.VirtualHost] = {}
         # Outstanding challenges
-        self._chall_out = set()  # type: Set[KeyAuthorizationAnnotatedChallenge]
+        self._chall_out: Set[KeyAuthorizationAnnotatedChallenge] = set()
         # List of vhosts configured per wildcard domain on this run.
         # used by deploy_cert() and enhance()
-        self._wildcard_vhosts = {}  # type: Dict[str, List[obj.VirtualHost]]
+        self._wildcard_vhosts: Dict[str, List[obj.VirtualHost]] = {}
         # Maps enhancements to vhosts we've enabled the enhancement for
-        self._enhanced_vhosts = defaultdict(set)  # type: DefaultDict[str, Set[obj.VirtualHost]]
+        self._enhanced_vhosts: DefaultDict[str, Set[obj.VirtualHost]] = defaultdict(set)
         # Temporary state for AutoHSTS enhancement
-        self._autohsts = {}  # type: Dict[str, Dict[str, Union[int, float]]]
+        self._autohsts: Dict[str, Dict[str, Union[int, float]]] = {}
         # Reverter save notes
         self.save_notes = ""
         # Should we use ParserNode implementation instead of the old behavior
         self.USE_PARSERNODE = use_parsernode
         # Saves the list of file paths that were parsed initially, and
         # not added to parser tree by self.conf("vhost-root") for example.
-        self.parsed_paths = []  # type: List[str]
+        self.parsed_paths: List[str] = []
         # These will be set in the prepare function
         self._prepared = False
         self.parser = None
@@ -833,7 +834,7 @@ class ApacheConfigurator(common.Installer):
         :rtype: set
 
         """
-        all_names = set()  # type: Set[str]
+        all_names: Set[str] = set()
 
         vhost_macro = []
 
@@ -997,8 +998,8 @@ class ApacheConfigurator(common.Installer):
 
         """
         # Search base config, and all included paths for VirtualHosts
-        file_paths = {}  # type: Dict[str, str]
+        file_paths: Dict[str, str] = {}
-        internal_paths = defaultdict(set)  # type: DefaultDict[str, Set[str]]
+        internal_paths: DefaultDict[str, Set[str]] = defaultdict(set)
         vhs = []
         # Make a list of parser paths because the parser_paths
         # dictionary may be modified during the loop.
@@ -2157,7 +2158,7 @@ class ApacheConfigurator(common.Installer):
         # There can be other RewriteRule directive lines in vhost config.
         # rewrite_args_dict keys are directive ids and the corresponding value
         # for each is a list of arguments to that directive.
-        rewrite_args_dict = defaultdict(list)  # type: DefaultDict[str, List[str]]
+        rewrite_args_dict: DefaultDict[str, List[str]] = defaultdict(list)
         pat = r'(.*directive\[\d+\]).*'
         for match in rewrite_path:
             m = re.match(pat, match)
@@ -2251,7 +2252,7 @@ class ApacheConfigurator(common.Installer):
         if ssl_vhost.aliases:
             serveralias = "ServerAlias " + " ".join(ssl_vhost.aliases)
 
-        rewrite_rule_args = []  # type: List[str]
+        rewrite_rule_args: List[str] = []
         if self.get_version() >= (2, 3, 9):
             rewrite_rule_args = constants.REWRITE_HTTPS_ARGS_WITH_END
         else:

certbot-apache/certbot_apache/_internal/http_01.py

@@ -57,7 +57,7 @@ class ApacheHttp01(common.ChallengePerformer):
         self.challenge_dir = os.path.join(
             self.configurator.config.work_dir,
             "http_challenges")
-        self.moded_vhosts = set()  # type: Set[VirtualHost]
+        self.moded_vhosts: Set[VirtualHost] = set()
 
     def perform(self):
         """Perform all HTTP-01 challenges."""
@@ -93,7 +93,7 @@ class ApacheHttp01(common.ChallengePerformer):
                     self.configurator.enable_mod(mod, temp=True)
 
     def _mod_config(self):
-        selected_vhosts = []  # type: List[VirtualHost]
+        selected_vhosts: List[VirtualHost] = []
         http_port = str(self.configurator.config.http01_port)
         for chall in self.achalls:
             # Search for matching VirtualHosts

certbot-apache/certbot_apache/_internal/obj.py

@@ -137,7 +137,7 @@ class VirtualHost:
 
     def get_names(self):
         """Return a set of all names."""
-        all_names = set()  # type: Set[str]
+        all_names: Set[str] = set()
         all_names.update(self.aliases)
         # Strip out any scheme:// and <port> field from servername
         if self.name is not None:
@@ -245,7 +245,7 @@ class VirtualHost:
 
         # already_found acts to keep everything very conservative.
         # Don't allow multiple ip:ports in same set.
-        already_found = set()  # type: Set[str]
+        already_found: Set[str] = set()
 
         for addr in vhost.addrs:
             for local_addr in self.addrs:

certbot-apache/certbot_apache/_internal/override_centos.py

@@ -102,9 +102,9 @@ class CentOSConfigurator(configurator.ApacheConfigurator):
 
         loadmods = self.parser.find_dir("LoadModule", "ssl_module", exclude=False)
 
-        correct_ifmods = []  # type: List[str]
+        correct_ifmods: List[str] = []
-        loadmod_args = []  # type: List[str]
+        loadmod_args: List[str] = []
-        loadmod_paths = []  # type: List[str]
+        loadmod_paths: List[str] = []
         for m in loadmods:
             noarg_path = m.rpartition("/")[0]
             path_args = self.parser.get_all_args(noarg_path)

certbot-apache/certbot_apache/_internal/parser.py

@@ -6,7 +6,6 @@ import re
 from typing import Dict
 from typing import List
 
-
 from certbot import errors
 from certbot.compat import os
 from certbot_apache._internal import apache_util
@@ -49,9 +48,9 @@ class ApacheParser:
                 "version 1.2.0 or higher, please make sure you have you have "
                 "those installed.")
 
-        self.modules = {}  # type: Dict[str, str]
+        self.modules: Dict[str, str] = {}
-        self.parser_paths = {}  # type: Dict[str, List[str]]
+        self.parser_paths: Dict[str, List[str]] = {}
-        self.variables = {}  # type: Dict[str, str]
+        self.variables: Dict[str, str] = {}
 
         # Find configuration root and make sure augeas can parse it.
         self.root = os.path.abspath(root)
@@ -264,7 +263,7 @@ class ApacheParser:
             the iteration issue.  Else... parse and enable mods at same time.
 
         """
-        mods = {}  # type: Dict[str, str]
+        mods: Dict[str, str] = {}
         matches = self.find_dir("LoadModule")
         iterator = iter(matches)
         # Make sure prev_size != cur_size for do: while: iteration
@@ -551,7 +550,7 @@ class ApacheParser:
         else:
             arg_suffix = "/*[self::arg=~regexp('%s')]" % case_i(arg)
 
-        ordered_matches = []  # type: List[str]
+        ordered_matches: List[str] = []
 
         # TODO: Wildcards should be included in alphabetical order
         # https://httpd.apache.org/docs/2.4/mod/core.html#include

certbot-apache/tests/augeasnode_test.py

@@ -107,7 +107,7 @@ class AugeasParserNodeTest(util.ApacheTest):  # pylint: disable=too-many-public-
 
     def test_set_parameters(self):
         servernames = self.config.parser_root.find_directives("servername")
-        names = []  # type: List[str]
+        names: List[str] = []
         for servername in servernames:
             names += servername.parameters
         self.assertFalse("going_to_set_this" in names)

certbot-apache/tests/http_01_test.py

@@ -26,7 +26,7 @@ class ApacheHttp01Test(util.ApacheTest):
         super(ApacheHttp01Test, self).setUp(*args, **kwargs)
 
         self.account_key = self.rsa512jwk
-        self.achalls = []  # type: List[achallenges.KeyAuthorizationAnnotatedChallenge]
+        self.achalls: List[achallenges.KeyAuthorizationAnnotatedChallenge] = []
         vh_truth = util.get_vh_truth(
             self.temp_dir, "debian_apache_2_4/multiple_vhosts")
         # Takes the vhosts for encryption-example.demo, certbot.demo

certbot-ci/certbot_integration_tests/nginx_tests/test_main.py

@@ -32,8 +32,8 @@ def test_context(request):
         '--preferred-challenges', 'http'
     ], {'default_server': False}),
 ], indirect=['context'])
-def test_certificate_deployment(certname_pattern, params, context):
+def test_certificate_deployment(certname_pattern: str, params: List[str],
-    # type: (str, List[str], nginx_context.IntegrationTestsContext) -> None
+                                context: nginx_context.IntegrationTestsContext) -> None:
     """
     Test various scenarios to deploy a certificate to nginx using certbot.
     """

certbot-ci/certbot_integration_tests/utils/acme_server.py

@@ -51,7 +51,7 @@ class ACMEServer:
         self._acme_type = 'pebble' if acme_server == 'pebble' else 'boulder'
         self._proxy = http_proxy
         self._workspace = tempfile.mkdtemp()
-        self._processes = []  # type: List[subprocess.Popen]
+        self._processes: List[subprocess.Popen] = []
         self._stdout = sys.stdout if stdout else open(os.devnull, 'w')
         self._dns_server = dns_server
         self._http_01_port = http_01_port

certbot-ci/certbot_integration_tests/utils/dns_server.py

@@ -38,7 +38,7 @@ class DNSServer:
 
         self.bind_root = tempfile.mkdtemp()
 
-        self.process = None  # type: subprocess.Popen
+        self.process: subprocess.Popen = None
 
         self.dns_xdist = {"address": BIND_BIND_ADDRESS[0], "port": BIND_BIND_ADDRESS[1]}
 
@@ -111,8 +111,7 @@ class DNSServer:
             self.stop()
             raise
 
-    def _wait_until_ready(self, attempts=30):
+    def _wait_until_ready(self, attempts: int = 30) -> None:
-        # type: (int) -> None
         """
         Polls the DNS server over TCP until it gets a response, or until
         it runs out of attempts and raises a ValueError.

certbot-compatibility-test/certbot_compatibility_test/configurators/nginx/common.py

@@ -68,7 +68,7 @@ def _get_server_root(config):
 
 def _get_names(config):
     """Returns all and testable domain names in config"""
-    all_names = set()  # type: Set[str]
+    all_names: Set[str] = set()
     for root, _dirs, files in os.walk(config):
         for this_file in files:
             update_names = _get_server_names(root, this_file)

certbot-compatibility-test/certbot_compatibility_test/test_driver.py

@@ -178,7 +178,7 @@ def test_enhancements(plugin, domains):
                      "enhancements")
         return False
 
-    domains_and_info = [(domain, []) for domain in domains]  # type: List[Tuple[str, List[bool]]]
+    domains_and_info: List[Tuple[str, List[bool]]] = [(domain, []) for domain in domains]
 
     for domain, info in domains_and_info:
         try:

certbot-dns-cloudflare/certbot_dns_cloudflare/_internal/dns_cloudflare.py

@@ -172,7 +172,7 @@ class _CloudflareClient:
         """
 
         zone_name_guesses = dns_common.base_domain_name_guesses(domain)
-        zones = []  # type: List[Dict[str, Any]]
+        zones: List[Dict[str, Any]] = []
         code = msg = None
 
         for zone_name in zone_name_guesses:

certbot-dns-route53/certbot_dns_route53/_internal/dns_route53.py

@@ -39,7 +39,7 @@ class Authenticator(dns_common.DNSAuthenticator):
     def __init__(self, *args, **kwargs):
         super(Authenticator, self).__init__(*args, **kwargs)
         self.r53 = boto3.client("route53")
-        self._resource_records = collections.defaultdict(list) # type: DefaultDict[str, List[Dict[str, str]]]
+        self._resource_records: DefaultDict[str, List[Dict[str, str]]] = collections.defaultdict(list)
 
     def more_info(self):  # pylint: disable=missing-function-docstring
         return "Solve a DNS01 challenge using AWS Route53"

certbot-nginx/certbot_nginx/_internal/configurator.py

@@ -112,8 +112,8 @@ class NginxConfigurator(common.Installer):
 
         # List of vhosts configured per wildcard domain on this run.
         # used by deploy_cert() and enhance()
-        self._wildcard_vhosts = {} # type: Dict[str, List[obj.VirtualHost]]
+        self._wildcard_vhosts: Dict[str, List[obj.VirtualHost]] = {}
-        self._wildcard_redirect_vhosts = {} # type: Dict[str, List[obj.VirtualHost]]
+        self._wildcard_redirect_vhosts: Dict[str, List[obj.VirtualHost]] = {}
 
         # Add number of outstanding challenges
         self._chall_out = 0
@@ -641,7 +641,7 @@ class NginxConfigurator(common.Installer):
         :rtype: set
 
         """
-        all_names = set()  # type: Set[str]
+        all_names: Set[str] = set()
 
         for vhost in self.parser.get_vhosts():
             try:
@@ -1222,7 +1222,7 @@ def nginx_restart(nginx_ctl, nginx_conf, sleep_duration):
 
     """
     try:
-        reload_output = u"" # type: Text
+        reload_output: Text = u""
         with tempfile.TemporaryFile() as out:
             proc = subprocess.Popen([nginx_ctl, "-c", nginx_conf, "-s", "reload"],
                                     env=util.env_no_snap_for_external_calls(),

certbot-nginx/certbot_nginx/_internal/constants.py

@@ -14,11 +14,11 @@ elif platform.system() in ('NetBSD',):
 else:
     server_root_tmp = LINUX_SERVER_ROOT
 
-CLI_DEFAULTS = dict(
+CLI_DEFAULTS: Dict[str, Any] = dict(
     server_root=server_root_tmp,
     ctl="nginx",
     sleep_seconds=1
-) # type: Dict[str, Any]
+)
 """CLI defaults."""
 
 

certbot-nginx/certbot_nginx/_internal/http_01.py

@@ -113,7 +113,7 @@ class NginxHttp01(common.ChallengePerformer):
         :returns: list of :class:`certbot_nginx._internal.obj.Addr` to apply
         :rtype: list
         """
-        addresses = [] # type: List[obj.Addr]
+        addresses: List[obj.Addr] = []
         default_addr = "%s" % self.configurator.config.http01_port
         ipv6_addr = "[::]:{0}".format(
             self.configurator.config.http01_port)

certbot-nginx/certbot_nginx/_internal/nginxparser.py

@@ -2,8 +2,8 @@
 # Forked from https://github.com/fatiherikli/nginxparser (MIT Licensed)
 import copy
 import logging
-from typing import Any  # pylint: disable=unused-import
+from typing import Any
-from typing import IO  # pylint: disable=unused-import
+from typing import IO
 
 from pyparsing import Combine
 from pyparsing import Forward
@@ -20,6 +20,7 @@ from pyparsing import ZeroOrMore
 
 logger = logging.getLogger(__name__)
 
+
 class RawNginxParser:
     # pylint: disable=pointless-statement
     """A class that parses nginx configuration with pyparsing."""
@@ -105,57 +106,9 @@ class RawNginxDumper:
         return ''.join(self)
 
 
-# Shortcut functions to respect Python's serialization interface
-# (like pyyaml, picker or json)
-
-def loads(source):
-    """Parses from a string.
-
-    :param str source: The string to parse
-    :returns: The parsed tree
-    :rtype: list
-
-    """
-    return UnspacedList(RawNginxParser(source).as_list())
-
-
-def load(_file):
-    """Parses from a file.
-
-    :param file _file: The file to parse
-    :returns: The parsed tree
-    :rtype: list
-
-    """
-    return loads(_file.read())
-
-
-def dumps(blocks):
-    # type: (UnspacedList) -> str
-    """Dump to a Unicode string.
-
-    :param UnspacedList block: The parsed tree
-    :rtype: str
-
-    """
-    return str(RawNginxDumper(blocks.spaced))
-
-
-def dump(blocks, _file):
-    # type: (UnspacedList, IO[Any]) -> None
-    """Dump to a file.
-
-    :param UnspacedList block: The parsed tree
-    :param IO[Any] _file: The file stream to dump to. It must be opened with
-                          Unicode encoding.
-    :rtype: None
-
-    """
-    _file.write(dumps(blocks))
-
-
 spacey = lambda x: (isinstance(x, str) and x.isspace()) or x == ''
 
+
 class UnspacedList(list):
     """Wrap a list [of lists], making any whitespace entries magically invisible"""
 
@@ -274,3 +227,50 @@ class UnspacedList(list):
                 idx -= 1
             pos += 1
         return idx0 + spaces
+
+
+# Shortcut functions to respect Python's serialization interface
+# (like pyyaml, picker or json)
+
+def loads(source):
+    """Parses from a string.
+
+    :param str source: The string to parse
+    :returns: The parsed tree
+    :rtype: list
+
+    """
+    return UnspacedList(RawNginxParser(source).as_list())
+
+
+def load(_file):
+    """Parses from a file.
+
+    :param file _file: The file to parse
+    :returns: The parsed tree
+    :rtype: list
+
+    """
+    return loads(_file.read())
+
+
+def dumps(blocks: UnspacedList) -> str:
+    """Dump to a Unicode string.
+
+    :param UnspacedList block: The parsed tree
+    :rtype: six.text_type
+
+    """
+    return str(RawNginxDumper(blocks.spaced))
+
+
+def dump(blocks: UnspacedList, _file: IO[Any]) -> None:
+    """Dump to a file.
+
+    :param UnspacedList block: The parsed tree
+    :param IO[Any] _file: The file stream to dump to. It must be opened with
+                          Unicode encoding.
+    :rtype: None
+
+    """
+    _file.write(dumps(blocks))

certbot-nginx/certbot_nginx/_internal/parser.py

@@ -32,7 +32,7 @@ class NginxParser:
     """
 
     def __init__(self, root):
-        self.parsed = {} # type: Dict[str, Union[List, nginxparser.UnspacedList]]
+        self.parsed: Dict[str, Union[List, nginxparser.UnspacedList]] = {}
         self.root = os.path.abspath(root)
         self.config_root = self._find_config_root()
 
@@ -94,7 +94,7 @@ class NginxParser:
         """
         servers = self._get_raw_servers()
 
-        addr_to_ssl = {} # type: Dict[Tuple[str, str], bool]
+        addr_to_ssl: Dict[Tuple[str, str], bool] = {}
         for filename in servers:
             for server, _ in servers[filename]:
                 # Parse the server block to save addr info
@@ -106,12 +106,11 @@ class NginxParser:
                     addr_to_ssl[addr_tuple] = addr.ssl or addr_to_ssl[addr_tuple]
         return addr_to_ssl
 
-    def _get_raw_servers(self):
+    def _get_raw_servers(self) -> Dict:
         # pylint: disable=cell-var-from-loop
-        # type: () -> Dict
         """Get a map of unparsed all server blocks
         """
-        servers = {} # type: Dict[str, Union[List, nginxparser.UnspacedList]]
+        servers: Dict[str, Union[List, nginxparser.UnspacedList]] = {}
         for filename in self.parsed:
             tree = self.parsed[filename]
             servers[filename] = []
@@ -741,9 +740,9 @@ def _parse_server_raw(server):
     :rtype: dict
 
     """
-    addrs = set() # type: Set[obj.Addr]
+    addrs: Set[obj.Addr] = set()
-    ssl = False # type: bool
+    ssl: bool = False
-    names = set() # type: Set[str]
+    names: Set[str] = set()
 
     apply_ssl_to_all_addrs = False
 

certbot-nginx/certbot_nginx/_internal/parser_obj.py

@@ -5,7 +5,6 @@ import abc
 import logging
 from typing import List
 
-
 from certbot import errors
 
 logger = logging.getLogger(__name__)
@@ -23,7 +22,7 @@ class Parsable:
     __metaclass__ = abc.ABCMeta
 
     def __init__(self, parent=None):
-        self._data = [] # type: List[object]
+        self._data: List[object] = []
         self._tabs = None
         self.parent = parent
 
@@ -182,7 +181,7 @@ class Statements(Parsable):
 
 def _space_list(list_):
     """ Inserts whitespace between adjacent non-whitespace tokens. """
-    spaced_statement = [] # type: List[str]
+    spaced_statement: List[str] = []
     for i in reversed(range(len(list_))):
         spaced_statement.insert(0, list_[i])
         if i > 0 and not list_[i].isspace() and not list_[i-1].isspace():
@@ -271,8 +270,8 @@ class Block(Parsable):
     """
     def __init__(self, parent=None):
         super(Block, self).__init__(parent)
-        self.names = None # type: Sentence
+        self.names: Sentence = None
-        self.contents = None # type: Block
+        self.contents: Block = None
 
     @staticmethod
     def should_parse(lists):

certbot-nginx/tests/parser_test.py

@@ -112,7 +112,7 @@ class NginxParserTest(util.NginxTest):
                    ([[[0], [3], [4]], [[5], [3], [0]]], [])]
 
         for mylist, result in mylists:
-            paths = [] # type: List[List[int]]
+            paths: List[List[int]] = []
             parser._do_for_subarray(mylist,
                                     lambda x: isinstance(x, list) and
                                     len(x) >= 1 and

certbot/certbot/_internal/account.py

@@ -229,8 +229,7 @@ class AccountFileStorage(interfaces.AccountStorage):
     def load(self, account_id):
         return self._load_for_server_path(account_id, self.config.server_path)
 
-    def save(self, account, client):
+    def save(self, account: Account, client: ClientBase) -> None:
-        # type: (Account, ClientBase) -> None
         """Create a new account.
 
         :param Account account: account to create
@@ -245,8 +244,7 @@ class AccountFileStorage(interfaces.AccountStorage):
         except IOError as error:
             raise errors.AccountStorageError(error)
 
-    def update_regr(self, account, client):
+    def update_regr(self, account: Account, client: ClientBase) -> None:
-        # type: (Account, ClientBase) -> None
         """Update the registration resource.
 
         :param Account account: account to update
@@ -259,8 +257,7 @@ class AccountFileStorage(interfaces.AccountStorage):
         except IOError as error:
             raise errors.AccountStorageError(error)
 
-    def update_meta(self, account):
+    def update_meta(self, account: Account) -> None:
-        # type: (Account) -> None
         """Update the meta resource.
 
         :param Account account: account to update
@@ -338,19 +335,16 @@ class AccountFileStorage(interfaces.AccountStorage):
 
         return dir_path
 
-    def _prepare(self, account):
+    def _prepare(self, account: Account) -> str:
-        # type: (Account) -> str
         account_dir_path = self._account_dir_path(account.id)
         util.make_or_verify_dir(account_dir_path, 0o700, self.config.strict_permissions)
         return account_dir_path
 
-    def _create(self, account, dir_path):
+    def _create(self, account: Account, dir_path: str) -> None:
-        # type: (Account, str) -> None
         with util.safe_open(self._key_path(dir_path), "w", chmod=0o400) as key_file:
             key_file.write(account.key.json_dumps())
 
-    def _update_regr(self, account, acme, dir_path):
+    def _update_regr(self, account: Account, acme: ClientBase, dir_path: str) -> None:
-        # type: (Account, ClientBase, str) -> None
         with open(self._regr_path(dir_path), "w") as regr_file:
             regr = account.regr
             # If we have a value for new-authz, save it for forwards

certbot/certbot/_internal/auth_handler.py

@@ -98,8 +98,7 @@ class AuthHandler:
 
             return authzrs_validated
 
-    def deactivate_valid_authorizations(self, orderr):
+    def deactivate_valid_authorizations(self, orderr: messages.OrderResource) -> Tuple[List, List]:
-        # type: (messages.OrderResource) -> Tuple[List, List]
         """
         Deactivate all `valid` authorizations in the order, so that they cannot be re-used
         in subsequent orders.
@@ -191,7 +190,7 @@ class AuthHandler:
         """
         pending_authzrs = [authzr for authzr in authzrs
                            if authzr.body.status != messages.STATUS_VALID]
-        achalls = []  # type: List[achallenges.AnnotatedChallenge]
+        achalls: List[achallenges.AnnotatedChallenge] = []
         if pending_authzrs:
             logger.info("Performing the following challenges:")
         for authzr in pending_authzrs:
@@ -428,7 +427,7 @@ _ERROR_HELP = {
 
 def _report_failed_authzrs(failed_authzrs, account_key):
     """Notifies the user about failed authorizations."""
-    problems = {}  # type: Dict[str, List[achallenges.AnnotatedChallenge]]
+    problems: Dict[str, List[achallenges.AnnotatedChallenge]] = {}
     failed_achalls = [challb_to_achall(challb, account_key, authzr.body.identifier.value)
                       for authzr in failed_authzrs for challb in authzr.body.challenges
                       if challb.error]

certbot/certbot/_internal/cert_manager.py

@@ -241,7 +241,7 @@ def match_and_check_overlaps(cli_config, acceptable_matches, match_func, rv_func
     def find_matches(candidate_lineage, return_value, acceptable_matches):
         """Returns a list of matches using _search_lineages."""
         acceptable_matches = [func(candidate_lineage) for func in acceptable_matches]
-        acceptable_matches_rv = []  # type: List[str]
+        acceptable_matches_rv: List[str] = []
         for item in acceptable_matches:
             if isinstance(item, list):
                 acceptable_matches_rv += item
@@ -364,7 +364,7 @@ def _report_human_readable(config, parsed_certs):
 
 def _describe_certs(config, parsed_certs, parse_failures):
     """Print information about the certs we know about"""
-    out = []  # type: List[str]
+    out: List[str] = []
 
     notify = out.append
 

certbot/certbot/_internal/cli/__init__.py

@@ -66,7 +66,7 @@ logger = logging.getLogger(__name__)
 
 
 # Global, to save us from a lot of argument passing within the scope of this module
-helpful_parser = None  # type: Optional[HelpfulArgumentParser]
+helpful_parser: Optional[HelpfulArgumentParser] = None
 
 
 def prepare_and_parse_args(plugins, args, detect_defaults=False):

certbot/certbot/_internal/cli/cli_utils.py

@@ -62,7 +62,7 @@ def config_help(name, hidden=False):
     """Extract the help message for an `.IConfig` attribute."""
     if hidden:
         return argparse.SUPPRESS
-    field = interfaces.IConfig.__getitem__(name)  # type: zope.interface.interface.Attribute
+    field: zope.interface.interface.Attribute = interfaces.IConfig.__getitem__(name)
     return field.__doc__
 
 

certbot/certbot/_internal/cli/helpful.py

@@ -105,9 +105,9 @@ class HelpfulArgumentParser:
         self.visible_topics = self.determine_help_topics(self.help_arg)
 
         # elements are added by .add_group()
-        self.groups = {}  # type: Dict[str, argparse._ArgumentGroup]
+        self.groups: Dict[str, argparse._ArgumentGroup] = {}
         # elements are added by .parse_args()
-        self.defaults = {}  # type: Dict[str, Any]
+        self.defaults: Dict[str, Any] = {}
 
         self.parser = configargparse.ArgParser(
             prog="certbot",

certbot/certbot/_internal/client.py

@@ -328,7 +328,7 @@ class Client:
             with open(old_keypath, "rb") as f:
                 keypath = old_keypath
                 keypem = f.read()
-            key = util.Key(file=keypath, pem=keypem) # type: Optional[util.Key]
+            key: Optional[util.Key] = util.Key(file=keypath, pem=keypem)
             logger.info("Reusing existing private key from %s.", old_keypath)
         else:
             # The key is set to None here but will be created below.
@@ -390,8 +390,8 @@ class Client:
             cert, chain = self.obtain_certificate_from_csr(csr, orderr)
             return cert, chain, key, csr
 
-    def _get_order_and_authorizations(self, csr_pem, best_effort):
+    def _get_order_and_authorizations(self, csr_pem: str,
-        # type: (str, bool) -> List[messages.OrderResource]
+                                      best_effort: bool) -> List[messages.OrderResource]:
         """Request a new order and complete its authorizations.
 
         :param str csr_pem: A CSR in PEM format.

certbot/certbot/_internal/eff.py

@@ -1,22 +1,21 @@
 """Subscribes users to the EFF newsletter."""
 import logging
-from typing import Optional  # pylint: disable=unused-import
+from typing import Optional
 
 import requests
 import zope.component
 
 from certbot import interfaces
 from certbot._internal import constants
-from certbot._internal.account import Account  # pylint: disable=unused-import
+from certbot._internal.account import Account
 from certbot._internal.account import AccountFileStorage
 from certbot.display import util as display_util
-from certbot.interfaces import IConfig  # pylint: disable=unused-import
+from certbot.interfaces import IConfig
 
 logger = logging.getLogger(__name__)
 
 
-def prepare_subscription(config, acc):
+def prepare_subscription(config: IConfig, acc: Account) -> None:
-    # type: (IConfig, Account) -> None
     """High level function to store potential EFF newsletter subscriptions.
 
     The user may be asked if they want to sign up for the newsletter if
@@ -44,8 +43,7 @@ def prepare_subscription(config, acc):
         storage.update_meta(acc)
 
 
-def handle_subscription(config, acc):
+def handle_subscription(config: IConfig, acc: Account) -> None:
-    # type: (IConfig, Account) -> None
     """High level function to take care of EFF newsletter subscriptions.
 
     Once subscription is handled, it will not be handled again.
@@ -64,8 +62,7 @@ def handle_subscription(config, acc):
         storage.update_meta(acc)
 
 
-def _want_subscription():
+def _want_subscription() -> bool:
-    # type: () -> bool
     """Does the user want to be subscribed to the EFF newsletter?
 
     :returns: True if we should subscribe the user, otherwise, False
@@ -82,8 +79,7 @@ def _want_subscription():
     return display.yesno(prompt, default=False)
 
 
-def subscribe(email):
+def subscribe(email: str) -> None:
-    # type: (str) -> None
     """Subscribe the user to the EFF mailing list.
 
     :param str email: the e-mail address to subscribe
@@ -98,8 +94,7 @@ def subscribe(email):
     _check_response(requests.post(url, data=data))
 
 
-def _check_response(response):
+def _check_response(response: requests.Response) -> None:
-    # type: (requests.Response) -> None
     """Check for errors in the server's response.
 
     If an error occurred, it will be reported to the user.
@@ -119,8 +114,7 @@ def _check_response(response):
         _report_failure('there was a problem with the server response')
 
 
-def _report_failure(reason=None):
+def _report_failure(reason: Optional[str] = None) -> None:
-    # type: (Optional[str]) -> None
     """Notify the user of failing to sign them up for the newsletter.
 
     :param reason: a phrase describing what the problem was

certbot/certbot/_internal/error_handler.py

@@ -77,9 +77,9 @@ class ErrorHandler:
     def __init__(self, func, *args, **kwargs):
         self.call_on_regular_exit = False
         self.body_executed = False
-        self.funcs = []  # type: List[Callable[[], Any]]
+        self.funcs: List[Callable[[], Any]] = []
-        self.prev_handlers = {}  # type: Dict[int, Union[int, None, Callable]]
+        self.prev_handlers: Dict[int, Union[int, None, Callable]] = {}
-        self.received_signals = []  # type: List[int]
+        self.received_signals: List[int] = []
         if func is not None:
             self.register(func, *args, **kwargs)
 
@@ -108,8 +108,7 @@ class ErrorHandler:
         self._call_signals()
         return retval
 
-    def register(self, func, *args, **kwargs):
+    def register(self, func: Callable, *args: Any, **kwargs: Any) -> None:
-        # type: (Callable, *Any, **Any) -> None
         """Sets func to be run with the given arguments during cleanup.
 
         :param function func: function to be called in case of an error

certbot/certbot/_internal/hooks.py

@@ -77,7 +77,7 @@ def pre_hook(config):
         _run_pre_hook_if_necessary(cmd)
 
 
-executed_pre_hooks = set()  # type: Set[str]
+executed_pre_hooks: Set[str] = set()
 
 
 def _run_pre_hook_if_necessary(command):
@@ -127,7 +127,7 @@ def post_hook(config):
         _run_hook("post-hook", cmd)
 
 
-post_hooks = []  # type: List[str]
+post_hooks: List[str] = []
 
 
 def _run_eventually(command):

certbot/certbot/_internal/lock.py

@@ -16,28 +16,9 @@ else:
     POSIX_MODE = True
 
 
-
 logger = logging.getLogger(__name__)
 
 
-def lock_dir(dir_path):
-    # type: (str) -> LockFile
-    """Place a lock file on the directory at dir_path.
-
-    The lock file is placed in the root of dir_path with the name
-    .certbot.lock.
-
-    :param str dir_path: path to directory
-
-    :returns: the locked LockFile object
-    :rtype: LockFile
-
-    :raises errors.LockError: if unable to acquire the lock
-
-    """
-    return LockFile(os.path.join(dir_path, '.certbot.lock'))
-
-
 class LockFile:
     """
     Platform independent file lock system.
@@ -52,8 +33,7 @@ class LockFile:
     LockFile is platform independent: it will proceed to the appropriate OS lock mechanism
     depending on Linux or Windows.
     """
-    def __init__(self, path):
+    def __init__(self, path: str) -> None:
-        # type: (str) -> None
         """
         Create a LockFile instance on the given file path, and acquire lock.
         :param str path: the path to the file that will hold a lock
@@ -64,8 +44,7 @@ class LockFile:
 
         self.acquire()
 
-    def __repr__(self):
+    def __repr__(self) -> str:
-        # type: () -> str
         repr_str = '{0}({1}) <'.format(self.__class__.__name__, self._path)
         if self.is_locked():
             repr_str += 'acquired>'
@@ -73,23 +52,20 @@ class LockFile:
             repr_str += 'released>'
         return repr_str
 
-    def acquire(self):
+    def acquire(self) -> None:
-        # type: () -> None
         """
         Acquire the lock on the file, forbidding any other Certbot instance to acquire it.
         :raises errors.LockError: if unable to acquire the lock
         """
         self._lock_mechanism.acquire()
 
-    def release(self):
+    def release(self) -> None:
-        # type: () -> None
         """
         Release the lock on the file, allowing any other Certbot instance to acquire it.
         """
         self._lock_mechanism.release()
 
-    def is_locked(self):
+    def is_locked(self) -> bool:
-        # type: () -> bool
         """
         Check if the file is currently locked.
         :return: True if the file is locked, False otherwise
@@ -98,17 +74,15 @@ class LockFile:
 
 
 class _BaseLockMechanism:
-    def __init__(self, path):
+    def __init__(self, path: str) -> None:
-        # type: (str) -> None
         """
         Create a lock file mechanism for Unix.
         :param str path: the path to the lock file
         """
         self._path = path
-        self._fd = None  # type: Optional[int]
+        self._fd: Optional[int] = None
 
-    def is_locked(self):
+    def is_locked(self) -> bool:
-        # type: () -> bool
         """Check if lock file is currently locked.
         :return: True if the lock file is locked
         :rtype: bool
@@ -129,8 +103,7 @@ class _UnixLockMechanism(_BaseLockMechanism):
     process exits. It cannot be used to provide synchronization between
     threads. It is based on the lock_file package by Martin Horcicka.
     """
-    def acquire(self):
+    def acquire(self) -> None:
-        # type: () -> None
         """Acquire the lock."""
         while self._fd is None:
             # Open the file
@@ -144,8 +117,7 @@ class _UnixLockMechanism(_BaseLockMechanism):
                 if self._fd is None:
                     os.close(fd)
 
-    def _try_lock(self, fd):
+    def _try_lock(self, fd: int) -> None:
-        # type: (int) -> None
         """
         Try to acquire the lock file without blocking.
         :param int fd: file descriptor of the opened file to lock
@@ -158,8 +130,7 @@ class _UnixLockMechanism(_BaseLockMechanism):
                 raise errors.LockError('Another instance of Certbot is already running.')
             raise
 
-    def _lock_success(self, fd):
+    def _lock_success(self, fd: int) -> bool:
-        # type: (int) -> bool
         """
         Did we successfully grab the lock?
         Because this class deletes the locked file when the lock is
@@ -185,8 +156,7 @@ class _UnixLockMechanism(_BaseLockMechanism):
         # the same device and inode, they're the same file.
         return stat1.st_dev == stat2.st_dev and stat1.st_ino == stat2.st_ino
 
-    def release(self):
+    def release(self) -> None:
-        # type: () -> None
         """Remove, close, and release the lock file."""
         # It is important the lock file is removed before it's released,
         # otherwise:
@@ -269,3 +239,20 @@ class _WindowsLockMechanism(_BaseLockMechanism):
                 logger.debug(str(e))
         finally:
             self._fd = None
+
+
+def lock_dir(dir_path: str) -> LockFile:
+    """Place a lock file on the directory at dir_path.
+
+    The lock file is placed in the root of dir_path with the name
+    .certbot.lock.
+
+    :param str dir_path: path to directory
+
+    :returns: the locked LockFile object
+    :rtype: LockFile
+
+    :raises errors.LockError: if unable to acquire the lock
+
+    """
+    return LockFile(os.path.join(dir_path, '.certbot.lock'))

certbot/certbot/_internal/main.py

@@ -4,7 +4,7 @@
 import functools
 import logging.handlers
 import sys
-from typing import Iterable  # pylint: disable=unused-import
+from typing import Iterable
 from typing import List
 from typing import Optional
 from typing import Tuple
@@ -145,8 +145,8 @@ def _get_and_save_cert(le_client, config, domains=None, certname=None, lineage=N
     return lineage
 
 
-def _handle_unexpected_key_type_migration(config, cert):
+def _handle_unexpected_key_type_migration(config: configuration.NamespaceConfig,
-    # type: (configuration.NamespaceConfig, storage.RenewableCert) -> None
+                                          cert: storage.RenewableCert) -> None:
     """
     This function ensures that the user will not implicitly migrate an existing key
     from one type to another in the situation where a certificate for that lineage
@@ -167,11 +167,10 @@ def _handle_unexpected_key_type_migration(config, cert):
             raise errors.Error(msg)
 
 
-def _handle_subset_cert_request(config,  # type: configuration.NamespaceConfig
+def _handle_subset_cert_request(config: configuration.NamespaceConfig,
-                                domains,  # type: List[str]
+                                domains: List[str],
-                                cert  # type: storage.RenewableCert
+                                cert: storage.RenewableCert
-                                ):
+                                ) -> Tuple[str, Optional[storage.RenewableCert]]:
-    # type: (...) -> Tuple[str, Optional[storage.RenewableCert]]
     """Figure out what to do if a previous cert had a subset of the names now requested
 
     :param config: Configuration object
@@ -218,10 +217,9 @@ def _handle_subset_cert_request(config,  # type: configuration.NamespaceConfig
     raise errors.Error(USER_CANCELLED)
 
 
-def _handle_identical_cert_request(config,  # type: configuration.NamespaceConfig
+def _handle_identical_cert_request(config: configuration.NamespaceConfig,
-                                   lineage,  # type: storage.RenewableCert
+                                   lineage: storage.RenewableCert,
-                                   ):
+                                   ) -> Tuple[str, Optional[storage.RenewableCert]]:
-    # type: (...) -> Tuple[str, Optional[storage.RenewableCert]]
     """Figure out what to do if a lineage has the same names as a previously obtained one
 
     :param config: Configuration object
@@ -337,11 +335,10 @@ def _find_cert(config, domains, certname):
     return (action != "reinstall"), lineage
 
 
-def _find_lineage_for_domains_and_certname(config,  # type: configuration.NamespaceConfig
+def _find_lineage_for_domains_and_certname(config: configuration.NamespaceConfig,
-                                           domains,  # type: List[str]
+                                           domains: List[str],
-                                           certname  # type: str
+                                           certname: str
-                                           ):
+                                           ) -> Tuple[str, Optional[storage.RenewableCert]]:
-    # type: (...) -> Tuple[str, Optional[storage.RenewableCert]]
     """Find appropriate lineage based on given domains and/or certname.
 
     :param config: Configuration object
@@ -758,7 +755,7 @@ def update_account(config, unused_plugins):
     cb_client = client.Client(config, acc, None, None, acme=acme)
     # Empty list of contacts in case the user is removing all emails
 
-    acc_contacts = () # type: Iterable[str]
+    acc_contacts: Iterable[str] = ()
     if config.email:
         acc_contacts = ['mailto:' + email for email in config.email.split(',')]
     # We rely on an exception to interrupt this process if it didn't work.
@@ -1362,8 +1359,8 @@ def set_displayer(config):
     """
     if config.quiet:
         config.noninteractive_mode = True
-        displayer = display_util.NoninteractiveDisplay(open(os.devnull, "w")) \
+        displayer: Union[None, display_util.NoninteractiveDisplay, display_util.FileDisplay] =\
-        # type: Union[None, display_util.NoninteractiveDisplay, display_util.FileDisplay]
+            display_util.NoninteractiveDisplay(open(os.devnull, "w"))
     elif config.noninteractive_mode:
         displayer = display_util.NoninteractiveDisplay(sys.stdout)
     else:

certbot/certbot/_internal/plugins/disco.py

@@ -1,9 +1,9 @@
 """Utilities for plugins discovery and selection."""
+from collections.abc import Mapping
 import itertools
 import logging
 import sys
 from typing import Dict
-from collections.abc import Mapping
 
 import pkg_resources
 import zope.interface
@@ -213,7 +213,7 @@ class PluginsRegistry(Mapping):
     @classmethod
     def find_all(cls):
         """Find plugins using setuptools entry points."""
-        plugins = {}  # type: Dict[str, PluginEntryPoint]
+        plugins: Dict[str, PluginEntryPoint] = {}
         plugin_paths_string = os.getenv('CERTBOT_PLUGIN_PATH')
         plugin_paths = plugin_paths_string.split(':') if plugin_paths_string else []
         # XXX should ensure this only happens once

certbot/certbot/_internal/plugins/manual.py

@@ -5,7 +5,7 @@ import zope.component
 import zope.interface
 
 from acme import challenges
-from certbot import achallenges  # pylint: disable=unused-import
+from certbot import achallenges
 from certbot import errors
 from certbot import interfaces
 from certbot import reverter
@@ -74,8 +74,7 @@ permitted by DNS standards.)
         super(Authenticator, self).__init__(*args, **kwargs)
         self.reverter = reverter.Reverter(self.config)
         self.reverter.recovery_routine()
-        self.env = {} \
+        self.env: Dict[achallenges.KeyAuthorizationAnnotatedChallenge, Dict[str, str]] = {}
-        # type: Dict[achallenges.KeyAuthorizationAnnotatedChallenge, Dict[str, str]]
         self.subsequent_dns_challenge = False
         self.subsequent_any_challenge = False
 

certbot/certbot/_internal/plugins/standalone.py

@@ -10,7 +10,7 @@ from typing import Set
 from typing import Tuple
 from typing import TYPE_CHECKING
 
-import OpenSSL  # pylint: disable=unused-import
+import OpenSSL
 import zope.interface
 
 from acme import challenges
@@ -42,7 +42,7 @@ class ServerManager:
 
     """
     def __init__(self, certs, http_01_resources):
-        self._instances = {}  # type: Dict[int, acme_standalone.BaseDualNetworkedServers]
+        self._instances: Dict[int, acme_standalone.BaseDualNetworkedServers] = {}
         self.certs = certs
         self.http_01_resources = http_01_resources
 
@@ -122,15 +122,14 @@ class Authenticator(common.Plugin):
     def __init__(self, *args, **kwargs):
         super(Authenticator, self).__init__(*args, **kwargs)
 
-        self.served = collections.defaultdict(set)  # type: ServedType
+        self.served: ServedType = collections.defaultdict(set)
 
         # Stuff below is shared across threads (i.e. servers read
         # values, main thread writes). Due to the nature of CPython's
         # GIL, the operations are safe, c.f.
         # https://docs.python.org/2/faq/library.html#what-kinds-of-global-value-mutation-are-thread-safe
-        self.certs = {}  # type: Dict[bytes, Tuple[OpenSSL.crypto.PKey, OpenSSL.crypto.X509]]
+        self.certs: Dict[bytes, Tuple[OpenSSL.crypto.PKey, OpenSSL.crypto.X509]] = {}
-        self.http_01_resources = set() \
+        self.http_01_resources: Set[acme_standalone.HTTP01RequestHandler.HTTP01Resource] = set()
-        # type: Set[acme_standalone.HTTP01RequestHandler.HTTP01Resource]
 
         self.servers = ServerManager(self.certs, self.http_01_resources)
 

certbot/certbot/_internal/plugins/webroot.py

@@ -12,10 +12,10 @@ import zope.component
 import zope.interface
 
 from acme import challenges
-from certbot import achallenges  # pylint: disable=unused-import
 from certbot import errors
 from certbot import interfaces
 from certbot._internal import cli
+from certbot.achallenges import KeyAuthorizationAnnotatedChallenge as AnnotatedChallenge
 from certbot.compat import filesystem
 from certbot.compat import os
 from certbot.display import ops
@@ -67,11 +67,10 @@ to serve all files under specified web root ({0})."""
 
     def __init__(self, *args, **kwargs):
         super(Authenticator, self).__init__(*args, **kwargs)
-        self.full_roots = {}  # type: Dict[str, str]
+        self.full_roots: Dict[str, str] = {}
-        self.performed = collections.defaultdict(set) \
+        self.performed: DefaultDict[str, Set[AnnotatedChallenge]] = collections.defaultdict(set)
-            # type: DefaultDict[str, Set[achallenges.KeyAuthorizationAnnotatedChallenge]]
         # stack of dirs successfully created by this authenticator
-        self._created_dirs = []  # type: List[str]
+        self._created_dirs: List[str] = []
 
     def prepare(self):  # pylint: disable=missing-function-docstring
         pass
@@ -224,7 +223,7 @@ to serve all files under specified web root ({0})."""
                 os.remove(validation_path)
                 self.performed[root_path].remove(achall)
 
-        not_removed = []  # type: List[str]
+        not_removed: List[str] = []
         while self._created_dirs:
             path = self._created_dirs.pop()
             try:

certbot/certbot/_internal/renewal.py

@@ -8,7 +8,7 @@ import sys
 import time
 import traceback
 from typing import List
-from typing import Optional  # pylint: disable=unused-import
+from typing import Optional
 
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.asymmetric import ec
@@ -22,7 +22,7 @@ from certbot import errors
 from certbot import interfaces
 from certbot import util
 from certbot._internal import cli
-from certbot._internal import client  # pylint: disable=unused-import
+from certbot._internal import client
 from certbot._internal import constants
 from certbot._internal import hooks
 from certbot._internal import storage
@@ -143,7 +143,7 @@ def _restore_plugin_configs(config, renewalparams):
     #      longer defined, stored copies of that parameter will be
     #      deserialized as strings by this logic even if they were
     #      originally meant to be some other type.
-    plugin_prefixes = []  # type: List[str]
+    plugin_prefixes: List[str] = []
     if renewalparams["authenticator"] == "webroot":
         _restore_webroot_config(config, renewalparams)
     else:
@@ -290,7 +290,7 @@ def _restore_str(name, value):
 
 
 def should_renew(config, lineage):
-    "Return true if any of the circumstances for automatic renewal apply."
+    """Return true if any of the circumstances for automatic renewal apply."""
     if config.renew_by_default:
         logger.debug("Auto-renewal forced with --force-renewal...")
         return True
@@ -305,7 +305,7 @@ def should_renew(config, lineage):
 
 
 def _avoid_invalidating_lineage(config, lineage, original_server):
-    "Do not renew a valid cert with one from a staging server!"
+    """Do not renew a valid cert with one from a staging server!"""
     # Some lineages may have begun with --staging, but then had production
     # certificates added to them
     with open(lineage.cert) as the_file:
@@ -323,8 +323,8 @@ def _avoid_invalidating_lineage(config, lineage, original_server):
                     "unless you use the --break-my-certs flag!".format(names))
 
 
-def renew_cert(config, domains, le_client, lineage):
+def renew_cert(config: interfaces.IConfig, domains: Optional[List[str]], le_client: client.Client,
-    # type: (interfaces.IConfig, Optional[List[str]], client.Client, storage.RenewableCert) -> None
+               lineage: storage.RenewableCert) -> None:
     """Renew a certificate lineage."""
     renewal_params = lineage.configuration["renewalparams"]
     original_server = renewal_params.get("server", cli.flag_default("server"))
@@ -351,14 +351,14 @@ def renew_cert(config, domains, le_client, lineage):
 
 
 def report(msgs, category):
-    "Format a results report for a category of renewal outcomes"
+    """Format a results report for a category of renewal outcomes"""
     lines = ("%s (%s)" % (m, category) for m in msgs)
     return "  " + "\n  ".join(lines)
 
 
-def _renew_describe_results(config, renew_successes, renew_failures,
+def _renew_describe_results(config: interfaces.IConfig, renew_successes: List[str],
-                            renew_skipped, parse_failures):
+                            renew_failures: List[str], renew_skipped: List[str],
-    # type: (interfaces.IConfig, List[str], List[str], List[str], List[str]) -> None
+                            parse_failures: List[str]) -> None:
     """
     Print a report to the terminal about the results of the renewal process.
 
@@ -511,8 +511,7 @@ def handle_renewal_request(config):
     logger.debug("no renewal failures")
 
 
-def _update_renewal_params_from_key(key_path, config):
+def _update_renewal_params_from_key(key_path: str, config: interfaces.IConfig) -> None:
-    # type: (str, interfaces.IConfig) -> None
     with open(key_path, 'rb') as file_h:
         key = load_pem_private_key(file_h.read(), password=None, backend=default_backend())
     if isinstance(key, rsa.RSAPrivateKey):

certbot/certbot/_internal/reporter.py

@@ -32,7 +32,7 @@ class Reporter:
     _msg_type = collections.namedtuple('ReporterMsg', 'priority text on_crash')
 
     def __init__(self, config):
-        self.messages = queue.PriorityQueue()  # type: queue.PriorityQueue[Reporter._msg_type]
+        self.messages: queue.PriorityQueue[Reporter._msg_type] = queue.PriorityQueue()
         self.config = config
 
     def add_message(self, msg, priority, on_crash=True):

certbot/certbot/_internal/snap_config.py

@@ -33,8 +33,7 @@ _ARCH_TRIPLET_MAP = {
 LOGGER = logging.getLogger(__name__)
 
 
-def prepare_env(cli_args):
+def prepare_env(cli_args: List[str]) -> List[str]:
-    # type: (List[str]) -> List[str]
     """
     Prepare runtime environment for a certbot execution in snap.
     :param list cli_args: List of command line arguments

certbot/certbot/compat/filesystem.py

@@ -35,8 +35,7 @@ class _WindowsUmask:
 _WINDOWS_UMASK = _WindowsUmask()
 
 
-def chmod(file_path, mode):
+def chmod(file_path: str, mode: int) -> None:
-    # type: (str, int) -> None
     """
     Apply a POSIX mode on given file_path:
 
@@ -57,8 +56,7 @@ def chmod(file_path, mode):
         _apply_win_mode(file_path, mode)
 
 
-def umask(mask):
+def umask(mask: int) -> int:
-    # type: (int) -> int
     """
     Set the current numeric umask and return the previous umask. On Linux, the built-in umask
     method is used. On Windows, our Certbot-side implementation is used.
@@ -84,8 +82,8 @@ def umask(mask):
 # Since copying and editing arbitrary DACL is very difficult, and since we actually know
 # the mode to apply at the time the owner of a file should change, it is easier to just
 # change the owner, then reapply the known mode, as copy_ownership_and_apply_mode() does.
-def copy_ownership_and_apply_mode(src, dst, mode, copy_user, copy_group):
+def copy_ownership_and_apply_mode(src: str, dst: str, mode: int,
-    # type: (str, str, int, bool, bool) -> None
+                                  copy_user: bool, copy_group: bool) -> None:
     """
     Copy ownership (user and optionally group on Linux) from the source to the
     destination, then apply given mode in compatible way for Linux and Windows.
@@ -117,8 +115,8 @@ def copy_ownership_and_apply_mode(src, dst, mode, copy_user, copy_group):
 # equivalent POSIX mode, because ownership and mode are copied altogether on the destination
 # file, so no recomputing of the DACL against the new owner is needed, as it would be
 # for a copy_ownership alone method.
-def copy_ownership_and_mode(src, dst, copy_user=True, copy_group=True):
+def copy_ownership_and_mode(src: str, dst: str,
-    # type: (str, str, bool, bool) -> None
+                            copy_user: bool = True, copy_group: bool = True) -> None:
     """
     Copy ownership (user and optionally group on Linux) and mode/DACL
     from the source to the destination.
@@ -142,8 +140,7 @@ def copy_ownership_and_mode(src, dst, copy_user=True, copy_group=True):
         _copy_win_mode(src, dst)
 
 
-def check_mode(file_path, mode):
+def check_mode(file_path: str, mode: int) -> bool:
-    # type: (str, int) -> bool
     """
     Check if the given mode matches the permissions of the given file.
     On Linux, will make a direct comparison, on Windows, mode will be compared against
@@ -160,8 +157,7 @@ def check_mode(file_path, mode):
     return _check_win_mode(file_path, mode)
 
 
-def check_owner(file_path):
+def check_owner(file_path: str) -> bool:
-    # type: (str) -> bool
     """
     Check if given file is owned by current user.
 
@@ -183,8 +179,7 @@ def check_owner(file_path):
     return _get_current_user() == user
 
 
-def check_permissions(file_path, mode):
+def check_permissions(file_path: str, mode: int) -> bool:
-    # type: (str, int) -> bool
     """
     Check if given file has the given mode and is owned by current user.
 
@@ -196,8 +191,7 @@ def check_permissions(file_path, mode):
     return check_owner(file_path) and check_mode(file_path, mode)
 
 
-def open(file_path, flags, mode=0o777):  # pylint: disable=redefined-builtin
+def open(file_path: str, flags: int, mode: int = 0o777) -> int:  # pylint: disable=redefined-builtin
-    # type: (str, int, int) -> int
     """
     Wrapper of original os.open function, that will ensure on Windows that given mode
     is correctly applied.
@@ -266,8 +260,7 @@ def open(file_path, flags, mode=0o777):  # pylint: disable=redefined-builtin
     return handle
 
 
-def makedirs(file_path, mode=0o777):
+def makedirs(file_path: str, mode: int = 0o777) -> None:
-    # type: (str, int) -> None
     """
     Rewrite of original os.makedirs function, that will ensure on Windows that given mode
     is correctly applied.
@@ -299,8 +292,7 @@ def makedirs(file_path, mode=0o777):
         umask(current_umask)
 
 
-def mkdir(file_path, mode=0o777):
+def mkdir(file_path: str, mode: int = 0o777) -> None:
-    # type: (str, int) -> None
     """
     Rewrite of original os.mkdir function, that will ensure on Windows that given mode
     is correctly applied.
@@ -331,8 +323,7 @@ def mkdir(file_path, mode=0o777):
     return None
 
 
-def replace(src, dst):
+def replace(src: str, dst: str) -> None:
-    # type: (str, str) -> None
     """
     Rename a file to a destination path and handles situations where the destination exists.
 
@@ -349,8 +340,7 @@ def replace(src, dst):
         os.rename(src, dst)
 
 
-def realpath(file_path):
+def realpath(file_path: str) -> str:
-    # type: (str) -> str
     """
     Find the real path for the given path. This method resolves symlinks, including
     recursive symlinks, and is protected against symlinks that creates an infinite loop.
@@ -371,7 +361,7 @@ def realpath(file_path):
             raise RuntimeError('Error, link {0} is a loop!'.format(original_path))
         return path
 
-    inspected_paths = []  # type: List[str]
+    inspected_paths: List[str] = []
     while os.path.islink(file_path):
         link_path = file_path
         file_path = os.readlink(file_path)
@@ -384,8 +374,7 @@ def realpath(file_path):
     return os.path.abspath(file_path)
 
 
-def readlink(link_path):
+def readlink(link_path: str) -> str:
-    # type: (str) -> str
     """
     Return a string representing the path to which the symbolic link points.
 
@@ -419,8 +408,7 @@ def readlink(link_path):
 # elevated privileges or not. However this is not a problem since certbot always
 # requires to be run under a privileged shell, so the user will always benefit
 # from the highest (privileged one) set of permissions on a given file.
-def is_executable(path):
+def is_executable(path: str) -> bool:
-    # type: (str) -> bool
     """
     Is path an executable file?
 
@@ -434,8 +422,7 @@ def is_executable(path):
     return _win_is_executable(path)
 
 
-def has_world_permissions(path):
+def has_world_permissions(path: str) -> bool:
-    # type: (str) -> bool
     """
     Check if everybody/world has any right (read/write/execute) on a file given its path.
 
@@ -456,8 +443,7 @@ def has_world_permissions(path):
     }))
 
 
-def compute_private_key_mode(old_key, base_mode):
+def compute_private_key_mode(old_key: str, base_mode: int) -> int:
-    # type: (str, int) -> int
     """
     Calculate the POSIX mode to apply to a private key given the previous private key.
 
@@ -478,8 +464,7 @@ def compute_private_key_mode(old_key, base_mode):
     return base_mode
 
 
-def has_same_ownership(path1, path2):
+def has_same_ownership(path1: str, path2: str) -> bool:
-    # type: (str, str) -> bool
     """
     Return True if the ownership of two files given their respective path is the same.
     On Windows, ownership is checked against owner only, since files do not have a group owner.
@@ -504,8 +489,7 @@ def has_same_ownership(path1, path2):
     return user1 == user2
 
 
-def has_min_permissions(path, min_mode):
+def has_min_permissions(path: str, min_mode: int) -> bool:
-    # type: (str, int) -> bool
     """
     Check if a file given its path has at least the permissions defined by the given minimal mode.
     On Windows, group permissions are ignored since files do not have a group owner.

certbot/certbot/compat/misc.py

@@ -27,8 +27,7 @@ logger = logging.getLogger(__name__)
 STANDARD_BINARY_DIRS = ["/usr/sbin", "/usr/local/bin", "/usr/local/sbin"] if POSIX_MODE else []
 
 
-def raise_for_non_administrative_windows_rights():
+def raise_for_non_administrative_windows_rights() -> None:
-    # type: () -> None
     """
     On Windows, raise if current shell does not have the administrative rights.
     Do nothing on Linux.
@@ -39,8 +38,7 @@ def raise_for_non_administrative_windows_rights():
         raise errors.Error('Error, certbot must be run on a shell with administrative rights.')
 
 
-def readline_with_timeout(timeout, prompt):
+def readline_with_timeout(timeout: float, prompt: str) -> str:
-    # type: (float, str) -> str
     """
     Read user input to return the first line entered, or raise after specified timeout.
 
@@ -81,8 +79,7 @@ LINUX_DEFAULT_FOLDERS = {
 }
 
 
-def get_default_folder(folder_type):
+def get_default_folder(folder_type: str) -> str:
-    # type: (str) -> str
     """
     Return the relevant default folder for the current OS
 
@@ -99,8 +96,7 @@ def get_default_folder(folder_type):
     return WINDOWS_DEFAULT_FOLDERS[folder_type]
 
 
-def underscores_for_unsupported_characters_in_path(path):
+def underscores_for_unsupported_characters_in_path(path: str) -> str:
-    # type: (str) -> str
     """
     Replace unsupported characters in path for current OS by underscores.
     :param str path: the path to normalize
@@ -116,8 +112,7 @@ def underscores_for_unsupported_characters_in_path(path):
     return drive + tail.replace(':', '_')
 
 
-def execute_command(cmd_name, shell_cmd, env=None):
+def execute_command(cmd_name: str, shell_cmd: str, env: Optional[dict] = None) -> Tuple[str, str]:
-    # type: (str, str, Optional[dict]) -> Tuple[str, str]
     """
     Run a command:
         - on Linux command will be run by the standard shell selected with Popen(shell=True)

certbot/certbot/crypto_util.py

@@ -7,7 +7,6 @@
 import hashlib
 import logging
 import re
-from typing import IO  # pylint: disable=unused-import
 import warnings
 
 # See https://github.com/pyca/cryptography/issues/4275
@@ -272,9 +271,9 @@ def verify_renewable_cert_sig(renewable_cert):
     :raises errors.Error: If signature verification fails.
     """
     try:
-        with open(renewable_cert.chain_path, 'rb') as chain_file:  # type: IO[bytes]
+        with open(renewable_cert.chain_path, 'rb') as chain_file:
             chain = x509.load_pem_x509_certificate(chain_file.read(), default_backend())
-        with open(renewable_cert.cert_path, 'rb') as cert_file:  # type: IO[bytes]
+        with open(renewable_cert.cert_path, 'rb') as cert_file:
             cert = x509.load_pem_x509_certificate(cert_file.read(), default_backend())
         pk = chain.public_key()
         with warnings.catch_warnings():
@@ -349,11 +348,11 @@ def verify_fullchain(renewable_cert):
     :raises errors.Error: If cert and chain do not combine to fullchain.
     """
     try:
-        with open(renewable_cert.chain_path) as chain_file:  # type: IO[str]
+        with open(renewable_cert.chain_path) as chain_file:
             chain = chain_file.read()
-        with open(renewable_cert.cert_path) as cert_file:  # type: IO[str]
+        with open(renewable_cert.cert_path) as cert_file:
             cert = cert_file.read()
-        with open(renewable_cert.fullchain_path) as fullchain_file:  # type: IO[str]
+        with open(renewable_cert.fullchain_path) as fullchain_file:
             fullchain = fullchain_file.read()
         if (cert + chain) != fullchain:
             error_str = "fullchain does not match cert + chain for {0}!"
@@ -487,7 +486,7 @@ def _notAfterBefore(cert_path, method):
 
     """
     # pylint: disable=redefined-outer-name
-    with open(cert_path, "rb") as f:  # type: IO[bytes]
+    with open(cert_path, "rb") as f:
         x509 = crypto.load_certificate(crypto.FILETYPE_PEM, f.read())
     # pyopenssl always returns bytes
     timestamp = method(x509)
@@ -564,7 +563,7 @@ def get_serial_from_cert(cert_path):
     :rtype: int
     """
     # pylint: disable=redefined-outer-name
-    with open(cert_path, "rb") as f:  # type: IO[bytes]
+    with open(cert_path, "rb") as f:
         x509 = crypto.load_certificate(crypto.FILETYPE_PEM, f.read())
     return x509.get_serial_number()
 

certbot/certbot/display/util.py

@@ -14,8 +14,8 @@ import sys
 import textwrap
 from typing import List
 
-import zope.interface
 import zope.component
+import zope.interface
 
 from certbot import errors
 from certbot import interfaces
@@ -98,8 +98,7 @@ def input_with_timeout(prompt=None, timeout=36000.0):
     return line.rstrip('\n')
 
 
-def notify(msg):
+def notify(msg: str) -> None:
-    # type: (str) -> None
     """Display a basic status message.
 
     :param str msg: message to display
@@ -636,8 +635,7 @@ def _parens_around_char(label):
     return "({first}){rest}".format(first=label[0], rest=label[1:])
 
 
-def summarize_domain_list(domains):
+def summarize_domain_list(domains: List[str]) -> str:
-    # type: (List[str]) -> str
     """Summarizes a list of domains in the format of:
         example.com.com and N more domains
     or if there is are only two domains:

certbot/certbot/ocsp.py

@@ -59,8 +59,7 @@ class RevocationChecker:
             else:
                 self.host_args = lambda host: ["Host", host]
 
-    def ocsp_revoked(self, cert):
+    def ocsp_revoked(self, cert: RenewableCert) -> bool:
-        # type: (RenewableCert) -> bool
         """Get revoked status for a particular cert version.
 
         .. todo:: Make this a non-blocking call
@@ -72,8 +71,7 @@ class RevocationChecker:
         """
         return self.ocsp_revoked_by_paths(cert.cert_path, cert.chain_path)
 
-    def ocsp_revoked_by_paths(self, cert_path, chain_path, timeout=10):
+    def ocsp_revoked_by_paths(self, cert_path: str, chain_path: str, timeout: int = 10) -> bool:
-        # type: (str, str, int) -> bool
         """Performs the OCSP revocation check
 
         :param str cert_path: Certificate filepath
@@ -102,8 +100,8 @@ class RevocationChecker:
             return self._check_ocsp_openssl_bin(cert_path, chain_path, host, url, timeout)
         return _check_ocsp_cryptography(cert_path, chain_path, url, timeout)
 
-    def _check_ocsp_openssl_bin(self, cert_path, chain_path, host, url, timeout):
+    def _check_ocsp_openssl_bin(self, cert_path: str, chain_path: str,
-        # type: (str, str, str, str, int) -> bool
+                                host: str, url: str, timeout: int) -> bool:
         # Minimal implementation of proxy selection logic as seen in, e.g., cURL
         # Some things that won't work, but may well be in use somewhere:
         # - username and password for proxy authentication
@@ -140,8 +138,7 @@ class RevocationChecker:
         return _translate_ocsp_query(cert_path, output, err)
 
 
-def _determine_ocsp_server(cert_path):
+def _determine_ocsp_server(cert_path: str) -> Tuple[Optional[str], Optional[str]]:
-    # type: (str) -> Tuple[Optional[str], Optional[str]]
     """Extract the OCSP server host from a certificate.
 
     :param str cert_path: Path to the cert we're checking OCSP for
@@ -171,8 +168,7 @@ def _determine_ocsp_server(cert_path):
     return None, None
 
 
-def _check_ocsp_cryptography(cert_path, chain_path, url, timeout):
+def _check_ocsp_cryptography(cert_path: str, chain_path: str, url: str, timeout: int) -> bool:
-    # type: (str, str, str, int) -> bool
     # Retrieve OCSP response
     with open(chain_path, 'rb') as file_handler:
         issuer = x509.load_pem_x509_certificate(file_handler.read(), default_backend())

certbot/certbot/plugins/common.py

@@ -9,7 +9,7 @@ from josepy import util as jose_util
 import pkg_resources
 import zope.interface
 
-from certbot import achallenges  # pylint: disable=unused-import
+from certbot import achallenges
 from certbot import crypto_util
 from certbot import errors
 from certbot import interfaces
@@ -313,8 +313,8 @@ class ChallengePerformer:
 
     def __init__(self, configurator):
         self.configurator = configurator
-        self.achalls = []  # type: List[achallenges.KeyAuthorizationAnnotatedChallenge]
+        self.achalls: List[achallenges.KeyAuthorizationAnnotatedChallenge] = []
-        self.indices = []  # type: List[int]
+        self.indices: List[int] = []
 
     def add_chall(self, achall, idx=None):
         """Store challenge to be performed when perform() is called.

certbot/certbot/plugins/dns_common_lexicon.py

@@ -116,8 +116,9 @@ class LexiconClient:
         return None
 
 
-def build_lexicon_config(lexicon_provider_name, lexicon_options, provider_options):
+def build_lexicon_config(lexicon_provider_name: str,
-    # type: (str, Dict, Dict) -> Union[ConfigResolver, Dict]
+                         lexicon_options: Dict, provider_options: Dict
+                         ) -> Union[ConfigResolver, Dict]:
     """
     Convenient function to build a Lexicon 2.x/3.x config object.
     :param str lexicon_provider_name: the name of the lexicon provider to use
@@ -126,7 +127,7 @@ def build_lexicon_config(lexicon_provider_name, lexicon_options, provider_option
     :return: configuration to apply to the provider
     :rtype: ConfigurationResolver or dict
     """
-    config = {'provider_name': lexicon_provider_name}  # type: Dict[str, Any]
+    config: Dict[str, Any] = {'provider_name': lexicon_provider_name}
     config.update(lexicon_options)
     if not ConfigResolver:
         # Lexicon 2.x

certbot/certbot/plugins/enhancements.py

@@ -156,7 +156,7 @@ class AutoHSTSEnhancement(object, metaclass=abc.ABCMeta):
 # This is used to configure internal new style enhancements in Certbot. These
 # enhancement interfaces need to be defined in this file. Please do not modify
 # this list from plugin code.
-_INDEX = [
+_INDEX: List[Dict[str, Any]] = [
     {
         "name": "AutoHSTS",
         "cli_help": "Gradually increasing max-age value for HTTP Strict Transport "+
@@ -171,4 +171,4 @@ _INDEX = [
         "deployer_function": "deploy_autohsts",
         "enable_function": "enable_autohsts"
     }
-]  # type: List[Dict[str, Any]]
+]

certbot/certbot/plugins/storage.py

@@ -42,7 +42,7 @@ class PluginStorage:
 
         :raises .errors.PluginStorageError: when unable to open or read the file
         """
-        data = {}  # type: Dict[str, Any]
+        data: Dict[str, Any] = {}
         filedata = ""
         try:
             with open(self._storagepath, 'r') as fh:

certbot/certbot/util.py

@@ -58,7 +58,7 @@ _INITIAL_PID = os.getpid()
 # the dict are attempted to be cleaned up at program exit. If the
 # program exits before the lock is cleaned up, it is automatically
 # released, but the file isn't deleted.
-_LOCKS = {}  # type: Dict[str, lock.LockFile]
+_LOCKS: Dict[str, lock.LockFile] = {}
 
 
 def env_no_snap_for_external_calls():
@@ -216,10 +216,10 @@ def safe_open(path, mode="w", chmod=None):
         if ``None``.
 
     """
-    open_args = ()  # type: Union[Tuple[()], Tuple[int]]
+    open_args: Union[Tuple[()], Tuple[int]] = ()
     if chmod is not None:
         open_args = (chmod,)
-    fdopen_args = ()  # type: Union[Tuple[()], Tuple[int]]
+    fdopen_args: Union[Tuple[()], Tuple[int]] = ()
     fd = filesystem.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, *open_args)
     return os.fdopen(fd, mode, *fdopen_args)
 
@@ -577,7 +577,7 @@ def is_wildcard_domain(domain):
     :rtype: bool
 
     """
-    wildcard_marker = b"*."  # type: Union[Text, bytes]
+    wildcard_marker: Union[Text, bytes] = b"*."
     if isinstance(domain, str):
         wildcard_marker = u"*."
     return domain.startswith(wildcard_marker)

certbot/tests/display/completer_test.py

@@ -30,7 +30,7 @@ class CompleterTest(test_util.TempDirTestCase):
         if self.tempdir[-1] != os.sep:
             self.tempdir += os.sep
 
-        self.paths = []  # type: List[str]
+        self.paths: List[str] = []
         # create some files and directories in temp_dir
         for c in string.ascii_lowercase:
             path = os.path.join(self.tempdir, c)

certbot/tests/error_handler_test.py

@@ -27,7 +27,7 @@ def set_signals(sig_handler_dict):
 def signal_receiver(signums):
     """Context manager to catch signals"""
     signals = []
-    prev_handlers = get_signals(signums)  # type: Dict[int, Union[int, None, Callable]]
+    prev_handlers: Dict[int, Union[int, None, Callable]] = get_signals(signums)
     set_signals({s: lambda s, _: signals.append(s) for s in signums})
     yield signals
     set_signals(prev_handlers)

certbot/tests/hook_test.py

@@ -267,7 +267,7 @@ class RunSavedPostHooksTest(HookTest):
 
     def setUp(self):
         super(RunSavedPostHooksTest, self).setUp()
-        self.eventually = []  # type: List[str]
+        self.eventually: List[str] = []
 
     def test_empty(self):
         self.assertFalse(self._call_with_mock_execute_and_eventually().called)

certbot/tests/log_test.py

@@ -43,7 +43,7 @@ class PreArgParseSetupTest(unittest.TestCase):
         mock_root_logger.setLevel.assert_called_once_with(logging.DEBUG)
         self.assertEqual(mock_root_logger.addHandler.call_count, 2)
 
-        memory_handler = None  # type: Optional[logging.handlers.MemoryHandler]
+        memory_handler: Optional[logging.handlers.MemoryHandler] = None
         for call in mock_root_logger.addHandler.call_args_list:
             handler = call[0][0]
             if memory_handler is None and isinstance(handler, logging.handlers.MemoryHandler):

certbot/tests/main_test.py

@@ -855,7 +855,7 @@ class MainTest(test_util.ConfigTestCase):
     @mock.patch('certbot._internal.main.plugins_disco')
     @mock.patch('certbot._internal.main.cli.HelpfulArgumentParser.determine_help_topics')
     def test_plugins_no_args(self, _det, mock_disco):
-        ifaces = []  # type: List[interfaces.IPlugin]
+        ifaces: List[interfaces.IPlugin] = []
         plugins = mock_disco.PluginsRegistry.find_all()
 
         stdout = io.StringIO()
@@ -870,7 +870,7 @@ class MainTest(test_util.ConfigTestCase):
     @mock.patch('certbot._internal.main.plugins_disco')
     @mock.patch('certbot._internal.main.cli.HelpfulArgumentParser.determine_help_topics')
     def test_plugins_no_args_unprivileged(self, _det, mock_disco):
-        ifaces = []  # type: List[interfaces.IPlugin]
+        ifaces: List[interfaces.IPlugin] = []
         plugins = mock_disco.PluginsRegistry.find_all()
 
         def throw_error(directory, mode, strict):
@@ -892,7 +892,7 @@ class MainTest(test_util.ConfigTestCase):
     @mock.patch('certbot._internal.main.plugins_disco')
     @mock.patch('certbot._internal.main.cli.HelpfulArgumentParser.determine_help_topics')
     def test_plugins_init(self, _det, mock_disco):
-        ifaces = []  # type: List[interfaces.IPlugin]
+        ifaces: List[interfaces.IPlugin] = []
         plugins = mock_disco.PluginsRegistry.find_all()
 
         stdout = io.StringIO()
@@ -910,7 +910,7 @@ class MainTest(test_util.ConfigTestCase):
     @mock.patch('certbot._internal.main.plugins_disco')
     @mock.patch('certbot._internal.main.cli.HelpfulArgumentParser.determine_help_topics')
     def test_plugins_prepare(self, _det, mock_disco):
-        ifaces = []  # type: List[interfaces.IPlugin]
+        ifaces: List[interfaces.IPlugin] = []
         plugins = mock_disco.PluginsRegistry.find_all()
 
         stdout = io.StringIO()

certbot/tests/plugins/disco_test.py

@@ -277,7 +277,7 @@ class PluginsRegistryTest(unittest.TestCase):
         self.plugin_ep.prepare.assert_called_once_with()
 
     def test_prepare_order(self):
-        order = []  # type: List[str]
+        order: List[str] = []
         plugins = dict(
             (c, mock.MagicMock(prepare=functools.partial(order.append, c)))
             for c in string.ascii_letters)

certbot/tests/plugins/selection_test.py

@@ -52,7 +52,7 @@ class PickPluginTest(unittest.TestCase):
         self.default = None
         self.reg = mock.MagicMock()
         self.question = "Question?"
-        self.ifaces = []  # type: List[interfaces.IPlugin]
+        self.ifaces: List[interfaces.IPlugin] = []
 
     def _call(self):
         from certbot._internal.plugins.selection import pick_plugin

certbot/tests/plugins/standalone_test.py

@@ -24,9 +24,8 @@ class ServerManagerTest(unittest.TestCase):
 
     def setUp(self):
         from certbot._internal.plugins.standalone import ServerManager
-        self.certs = {}  # type: Dict[bytes, Tuple[OpenSSL.crypto.PKey, OpenSSL.crypto.X509]]
+        self.certs: Dict[bytes, Tuple[OpenSSL.crypto.PKey, OpenSSL.crypto.X509]] = {}
-        self.http_01_resources = {} \
+        self.http_01_resources: Set[acme_standalone.HTTP01RequestHandler.HTTP01Resource] = {}
-        # type: Set[acme_standalone.HTTP01RequestHandler.HTTP01Resource]
         self.mgr = ServerManager(self.certs, self.http_01_resources)
 
     def test_init(self):