diff --git a/acme/acme/challenges.py b/acme/acme/challenges.py index 58b457e0d..82f7ae1b8 100644 --- a/acme/acme/challenges.py +++ b/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) diff --git a/acme/acme/client.py b/acme/acme/client.py index b1b2089cf..f5aa1ff9e 100644 --- a/acme/acme/client.py +++ b/acme/acme/client.py @@ -112,8 +112,9 @@ class ClientBase: """ return self.update_registration(regr, update={'status': 'deactivated'}) - def deactivate_authorization(self, authzr): - # type: (messages.AuthorizationResource) -> messages.AuthorizationResource + def deactivate_authorization(self, + 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 diff --git a/acme/acme/crypto_util.py b/acme/acme/crypto_util.py index 04faf6912..749478bf5 100644 --- a/acme/acme/crypto_util.py +++ b/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") diff --git a/acme/acme/messages.py b/acme/acme/messages.py index 0d73037ae..5e7e22c34 100644 --- a/acme/acme/messages.py +++ b/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.""" diff --git a/acme/acme/standalone.py b/acme/acme/standalone.py index 123899470..0b5a8b5c7 100644 --- a/acme/acme/standalone.py +++ b/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.servers = [] # type: List[ACMEServerMixin] + self.threads: List[threading.Thread] = [] + self.servers: List[ACMEServerMixin] = [] # Must try True first. # Ubuntu, for example, will fail to bind to IPv4 if we've already bound diff --git a/acme/tests/client_test.py b/acme/tests/client_test.py index 6f9aecda2..14247335c 100644 --- a/acme/tests/client_test.py +++ b/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') diff --git a/acme/tests/crypto_util_test.py b/acme/tests/crypto_util_test.py index f271ad37d..8075b68ed 100644 --- a/acme/tests/crypto_util_test.py +++ b/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) diff --git a/acme/tests/messages_test.py b/acme/tests/messages_test.py index 74d1737ec..99a3a9ce4 100644 --- a/acme/tests/messages_test.py +++ b/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') diff --git a/acme/tests/standalone_test.py b/acme/tests/standalone_test.py index e6aa8f2d6..a02d008a0 100644 --- a/acme/tests/standalone_test.py +++ b/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) diff --git a/certbot-apache/certbot_apache/_internal/augeasparser.py b/certbot-apache/certbot_apache/_internal/augeasparser.py index 461b5cb7b..4549e935a 100644 --- a/certbot-apache/certbot_apache/_internal/augeasparser.py +++ b/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] diff --git a/certbot-apache/certbot_apache/_internal/configurator.py b/certbot-apache/certbot_apache/_internal/configurator.py index ecd5c46de..29648d4c1 100644 --- a/certbot-apache/certbot_apache/_internal/configurator.py +++ b/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] - internal_paths = defaultdict(set) # type: DefaultDict[str, Set[str]] + file_paths: Dict[str, 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: diff --git a/certbot-apache/certbot_apache/_internal/http_01.py b/certbot-apache/certbot_apache/_internal/http_01.py index 4bd935d2d..66d888e97 100644 --- a/certbot-apache/certbot_apache/_internal/http_01.py +++ b/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 diff --git a/certbot-apache/certbot_apache/_internal/obj.py b/certbot-apache/certbot_apache/_internal/obj.py index 5e4680e1b..21fd042f6 100644 --- a/certbot-apache/certbot_apache/_internal/obj.py +++ b/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 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: diff --git a/certbot-apache/certbot_apache/_internal/override_centos.py b/certbot-apache/certbot_apache/_internal/override_centos.py index 4cde65d57..54edfc911 100644 --- a/certbot-apache/certbot_apache/_internal/override_centos.py +++ b/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] - loadmod_args = [] # type: List[str] - loadmod_paths = [] # type: List[str] + correct_ifmods: List[str] = [] + loadmod_args: List[str] = [] + loadmod_paths: List[str] = [] for m in loadmods: noarg_path = m.rpartition("/")[0] path_args = self.parser.get_all_args(noarg_path) diff --git a/certbot-apache/certbot_apache/_internal/parser.py b/certbot-apache/certbot_apache/_internal/parser.py index eb8f267f9..1a6af4c4b 100644 --- a/certbot-apache/certbot_apache/_internal/parser.py +++ b/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.parser_paths = {} # type: Dict[str, List[str]] - self.variables = {} # type: Dict[str, str] + self.modules: Dict[str, str] = {} + self.parser_paths: Dict[str, List[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 diff --git a/certbot-apache/tests/augeasnode_test.py b/certbot-apache/tests/augeasnode_test.py index abe72a5d0..83224fc98 100644 --- a/certbot-apache/tests/augeasnode_test.py +++ b/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) diff --git a/certbot-apache/tests/http_01_test.py b/certbot-apache/tests/http_01_test.py index 696cd4a54..06e19e445 100644 --- a/certbot-apache/tests/http_01_test.py +++ b/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 diff --git a/certbot-ci/certbot_integration_tests/nginx_tests/test_main.py b/certbot-ci/certbot_integration_tests/nginx_tests/test_main.py index 4c746e1ee..2c52c8523 100644 --- a/certbot-ci/certbot_integration_tests/nginx_tests/test_main.py +++ b/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): - # type: (str, List[str], nginx_context.IntegrationTestsContext) -> None +def test_certificate_deployment(certname_pattern: str, params: List[str], + context: nginx_context.IntegrationTestsContext) -> None: """ Test various scenarios to deploy a certificate to nginx using certbot. """ diff --git a/certbot-ci/certbot_integration_tests/utils/acme_server.py b/certbot-ci/certbot_integration_tests/utils/acme_server.py index 706c3ebae..ceebbe7ed 100755 --- a/certbot-ci/certbot_integration_tests/utils/acme_server.py +++ b/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 diff --git a/certbot-ci/certbot_integration_tests/utils/dns_server.py b/certbot-ci/certbot_integration_tests/utils/dns_server.py index eaa601f1b..31e7aee18 100644 --- a/certbot-ci/certbot_integration_tests/utils/dns_server.py +++ b/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): - # type: (int) -> None + def _wait_until_ready(self, attempts: int = 30) -> None: """ Polls the DNS server over TCP until it gets a response, or until it runs out of attempts and raises a ValueError. diff --git a/certbot-compatibility-test/certbot_compatibility_test/configurators/nginx/common.py b/certbot-compatibility-test/certbot_compatibility_test/configurators/nginx/common.py index 58b72c205..d7b7a120e 100644 --- a/certbot-compatibility-test/certbot_compatibility_test/configurators/nginx/common.py +++ b/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) diff --git a/certbot-compatibility-test/certbot_compatibility_test/test_driver.py b/certbot-compatibility-test/certbot_compatibility_test/test_driver.py index fe634ba9f..4bb16e63f 100644 --- a/certbot-compatibility-test/certbot_compatibility_test/test_driver.py +++ b/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: diff --git a/certbot-dns-cloudflare/certbot_dns_cloudflare/_internal/dns_cloudflare.py b/certbot-dns-cloudflare/certbot_dns_cloudflare/_internal/dns_cloudflare.py index 583b41410..034d7bbb2 100644 --- a/certbot-dns-cloudflare/certbot_dns_cloudflare/_internal/dns_cloudflare.py +++ b/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: diff --git a/certbot-dns-route53/certbot_dns_route53/_internal/dns_route53.py b/certbot-dns-route53/certbot_dns_route53/_internal/dns_route53.py index 4837e2036..4dda13f1d 100644 --- a/certbot-dns-route53/certbot_dns_route53/_internal/dns_route53.py +++ b/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" diff --git a/certbot-nginx/certbot_nginx/_internal/configurator.py b/certbot-nginx/certbot_nginx/_internal/configurator.py index 33baf076a..aab489315 100644 --- a/certbot-nginx/certbot_nginx/_internal/configurator.py +++ b/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_redirect_vhosts = {} # type: Dict[str, List[obj.VirtualHost]] + self._wildcard_vhosts: 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(), diff --git a/certbot-nginx/certbot_nginx/_internal/constants.py b/certbot-nginx/certbot_nginx/_internal/constants.py index 7b4111577..52f4d8238 100644 --- a/certbot-nginx/certbot_nginx/_internal/constants.py +++ b/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.""" diff --git a/certbot-nginx/certbot_nginx/_internal/http_01.py b/certbot-nginx/certbot_nginx/_internal/http_01.py index ec610f15b..abcdfbd53 100644 --- a/certbot-nginx/certbot_nginx/_internal/http_01.py +++ b/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) diff --git a/certbot-nginx/certbot_nginx/_internal/nginxparser.py b/certbot-nginx/certbot_nginx/_internal/nginxparser.py index 1aee1f329..787f7c363 100644 --- a/certbot-nginx/certbot_nginx/_internal/nginxparser.py +++ b/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 IO # pylint: disable=unused-import +from typing import Any +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)) diff --git a/certbot-nginx/certbot_nginx/_internal/parser.py b/certbot-nginx/certbot_nginx/_internal/parser.py index ed7e211d5..8b353c095 100644 --- a/certbot-nginx/certbot_nginx/_internal/parser.py +++ b/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] - ssl = False # type: bool - names = set() # type: Set[str] + addrs: Set[obj.Addr] = set() + ssl: bool = False + names: Set[str] = set() apply_ssl_to_all_addrs = False diff --git a/certbot-nginx/certbot_nginx/_internal/parser_obj.py b/certbot-nginx/certbot_nginx/_internal/parser_obj.py index 9efd6651a..37392ea80 100644 --- a/certbot-nginx/certbot_nginx/_internal/parser_obj.py +++ b/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.contents = None # type: Block + self.names: Sentence = None + self.contents: Block = None @staticmethod def should_parse(lists): diff --git a/certbot-nginx/tests/parser_test.py b/certbot-nginx/tests/parser_test.py index 23fe390ad..35173d094 100644 --- a/certbot-nginx/tests/parser_test.py +++ b/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 diff --git a/certbot/certbot/_internal/account.py b/certbot/certbot/_internal/account.py index 07ae95e9d..c5667a865 100644 --- a/certbot/certbot/_internal/account.py +++ b/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): - # type: (Account, ClientBase) -> None + def save(self, account: Account, client: 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): - # type: (Account, ClientBase) -> None + def update_regr(self, account: Account, client: 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): - # type: (Account) -> None + def update_meta(self, account: 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): - # type: (Account) -> str + def _prepare(self, account: 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): - # type: (Account, str) -> None + def _create(self, account: Account, dir_path: 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): - # type: (Account, ClientBase, str) -> None + def _update_regr(self, account: Account, acme: ClientBase, dir_path: 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 diff --git a/certbot/certbot/_internal/auth_handler.py b/certbot/certbot/_internal/auth_handler.py index 2801e3c2f..c2f323a36 100644 --- a/certbot/certbot/_internal/auth_handler.py +++ b/certbot/certbot/_internal/auth_handler.py @@ -98,8 +98,7 @@ class AuthHandler: return authzrs_validated - def deactivate_valid_authorizations(self, orderr): - # type: (messages.OrderResource) -> Tuple[List, List] + def deactivate_valid_authorizations(self, orderr: 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] diff --git a/certbot/certbot/_internal/cert_manager.py b/certbot/certbot/_internal/cert_manager.py index 2ac9fc538..8fab5735a 100644 --- a/certbot/certbot/_internal/cert_manager.py +++ b/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 diff --git a/certbot/certbot/_internal/cli/__init__.py b/certbot/certbot/_internal/cli/__init__.py index 30f444e5b..b64ab0e9c 100644 --- a/certbot/certbot/_internal/cli/__init__.py +++ b/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): diff --git a/certbot/certbot/_internal/cli/cli_utils.py b/certbot/certbot/_internal/cli/cli_utils.py index f1f9d0f3c..1f30261c1 100644 --- a/certbot/certbot/_internal/cli/cli_utils.py +++ b/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__ diff --git a/certbot/certbot/_internal/cli/helpful.py b/certbot/certbot/_internal/cli/helpful.py index cbac4e232..d36147089 100644 --- a/certbot/certbot/_internal/cli/helpful.py +++ b/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", diff --git a/certbot/certbot/_internal/client.py b/certbot/certbot/_internal/client.py index b9137e249..ba696bf85 100644 --- a/certbot/certbot/_internal/client.py +++ b/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): - # type: (str, bool) -> List[messages.OrderResource] + def _get_order_and_authorizations(self, csr_pem: str, + best_effort: bool) -> List[messages.OrderResource]: """Request a new order and complete its authorizations. :param str csr_pem: A CSR in PEM format. diff --git a/certbot/certbot/_internal/eff.py b/certbot/certbot/_internal/eff.py index 6212c24ee..b01e2dd61 100644 --- a/certbot/certbot/_internal/eff.py +++ b/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): - # type: (IConfig, Account) -> None +def prepare_subscription(config: IConfig, acc: 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): - # type: (IConfig, Account) -> None +def handle_subscription(config: IConfig, acc: 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(): - # type: () -> bool +def _want_subscription() -> 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): - # type: (str) -> None +def subscribe(email: 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): - # type: (requests.Response) -> None +def _check_response(response: 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): - # type: (Optional[str]) -> None +def _report_failure(reason: Optional[str] = None) -> None: """Notify the user of failing to sign them up for the newsletter. :param reason: a phrase describing what the problem was diff --git a/certbot/certbot/_internal/error_handler.py b/certbot/certbot/_internal/error_handler.py index cb8301980..01cc92b42 100644 --- a/certbot/certbot/_internal/error_handler.py +++ b/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.prev_handlers = {} # type: Dict[int, Union[int, None, Callable]] - self.received_signals = [] # type: List[int] + self.funcs: List[Callable[[], Any]] = [] + self.prev_handlers: Dict[int, Union[int, None, Callable]] = {} + 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): - # type: (Callable, *Any, **Any) -> None + def register(self, func: Callable, *args: Any, **kwargs: 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 diff --git a/certbot/certbot/_internal/hooks.py b/certbot/certbot/_internal/hooks.py index b695d9930..b9f1f1531 100644 --- a/certbot/certbot/_internal/hooks.py +++ b/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): diff --git a/certbot/certbot/_internal/lock.py b/certbot/certbot/_internal/lock.py index f878946cb..d00302598 100644 --- a/certbot/certbot/_internal/lock.py +++ b/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): - # type: (str) -> None + def __init__(self, path: 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): - # type: () -> str + def __repr__(self) -> 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): - # type: () -> None + def acquire(self) -> 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): - # type: () -> None + def release(self) -> None: """ Release the lock on the file, allowing any other Certbot instance to acquire it. """ self._lock_mechanism.release() - def is_locked(self): - # type: () -> bool + def is_locked(self) -> 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): - # type: (str) -> None + def __init__(self, path: 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): - # type: () -> bool + def is_locked(self) -> 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): - # type: () -> None + def acquire(self) -> 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): - # type: (int) -> None + def _try_lock(self, fd: 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): - # type: (int) -> bool + def _lock_success(self, fd: 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): - # type: () -> None + def release(self) -> 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')) diff --git a/certbot/certbot/_internal/main.py b/certbot/certbot/_internal/main.py index ba8d94bca..ac4d18449 100644 --- a/certbot/certbot/_internal/main.py +++ b/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): - # type: (configuration.NamespaceConfig, storage.RenewableCert) -> None +def _handle_unexpected_key_type_migration(config: configuration.NamespaceConfig, + 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 - domains, # type: List[str] - cert # type: storage.RenewableCert - ): - # type: (...) -> Tuple[str, Optional[storage.RenewableCert]] +def _handle_subset_cert_request(config: configuration.NamespaceConfig, + domains: List[str], + cert: storage.RenewableCert + ) -> 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 - lineage, # type: storage.RenewableCert - ): - # type: (...) -> Tuple[str, Optional[storage.RenewableCert]] +def _handle_identical_cert_request(config: configuration.NamespaceConfig, + lineage: storage.RenewableCert, + ) -> 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 - domains, # type: List[str] - certname # type: str - ): - # type: (...) -> Tuple[str, Optional[storage.RenewableCert]] +def _find_lineage_for_domains_and_certname(config: configuration.NamespaceConfig, + domains: List[str], + certname: str + ) -> 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")) \ - # type: Union[None, display_util.NoninteractiveDisplay, display_util.FileDisplay] + displayer: 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: diff --git a/certbot/certbot/_internal/plugins/disco.py b/certbot/certbot/_internal/plugins/disco.py index 27b2605e0..4097544e8 100644 --- a/certbot/certbot/_internal/plugins/disco.py +++ b/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 diff --git a/certbot/certbot/_internal/plugins/manual.py b/certbot/certbot/_internal/plugins/manual.py index c28e333b9..ed2e0559e 100644 --- a/certbot/certbot/_internal/plugins/manual.py +++ b/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 = {} \ - # type: Dict[achallenges.KeyAuthorizationAnnotatedChallenge, Dict[str, str]] + self.env: Dict[achallenges.KeyAuthorizationAnnotatedChallenge, Dict[str, str]] = {} self.subsequent_dns_challenge = False self.subsequent_any_challenge = False diff --git a/certbot/certbot/_internal/plugins/standalone.py b/certbot/certbot/_internal/plugins/standalone.py index f511c60e9..833651346 100644 --- a/certbot/certbot/_internal/plugins/standalone.py +++ b/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.http_01_resources = set() \ - # type: Set[acme_standalone.HTTP01RequestHandler.HTTP01Resource] + self.certs: Dict[bytes, Tuple[OpenSSL.crypto.PKey, OpenSSL.crypto.X509]] = {} + self.http_01_resources: Set[acme_standalone.HTTP01RequestHandler.HTTP01Resource] = set() self.servers = ServerManager(self.certs, self.http_01_resources) diff --git a/certbot/certbot/_internal/plugins/webroot.py b/certbot/certbot/_internal/plugins/webroot.py index 730241066..c5b436b65 100644 --- a/certbot/certbot/_internal/plugins/webroot.py +++ b/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.performed = collections.defaultdict(set) \ - # type: DefaultDict[str, Set[achallenges.KeyAuthorizationAnnotatedChallenge]] + self.full_roots: Dict[str, str] = {} + self.performed: DefaultDict[str, Set[AnnotatedChallenge]] = collections.defaultdict(set) # 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: diff --git a/certbot/certbot/_internal/renewal.py b/certbot/certbot/_internal/renewal.py index a09edfa1e..f29709ef4 100644 --- a/certbot/certbot/_internal/renewal.py +++ b/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): - # type: (interfaces.IConfig, Optional[List[str]], client.Client, storage.RenewableCert) -> None +def renew_cert(config: interfaces.IConfig, domains: Optional[List[str]], le_client: client.Client, + 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, - renew_skipped, parse_failures): - # type: (interfaces.IConfig, List[str], List[str], List[str], List[str]) -> None +def _renew_describe_results(config: interfaces.IConfig, renew_successes: List[str], + renew_failures: List[str], renew_skipped: List[str], + 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): - # type: (str, interfaces.IConfig) -> None +def _update_renewal_params_from_key(key_path: str, config: 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): diff --git a/certbot/certbot/_internal/reporter.py b/certbot/certbot/_internal/reporter.py index 3093ef8ca..0268fbf34 100644 --- a/certbot/certbot/_internal/reporter.py +++ b/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): diff --git a/certbot/certbot/_internal/snap_config.py b/certbot/certbot/_internal/snap_config.py index f2fe035db..1ba53cad4 100644 --- a/certbot/certbot/_internal/snap_config.py +++ b/certbot/certbot/_internal/snap_config.py @@ -33,8 +33,7 @@ _ARCH_TRIPLET_MAP = { LOGGER = logging.getLogger(__name__) -def prepare_env(cli_args): - # type: (List[str]) -> List[str] +def prepare_env(cli_args: List[str]) -> List[str]: """ Prepare runtime environment for a certbot execution in snap. :param list cli_args: List of command line arguments diff --git a/certbot/certbot/compat/filesystem.py b/certbot/certbot/compat/filesystem.py index aa6fe9134..25e1687da 100644 --- a/certbot/certbot/compat/filesystem.py +++ b/certbot/certbot/compat/filesystem.py @@ -35,8 +35,7 @@ class _WindowsUmask: _WINDOWS_UMASK = _WindowsUmask() -def chmod(file_path, mode): - # type: (str, int) -> None +def chmod(file_path: str, mode: 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): - # type: (int) -> int +def umask(mask: 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): - # type: (str, str, int, bool, bool) -> None +def copy_ownership_and_apply_mode(src: str, dst: str, mode: int, + 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): - # type: (str, str, bool, bool) -> None +def copy_ownership_and_mode(src: str, dst: str, + 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): - # type: (str, int) -> bool +def check_mode(file_path: str, mode: 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): - # type: (str) -> bool +def check_owner(file_path: 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): - # type: (str, int) -> bool +def check_permissions(file_path: str, mode: 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 - # type: (str, int, int) -> int +def open(file_path: str, flags: int, mode: int = 0o777) -> int: # pylint: disable=redefined-builtin """ 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): - # type: (str, int) -> None +def makedirs(file_path: str, mode: int = 0o777) -> 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): - # type: (str, int) -> None +def mkdir(file_path: str, mode: int = 0o777) -> 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): - # type: (str, str) -> None +def replace(src: str, dst: 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): - # type: (str) -> str +def realpath(file_path: 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): - # type: (str) -> str +def readlink(link_path: 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): - # type: (str) -> bool +def is_executable(path: 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): - # type: (str) -> bool +def has_world_permissions(path: 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): - # type: (str, int) -> int +def compute_private_key_mode(old_key: str, base_mode: 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): - # type: (str, str) -> bool +def has_same_ownership(path1: str, path2: 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): - # type: (str, int) -> bool +def has_min_permissions(path: str, min_mode: 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. diff --git a/certbot/certbot/compat/misc.py b/certbot/certbot/compat/misc.py index 297df80fc..3d2624906 100644 --- a/certbot/certbot/compat/misc.py +++ b/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(): - # type: () -> None +def raise_for_non_administrative_windows_rights() -> 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): - # type: (float, str) -> str +def readline_with_timeout(timeout: float, prompt: 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): - # type: (str) -> str +def get_default_folder(folder_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): - # type: (str) -> str +def underscores_for_unsupported_characters_in_path(path: 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): - # type: (str, str, Optional[dict]) -> Tuple[str, str] +def execute_command(cmd_name: str, shell_cmd: str, env: Optional[dict] = None) -> Tuple[str, str]: """ Run a command: - on Linux command will be run by the standard shell selected with Popen(shell=True) diff --git a/certbot/certbot/crypto_util.py b/certbot/certbot/crypto_util.py index 5bc9251a4..445618ea0 100644 --- a/certbot/certbot/crypto_util.py +++ b/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() diff --git a/certbot/certbot/display/util.py b/certbot/certbot/display/util.py index 1e85ae808..9443ae266 100644 --- a/certbot/certbot/display/util.py +++ b/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): - # type: (str) -> None +def notify(msg: 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): - # type: (List[str]) -> str +def summarize_domain_list(domains: 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: diff --git a/certbot/certbot/ocsp.py b/certbot/certbot/ocsp.py index 9a18b32f2..0a842e108 100644 --- a/certbot/certbot/ocsp.py +++ b/certbot/certbot/ocsp.py @@ -59,8 +59,7 @@ class RevocationChecker: else: self.host_args = lambda host: ["Host", host] - def ocsp_revoked(self, cert): - # type: (RenewableCert) -> bool + def ocsp_revoked(self, cert: 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): - # type: (str, str, int) -> bool + def ocsp_revoked_by_paths(self, cert_path: str, chain_path: str, timeout: int = 10) -> 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): - # type: (str, str, str, str, int) -> bool + def _check_ocsp_openssl_bin(self, cert_path: str, chain_path: str, + 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): - # type: (str) -> Tuple[Optional[str], Optional[str]] +def _determine_ocsp_server(cert_path: 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): - # type: (str, str, str, int) -> bool +def _check_ocsp_cryptography(cert_path: str, chain_path: str, url: str, timeout: int) -> bool: # Retrieve OCSP response with open(chain_path, 'rb') as file_handler: issuer = x509.load_pem_x509_certificate(file_handler.read(), default_backend()) diff --git a/certbot/certbot/plugins/common.py b/certbot/certbot/plugins/common.py index 5e1c29669..06acefc69 100644 --- a/certbot/certbot/plugins/common.py +++ b/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.indices = [] # type: List[int] + self.achalls: List[achallenges.KeyAuthorizationAnnotatedChallenge] = [] + self.indices: List[int] = [] def add_chall(self, achall, idx=None): """Store challenge to be performed when perform() is called. diff --git a/certbot/certbot/plugins/dns_common_lexicon.py b/certbot/certbot/plugins/dns_common_lexicon.py index 5fec5e0d6..10efa5f85 100644 --- a/certbot/certbot/plugins/dns_common_lexicon.py +++ b/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): - # type: (str, Dict, Dict) -> Union[ConfigResolver, Dict] +def build_lexicon_config(lexicon_provider_name: str, + 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 diff --git a/certbot/certbot/plugins/enhancements.py b/certbot/certbot/plugins/enhancements.py index 3c3db7e71..c1de4d44f 100644 --- a/certbot/certbot/plugins/enhancements.py +++ b/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]] +] diff --git a/certbot/certbot/plugins/storage.py b/certbot/certbot/plugins/storage.py index b5dd6a502..0d32d6850 100644 --- a/certbot/certbot/plugins/storage.py +++ b/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: diff --git a/certbot/certbot/util.py b/certbot/certbot/util.py index 0643d5ba7..5f4a08dc7 100644 --- a/certbot/certbot/util.py +++ b/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) diff --git a/certbot/tests/display/completer_test.py b/certbot/tests/display/completer_test.py index 8ae6290bf..4b08fe3a1 100644 --- a/certbot/tests/display/completer_test.py +++ b/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) diff --git a/certbot/tests/error_handler_test.py b/certbot/tests/error_handler_test.py index 9ccd63a3a..265a42023 100644 --- a/certbot/tests/error_handler_test.py +++ b/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) diff --git a/certbot/tests/hook_test.py b/certbot/tests/hook_test.py index 06a641216..b0522bd52 100644 --- a/certbot/tests/hook_test.py +++ b/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) diff --git a/certbot/tests/log_test.py b/certbot/tests/log_test.py index 088faa0fb..33d89c3ea 100644 --- a/certbot/tests/log_test.py +++ b/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): diff --git a/certbot/tests/main_test.py b/certbot/tests/main_test.py index 7668a1b6a..d4afed858 100644 --- a/certbot/tests/main_test.py +++ b/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() diff --git a/certbot/tests/plugins/disco_test.py b/certbot/tests/plugins/disco_test.py index a97de24b9..64829c8cc 100644 --- a/certbot/tests/plugins/disco_test.py +++ b/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) diff --git a/certbot/tests/plugins/selection_test.py b/certbot/tests/plugins/selection_test.py index a5de99e60..027ebdc35 100644 --- a/certbot/tests/plugins/selection_test.py +++ b/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 diff --git a/certbot/tests/plugins/standalone_test.py b/certbot/tests/plugins/standalone_test.py index 596cee622..80347d35a 100644 --- a/certbot/tests/plugins/standalone_test.py +++ b/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.http_01_resources = {} \ - # type: Set[acme_standalone.HTTP01RequestHandler.HTTP01Resource] + self.certs: Dict[bytes, Tuple[OpenSSL.crypto.PKey, OpenSSL.crypto.X509]] = {} + self.http_01_resources: Set[acme_standalone.HTTP01RequestHandler.HTTP01Resource] = {} self.mgr = ServerManager(self.certs, self.http_01_resources) def test_init(self):