Enable mypy strict mode (#8766)

Built on top of #8748, this PR reenables mypy strict mode and adds the appropriate corrections to pass the types checks.

* Upgrade mypy

* First step for acme

* Cast for the rescue

* Fixing types for certbot

* Fix typing for certbot-nginx

* Finalize type fixes, configure no optional strict check for mypy in tox

* Align requirements

* Isort

* Pylint

* Protocol for python 3.6

* Use Python 3.9 for mypy, make code compatible with Python 3.8<

* Pylint and mypy

* Pragma no cover

* Pythonic NotImplemented constant

* More type definitions

* Add comments

* Simplify typing logic

* Use vararg tuple

* Relax constraints on mypy

* Add more type

* Do not silence error if target is not defined

* Conditionally import Protocol for type checking only

* Clean up imports

* Add comments

* Align python version linting with mypy and coverage

* Just ignore types in an unused module

* Add comments

* Fix lint

* Work in progress

* Finish type control

* Isort

* Fix pylint

* Fix imports

* Fix cli subparser

* Some fixes

* Coverage

* Remove --no-strict-optional (obviously...)

* Update certbot-apache/certbot_apache/_internal/configurator.py

Co-authored-by: Brad Warren <bmw@users.noreply.github.com>

* Update certbot/certbot/_internal/display/completer.py

Co-authored-by: Brad Warren <bmw@users.noreply.github.com>

* Cleanup dns_google

* Improve lock controls and fix subparser

* Use the expected interfaces

* Fix code

Co-authored-by: Brad Warren <bmw@users.noreply.github.com>
This commit is contained in:
Adrien Ferrand
2021-04-05 15:04:21 -07:00
committed by GitHub
co-authored by Brad Warren
parent 0f9f902b6e
commit c438a397a0
32 changed files with 164 additions and 74 deletions
@@ -12,6 +12,7 @@ from typing import cast
from typing import DefaultDict
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from typing import Union
@@ -36,6 +37,9 @@ from certbot_apache._internal import dualparser
from certbot_apache._internal import http_01
from certbot_apache._internal import obj
from certbot_apache._internal import parser
from certbot_apache._internal.dualparser import DualBlockNode
from certbot_apache._internal.obj import VirtualHost
from certbot_apache._internal.parser import ApacheParser
try:
import apacheconfig
@@ -232,11 +236,11 @@ class ApacheConfigurator(common.Installer):
self.parsed_paths: List[str] = []
# These will be set in the prepare function
self._prepared = False
self.parser = None
self.parser_root = None
self.parser: ApacheParser
self.parser_root: Optional[DualBlockNode] = None
self.version = version
self._openssl_version = openssl_version
self.vhosts = None
self.vhosts: List[VirtualHost]
self.options = copy.deepcopy(self.OS_DEFAULTS)
self._enhance_func = {"redirect": self._enable_redirect,
"ensure-http-header": self._set_http_header,
@@ -345,8 +349,9 @@ class ApacheConfigurator(common.Installer):
"augeaspath": self.parser.get_root_augpath(),
"ac_ast": None}
if self.USE_PARSERNODE:
self.parser_root = self.get_parsernode_root(pn_meta)
self.parsed_paths = self.parser_root.parsed_paths()
parser_root = self.get_parsernode_root(pn_meta)
self.parser_root = parser_root
self.parsed_paths = parser_root.parsed_paths()
# Check for errors in parsing files with Augeas
self.parser.check_parsing_errors("httpd.aug")
@@ -408,7 +413,7 @@ class ApacheConfigurator(common.Installer):
super(ApacheConfigurator, self).recovery_routine()
# Reload configuration after these changes take effect if needed
# ie. ApacheParser has been initialized.
if self.parser:
if hasattr(self, "parser"):
# TODO: wrap into non-implementation specific parser interface
self.parser.aug.load()
@@ -1051,6 +1056,9 @@ class ApacheConfigurator(common.Installer):
:rtype: list
"""
if not self.parser_root:
raise errors.Error("This ApacheConfigurator instance is not" # pragma: no cover
" configured to use a node parser.")
vhs = []
vhosts = self.parser_root.find_blocks("VirtualHost", exclude=False)
for vhblock in vhosts:
@@ -119,8 +119,9 @@ class CentOSConfigurator(configurator.ApacheConfigurator):
else:
loadmod_args = path_args
if self.parser.not_modssl_ifmodule(noarg_path): # pylint: disable=no-member
if self.parser.loc["default"] in noarg_path:
centos_parser: CentOSParser = cast(CentOSParser, self.parser)
if centos_parser.not_modssl_ifmodule(noarg_path):
if centos_parser.loc["default"] in noarg_path:
# LoadModule already in the main configuration file
if ("ifmodule/" in noarg_path.lower() or
"ifmodule[1]" in noarg_path.lower()):
@@ -5,12 +5,18 @@ import logging
import re
from typing import Dict
from typing import List
from typing import Optional
from certbot import errors
from certbot.compat import os
from certbot_apache._internal import apache_util
from certbot_apache._internal import constants
try:
from augeas import Augeas
except ImportError: # pragma: no cover
Augeas = None # type: ignore
logger = logging.getLogger(__name__)
@@ -39,8 +45,7 @@ class ApacheParser:
self.configurator = configurator
# Initialize augeas
self.aug = None
self.init_augeas()
self.aug = init_augeas()
if not self.check_aug_version():
raise errors.NotSupportedError(
@@ -48,7 +53,7 @@ class ApacheParser:
"version 1.2.0 or higher, please make sure you have you have "
"those installed.")
self.modules: Dict[str, str] = {}
self.modules: Dict[str, Optional[str]] = {}
self.parser_paths: Dict[str, List[str]] = {}
self.variables: Dict[str, str] = {}
@@ -83,23 +88,6 @@ class ApacheParser:
if self.find_dir("Define", exclude=False):
raise errors.PluginError("Error parsing runtime variables")
def init_augeas(self):
""" Initialize the actual Augeas instance """
try:
import augeas
except ImportError: # pragma: no cover
raise errors.NoInstallationError("Problem in Augeas installation")
self.aug = augeas.Augeas(
# specify a directory to load our preferred lens from
loadpath=constants.AUGEAS_LENS_DIR,
# Do not save backup (we do it ourselves), do not load
# anything by default
flags=(augeas.Augeas.NONE |
augeas.Augeas.NO_MODL_AUTOLOAD |
augeas.Augeas.ENABLE_SPAN))
def check_parsing_errors(self, lens):
"""Verify Augeas can parse all of the lens files.
@@ -949,3 +937,19 @@ def get_aug_path(file_path):
"""
return "/files%s" % file_path
def init_augeas() -> Augeas:
""" Initialize the actual Augeas instance """
if not Augeas: # pragma: no cover
raise errors.NoInstallationError("Problem in Augeas installation")
return Augeas(
# specify a directory to load our preferred lens from
loadpath=constants.AUGEAS_LENS_DIR,
# Do not save backup (we do it ourselves), do not load
# anything by default
flags=(Augeas.NONE |
Augeas.NO_MODL_AUTOLOAD |
Augeas.ENABLE_SPAN))
+1 -1
View File
@@ -339,7 +339,7 @@ class ParserInitTest(util.ApacheTest):
shutil.rmtree(self.config_dir)
shutil.rmtree(self.work_dir)
@mock.patch("certbot_apache._internal.parser.ApacheParser.init_augeas")
@mock.patch("certbot_apache._internal.parser.init_augeas")
def test_prepare_no_augeas(self, mock_init_augeas):
from certbot_apache._internal.parser import ApacheParser
mock_init_augeas.side_effect = errors.NoInstallationError
@@ -17,7 +17,6 @@ class IntegrationTestsContext(certbot_context.IntegrationTestsContext):
self.request = request
self._dns_xdist = None
if hasattr(request.config, 'workerinput'): # Worker node
self._dns_xdist = request.config.workerinput['dns_xdist']
else: # Primary node
@@ -45,7 +44,6 @@ class IntegrationTestsContext(certbot_context.IntegrationTestsContext):
src_file = resource_filename('certbot_integration_tests',
'assets/bind-config/rfc2136-credentials-{}.ini.tpl'
.format(label))
contents = None
with open(src_file, 'r') as f:
contents = f.read().format(
@@ -8,6 +8,7 @@ import subprocess
import sys
import tempfile
import time
from typing import Optional
from pkg_resources import resource_filename
@@ -38,7 +39,7 @@ class DNSServer:
self.bind_root = tempfile.mkdtemp()
self.process: subprocess.Popen = None
self.process: Optional[subprocess.Popen] = None
self.dns_xdist = {"address": BIND_BIND_ADDRESS[0], "port": BIND_BIND_ADDRESS[1]}
@@ -119,6 +120,9 @@ class DNSServer:
but otherwise the contents are ignored.
:param int attempts: The number of attempts to make.
"""
if not self.process:
raise ValueError("DNS server has not been started. Please run start() first.")
for _ in range(attempts):
if self.process.poll():
raise ValueError("BIND9 server stopped unexpectedly")
@@ -33,7 +33,9 @@ class Proxy:
self.args = args
self.http_port = 80
self.https_port = 443
self._configurator = self._all_names = self._test_names = None
self._configurator = None
self._all_names = None
self._test_names = None
def __getattr__(self, name):
"""Wraps the configurator methods"""
@@ -93,5 +95,7 @@ class Proxy:
"""Installs cert"""
cert_path, key_path, chain_path = self.copy_certs_and_keys(
cert_path, key_path, chain_path)
if not self._configurator:
raise ValueError("Configurator plugin is not set.")
self._configurator.deploy_cert(
domain, cert_path, key_path, chain_path, fullchain_path)
@@ -3,6 +3,7 @@ import logging
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
import CloudFlare
import zope.interface
@@ -10,6 +11,7 @@ import zope.interface
from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -30,7 +32,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -79,6 +81,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_cloudflare_client().del_txt_record(domain, validation_name, validation)
def _get_cloudflare_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
if self.credentials.conf('api-token'):
return _CloudflareClient(None, self.credentials.conf('api-token'))
return _CloudflareClient(self.credentials.conf('email'), self.credentials.conf('api-key'))
@@ -1,5 +1,6 @@
"""DNS Authenticator for CloudXNS DNS."""
import logging
from typing import Optional
from lexicon.providers import cloudxns
import zope.interface
@@ -8,6 +9,7 @@ from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -27,7 +29,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -56,6 +58,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_cloudxns_client().del_txt_record(domain, validation_name, validation)
def _get_cloudxns_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _CloudXNSLexiconClient(self.credentials.conf('api-key'),
self.credentials.conf('secret-key'),
self.ttl)
@@ -1,5 +1,6 @@
"""DNS Authenticator for DigitalOcean."""
import logging
from typing import Optional
import digitalocean
import zope.interface
@@ -7,6 +8,7 @@ import zope.interface
from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -25,7 +27,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -53,6 +55,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_digitalocean_client().del_txt_record(domain, validation_name, validation)
def _get_digitalocean_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _DigitalOceanClient(self.credentials.conf('token'))
@@ -1,5 +1,6 @@
"""DNS Authenticator for DNSimple DNS."""
import logging
from typing import Optional
from lexicon.providers import dnsimple
import zope.interface
@@ -8,6 +9,7 @@ from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -27,7 +29,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -54,6 +56,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_dnsimple_client().del_txt_record(domain, validation_name, validation)
def _get_dnsimple_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _DNSimpleLexiconClient(self.credentials.conf('token'), self.ttl)
@@ -1,5 +1,6 @@
"""DNS Authenticator for DNS Made Easy DNS."""
import logging
from typing import Optional
from lexicon.providers import dnsmadeeasy
import zope.interface
@@ -8,6 +9,7 @@ from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -28,7 +30,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -58,6 +60,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_dnsmadeeasy_client().del_txt_record(domain, validation_name, validation)
def _get_dnsmadeeasy_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _DNSMadeEasyLexiconClient(self.credentials.conf('api-key'),
self.credentials.conf('secret-key'),
self.ttl)
@@ -1,12 +1,15 @@
"""DNS Authenticator for Gehirn Infrastructure Service DNS."""
import logging
from typing import Optional
from lexicon.providers import gehirn
import zope.interface
from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -27,7 +30,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -57,6 +60,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_gehirn_client().del_txt_record(domain, validation_name, validation)
def _get_gehirn_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _GehirnLexiconClient(
self.credentials.conf('api-token'),
self.credentials.conf('api-secret'),
@@ -32,10 +32,6 @@ class Authenticator(dns_common.DNSAuthenticator):
'for DNS).')
ttl = 60
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
super(Authenticator, cls).add_parser_arguments(add, default_propagation_seconds=60)
@@ -1,6 +1,7 @@
"""DNS Authenticator for Linode."""
import logging
import re
from typing import Optional
from lexicon.providers import linode
from lexicon.providers import linode4
@@ -10,6 +11,7 @@ from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -28,7 +30,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -56,6 +58,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_linode_client().del_txt_record(domain, validation_name, validation)
def _get_linode_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
api_key = self.credentials.conf('key')
api_version = self.credentials.conf('version')
if api_version == '':
@@ -1,5 +1,6 @@
"""DNS Authenticator for LuaDNS DNS."""
import logging
from typing import Optional
from lexicon.providers import luadns
import zope.interface
@@ -8,6 +9,7 @@ from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -27,7 +29,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -55,6 +57,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_luadns_client().del_txt_record(domain, validation_name, validation)
def _get_luadns_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _LuaDNSLexiconClient(self.credentials.conf('email'),
self.credentials.conf('token'),
self.ttl)
@@ -1,5 +1,6 @@
"""DNS Authenticator for NS1 DNS."""
import logging
from typing import Optional
from lexicon.providers import nsone
import zope.interface
@@ -8,6 +9,7 @@ from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -27,7 +29,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -54,6 +56,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_nsone_client().del_txt_record(domain, validation_name, validation)
def _get_nsone_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _NS1LexiconClient(self.credentials.conf('api-key'), self.ttl)
@@ -1,5 +1,6 @@
"""DNS Authenticator for OVH DNS."""
import logging
from typing import Optional
from lexicon.providers import ovh
import zope.interface
@@ -8,6 +9,7 @@ from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -27,7 +29,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -60,6 +62,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_ovh_client().del_txt_record(domain, validation_name, validation)
def _get_ovh_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _OVHLexiconClient(
self.credentials.conf('endpoint'),
self.credentials.conf('application-key'),
@@ -1,5 +1,6 @@
"""DNS Authenticator using RFC 2136 Dynamic Updates."""
import logging
from typing import Optional
import dns.flags
import dns.message
@@ -15,6 +16,7 @@ import zope.interface
from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -44,7 +46,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -80,6 +82,8 @@ class Authenticator(dns_common.DNSAuthenticator):
self._get_rfc2136_client().del_txt_record(validation_name, validation)
def _get_rfc2136_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _RFC2136Client(self.credentials.conf('server'),
int(self.credentials.conf('port') or self.PORT),
self.credentials.conf('name'),
@@ -1,12 +1,15 @@
"""DNS Authenticator for Sakura Cloud DNS."""
import logging
from typing import Optional
from lexicon.providers import sakuracloud
import zope.interface
from certbot import errors
from certbot import interfaces
from certbot.plugins import dns_common
from certbot.plugins import dns_common_lexicon
from certbot.plugins.dns_common import CredentialsConfiguration
logger = logging.getLogger(__name__)
@@ -27,7 +30,7 @@ class Authenticator(dns_common.DNSAuthenticator):
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.credentials = None
self.credentials: Optional[CredentialsConfiguration] = None
@classmethod
def add_parser_arguments(cls, add): # pylint: disable=arguments-differ
@@ -60,6 +63,8 @@ class Authenticator(dns_common.DNSAuthenticator):
domain, validation_name, validation)
def _get_sakuracloud_client(self):
if not self.credentials: # pragma: no cover
raise errors.Error("Plugin has not been prepared.")
return _SakuraCloudLexiconClient(
self.credentials.conf('api-token'),
self.credentials.conf('api-secret'),
@@ -119,7 +119,6 @@ class NginxConfigurator(common.Installer):
self._chall_out = 0
# These will be set in the prepare function
self.parser: Optional[parser.NginxParser] = None
self.version = version
self.openssl_version = openssl_version
self._enhance_func = {"redirect": self._enable_redirect,
@@ -127,6 +126,7 @@ class NginxConfigurator(common.Installer):
"staple-ocsp": self._enable_ocsp_stapling}
self.reverter.recovery_routine()
self.parser: parser.NginxParser
@property
def mod_ssl_conf_src(self):
+1 -2
View File
@@ -32,8 +32,7 @@ def _create_subparsers(helpful):
" Currently --csr only works with the 'certonly' subcommand.")
helpful.add("revoke",
"--reason", dest="reason",
choices=CaseInsensitiveList(sorted(constants.REVOCATION_REASONS,
key=constants.REVOCATION_REASONS.get)),
choices=CaseInsensitiveList(constants.REVOCATION_REASONS.keys()),
action=_EncodeReasonAction, default=flag_default("reason"),
help="Specify reason for revoking certificate. (default: unspecified)")
helpful.add("revoke",
+4
View File
@@ -254,6 +254,7 @@ class Client:
acme = acme_from_config_key(config, self.account.key, self.account.regr)
self.acme = acme
self.auth_handler: Optional[auth_handler.AuthHandler]
if auth is not None:
self.auth_handler = auth_handler.AuthHandler(
auth, self.acme, self.account, self.config.pref_challs)
@@ -407,6 +408,9 @@ class Client:
raise errors.Error("The currently selected ACME CA endpoint does"
" not support issuing wildcard certificates.")
if not self.auth_handler:
raise errors.Error("No authorization handler has been set.")
# For a dry run, ensure we have an order with fresh authorizations
if orderr and self.config.dry_run:
deactivated, failed = self.auth_handler.deactivate_valid_authorizations(orderr)
@@ -1,5 +1,8 @@
"""Provides Tab completion when prompting users for a path."""
import glob
from typing import Callable
from typing import Iterator
from typing import Optional
# readline module is not available on all systems
try:
@@ -26,7 +29,9 @@ class Completer:
"""
def __init__(self):
self._iter = self._original_completer = self._original_delims = None
self._iter: Iterator[str]
self._original_completer: Optional[Callable]
self._original_delims: str
def complete(self, text, state):
"""Provides path completion for use with readline.
+2
View File
@@ -223,6 +223,8 @@ class _WindowsLockMechanism(_BaseLockMechanism):
def release(self):
"""Release the lock."""
try:
if not self._fd:
raise errors.Error("The lock has not been acquired first.")
# This "type: ignore" is currently needed because msvcrt methods
# are only defined on Windows. See
# https://github.com/python/typeshed/blob/16ae4c61201cd8b96b8b22cdfb2ab9e89ba5bcf2/stdlib/msvcrt.pyi.
+23 -13
View File
@@ -1,18 +1,20 @@
"""Utilities for plugins discovery and selection."""
from collections.abc import Mapping
import itertools
import logging
import sys
from collections.abc import Mapping
from typing import Dict
from typing import Optional
from typing import Union
import pkg_resources
import zope.interface
import zope.interface.verify
from certbot import errors
from certbot import interfaces
from certbot._internal import constants
from certbot.compat import os
from certbot.errors import Error
logger = logging.getLogger(__name__)
@@ -44,15 +46,15 @@ class PluginEntryPoint:
# this object is mutable, don't allow it to be hashed!
__hash__ = None # type: ignore
def __init__(self, entry_point, with_prefix=False):
def __init__(self, entry_point: pkg_resources.EntryPoint, with_prefix=False):
self.name = self.entry_point_to_plugin_name(entry_point, with_prefix)
self.plugin_cls = entry_point.load()
self.plugin_cls: interfaces.IPluginFactory = entry_point.load()
self.entry_point = entry_point
self.warning_message = None
self._initialized = None
self._prepared = None
self.warning_message: Optional[str] = None
self._initialized: Optional[interfaces.IPlugin] = None
self._prepared: Optional[Union[bool, Error]] = None
self._hidden = False
self._long_description = None
self._long_description: Optional[str] = None
def check_name(self, name):
"""Check if the name refers to this plugin."""
@@ -118,12 +120,15 @@ class PluginEntryPoint:
"""Memoized plugin initialization."""
if not self.initialized:
self.entry_point.require() # fetch extras!
self._initialized = self.plugin_cls(config, self.name)
# TODO: remove type ignore once the interface becomes a proper
# abstract class (using abc) that mypy understands.
self._initialized = self.plugin_cls(config, self.name) # type: ignore
return self._initialized
def verify(self, ifaces):
"""Verify that the plugin conforms to the specified interfaces."""
assert self.initialized
if not self.initialized:
raise ValueError("Plugin is not initialized.")
for iface in ifaces: # zope.interface.providedBy(plugin)
try:
zope.interface.verify.verifyObject(iface, self.init())
@@ -144,10 +149,13 @@ class PluginEntryPoint:
def prepare(self):
"""Memoized plugin preparation."""
assert self.initialized
if self._initialized is None:
raise ValueError("Plugin is not initialized.")
if self._prepared is None:
try:
self._initialized.prepare()
# TODO: remove type ignore once the interface becomes a proper
# abstract class (using abc) that mypy understands.
self._initialized.prepare() # type: ignore
except errors.MisconfigurationError as error:
logger.debug("Misconfigured %r: %s", self, error, exc_info=True)
self._prepared = error
@@ -247,8 +255,10 @@ class PluginsRegistry(Mapping):
plugin_ep = PluginEntryPoint(entry_point, with_prefix)
if plugin_ep.name in plugins:
other_ep = plugins[plugin_ep.name]
plugin1 = plugin_ep.entry_point.dist.key if plugin_ep.entry_point.dist else "unknown"
plugin2 = other_ep.entry_point.dist.key if other_ep.entry_point.dist else "unknown"
raise Exception("Duplicate plugin name {0} from {1} and {2}.".format(
plugin_ep.name, plugin_ep.entry_point.dist.key, other_ep.entry_point.dist.key))
plugin_ep.name, plugin1, plugin2))
if interfaces.IPluginFactory.providedBy(plugin_ep.plugin_cls):
plugins[plugin_ep.name] = plugin_ep
else: # pragma: no cover
+2 -1
View File
@@ -1,5 +1,6 @@
"""Certbot client interfaces."""
import abc
from typing import Optional
import zope.interface
@@ -327,7 +328,7 @@ class IInstaller(IPlugin):
"""
def save(title=None, temporary=False):
def save(title: Optional[str] = None, temporary: bool = False):
"""Saves all changes to the configuration files.
Both title and temporary are needed because a save may be
+2 -1
View File
@@ -142,7 +142,8 @@ class DNSAuthenticator(common.Plugin):
setattr(self.config, self.dest(key), os.path.abspath(os.path.expanduser(new_value)))
def _configure_credentials(self, key, label, required_variables=None, validator=None):
def _configure_credentials(self, key, label, required_variables=None,
validator=None) -> 'CredentialsConfiguration':
"""
As `_configure_file`, but for a credential configuration file.
@@ -17,8 +17,10 @@ from certbot.plugins import dns_common
# if Lexicon is not available, obviously.
try:
from lexicon.config import ConfigResolver
from lexicon.providers.base import Provider
except ImportError:
ConfigResolver = None # type: ignore
Provider = None # type: ignore
logger = logging.getLogger(__name__)
@@ -29,7 +31,7 @@ class LexiconClient:
"""
def __init__(self):
self.provider = None
self.provider: Provider
def add_txt_record(self, domain, record_name, record_content):
"""
+2 -2
View File
@@ -26,8 +26,8 @@ class PluginStorage:
self._config = config
self._classkey = classkey
self._initialized = False
self._data = None
self._storagepath = None
self._data: Dict
self._storagepath: str
def _initialize_storage(self):
"""Initializes PluginStorage data and reads current state from the disk
+3 -2
View File
@@ -31,7 +31,7 @@ class PluginStorageTest(test_util.ConfigTestCase):
# When unable to read file that exists
mock_open = mock.mock_open()
mock_open.side_effect = IOError
self.plugin.storage.storagepath = os.path.join(self.config.config_dir,
self.plugin.storage._storagepath = os.path.join(self.config.config_dir,
".pluginstorage.json")
with mock.patch("builtins.open", mock_open):
with mock.patch('certbot.compat.os.path.isfile', return_value=True):
@@ -67,7 +67,7 @@ class PluginStorageTest(test_util.ConfigTestCase):
with mock.patch("certbot.plugins.storage.logger.error") as mock_log:
# Set data as something that can't be serialized
self.plugin.storage._initialized = True # pylint: disable=protected-access
self.plugin.storage.storagepath = "/tmp/whatever"
self.plugin.storage._storagepath = "/tmp/whatever"
self.plugin.storage._data = self.plugin_cls # pylint: disable=protected-access
self.assertRaises(errors.PluginStorageError,
self.plugin.storage.save)
@@ -80,6 +80,7 @@ class PluginStorageTest(test_util.ConfigTestCase):
with mock.patch("certbot.plugins.storage.logger.error") as mock_log:
self.plugin.storage._data = {"valid": "data"} # pylint: disable=protected-access
self.plugin.storage._initialized = True # pylint: disable=protected-access
self.plugin.storage._storagepath = "/tmp/whatever"
self.assertRaises(errors.PluginStorageError,
self.plugin.storage.save)
self.assertTrue("Could not write" in mock_log.call_args[0][0])
+1 -1
View File
@@ -156,7 +156,7 @@ commands =
basepython = python3
commands =
{[base]install_packages}
mypy --no-strict-optional {[base]source_paths}
mypy {[base]source_paths}
[testenv:apacheconftest]
commands =