starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-31 18:34:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/2.0.x' into ecdsa-default-flag

Browse Source
This commit is contained in:
Alex Zorin
2022-09-27 12:38:20 +10:00
parent 82b6e15be7 212c2ba990
commit f6d532a15b
134 changed files with 1120 additions and 5789 deletions
Download Patch File Download Diff File Expand all files Collapse all files
certbot/CHANGELOG.md
+30 -1
View File
@@ -2,7 +2,7 @@
Certbot adheres to [Semantic Versioning](https://semver.org/).
## 1.30.0 - master
## 1.31.0 - master
### Added
@@ -18,6 +18,35 @@ Certbot adheres to [Semantic Versioning](https://semver.org/).
More details about these changes can be found on our GitHub repo.
## 1.30.0 - 2022-09-07
### Added
*
### Changed
* `acme.client.ClientBase`, `acme.messages.Authorization.resolved_combinations`,
`acme.messages.Authorization.combinations`, `acme.mixins`, `acme.fields.resource`,
and `acme.fields.Resource` are deprecated and will be removed in a future release.
* `acme.messages.OLD_ERROR_PREFIX` (`urn:acme:error:`) is deprecated and support for
the old ACME error prefix in Certbot will be removed in the next major release of
Certbot.
* `acme.messages.Directory.register` is deprecated and will be removed in the next
major release of Certbot. Furthermore, `.Directory` will only support lookups
by the exact resource name string in the ACME directory (e.g. `directory['newOrder']`).
* The `certbot-dns-cloudxns` plugin is now deprecated and will be removed in the
next major release of Certbot.
* The `source_address` argument for `acme.client.ClientNetwork` is deprecated
and support for it will be removed in the next major release.
* Add UI text suggesting users create certs for multiple domains, when possible
### Fixed
*
More details about these changes can be found on our GitHub repo.
## 1.29.0 - 2022-07-05
### Added
certbot/certbot/__init__.py
+1 -1
View File
@@ -1,3 +1,3 @@
"""Certbot client."""
# version number like 1.2.3a0, must have at least 2 parts, like 1.2
__version__ = '1.30.0.dev0'
__version__ = '1.31.0.dev0'
certbot/certbot/_internal/account.py
+7 -7
View File
@@ -20,7 +20,7 @@ import pytz
from acme import fields as acme_fields
from acme import messages
from acme.client import ClientBase
from acme.client import ClientV2
from certbot import configuration
from certbot import errors
from certbot import interfaces
@@ -114,7 +114,7 @@ class AccountMemoryStorage(interfaces.AccountStorage):
def find_all(self) -> List[Account]:
return list(self.accounts.values())
def save(self, account: Account, client: ClientBase) -> None:
def save(self, account: Account, client: ClientV2) -> None:
if account.id in self.accounts:
logger.debug("Overwriting account: %s", account.id)
self.accounts[account.id] = account
@@ -243,11 +243,11 @@ class AccountFileStorage(interfaces.AccountStorage):
def load(self, account_id: str) -> Account:
return self._load_for_server_path(account_id, self.config.server_path)
def save(self, account: Account, client: ClientBase) -> None:
def save(self, account: Account, client: ClientV2) -> None:
"""Create a new account.
:param Account account: account to create
:param ClientBase client: ACME client associated to the account
:param ClientV2 client: ACME client associated to the account
"""
try:
@@ -258,11 +258,11 @@ class AccountFileStorage(interfaces.AccountStorage):
except IOError as error:
raise errors.AccountStorageError(error)
def update_regr(self, account: Account, client: ClientBase) -> None:
def update_regr(self, account: Account, client: ClientV2) -> None:
"""Update the registration resource.
:param Account account: account to update
:param ClientBase client: ACME client associated to the account
:param ClientV2 client: ACME client associated to the account
"""
try:
@@ -358,7 +358,7 @@ class AccountFileStorage(interfaces.AccountStorage):
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: Account, acme: ClientBase, dir_path: str) -> None:
def _update_regr(self, account: Account, acme: ClientV2, 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
certbot/certbot/_internal/auth_handler.py
+5 -52
View File
@@ -36,7 +36,7 @@ class AuthHandler:
:class:`~acme.challenges.Challenge` types
:type auth: certbot.interfaces.Authenticator
:ivar acme.client.BackwardsCompatibleClientV2 acme_client: ACME client API.
:ivar acme.client.ClientV2 acme_client: ACME client API.
:ivar account: Client's Account
:type account: :class:`certbot._internal.account.Account`
@@ -226,15 +226,10 @@ class AuthHandler:
logger.info("Performing the following challenges:")
for authzr in pending_authzrs:
authzr_challenges = authzr.body.challenges
if self.acme.acme_version == 1:
combinations = authzr.body.combinations
else:
combinations = tuple((i,) for i in range(len(authzr_challenges)))
path = gen_challenge_path(
authzr_challenges,
self._get_chall_pref(authzr.body.identifier.value),
combinations)
self._get_chall_pref(authzr.body.identifier.value))
achalls.extend(self._challenge_factory(authzr, path))
@@ -387,12 +382,9 @@ def challb_to_achall(challb: messages.ChallengeBody, account_key: josepy.JWK,
def gen_challenge_path(challbs: List[messages.ChallengeBody],
preferences: List[Type[challenges.Challenge]],
combinations: Tuple[Tuple[int, ...], ...]) -> Tuple[int, ...]:
preferences: List[Type[challenges.Challenge]]) -> Tuple[int, ...]:
"""Generate a plan to get authority over the identity.
.. todo:: This can be possibly be rewritten to use resolved_combinations.
:param tuple challbs: A tuple of challenges
(:class:`acme.messages.Challenge`) from
:class:`acme.messages.AuthorizationResource` to be
@@ -402,10 +394,6 @@ def gen_challenge_path(challbs: List[messages.ChallengeBody],
:param list preferences: List of challenge preferences for domain
(:class:`acme.challenges.Challenge` subclasses)
:param tuple combinations: A collection of sets of challenges from
:class:`acme.messages.Challenge`, each of which would
be sufficient to prove possession of the identifier.
:returns: list of indices from ``challenges``.
:rtype: list
@@ -413,21 +401,6 @@ def gen_challenge_path(challbs: List[messages.ChallengeBody],
path cannot be created that satisfies the CA given the preferences and
combinations.
"""
if combinations:
return _find_smart_path(challbs, preferences, combinations)
return _find_dumb_path(challbs, preferences)
def _find_smart_path(challbs: List[messages.ChallengeBody],
preferences: List[Type[challenges.Challenge]],
combinations: Tuple[Tuple[int, ...], ...]
) -> Tuple[int, ...]:
"""Find challenge path with server hints.
Can be called if combinations is included. Function uses a simple
ranking system to choose the combo with the lowest cost.
"""
chall_cost = {}
max_cost = 1
@@ -441,6 +414,8 @@ def _find_smart_path(challbs: List[messages.ChallengeBody],
# Set above completing all of the available challenges
best_combo_cost = max_cost
combinations = tuple((i,) for i in range(len(challbs)))
combo_total = 0
for combo in combinations:
for challenge_index in combo:
@@ -459,28 +434,6 @@ def _find_smart_path(challbs: List[messages.ChallengeBody],
return best_combo
def _find_dumb_path(challbs: List[messages.ChallengeBody],
preferences: List[Type[challenges.Challenge]]) -> Tuple[int, ...]:
"""Find challenge path without server hints.
Should be called if the combinations hint is not included by the
server. This function either returns a path containing all
challenges provided by the CA or raises an exception.
"""
path = []
for i, challb in enumerate(challbs):
# supported is set to True if the challenge type is supported
supported = next((True for pref_c in preferences
if isinstance(challb.chall, pref_c)), False)
if supported:
path.append(i)
else:
raise _report_no_chall_path(challbs)
return tuple(path)
def _report_no_chall_path(challbs: List[messages.ChallengeBody]) -> errors.AuthorizationError:
"""Logs and return a raisable error reporting that no satisfiable chall path exists.
certbot/certbot/_internal/cli/plugins_parsing.py
-4
View File
@@ -45,10 +45,6 @@ def _plugins_parsing(helpful: "helpful.HelpfulArgumentParser",
default=flag_default("dns_cloudflare"),
help=("Obtain certificates using a DNS TXT record (if you are "
"using Cloudflare for DNS)."))
helpful.add(["plugins", "certonly"], "--dns-cloudxns", action="store_true",
default=flag_default("dns_cloudxns"),
help=("Obtain certificates using a DNS TXT record (if you are "
"using CloudXNS for DNS)."))
helpful.add(["plugins", "certonly"], "--dns-digitalocean", action="store_true",
default=flag_default("dns_digitalocean"),
help=("Obtain certificates using a DNS TXT record (if you are "
certbot/certbot/_internal/client.py
+10 -24
View File
@@ -10,7 +10,6 @@ from typing import IO
from typing import List
from typing import Optional
from typing import Tuple
import warnings
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric.rsa import generate_private_key
@@ -70,16 +69,8 @@ def acme_from_config_key(config: configuration.NamespaceConfig, key: jose.JWK,
verify_ssl=(not config.no_verify_ssl),
user_agent=determine_user_agent(config))
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
client = acme_client.BackwardsCompatibleClientV2(net, key, config.server)
if client.acme_version == 1:
logger.warning(
"Certbot is configured to use an ACMEv1 server (%s). ACMEv1 support is deprecated"
" and will soon be removed. See https://community.letsencrypt.org/t/143839 for "
"more information.", config.server)
return cast(acme_client.ClientV2, client)
directory = acme_client.ClientV2.get_directory(config.server, net)
return acme_client.ClientV2(directory, net)
def determine_user_agent(config: configuration.NamespaceConfig) -> str:
@@ -256,18 +247,13 @@ def perform_registration(acme: acme_client.ClientV2, config: configuration.Names
" Please use --eab-kid and --eab-hmac-key.")
raise errors.Error(msg)
tos = acme.directory.meta.terms_of_service
if tos_cb and tos:
tos_cb(tos)
try:
newreg = messages.NewRegistration.from_data(
email=config.email, external_account_binding=eab)
# Until ACME v1 support is removed from Certbot, we actually need the provided
# ACME client to be a wrapper of type BackwardsCompatibleClientV2.
# TODO: Remove this cast and rewrite the logic when the client is actually a ClientV2
try:
return cast(acme_client.BackwardsCompatibleClientV2,
acme).new_account_and_tos(newreg, tos_cb)
except AttributeError:
raise errors.Error("The ACME client must be an instance of "
"acme.client.BackwardsCompatibleClientV2")
return acme.new_account(messages.NewRegistration.from_data(
email=config.email, terms_of_service_agreed=True, external_account_binding=eab))
except messages.Error as e:
if e.code in ("invalidEmail", "invalidContact"):
if config.noninteractive_mode:
@@ -291,8 +277,8 @@ class Client:
:ivar .Authenticator auth: Prepared (`.Authenticator.prepare`)
authenticator that can solve ACME challenges.
:ivar .Installer installer: Installer.
:ivar acme.client.BackwardsCompatibleClientV2 acme: Optional ACME
client API handle. You might already have one from `register`.
:ivar acme.client.ClientV2 acme: Optional ACME client API handle. You might
already have one from `register`.
"""
certbot/certbot/_internal/constants.py
-1
View File
@@ -112,7 +112,6 @@ CLI_DEFAULTS: Dict[str, Any] = dict( # noqa
manual=False,
webroot=False,
dns_cloudflare=False,
dns_cloudxns=False,
dns_digitalocean=False,
dns_dnsimple=False,
dns_dnsmadeeasy=False,
certbot/certbot/_internal/display/obj.py
+1 -14
View File
@@ -10,11 +10,7 @@ from typing import Tuple
from typing import TypeVar
from typing import Union
import zope.component
import zope.interface
from certbot import errors
from certbot import interfaces
from certbot._internal import constants
from certbot._internal.display import completer
from certbot._internal.display import util
@@ -34,6 +30,7 @@ SIDE_FRAME = ("- " * 39) + "-"
"""Display boundary (alternates spaces, so when copy-pasted, markdown doesn't interpret
it as a heading)"""
# This class holds the global state of the display service. Using this class
# eliminates potential gotchas that exist if self.display was just a global
# variable. In particular, in functions `_DISPLAY = <value>` would create a
@@ -50,9 +47,6 @@ _SERVICE = _DisplayService()
T = TypeVar("T")
# This use of IDisplay can be removed when this class is no longer accessible
# through the public API in certbot.display.util.
@zope.interface.implementer(interfaces.IDisplay)
class FileDisplay:
"""File-based display."""
# see https://github.com/certbot/certbot/issues/3915
@@ -410,9 +404,6 @@ class FileDisplay:
return OK, selection
# This use of IDisplay can be removed when this class is no longer accessible
# through the public API in certbot.display.util.
@zope.interface.implementer(interfaces.IDisplay)
class NoninteractiveDisplay:
"""A display utility implementation that never asks for interactive user input"""
@@ -573,8 +564,4 @@ def set_display(display: Union[FileDisplay, NoninteractiveDisplay]) -> None:
:param Union[FileDisplay, NoninteractiveDisplay] display: the display service
"""
# This call is done only for retro-compatibility purposes.
# TODO: Remove this call once zope dependencies are removed from Certbot.
zope.component.provideUtility(display, interfaces.IDisplay)
_SERVICE.display = display
certbot/certbot/_internal/lock.py
+3 -3
View File
@@ -194,7 +194,7 @@ class _WindowsLockMechanism(_BaseLockMechanism):
low level APIs, and Python does not do it. As of Python 3.7 and below, Python developers
state that deleting a file opened by a process from another process is not possible with
os.open and io.open.
Consequently, mscvrt.locking is sufficient to obtain an effective lock, and the race
Consequently, msvcrt.locking is sufficient to obtain an effective lock, and the race
condition encountered on Linux is not possible on Windows, leading to a simpler workflow.
"""
def acquire(self) -> None:
@@ -209,7 +209,7 @@ class _WindowsLockMechanism(_BaseLockMechanism):
# 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.
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) # type: ignore
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) # type: ignore # pylint: disable=used-before-assignment
except (IOError, OSError) as err:
if fd:
os.close(fd)
@@ -229,7 +229,7 @@ class _WindowsLockMechanism(_BaseLockMechanism):
# 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.
msvcrt.locking(self._fd, msvcrt.LK_UNLCK, 1) # type: ignore
msvcrt.locking(self._fd, msvcrt.LK_UNLCK, 1) # type: ignore # pylint: disable=used-before-assignment
os.close(self._fd)
try:
certbot/certbot/_internal/main.py
+11 -25
View File
@@ -17,8 +17,6 @@ from typing import Union
import configobj
import josepy as jose
import zope.component
import zope.interface
from acme import client as acme_client
from acme import errors as acme_errors
@@ -38,7 +36,6 @@ from certbot._internal import eff
from certbot._internal import hooks
from certbot._internal import log
from certbot._internal import renewal
from certbot._internal import reporter
from certbot._internal import snap_config
from certbot._internal import storage
from certbot._internal import updater
@@ -1165,15 +1162,14 @@ def plugins_cmd(config: configuration.NamespaceConfig,
return
filtered.init(config)
verified = filtered.verify(ifaces)
logger.debug("Verified plugins: %r", verified)
logger.debug("Filtered plugins: %r", filtered)
if not config.prepare:
notify(str(verified))
notify(str(filtered))
return
verified.prepare()
available = verified.available()
filtered.prepare()
available = filtered.available()
logger.debug("Prepared plugins: %s", available)
notify(str(available))
@@ -1654,8 +1650,8 @@ def make_or_verify_needed_dirs(config: configuration.NamespaceConfig) -> None:
@contextmanager
def make_displayer(config: configuration.NamespaceConfig
) -> Generator[Union[display_util.NoninteractiveDisplay,
display_util.FileDisplay], None, None]:
) -> Generator[Union[display_obj.NoninteractiveDisplay,
display_obj.FileDisplay], None, None]:
"""Creates a display object appropriate to the flags in the supplied config.
:param config: Configuration object
@@ -1663,18 +1659,18 @@ def make_displayer(config: configuration.NamespaceConfig
:returns: Display object
"""
displayer: Union[None, display_util.NoninteractiveDisplay,
display_util.FileDisplay] = None
displayer: Union[None, display_obj.NoninteractiveDisplay,
display_obj.FileDisplay] = None
devnull: Optional[IO] = None
if config.quiet:
config.noninteractive_mode = True
devnull = open(os.devnull, "w") # pylint: disable=consider-using-with
displayer = display_util.NoninteractiveDisplay(devnull)
displayer = display_obj.NoninteractiveDisplay(devnull)
elif config.noninteractive_mode:
displayer = display_util.NoninteractiveDisplay(sys.stdout)
displayer = display_obj.NoninteractiveDisplay(sys.stdout)
else:
displayer = display_util.FileDisplay(
displayer = display_obj.FileDisplay(
sys.stdout, config.force_interactive)
try:
@@ -1716,10 +1712,6 @@ def main(cli_args: List[str] = None) -> Optional[Union[str, int]]:
args = cli.prepare_and_parse_args(plugins, cli_args)
config = configuration.NamespaceConfig(args)
# This call is done only for retro-compatibility purposes.
# TODO: Remove this call once zope dependencies are removed from Certbot.
zope.component.provideUtility(config, interfaces.IConfig)
# On windows, shell without administrative right cannot create symlinks required by certbot.
# So we check the rights before continuing.
misc.raise_for_non_administrative_windows_rights()
@@ -1732,12 +1724,6 @@ def main(cli_args: List[str] = None) -> Optional[Union[str, int]]:
if config.func != plugins_cmd: # pylint: disable=comparison-with-callable
raise
# These calls are done only for retro-compatibility purposes.
# TODO: Remove these calls once zope dependencies are removed from Certbot.
report = reporter.Reporter(config)
zope.component.provideUtility(report, interfaces.IReporter)
util.atexit_register(report.print_messages)
with make_displayer(config) as displayer:
display_obj.set_display(displayer)
certbot/certbot/_internal/plugins/disco.py
+2 -105
View File
@@ -12,11 +12,8 @@ from typing import Mapping
from typing import Optional
from typing import Type
from typing import Union
import warnings
import pkg_resources
import zope.interface
import zope.interface.verify
from certbot import configuration
from certbot import errors
@@ -31,7 +28,6 @@ PREFIX_FREE_DISTRIBUTIONS = [
"certbot",
"certbot-apache",
"certbot-dns-cloudflare",
"certbot-dns-cloudxns",
"certbot-dns-digitalocean",
"certbot-dns-dnsimple",
"certbot-dns-dnsmadeeasy",
@@ -116,7 +112,7 @@ class PluginEntryPoint:
def ifaces(self, *ifaces_groups: Iterable[Type]) -> bool:
"""Does plugin implements specified interface groups?"""
return not ifaces_groups or any(
all(_implements(self.plugin_cls, iface)
all(issubclass(self.plugin_cls, iface)
for iface in ifaces)
for ifaces in ifaces_groups)
@@ -134,16 +130,6 @@ class PluginEntryPoint:
self._initialized = self.plugin_cls(config, self.name)
return self._initialized
def verify(self, ifaces: Iterable[Type]) -> bool:
"""Verify that the plugin conforms to the specified interfaces."""
if not self.initialized:
raise ValueError("Plugin is not initialized.")
for iface in ifaces: # zope.interface.providedBy(plugin)
if not _verify(self.init(), self.plugin_cls, iface):
return False
return True
@property
def prepared(self) -> bool:
"""Has the plugin been prepared already?"""
@@ -265,7 +251,7 @@ class PluginsRegistry(Mapping):
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, plugin1, plugin2))
if _provides(plugin_ep.plugin_cls, interfaces.Plugin):
if issubclass(plugin_ep.plugin_cls, interfaces.Plugin):
plugins[plugin_ep.name] = plugin_ep
else: # pragma: no cover
logger.warning(
@@ -300,10 +286,6 @@ class PluginsRegistry(Mapping):
"""Filter plugins based on interfaces."""
return self.filter(lambda p_ep: p_ep.ifaces(*ifaces_groups))
def verify(self, ifaces: Iterable[Type]) -> "PluginsRegistry":
"""Filter plugins based on verification."""
return self.filter(lambda p_ep: p_ep.verify(ifaces))
def prepare(self) -> List[Union[bool, Error]]:
"""Prepare all plugins in the registry."""
return [plugin_ep.prepare() for plugin_ep in self._plugins.values()]
@@ -342,88 +324,3 @@ class PluginsRegistry(Mapping):
if not self._plugins:
return "No plugins"
return "\n\n".join(str(p_ep) for p_ep in self._plugins.values())
_DEPRECATION_PLUGIN = ("Zope interface certbot.interfaces.IPlugin is deprecated, "
"use ABC certbot.interface.Plugin instead.")
_DEPRECATION_AUTHENTICATOR = ("Zope interface certbot.interfaces.IAuthenticator is deprecated, "
"use ABC certbot.interface.Authenticator instead.")
_DEPRECATION_INSTALLER = ("Zope interface certbot.interfaces.IInstaller is deprecated, "
"use ABC certbot.interface.Installer instead.")
_DEPRECATION_FACTORY = ("Zope interface certbot.interfaces.IPluginFactory is deprecated, "
"use ABC certbot.interface.Plugin instead.")
def _provides(target_class: Type[interfaces.Plugin], iface: Type) -> bool:
if issubclass(target_class, iface):
return True
if iface == interfaces.Plugin and interfaces.IPluginFactory.providedBy(target_class):
logging.warning(_DEPRECATION_FACTORY)
warnings.warn(_DEPRECATION_FACTORY, DeprecationWarning)
return True
return False
def _implements(target_class: Type[interfaces.Plugin], iface: Type) -> bool:
if issubclass(target_class, iface):
return True
if iface == interfaces.Plugin and interfaces.IPlugin.implementedBy(target_class):
logging.warning(_DEPRECATION_PLUGIN)
warnings.warn(_DEPRECATION_PLUGIN, DeprecationWarning)
return True
if iface == interfaces.Authenticator and interfaces.IAuthenticator.implementedBy(target_class):
logging.warning(_DEPRECATION_AUTHENTICATOR)
warnings.warn(_DEPRECATION_AUTHENTICATOR, DeprecationWarning)
return True
if iface == interfaces.Installer and interfaces.IInstaller.implementedBy(target_class):
logging.warning(_DEPRECATION_INSTALLER)
warnings.warn(_DEPRECATION_INSTALLER, DeprecationWarning)
return True
return False
def _verify(target_instance: interfaces.Plugin, target_class: Type[interfaces.Plugin],
iface: Type) -> bool:
if issubclass(target_class, iface):
# No need to trigger some verify logic for ABCs: when the object is instantiated,
# an error would be raised if implementation is not done properly.
# So the checks have been done effectively when the plugin has been initialized.
return True
zope_iface: Optional[Type[zope.interface.Interface]] = None
message = ""
if iface == interfaces.Plugin:
zope_iface = interfaces.IPlugin
message = _DEPRECATION_PLUGIN
if iface == interfaces.Authenticator:
zope_iface = interfaces.IAuthenticator
message = _DEPRECATION_AUTHENTICATOR
if iface == interfaces.Installer:
zope_iface = interfaces.IInstaller
message = _DEPRECATION_INSTALLER
if not zope_iface:
raise ValueError(f"Unexpected type: {iface.__name__}")
try:
zope.interface.verify.verifyObject(zope_iface, target_instance)
logging.warning(message)
warnings.warn(message, DeprecationWarning)
return True
except zope.interface.exceptions.BrokenImplementation as error:
if zope_iface.implementedBy(target_class):
logger.debug(
"%s implements %s but object does not verify: %s",
target_class, zope_iface.__name__, error, exc_info=True)
return False
certbot/certbot/_internal/plugins/selection.py
+3 -7
View File
@@ -65,7 +65,6 @@ def get_unprepared_installer(config: configuration.NamespaceConfig,
return None
installers = plugins.filter(lambda p_ep: p_ep.check_name(req_inst))
installers.init(config)
installers = installers.verify((interfaces.Installer,))
if len(installers) > 1:
raise errors.PluginSelectionError(
"Found multiple installers with the name %s, Certbot is unable to "
@@ -116,9 +115,8 @@ def pick_plugin(config: configuration.NamespaceConfig, default: Optional[str],
filtered = plugins.visible().ifaces(ifaces)
filtered.init(config)
verified = filtered.verify(ifaces)
verified.prepare()
prepared = verified.available()
filtered.prepare()
prepared = filtered.available()
if len(prepared) > 1:
logger.debug("Multiple candidate plugins: %s", prepared)
@@ -168,7 +166,7 @@ def choose_plugin(prepared: List[disco.PluginEntryPoint],
return None
noninstaller_plugins = ["webroot", "manual", "standalone", "dns-cloudflare", "dns-cloudxns",
noninstaller_plugins = ["webroot", "manual", "standalone", "dns-cloudflare",
"dns-digitalocean", "dns-dnsimple", "dns-dnsmadeeasy", "dns-gehirn",
"dns-google", "dns-linode", "dns-luadns", "dns-nsone", "dns-ovh",
"dns-rfc2136", "dns-route53", "dns-sakuracloud"]
@@ -316,8 +314,6 @@ def cli_plugin_requests(config: configuration.NamespaceConfig
req_auth = set_configurator(req_auth, "manual")
if config.dns_cloudflare:
req_auth = set_configurator(req_auth, "dns-cloudflare")
if config.dns_cloudxns:
req_auth = set_configurator(req_auth, "dns-cloudxns")
if config.dns_digitalocean:
req_auth = set_configurator(req_auth, "dns-digitalocean")
if config.dns_dnsimple:
certbot/certbot/_internal/renewal.py
+45 -5
View File
@@ -19,12 +19,10 @@ from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import load_pem_private_key
import zope.component
from certbot import configuration
from certbot import crypto_util
from certbot import errors
from certbot import interfaces
from certbot import util
from certbot._internal import cli
from certbot._internal import client
@@ -326,12 +324,57 @@ def _avoid_invalidating_lineage(config: configuration.NamespaceConfig,
"unless you use the --break-my-certs flag!")
def _avoid_reuse_key_conflicts(config: configuration.NamespaceConfig,
lineage: storage.RenewableCert) -> None:
"""Don't allow combining --reuse-key with any flags that would conflict
with key reuse (--key-type, --rsa-key-size, --elliptic-curve), unless
--new-key is also set.
"""
# If --no-reuse-key is set, no conflict
if cli.set_by_cli("reuse_key") and not config.reuse_key:
return
# If reuse_key is not set on the lineage and --reuse-key is not
# set on the CLI, no conflict.
if not lineage.reuse_key and not config.reuse_key:
return
# If --new-key is set, no conflict
if config.new_key:
return
kt = config.key_type.lower()
# The remaining cases where conflicts are present:
# - --key-type is set on the CLI and doesn't match the stored private key
# - It's an RSA key and --rsa-key-size is set and doesn't match
# - It's an ECDSA key and --eliptic-curve is set and doesn't match
potential_conflicts = [
("--key-type",
lambda: kt != lineage.private_key_type.lower()),
("--rsa-key-type",
lambda: kt == "rsa" and config.rsa_key_size != lineage.rsa_key_size),
("--elliptic-curve",
lambda: kt == "ecdsa" and lineage.elliptic_curve and \
config.elliptic_curve.lower() != lineage.elliptic_curve.lower())
]
for conflict in potential_conflicts:
if conflict[1]():
raise errors.Error(
f"Unable to change the {conflict[0]} of this certificate because --reuse-key "
"is set. To stop reusing the private key, specify --no-reuse-key. "
"To change the private key this one time and then reuse it in future, "
"add --new-key.")
def renew_cert(config: configuration.NamespaceConfig, 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"))
_avoid_invalidating_lineage(config, lineage, original_server)
_avoid_reuse_key_conflicts(config, lineage)
if not domains:
domains = lineage.names()
# The private key is the existing lineage private key if reuse_key is set.
@@ -460,9 +503,6 @@ def handle_renewal_request(config: configuration.NamespaceConfig) -> None:
if not renewal_candidate:
parse_failures.append(renewal_file)
else:
# This call is done only for retro-compatibility purposes.
# TODO: Remove this call once zope dependencies are removed from Certbot.
zope.component.provideUtility(lineage_config, interfaces.IConfig)
renewal_candidate.ensure_deployed()
from certbot._internal import main
plugins = plugins_disco.PluginsRegistry.find_all()
certbot/certbot/_internal/reporter.py
-94
View File
@@ -1,94 +0,0 @@
"""Collects and displays information to the user."""
import collections
import logging
import queue
import sys
import textwrap
from certbot import configuration
from certbot import util
logger = logging.getLogger(__name__)
class Reporter:
"""Collects and displays information to the user.
:ivar `queue.PriorityQueue` messages: Messages to be displayed to
the user.
"""
HIGH_PRIORITY = 0
"""High priority constant. See `add_message`."""
MEDIUM_PRIORITY = 1
"""Medium priority constant. See `add_message`."""
LOW_PRIORITY = 2
"""Low priority constant. See `add_message`."""
_msg_type = collections.namedtuple('_msg_type', 'priority text on_crash')
def __init__(self, config: configuration.NamespaceConfig) -> None:
self.messages: "queue.PriorityQueue[Reporter._msg_type]" = queue.PriorityQueue()
self.config = config
def add_message(self, msg: str, priority: int, on_crash: bool = True) -> None:
"""Adds msg to the list of messages to be printed.
:param str msg: Message to be displayed to the user.
:param int priority: One of `HIGH_PRIORITY`, `MEDIUM_PRIORITY`,
or `LOW_PRIORITY`.
:param bool on_crash: Whether or not the message should be
printed if the program exits abnormally.
"""
assert self.HIGH_PRIORITY <= priority <= self.LOW_PRIORITY
self.messages.put(self._msg_type(priority, msg, on_crash))
logger.debug("Reporting to user: %s", msg)
def print_messages(self) -> None:
"""Prints messages to the user and clears the message queue.
If there is an unhandled exception, only messages for which
``on_crash`` is ``True`` are printed.
"""
bold_on = False
if not self.messages.empty():
no_exception = sys.exc_info()[0] is None
bold_on = sys.stdout.isatty()
if not self.config.quiet:
if bold_on:
print(util.ANSI_SGR_BOLD)
print('IMPORTANT NOTES:')
first_wrapper = textwrap.TextWrapper(
initial_indent=' - ',
subsequent_indent=(' ' * 3),
break_long_words=False,
break_on_hyphens=False)
next_wrapper = textwrap.TextWrapper(
initial_indent=first_wrapper.subsequent_indent,
subsequent_indent=first_wrapper.subsequent_indent,
break_long_words=False,
break_on_hyphens=False)
while not self.messages.empty():
msg = self.messages.get()
if self.config.quiet:
# In --quiet mode, we only print high priority messages that
# are flagged for crash cases
if not (msg.priority == self.HIGH_PRIORITY and msg.on_crash):
continue
if no_exception or msg.on_crash:
if bold_on and msg.priority > self.HIGH_PRIORITY:
if not self.config.quiet:
sys.stdout.write(util.ANSI_SGR_RESET)
bold_on = False
lines = msg.text.splitlines()
print(first_wrapper.fill(lines[0]))
if len(lines) > 1:
print("\n".join(
next_wrapper.fill(line) for line in lines[1:]))
if bold_on and not self.config.quiet:
sys.stdout.write(util.ANSI_SGR_RESET)
certbot/certbot/_internal/storage.py
+41 -8
View File
@@ -12,10 +12,12 @@ from typing import List
from typing import Mapping
from typing import Optional
from typing import Tuple
from typing import Union
import configobj
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey
from cryptography.hazmat.primitives.serialization import load_pem_private_key
import parsedatetime
import pkg_resources
@@ -569,6 +571,12 @@ class RenewableCert(interfaces.RenewableCert):
return util.is_staging(self.server)
return False
@property
def reuse_key(self) -> bool:
"""Returns whether this certificate is configured to reuse its private key"""
return "reuse_key" in self.configuration["renewalparams"] and \
self.configuration["renewalparams"].as_bool("reuse_key")
def _check_symlinks(self) -> None:
"""Raises an exception if a symlink doesn't exist"""
for kind in ALL_FOUR:
@@ -1115,22 +1123,47 @@ class RenewableCert(interfaces.RenewableCert):
target, values)
return cls(new_config.filename, cli_config)
@property
def private_key_type(self) -> str:
"""
:returns: The type of algorithm for the private, RSA or ECDSA
:rtype: str
"""
def _private_key(self) -> Union[RSAPrivateKey, EllipticCurvePrivateKey]:
with open(self.configuration["privkey"], "rb") as priv_key_file:
key = load_pem_private_key(
data=priv_key_file.read(),
password=None,
backend=default_backend()
)
return key
@property
def private_key_type(self) -> str:
"""
:returns: The type of algorithm for the private, RSA or ECDSA
:rtype: str
"""
key = self._private_key()
if isinstance(key, RSAPrivateKey):
return "RSA"
else:
return "ECDSA"
return "ECDSA"
@property
def rsa_key_size(self) -> Optional[int]:
"""
:returns: If the private key is an RSA key, its size.
:rtype: int
"""
key = self._private_key()
if isinstance(key, RSAPrivateKey):
return key.key_size
return None
@property
def elliptic_curve(self) -> Optional[str]:
"""
:returns: If the private key is an elliptic key, the name of its curve.
:rtype: str
"""
key = self._private_key()
if isinstance(key, EllipticCurvePrivateKey):
return key.curve.name
return None
def save_successor(self, prior_version: int, new_cert: bytes, new_privkey: bytes,
new_chain: bytes, cli_config: configuration.NamespaceConfig) -> int:
certbot/certbot/achallenges.py
+2 -2
View File
@@ -47,7 +47,7 @@ class AnnotatedChallenge(jose.ImmutableMap):
class KeyAuthorizationAnnotatedChallenge(AnnotatedChallenge):
"""Client annotated `KeyAuthorizationChallenge` challenge."""
__slots__ = ('challb', 'domain', 'account_key')
__slots__ = ('challb', 'domain', 'account_key') # pylint: disable=redefined-slots-in-subclass
def response_and_validation(self, *args: Any, **kwargs: Any) -> Any:
"""Generate response and validation."""
@@ -57,5 +57,5 @@ class KeyAuthorizationAnnotatedChallenge(AnnotatedChallenge):
class DNS(AnnotatedChallenge):
"""Client annotated "dns" ACME challenge."""
__slots__ = ('challb', 'domain')
__slots__ = ('challb', 'domain') # pylint: disable=redefined-slots-in-subclass
acme_type = challenges.DNS
certbot/certbot/compat/misc.py
+2 -36
View File
@@ -10,7 +10,6 @@ import subprocess
import sys
from typing import Optional
from typing import Tuple
import warnings
from certbot import errors
from certbot.compat import os
@@ -144,8 +143,8 @@ def execute_command_status(cmd_name: str, shell_cmd: str,
subprocess.run(shell=True)
- on Windows command will be run in a Powershell shell
This differs from execute_command: it returns the exit code, and does not log the result
and output of the command.
This function returns the exit code, and does not log the result and output
of the command.
:param str cmd_name: the user facing name of the hook being run
:param str shell_cmd: shell command to execute
@@ -168,36 +167,3 @@ def execute_command_status(cmd_name: str, shell_cmd: str,
# bytes in Python 3
out, err = proc.stdout, proc.stderr
return proc.returncode, err, out
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
subprocess.run(shell=True)
- on Windows command will be run in a Powershell shell
This differs from execute_command: it returns the exit code, and does not log the result
and output of the command.
:param str cmd_name: the user facing name of the hook being run
:param str shell_cmd: shell command to execute
:param dict env: environ to pass into subprocess.run
:returns: `tuple` (`str` stderr, `str` stdout)
"""
# Deprecation per https://github.com/certbot/certbot/issues/8854
warnings.warn(
"execute_command will be deprecated in the future, use execute_command_status instead",
PendingDeprecationWarning
)
returncode, err, out = execute_command_status(cmd_name, shell_cmd, env)
base_cmd = os.path.basename(shell_cmd.split(None, 1)[0])
if out:
logger.info('Output from %s command %s:\n%s', cmd_name, base_cmd, out)
if returncode != 0:
logger.error('%s command "%s" returned error code %d',
cmd_name, shell_cmd, returncode)
if err:
logger.error('Error output from %s command %s:\n%s', cmd_name, base_cmd, err)
return err, out
certbot/certbot/configuration.py
+5 -1
View File
@@ -170,7 +170,11 @@ class NamespaceConfig:
@property
def no_verify_ssl(self) -> bool:
"""Disable verification of the ACME server's certificate."""
"""Disable verification of the ACME server's certificate.
The root certificates trusted by Certbot can be overriden by setting the
REQUESTS_CA_BUNDLE environment variable.
"""
return self.namespace.no_verify_ssl
@property
certbot/certbot/crypto_util.py
-64
View File
@@ -15,7 +15,6 @@ from typing import Set
from typing import Tuple
from typing import TYPE_CHECKING
from typing import Union
import warnings
from cryptography import x509
from cryptography.exceptions import InvalidSignature
@@ -35,7 +34,6 @@ import josepy
from OpenSSL import crypto
from OpenSSL import SSL
import pyrfc3339
import zope.component
from acme import crypto_util as acme_crypto_util
from certbot import errors
@@ -100,41 +98,6 @@ def generate_key(key_size: int, key_dir: str, key_type: str = "rsa",
return util.Key(key_path, key_pem)
# TODO: Remove this call once zope dependencies are removed from Certbot.
def init_save_key(key_size: int, key_dir: str, key_type: str = "rsa",
elliptic_curve: str = "secp256r1",
keyname: str = "key-certbot.pem") -> util.Key:
"""Initializes and saves a privkey.
Inits key and saves it in PEM format on the filesystem.
.. note:: keyname is the attempted filename, it may be different if a file
already exists at the path.
.. deprecated:: 1.16.0
Use :func:`generate_key` instead.
:param int key_size: key size in bits if key size is rsa.
:param str key_dir: Key save directory.
:param str key_type: Key Type [rsa, ecdsa]
:param str elliptic_curve: Name of the elliptic curve if key type is ecdsa.
:param str keyname: Filename of key
:returns: Key
:rtype: :class:`certbot.util.Key`
:raises ValueError: If unable to generate the key given key_size.
"""
warnings.warn("certbot.crypto_util.init_save_key is deprecated, please use "
"certbot.crypto_util.generate_key instead.", DeprecationWarning)
config = zope.component.getUtility(interfaces.IConfig)
return generate_key(key_size, key_dir, key_type=key_type, elliptic_curve=elliptic_curve,
keyname=keyname, strict_permissions=config.strict_permissions)
def generate_csr(privkey: util.Key, names: Union[List[str], Set[str]], path: str,
must_staple: bool = False, strict_permissions: bool = True) -> util.CSR:
"""Initialize a CSR with the given private key.
@@ -165,33 +128,6 @@ def generate_csr(privkey: util.Key, names: Union[List[str], Set[str]], path: str
return util.CSR(csr_filename, csr_pem, "pem")
# TODO: Remove this call once zope dependencies are removed from Certbot.
def init_save_csr(privkey: util.Key, names: Set[str], path: str) -> util.CSR:
"""Initialize a CSR with the given private key.
.. deprecated:: 1.16.0
Use :func:`generate_csr` instead.
:param privkey: Key to include in the CSR
:type privkey: :class:`certbot.util.Key`
:param set names: `str` names to include in the CSR
:param str path: Certificate save directory.
:returns: CSR
:rtype: :class:`certbot.util.CSR`
"""
warnings.warn("certbot.crypto_util.init_save_csr is deprecated, please use "
"certbot.crypto_util.generate_csr instead.", DeprecationWarning)
config = zope.component.getUtility(interfaces.IConfig)
return generate_csr(privkey, names, path, must_staple=config.must_staple,
strict_permissions=config.strict_permissions)
# WARNING: the csr and private key file are possible attack vectors for TOCTOU
# We should either...
# A. Do more checks to verify that the CSR is trusted/valid
certbot/certbot/display/ops.py
+4 -1
View File
@@ -181,7 +181,10 @@ def _filter_names(names: Iterable[str],
if override_question:
question = override_question
else:
question = "Which names would you like to activate HTTPS for?"
question = (
"Which names would you like to activate HTTPS for?\n"
"We recommend selecting either all domains, or all domains in a VirtualHost/server "
"block.")
code, names = display_util.checklist(
question, tags=sorted_names, cli_flag="--domains", force_interactive=True)
return code, [str(s) for s in names]
certbot/certbot/display/util.py
-56
View File
@@ -9,26 +9,12 @@ should be used whenever:
Other messages can use the `logging` module. See `log.py`.
"""
import sys
from types import ModuleType
from typing import Any
from typing import cast
from typing import List
from typing import Optional
from typing import Tuple
from typing import Union
import warnings
from certbot._internal.display import obj
# These specific imports from certbot._internal.display.obj and
# certbot._internal.display.util are done to not break the public API of this
# module.
from certbot._internal.display.obj import FileDisplay # pylint: disable=unused-import
from certbot._internal.display.obj import NoninteractiveDisplay # pylint: disable=unused-import
from certbot._internal.display.obj import SIDE_FRAME # pylint: disable=unused-import
from certbot._internal.display.util import input_with_timeout # pylint: disable=unused-import
from certbot._internal.display.util import separate_list_input # pylint: disable=unused-import
from certbot._internal.display.util import summarize_domain_list # pylint: disable=unused-import
# These constants are defined this way to make them easier to document with
# Sphinx and to not couple our public docstrings to our internal ones.
@@ -38,17 +24,8 @@ OK = obj.OK
CANCEL = obj.CANCEL
"""Display exit code for a user canceling the display."""
# These constants are unused and should be removed in a major release of
# Certbot.
WIDTH = 72
HELP = "help"
"""Display exit code when for when the user requests more help. (UNUSED)"""
ESC = "esc"
"""Display exit code when the user hits Escape (UNUSED)"""
def notify(msg: str) -> None:
"""Display a basic status message.
@@ -204,36 +181,3 @@ def assert_valid_call(prompt: str, default: str, cli_flag: str, force_interactiv
msg += ("\nYou can set an answer to "
"this prompt with the {0} flag".format(cli_flag))
assert default is not None or force_interactive, msg
# This class takes a similar approach to the cryptography project to deprecate attributes
# in public modules. See the _ModuleWithDeprecation class here:
# https://github.com/pyca/cryptography/blob/91105952739442a74582d3e62b3d2111365b0dc7/src/cryptography/utils.py#L129
class _DisplayUtilDeprecationModule:
"""
Internal class delegating to a module, and displaying warnings when attributes
related to deprecated attributes in the certbot.display.util module.
"""
def __init__(self, module: ModuleType) -> None:
self.__dict__['_module'] = module
def __getattr__(self, attr: str) -> Any:
if attr in ('FileDisplay', 'NoninteractiveDisplay', 'SIDE_FRAME', 'input_with_timeout',
'separate_list_input', 'summarize_domain_list', 'WIDTH', 'HELP', 'ESC'):
warnings.warn('{0} attribute in certbot.display.util module is deprecated '
'and will be removed soon.'.format(attr),
DeprecationWarning, stacklevel=2)
return getattr(self._module, attr)
def __setattr__(self, attr: str, value: Any) -> None: # pragma: no cover
setattr(self._module, attr, value)
def __delattr__(self, attr: str) -> None: # pragma: no cover
delattr(self._module, attr)
def __dir__(self) -> List[str]: # pragma: no cover
return ['_module'] + dir(self._module)
# Patching ourselves to warn about deprecation and planned removal of some elements in the module.
sys.modules[__name__] = cast(ModuleType, _DisplayUtilDeprecationModule(sys.modules[__name__]))
certbot/certbot/interfaces.py
+3 -70
View File
@@ -2,23 +2,17 @@
from abc import ABCMeta
from abc import abstractmethod
from argparse import ArgumentParser
import sys
from types import ModuleType
from typing import Any
from typing import Union
from typing import cast
from typing import Iterable
from typing import List
from typing import Optional
from typing import Type
from typing import TYPE_CHECKING
import warnings
import zope.interface
from typing import Union
from acme.challenges import Challenge
from acme.challenges import ChallengeResponse
from acme.client import ClientBase
from acme.client import ClientV2
from certbot import configuration
from certbot.achallenges import AnnotatedChallenge
@@ -53,7 +47,7 @@ class AccountStorage(metaclass=ABCMeta):
raise NotImplementedError()
@abstractmethod
def save(self, account: 'Account', client: ClientBase) -> None: # pragma: no cover
def save(self, account: 'Account', client: ClientV2) -> None: # pragma: no cover
"""Save account.
:raises .AccountStorageError: if account could not be saved
@@ -62,18 +56,6 @@ class AccountStorage(metaclass=ABCMeta):
raise NotImplementedError()
class IConfig(zope.interface.Interface): # pylint: disable=inherit-non-class
"""Deprecated, use certbot.configuration.NamespaceConfig instead."""
class IPluginFactory(zope.interface.Interface): # pylint: disable=inherit-non-class
"""Deprecated, use certbot.interfaces.Plugin as ABC instead."""
class IPlugin(zope.interface.Interface): # pylint: disable=inherit-non-class
"""Deprecated, use certbot.interfaces.Plugin as ABC instead."""
class Plugin(metaclass=ABCMeta):
"""Certbot plugin.
@@ -168,10 +150,6 @@ class Plugin(metaclass=ABCMeta):
"""
class IAuthenticator(IPlugin): # pylint: disable=inherit-non-class
"""Deprecated, use certbot.interfaces.Authenticator as ABC instead."""
class Authenticator(Plugin):
"""Generic Certbot Authenticator.
@@ -231,10 +209,6 @@ class Authenticator(Plugin):
"""
class IInstaller(IPlugin): # pylint: disable=inherit-non-class
"""Deprecated, use certbot.interfaces.Installer as ABC instead."""
class Installer(Plugin):
"""Generic Certbot Installer Interface.
@@ -362,14 +336,6 @@ class Installer(Plugin):
"""
class IDisplay(zope.interface.Interface): # pylint: disable=inherit-non-class
"""Deprecated, use your own Display implementation instead."""
class IReporter(zope.interface.Interface): # pylint: disable=inherit-non-class
"""Deprecated, use your own Reporter implementation instead."""
class RenewableCert(metaclass=ABCMeta):
"""Interface to a certificate lineage."""
@@ -499,36 +465,3 @@ class RenewDeployer(metaclass=ABCMeta):
:type lineage: RenewableCert
"""
# This class takes a similar approach to the cryptography project to deprecate attributes
# in public modules. See the _ModuleWithDeprecation class here:
# https://github.com/pyca/cryptography/blob/91105952739442a74582d3e62b3d2111365b0dc7/src/cryptography/utils.py#L129
class _ZopeInterfacesDeprecationModule:
"""
Internal class delegating to a module, and displaying warnings when
attributes related to Zope interfaces are accessed.
"""
def __init__(self, module: ModuleType) -> None:
self.__dict__['_module'] = module
def __getattr__(self, attr: str) -> None:
if attr in ('IConfig', 'IPlugin', 'IPluginFactory', 'IAuthenticator',
'IInstaller', 'IDisplay', 'IReporter'):
warnings.warn('{0} attribute in certbot.interfaces module is deprecated '
'and will be removed soon.'.format(attr),
DeprecationWarning, stacklevel=2)
return getattr(self._module, attr)
def __setattr__(self, attr: str, value: Any) -> None: # pragma: no cover
setattr(self._module, attr, value)
def __delattr__(self, attr: str) -> None: # pragma: no cover
delattr(self._module, attr)
def __dir__(self) -> List[str]: # pragma: no cover
return ['_module'] + dir(self._module)
# Patching ourselves to warn about Zope interfaces deprecation and planned removal.
sys.modules[__name__] = cast(ModuleType, _ZopeInterfacesDeprecationModule(sys.modules[__name__]))
certbot/certbot/plugins/dns_test_common.py
-2
View File
@@ -38,14 +38,12 @@ class _AuthenticatorCallableTestCase(Protocol):
See
https://docs.python.org/3/library/unittest.html#unittest.TestCase.assertTrue
"""
...
def assertEqual(self, *unused_args: Any) -> None:
"""
See
https://docs.python.org/3/library/unittest.html#unittest.TestCase.assertEqual
"""
...
class BaseAuthenticatorTest:
certbot/certbot/plugins/dns_test_common_lexicon.py
-1
View File
@@ -57,7 +57,6 @@ class _LexiconAwareTestCase(Protocol):
See
https://docs.python.org/3/library/unittest.html#unittest.TestCase.assertRaises
"""
...
# These classes are intended to be subclassed/mixed in, so not all members are defined.
certbot/certbot/tests/acme_util.py
+1 -12
View File
@@ -3,7 +3,6 @@ import datetime
from typing import Any
from typing import Dict
from typing import Iterable
from typing import Tuple
import josepy as jose
@@ -24,12 +23,6 @@ DNS01_2 = challenges.DNS01(token=b"cafecafecafecafecafecafe0feedbac")
CHALLENGES = [HTTP01, DNS01]
def gen_combos(challbs: Iterable[messages.ChallengeBody]) -> Tuple[Tuple[int], ...]:
"""Generate natural combinations for challbs."""
# completing a single DV challenge satisfies the CA
return tuple((i,) for i, _ in enumerate(challbs))
def chall_to_challb(chall: challenges.Challenge, status: messages.Status) -> messages.ChallengeBody:
"""Return ChallengeBody from Challenge."""
kwargs = {
@@ -61,15 +54,13 @@ ACHALLENGES = [HTTP01_A, DNS01_A]
def gen_authzr(authz_status: messages.Status, domain: str, challs: Iterable[challenges.Challenge],
statuses: Iterable[messages.Status],
combos: bool = True) -> messages.AuthorizationResource:
statuses: Iterable[messages.Status]) -> messages.AuthorizationResource:
"""Generate an authorization resource.
:param authz_status: Status object
:type authz_status: :class:`acme.messages.Status`
:param list challs: Challenge objects
:param list statuses: status of each challenge object
:param bool combos: Whether or not to add combinations
"""
challbs = tuple(
@@ -81,8 +72,6 @@ def gen_authzr(authz_status: messages.Status, domain: str, challs: Iterable[chal
typ=messages.IDENTIFIER_FQDN, value=domain),
"challenges": challbs,
}
if combos:
authz_kwargs.update({"combinations": gen_combos(challbs)})
if authz_status == messages.STATUS_VALID:
authz_kwargs.update({
"status": authz_status,
certbot/certbot/tests/util.py
+12 -51
View File
@@ -198,56 +198,18 @@ def make_lineage(config_dir: str, testfile: str, ec: bool = False) -> str:
return conf_path
def patch_get_utility(target: str = 'zope.component.getUtility') -> mock.MagicMock:
"""Deprecated, patch certbot.display.util directly or use patch_display_util instead.
:param str target: path to patch
:returns: mock zope.component.getUtility
:rtype: mock.MagicMock
"""
warnings.warn('Decorator certbot.tests.util.patch_get_utility is deprecated. You should now '
'patch certbot.display.util yourself directly or use '
'certbot.tests.util.patch_display_util as a temporary workaround.')
return cast(mock.MagicMock, mock.patch(target, new_callable=_create_display_util_mock))
def patch_get_utility_with_stdout(target: str = 'zope.component.getUtility',
stdout: Optional[IO] = None) -> mock.MagicMock:
"""Deprecated, patch certbot.display.util directly
or use patch_display_util_with_stdout instead.
:param str target: path to patch
:param object stdout: object to write standard output to; it is
expected to have a `write` method
:returns: mock zope.component.getUtility
:rtype: mock.MagicMock
"""
warnings.warn('Decorator certbot.tests.util.patch_get_utility_with_stdout is deprecated. You '
'should now patch certbot.display.util yourself directly or use '
'use certbot.tests.util.patch_display_util_with_stdout as a temporary '
'workaround.')
stdout = stdout if stdout else io.StringIO()
freezable_mock = _create_display_util_mock_with_stdout(stdout)
return cast(mock.MagicMock, mock.patch(target, new=freezable_mock))
def patch_display_util() -> mock.MagicMock:
"""Patch certbot.display.util to use a special mock display utility.
The mock display utility works like a regular mock object, except it also
also asserts that methods are called with valid arguments.
The mock created by this patch mocks out Certbot internals so this can be
used like the old patch_get_utility function. That is, the mock object will
be called by the certbot.display.util functions and the mock returned by
that call will be used as the display utility. This was done to simplify
the transition from zope.component and mocking certbot.display.util
functions directly in test code should be preferred over using this
function in the future.
The mock created by this patch mocks out Certbot internals. That is, the
mock object will be called by the certbot.display.util functions and the
mock returned by that call will be used as the display utility. This was
done to simplify the transition from zope.component and mocking
certbot.display.util functions directly in test code should be preferred
over using this function in the future.
See https://github.com/certbot/certbot/issues/8948
@@ -267,13 +229,12 @@ def patch_display_util_with_stdout(
The mock display utility works like a regular mock object, except it also
asserts that methods are called with valid arguments.
The mock created by this patch mocks out Certbot internals so this can be
used like the old patch_get_utility function. That is, the mock object will
be called by the certbot.display.util functions and the mock returned by
that call will be used as the display utility. This was done to simplify
the transition from zope.component and mocking certbot.display.util
functions directly in test code should be preferred over using this
function in the future.
The mock created by this patch mocks out Certbot internals. That is, the
mock object will be called by the certbot.display.util functions and the
mock returned by that call will be used as the display utility. This was
done to simplify the transition from zope.component and mocking
certbot.display.util functions directly in test code should be preferred
over using this function in the future.
See https://github.com/certbot/certbot/issues/8948
certbot/certbot/util.py
-22
View File
@@ -17,7 +17,6 @@ from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from typing import TYPE_CHECKING
from typing import Union
import warnings
@@ -33,9 +32,6 @@ _USE_DISTRO = sys.platform.startswith('linux')
if _USE_DISTRO:
import distro
if TYPE_CHECKING:
import distutils.version
logger = logging.getLogger(__name__)
@@ -611,24 +607,6 @@ def is_wildcard_domain(domain: Union[str, bytes]) -> bool:
return domain.startswith(b"*.")
def get_strict_version(normalized: str) -> "distutils.version.StrictVersion":
"""Converts a normalized version to a strict version.
:param str normalized: normalized version string
:returns: An equivalent strict version
:rtype: distutils.version.StrictVersion
"""
warnings.warn("certbot.util.get_strict_version is deprecated and will be "
"removed in a future release.", DeprecationWarning)
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
import distutils.version
# strict version ending with "a" and a number designates a pre-release
return distutils.version.StrictVersion(normalized.replace(".dev", "a"))
def is_staging(srv: str) -> bool:
"""
Determine whether a given ACME server is a known test / staging server.
certbot/docs/cli-help.txt
+4 -2
View File
@@ -126,7 +126,7 @@ optional arguments:
case, and to know when to deprecate support for past
Python versions and flags. If you wish to hide this
information from the Let's Encrypt server, set this to
"". (default: CertbotACMEClient/1.29.0 (certbot;
"". (default: CertbotACMEClient/1.30.0 (certbot;
OS_NAME OS_VERSION) Authenticator/XXX Installer/YYY
(SUBCOMMAND; flags: FLAGS) Py/major.minor.patchlevel).
The flags encoded in the user agent are: --duplicate,
@@ -236,7 +236,9 @@ testing:
(default: False)
--debug Show tracebacks in case of errors (default: False)
--no-verify-ssl Disable verification of the ACME server's certificate.
(default: False)
The root certificates trusted by Certbot can be
overriden by setting the REQUESTS_CA_BUNDLE
environment variable. (default: False)
--http-01-port HTTP01_PORT
Port used in the http-01 challenge. This only affects
the port Certbot listens on. A conforming ACME server
certbot/docs/conf.py
+1 -1
View File
@@ -84,7 +84,7 @@ release = meta['version']
#
# This is also used if you do content translation via gettext catalogs.
# Usually you set "language" from the command line for these cases.
language = None
language = 'en'
# There are two options for replacing |today|: either, you set today to some
# non-false value, then it is used:
certbot/docs/contributing.rst
+3
View File
@@ -500,6 +500,9 @@ Submitting a pull request
Steps:
0. We recommend you talk with us in a GitHub issue or :ref:`Mattermost <ask for
help>` before writing a pull request to ensure the changes you're making is
something we have the time and interest to review.
1. Write your code! When doing this, you should add :ref:`mypy type annotations
<type annotations>` for any functions you add or modify. You can check that
you've done this correctly by running ``tox -e mypy`` on a machine that has
certbot/docs/packaging.rst
-1
View File
@@ -12,7 +12,6 @@ We release packages and upload them to PyPI (wheels and source tarballs).
- https://pypi.python.org/pypi/certbot-apache
- https://pypi.python.org/pypi/certbot-nginx
- https://pypi.python.org/pypi/certbot-dns-cloudflare
- https://pypi.python.org/pypi/certbot-dns-cloudxns
- https://pypi.python.org/pypi/certbot-dns-digitalocean
- https://pypi.python.org/pypi/certbot-dns-dnsimple
- https://pypi.python.org/pypi/certbot-dns-dnsmadeeasy
certbot/docs/using.rst
+15 -1
View File
@@ -206,7 +206,6 @@ use the DNS plugins on your system.
Once installed, you can find documentation on how to use each plugin at:
* `certbot-dns-cloudflare <https://certbot-dns-cloudflare.readthedocs.io>`_
* `certbot-dns-cloudxns <https://certbot-dns-cloudxns.readthedocs.io>`_
* `certbot-dns-digitalocean <https://certbot-dns-digitalocean.readthedocs.io>`_
* `certbot-dns-dnsimple <https://certbot-dns-dnsimple.readthedocs.io>`_
* `certbot-dns-dnsmadeeasy <https://certbot-dns-dnsmadeeasy.readthedocs.io>`_
@@ -316,6 +315,8 @@ dns-lightsail_ Y N DNS Authentication using Amazon Lightsail DNS API
dns-inwx_ Y Y DNS Authentication for INWX through the XML API
dns-azure_ Y N DNS Authentication using Azure DNS
dns-godaddy_ Y N DNS Authentication using Godaddy DNS
dns-yandexcloud_ Y N DNS Authentication using Yandex Cloud DNS
dns-bunny_ Y N DNS Authentication using BunnyDNS
njalla_ Y N DNS Authentication for njalla
DuckDNS_ Y N DNS Authentication for DuckDNS
Porkbun_ Y N DNS Authentication for Porkbun
@@ -336,6 +337,8 @@ Infomaniak_ Y N DNS Authentication using Infomaniak Domains API
.. _dns-inwx: https://github.com/oGGy990/certbot-dns-inwx/
.. _dns-azure: https://github.com/binkhq/certbot-dns-azure
.. _dns-godaddy: https://github.com/miigotu/certbot-dns-godaddy
.. _dns-yandexcloud: https://github.com/PykupeJIbc/certbot-dns-yandexcloud
.. _dns-bunny: https://github.com/mwt/certbot-dns-bunny
.. _njalla: https://github.com/chaptergy/certbot-dns-njalla
.. _DuckDNS: https://github.com/infinityofspace/certbot_dns_duckdns
.. _Porkbun: https://github.com/infinityofspace/certbot_dns_porkbun
@@ -558,6 +561,11 @@ If you need to delete a certificate, use the ``delete`` subcommand.
.. note:: Read this and the `Safely deleting certificates`_ sections carefully. This is an irreversible operation and must
be done with care.
Certbot does not automatically revoke a certificate before deleting it. If you're no longer using a certificate and don't
plan to use it anywhere else, you may want to follow the instructions in `Revoking certificates`_ instead. Generally, there's
no need to revoke a certificate if its private key has not been compromised, but you may still receive expiration emails
from Let's Encrypt unless you revoke.
.. note:: Do not manually delete certificate files from inside ``/etc/letsencrypt/``. Always use the ``delete`` subcommand.
A certificate may be deleted by providing its name with ``--cert-name``. \
@@ -1078,6 +1086,12 @@ ACME directory. For example, if you would like to use Let's Encrypt's
staging server, you would add ``--server
https://acme-staging-v02.api.letsencrypt.org/directory`` to the command line.
If Certbot does not trust the SSL certificate used by the ACME server, you
can use the `REQUESTS_CA_BUNDLE
<https://requests.readthedocs.io/en/latest/user/advanced/#ssl-cert-verification>`_
environment variable to override the root certificates trusted by Certbot. Certbot
uses the ``requests`` library, which does not use the operating system trusted root store.
If you use ``--server`` to specify an ACME CA that implements the standardized
version of the spec, you may be able to obtain a certificate for a
wildcard domain. Some CAs (such as Let's Encrypt) require that domain
certbot/setup.py
-2
View File
@@ -60,8 +60,6 @@ install_requires = [
# installation on Linux.
'pywin32>=300 ; sys_platform == "win32"',
f'setuptools>={min_setuptools_version}',
'zope.component',
'zope.interface',
]
dev_extras = [
certbot/tests/auth_handler_test.py
+25 -85
View File
@@ -33,7 +33,7 @@ class ChallengeFactoryTest(unittest.TestCase):
self.authzr = acme_util.gen_authzr(
messages.STATUS_PENDING, "test", acme_util.CHALLENGES,
[messages.STATUS_PENDING] * 6, False)
[messages.STATUS_PENDING] * 6)
def test_all(self):
achalls = self.handler._challenge_factory(
@@ -70,8 +70,7 @@ class HandleAuthorizationsTest(unittest.TestCase):
self.mock_display = mock.Mock()
self.mock_config = mock.Mock(debug_challenges=False)
with mock.patch("zope.component.provideUtility"):
display_obj.set_display(self.mock_display)
display_obj.set_display(self.mock_display)
self.mock_auth = mock.MagicMock(name="Authenticator")
@@ -81,7 +80,6 @@ class HandleAuthorizationsTest(unittest.TestCase):
self.mock_account = mock.MagicMock()
self.mock_net = mock.MagicMock(spec=acme_client.ClientV2)
self.mock_net.acme_version = 1
self.mock_net.retry_after.side_effect = acme_client.ClientV2.retry_after
self.handler = AuthHandler(
@@ -92,8 +90,8 @@ class HandleAuthorizationsTest(unittest.TestCase):
def tearDown(self):
logging.disable(logging.NOTSET)
def _test_name1_http_01_1_common(self, combos):
authzr = gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES, combos=combos)
def _test_name1_http_01_1_common(self):
authzr = gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES)
mock_order = mock.MagicMock(authorizations=[authzr])
self.mock_net.poll.side_effect = _gen_mock_on_poll(retry=1, wait_value=30)
@@ -117,39 +115,14 @@ class HandleAuthorizationsTest(unittest.TestCase):
self.assertEqual(len(authzr), 1)
def test_name1_http_01_1_acme_1(self):
self._test_name1_http_01_1_common(combos=True)
def test_name1_http_01_1_acme_2(self):
self.mock_net.acme_version = 2
self._test_name1_http_01_1_common(combos=False)
def test_name1_http_01_1_dns_1_acme_1(self):
self.mock_net.poll.side_effect = _gen_mock_on_poll()
self.mock_auth.get_chall_pref.return_value.append(challenges.DNS01)
authzr = gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES, combos=False)
mock_order = mock.MagicMock(authorizations=[authzr])
authzr = self.handler.handle_authorizations(mock_order, self.mock_config)
self.assertEqual(self.mock_net.answer_challenge.call_count, 2)
self.assertEqual(self.mock_net.poll.call_count, 1)
self.assertEqual(self.mock_auth.cleanup.call_count, 1)
# Test if list first element is http-01, use typ because it is an achall
for achall in self.mock_auth.cleanup.call_args[0][0]:
self.assertIn(achall.typ, ["http-01", "dns-01"])
# Length of authorizations list
self.assertEqual(len(authzr), 1)
self._test_name1_http_01_1_common()
def test_name1_http_01_1_dns_1_acme_2(self):
self.mock_net.acme_version = 2
self.mock_net.poll.side_effect = _gen_mock_on_poll()
self.mock_auth.get_chall_pref.return_value.append(challenges.DNS01)
authzr = gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES, combos=False)
authzr = gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES)
mock_order = mock.MagicMock(authorizations=[authzr])
authzr = self.handler.handle_authorizations(mock_order, self.mock_config)
@@ -165,7 +138,7 @@ class HandleAuthorizationsTest(unittest.TestCase):
# Length of authorizations list
self.assertEqual(len(authzr), 1)
def _test_name3_http_01_3_common(self, combos):
def test_name3_http_01_3_common_acme_2(self):
authzrs = [gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES),
gen_dom_authzr(domain="1", challs=acme_util.CHALLENGES),
gen_dom_authzr(domain="2", challs=acme_util.CHALLENGES)]
@@ -183,13 +156,6 @@ class HandleAuthorizationsTest(unittest.TestCase):
self.assertEqual(len(authzr), 3)
def test_name3_http_01_3_common_acme_1(self):
self._test_name3_http_01_3_common(combos=True)
def test_name3_http_01_3_common_acme_2(self):
self.mock_net.acme_version = 2
self._test_name3_http_01_3_common(combos=False)
def test_debug_challenges(self):
config = mock.Mock(debug_challenges=True, verbose_count=0)
authzrs = [gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES)]
@@ -269,8 +235,8 @@ class HandleAuthorizationsTest(unittest.TestCase):
self.assertRaises(errors.AuthorizationError, self.handler.handle_authorizations,
mock_order, self.mock_config)
def _test_preferred_challenge_choice_common(self, combos):
authzrs = [gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES, combos=combos)]
def test_preferred_challenge_choice_common_acme_2(self):
authzrs = [gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES)]
mock_order = mock.MagicMock(authorizations=authzrs)
self.mock_auth.get_chall_pref.return_value.append(challenges.HTTP01)
@@ -285,28 +251,14 @@ class HandleAuthorizationsTest(unittest.TestCase):
self.assertEqual(
self.mock_auth.cleanup.call_args[0][0][0].typ, "http-01")
def test_preferred_challenge_choice_common_acme_1(self):
self._test_preferred_challenge_choice_common(combos=True)
def test_preferred_challenge_choice_common_acme_2(self):
self.mock_net.acme_version = 2
self._test_preferred_challenge_choice_common(combos=False)
def _test_preferred_challenges_not_supported_common(self, combos):
authzrs = [gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES, combos=combos)]
def test_preferred_challenges_not_supported_acme_2(self):
authzrs = [gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES)]
mock_order = mock.MagicMock(authorizations=authzrs)
self.handler.pref_challs.append(challenges.DNS01.typ)
self.assertRaises(
errors.AuthorizationError, self.handler.handle_authorizations,
mock_order, self.mock_config)
def test_preferred_challenges_not_supported_acme_1(self):
self._test_preferred_challenges_not_supported_common(combos=True)
def test_preferred_challenges_not_supported_acme_2(self):
self.mock_net.acme_version = 2
self._test_preferred_challenges_not_supported_common(combos=False)
def test_dns_only_challenge_not_supported(self):
authzrs = [gen_dom_authzr(domain="0", challs=[acme_util.DNS01])]
mock_order = mock.MagicMock(authorizations=authzrs)
@@ -317,7 +269,7 @@ class HandleAuthorizationsTest(unittest.TestCase):
def test_perform_error(self):
self.mock_auth.perform.side_effect = errors.AuthorizationError
authzr = gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES, combos=True)
authzr = gen_dom_authzr(domain="0", challs=acme_util.CHALLENGES)
mock_order = mock.MagicMock(authorizations=[authzr])
self.assertRaises(errors.AuthorizationError, self.handler.handle_authorizations,
mock_order, self.mock_config)
@@ -392,7 +344,7 @@ class HandleAuthorizationsTest(unittest.TestCase):
authzr = acme_util.gen_authzr(
messages.STATUS_PENDING, "0",
[acme_util.DNS01],
[messages.STATUS_PENDING], False)
[messages.STATUS_PENDING])
mock_order = mock.MagicMock(authorizations=[authzr])
self.assertRaises(
errors.AuthorizationError, self.handler.handle_authorizations,
@@ -404,7 +356,7 @@ class HandleAuthorizationsTest(unittest.TestCase):
authzr = acme_util.gen_authzr(
messages.STATUS_VALID, "0",
[acme_util.DNS01],
[messages.STATUS_VALID], False)
[messages.STATUS_VALID])
mock_order = mock.MagicMock(authorizations=[authzr])
self.handler.handle_authorizations(mock_order, self.mock_config)
@@ -426,7 +378,7 @@ class HandleAuthorizationsTest(unittest.TestCase):
("is_valid_but_will_fail", messages.STATUS_VALID)]
to_deactivate = [acme_util.gen_authzr(a[1], a[0], [acme_util.HTTP01],
[a[1], False]) for a in to_deactivate]
[a[1]]) for a in to_deactivate]
orderr = mock.MagicMock(authorizations=to_deactivate)
self.mock_net.deactivate_authorization.side_effect = _mock_deactivate
@@ -452,8 +404,7 @@ def _gen_mock_on_poll(status=messages.STATUS_VALID, retry=0, wait_value=1):
effective_status,
authzr.body.identifier.value,
[challb.chall for challb in authzr.body.challenges],
[effective_status] * len(authzr.body.challenges),
authzr.body.combinations)
[effective_status] * len(authzr.body.challenges))
return updated_azr, mock.MagicMock(headers={'Retry-After': str(wait_value)})
return _mock
@@ -477,8 +428,6 @@ class ChallbToAchallTest(unittest.TestCase):
class GenChallengePathTest(unittest.TestCase):
"""Tests for certbot._internal.auth_handler.gen_challenge_path.
.. todo:: Add more tests for dumb_path... depending on what we want to do.
"""
def setUp(self):
logging.disable(logging.FATAL)
@@ -487,34 +436,25 @@ class GenChallengePathTest(unittest.TestCase):
logging.disable(logging.NOTSET)
@classmethod
def _call(cls, challbs, preferences, combinations):
def _call(cls, challbs, preferences):
from certbot._internal.auth_handler import gen_challenge_path
return gen_challenge_path(challbs, preferences, combinations)
return gen_challenge_path(challbs, preferences)
def test_common_case(self):
"""Given DNS01 and HTTP01 with appropriate combos."""
challbs = (acme_util.DNS01_P, acme_util.HTTP01_P)
prefs = [challenges.DNS01, challenges.HTTP01]
combos = ((0,), (1,))
# Smart then trivial dumb path test
self.assertEqual(self._call(challbs, prefs, combos), (0,))
self.assertTrue(self._call(challbs, prefs, None))
# Rearrange order...
self.assertEqual(self._call(challbs[::-1], prefs, combos), (1,))
self.assertTrue(self._call(challbs[::-1], prefs, None))
self.assertEqual(self._call(challbs, prefs), (0,))
self.assertEqual(self._call(challbs[::-1], prefs), (1,))
def test_not_supported(self):
challbs = (acme_util.DNS01_P, acme_util.HTTP01_P)
challbs = (acme_util.DNS01_P,)
prefs = [challenges.HTTP01]
combos = ((0, 1),)
# smart path fails because no challs in perfs satisfies combos
# smart path fails because no challs in prefs satisfies combos
self.assertRaises(
errors.AuthorizationError, self._call, challbs, prefs, combos)
# dumb path fails because all challbs are not supported
self.assertRaises(
errors.AuthorizationError, self._call, challbs, prefs, None)
errors.AuthorizationError, self._call, challbs, prefs)
class ReportFailedAuthzrsTest(unittest.TestCase):
@@ -615,11 +555,11 @@ def gen_auth_resp(chall_list):
for chall in chall_list]
def gen_dom_authzr(domain, challs, combos=True):
def gen_dom_authzr(domain, challs):
"""Generates new authzr for domains."""
return acme_util.gen_authzr(
messages.STATUS_PENDING, domain, challs,
[messages.STATUS_PENDING] * len(challs), combos)
[messages.STATUS_PENDING] * len(challs))
if __name__ == "__main__":
certbot/tests/cli_test.py
+1 -1
View File
@@ -85,7 +85,7 @@ class ParseTest(unittest.TestCase):
@staticmethod
def parse(*args, **kwargs):
"""Mocks zope.component.getUtility and calls _unmocked_parse."""
"""Mocks certbot._internal.display.obj.get_display and calls _unmocked_parse."""
with test_util.patch_display_util():
return ParseTest._unmocked_parse(*args, **kwargs)
certbot/tests/client_test.py
+26 -16
View File
@@ -69,13 +69,12 @@ class RegisterTest(test_util.ConfigTestCase):
self.config.register_unsafely_without_email = False
self.config.email = "alias@example.com"
self.account_storage = account.AccountMemoryStorage()
with mock.patch("zope.component.provideUtility"):
display_obj.set_display(MagicMock())
self.tos_cb = mock.MagicMock()
display_obj.set_display(MagicMock())
def _call(self):
from certbot._internal.client import register
tos_cb = mock.MagicMock()
return register(self.config, self.account_storage, tos_cb)
return register(self.config, self.account_storage, self.tos_cb)
@staticmethod
def _public_key_mock():
@@ -98,31 +97,42 @@ class RegisterTest(test_util.ConfigTestCase):
@staticmethod
@contextlib.contextmanager
def _patched_acme_client():
# This function is written this way to avoid deprecation warnings that
# are raised when BackwardsCompatibleClientV2 is accessed on the real
# acme.client module.
with mock.patch('certbot._internal.client.acme_client') as mock_acme_client:
yield mock_acme_client.BackwardsCompatibleClientV2
yield mock_acme_client.ClientV2
def test_no_tos(self):
with self._patched_acme_client() as mock_client:
mock_client.new_account_and_tos().terms_of_service = "http://tos"
mock_client.new_account().terms_of_service = "http://tos"
mock_client().external_account_required.side_effect = self._false_mock
with mock.patch("certbot._internal.eff.prepare_subscription") as mock_prepare:
mock_client().new_account_and_tos.side_effect = errors.Error
mock_client().new_account.side_effect = errors.Error
self.assertRaises(errors.Error, self._call)
self.assertIs(mock_prepare.called, False)
mock_client().new_account_and_tos.side_effect = None
mock_client().new_account.side_effect = None
self._call()
self.assertIs(mock_prepare.called, True)
@mock.patch('certbot._internal.eff.prepare_subscription')
def test_empty_meta(self, unused_mock_prepare):
# Test that we can handle an ACME server which does not implement the 'meta'
# directory object (for terms-of-service handling).
with self._patched_acme_client() as mock_client:
from acme.messages import Directory
mock_client().directory = Directory.from_json({})
mock_client().external_account_required.side_effect = self._false_mock
self._call()
self.assertIs(self.tos_cb.called, False)
@test_util.patch_display_util()
def test_it(self, unused_mock_get_utility):
with self._patched_acme_client() as mock_client:
mock_client().external_account_required.side_effect = self._false_mock
with mock.patch("certbot._internal.eff.handle_subscription"):
self._call()
self.assertIs(self.tos_cb.called, True)
@mock.patch("certbot._internal.client.display_ops.get_email")
def test_email_retry(self, mock_get_email):
@@ -133,7 +143,7 @@ class RegisterTest(test_util.ConfigTestCase):
with self._patched_acme_client() as mock_client:
mock_client().external_account_required.side_effect = self._false_mock
with mock.patch("certbot._internal.eff.prepare_subscription") as mock_prepare:
mock_client().new_account_and_tos.side_effect = [mx_err, mock.MagicMock()]
mock_client().new_account.side_effect = [mx_err, mock.MagicMock()]
self._call()
self.assertEqual(mock_get_email.call_count, 1)
self.assertIs(mock_prepare.called, True)
@@ -146,7 +156,7 @@ class RegisterTest(test_util.ConfigTestCase):
with self._patched_acme_client() as mock_client:
mock_client().external_account_required.side_effect = self._false_mock
with mock.patch("certbot._internal.eff.handle_subscription"):
mock_client().new_account_and_tos.side_effect = [mx_err, mock.MagicMock()]
mock_client().new_account.side_effect = [mx_err, mock.MagicMock()]
self.assertRaises(errors.Error, self._call)
def test_needs_email(self):
@@ -176,7 +186,7 @@ class RegisterTest(test_util.ConfigTestCase):
# check Certbot did not ask the user to provide an email
self.assertIs(mock_get_email.called, False)
# check Certbot created an account with no email. Contact should return empty
self.assertFalse(mock_client().new_account_and_tos.call_args[0][0].contact)
self.assertFalse(mock_client().new_account.call_args[0][0].contact)
@test_util.patch_display_util()
def test_with_eab_arguments(self, unused_mock_get_utility):
@@ -228,7 +238,7 @@ class RegisterTest(test_util.ConfigTestCase):
)
mock_client().external_account_required.side_effect = self._false_mock
with mock.patch("certbot._internal.eff.handle_subscription") as mock_handle:
mock_client().new_account_and_tos.side_effect = [mx_err, mock.MagicMock()]
mock_client().new_account.side_effect = [mx_err, mock.MagicMock()]
self.assertRaises(messages.Error, self._call)
self.assertIs(mock_handle.called, False)
@@ -245,7 +255,7 @@ class ClientTestCommon(test_util.ConfigTestCase):
from certbot._internal.client import Client
with mock.patch("certbot._internal.client.acme_client") as acme:
self.acme_client = acme.BackwardsCompatibleClientV2
self.acme_client = acme.ClientV2
self.acme = self.acme_client.return_value = mock.MagicMock()
self.client_network = acme.ClientNetwork
self.client = Client(
certbot/tests/compat/misc_test.py
+7 -46
View File
@@ -9,52 +9,7 @@ import warnings
from certbot.compat import os
class ExecuteTest(unittest.TestCase):
"""Tests for certbot.compat.misc.execute_command."""
@classmethod
def _call(cls, *args, **kwargs):
from certbot.compat.misc import execute_command
# execute_command is superseded by execute_command_status
with warnings.catch_warnings():
warnings.simplefilter('ignore', category=PendingDeprecationWarning)
return execute_command(*args, **kwargs)
def test_it(self):
for returncode in range(0, 2):
for stdout in ("", "Hello World!",):
for stderr in ("", "Goodbye Cruel World!"):
self._test_common(returncode, stdout, stderr)
def _test_common(self, returncode, stdout, stderr):
given_command = "foo"
given_name = "foo-hook"
with mock.patch("certbot.compat.misc.subprocess.run") as mock_run:
mock_run.return_value.stdout = stdout
mock_run.return_value.stderr = stderr
mock_run.return_value.returncode = returncode
with mock.patch("certbot.compat.misc.logger") as mock_logger:
self.assertEqual(self._call(given_name, given_command), (stderr, stdout))
executed_command = mock_run.call_args[1].get(
"args", mock_run.call_args[0][0])
if os.name == 'nt':
expected_command = ['powershell.exe', '-Command', given_command]
else:
expected_command = given_command
self.assertEqual(executed_command, expected_command)
mock_logger.info.assert_any_call("Running %s command: %s",
given_name, given_command)
if stdout:
mock_logger.info.assert_any_call(mock.ANY, mock.ANY,
mock.ANY, stdout)
if stderr or returncode:
self.assertIs(mock_logger.error.called, True)
class ExecuteStatusTest(ExecuteTest):
class ExecuteStatusTest(unittest.TestCase):
"""Tests for certbot.compat.misc.execute_command_status."""
@classmethod
@@ -84,6 +39,12 @@ class ExecuteStatusTest(ExecuteTest):
mock_logger.info.assert_any_call("Running %s command: %s",
given_name, given_command)
def test_it(self):
for returncode in range(0, 2):
for stdout in ("", "Hello World!",):
for stderr in ("", "Goodbye Cruel World!"):
self._test_common(returncode, stdout, stderr)
if __name__ == "__main__":
unittest.main() # pragma: no cover
certbot/tests/crypto_util_test.py
-37
View File
@@ -2,8 +2,6 @@
import logging
import unittest
import certbot.util
try:
import mock
except ImportError: # pragma: no cover
@@ -67,23 +65,6 @@ class GenerateKeyTest(test_util.TempDirTestCase):
self.assertRaises(ValueError, self._call, 431, self.workdir)
class InitSaveKey(unittest.TestCase):
"""Test for certbot.crypto_util.init_save_key."""
@mock.patch("certbot.crypto_util.generate_key")
@mock.patch("certbot.crypto_util.zope.component")
def test_it(self, mock_zope, mock_generate):
from certbot.crypto_util import init_save_key
mock_zope.getUtility.return_value = mock.MagicMock(strict_permissions=True)
with self.assertWarns(DeprecationWarning):
init_save_key(4096, "/some/path")
mock_generate.assert_called_with(4096, "/some/path", elliptic_curve="secp256r1",
key_type="rsa", keyname="key-certbot.pem",
strict_permissions=True)
class GenerateCSRTest(test_util.TempDirTestCase):
"""Tests for certbot.crypto_util.generate_csr."""
@mock.patch('acme.crypto_util.make_csr')
@@ -100,24 +81,6 @@ class GenerateCSRTest(test_util.TempDirTestCase):
self.assertIn('csr-certbot.pem', csr.file)
class InitSaveCsr(unittest.TestCase):
"""Tests for certbot.crypto_util.init_save_csr."""
@mock.patch("certbot.crypto_util.generate_csr")
@mock.patch("certbot.crypto_util.zope.component")
def test_it(self, mock_zope, mock_generate):
from certbot.crypto_util import init_save_csr
mock_zope.getUtility.return_value = mock.MagicMock(must_staple=True,
strict_permissions=True)
key = certbot.util.Key(file=None, pem=None)
with self.assertWarns(DeprecationWarning):
init_save_csr(key, {"dummy"}, "/some/path")
mock_generate.assert_called_with(key, {"dummy"}, "/some/path",
must_staple=True, strict_permissions=True)
class ValidCSRTest(unittest.TestCase):
"""Tests for certbot.crypto_util.valid_csr."""
certbot/tests/display/obj_test.py
+1 -1
View File
@@ -275,7 +275,7 @@ class NoninteractiveDisplayTest(unittest.TestCase):
"""Test non-interactive display. These tests are pretty easy!"""
def setUp(self):
self.mock_stdout = mock.MagicMock()
self.displayer = display_util.NoninteractiveDisplay(self.mock_stdout)
self.displayer = display_obj.NoninteractiveDisplay(self.mock_stdout)
@mock.patch("certbot._internal.display.obj.logger")
def test_notification_no_pause(self, mock_logger):
certbot/tests/main_test.py
+41 -22
View File
@@ -388,7 +388,7 @@ class RevokeTest(test_util.TempDirTestCase):
mock.patch('certbot._internal.main._determine_account'),
mock.patch('certbot._internal.main.display_ops.success_revocation')
]
self.mock_acme_client = patches[0].start().BackwardsCompatibleClientV2
self.mock_acme_client = patches[0].start().ClientV2
patches[1].start()
self.mock_determine_account = patches[2].start()
self.mock_success_revoke = patches[3].start()
@@ -418,12 +418,19 @@ class RevokeTest(test_util.TempDirTestCase):
from certbot._internal.main import revoke
revoke(config, plugins)
def _mock_set_by_cli(self, mocked: mock.MagicMock, key: str, value: bool) -> None:
def set_by_cli(k: str) -> bool:
if key == k:
return value
return mock.DEFAULT
mocked.side_effect = set_by_cli
@mock.patch('certbot._internal.main._delete_if_appropriate')
@mock.patch('certbot._internal.main.client.acme_client')
def test_revoke_with_reason(self, mock_acme_client,
mock_delete_if_appropriate):
mock_delete_if_appropriate.return_value = False
mock_revoke = mock_acme_client.BackwardsCompatibleClientV2().revoke
mock_revoke = mock_acme_client.ClientV2().revoke
expected = []
for reason, code in constants.REVOCATION_REASONS.items():
args = 'revoke --cert-path={0} --reason {1}'.format(self.tmp_cert_path, reason).split()
@@ -438,42 +445,56 @@ class RevokeTest(test_util.TempDirTestCase):
@mock.patch('certbot._internal.main._delete_if_appropriate')
@mock.patch('certbot._internal.storage.RenewableCert')
@mock.patch('certbot._internal.storage.renewal_file_for_certname')
def test_revoke_by_certname(self, unused_mock_renewal_file_for_certname,
mock_cert, mock_delete_if_appropriate):
@mock.patch('certbot._internal.client.acme_from_config_key')
@mock.patch('certbot._internal.cli.set_by_cli')
def test_revoke_by_certname(self, mock_set_by_cli, mock_acme_from_config,
unused_mock_renewal_file_for_certname, mock_cert,
mock_delete_if_appropriate):
self._mock_set_by_cli(mock_set_by_cli, "server", False)
mock_acme_from_config.return_value = self.mock_acme_client
mock_cert.return_value = mock.MagicMock(cert_path=self.tmp_cert_path,
server="https://acme.example")
args = 'revoke --cert-name=example.com'.split()
mock_delete_if_appropriate.return_value = False
self._call(args)
self.mock_acme_client.assert_called_once_with(mock.ANY, mock.ANY, 'https://acme.example')
self.assertEqual(mock_acme_from_config.call_args_list[0][0][0].server,
'https://acme.example')
self.mock_success_revoke.assert_called_once_with(self.tmp_cert_path)
@mock.patch('certbot._internal.main._delete_if_appropriate')
@mock.patch('certbot._internal.storage.RenewableCert')
@mock.patch('certbot._internal.storage.renewal_file_for_certname')
def test_revoke_by_certname_and_server(self, unused_mock_renewal_file_for_certname,
mock_cert, mock_delete_if_appropriate):
@mock.patch('certbot._internal.client.acme_from_config_key')
@mock.patch('certbot._internal.cli.set_by_cli')
def test_revoke_by_certname_and_server(self, mock_set_by_cli, mock_acme_from_config,
unused_mock_renewal_file_for_certname, mock_cert,
mock_delete_if_appropriate):
"""Revoking with --server should use the server from the CLI"""
self._mock_set_by_cli(mock_set_by_cli, "server", True)
mock_cert.return_value = mock.MagicMock(cert_path=self.tmp_cert_path,
server="https://acme.example")
args = 'revoke --cert-name=example.com --server https://other.example'.split()
mock_delete_if_appropriate.return_value = False
self._call(args)
self.mock_acme_client.assert_called_once_with(mock.ANY, mock.ANY, 'https://other.example')
self.assertEqual(mock_acme_from_config.call_args_list[0][0][0].server,
'https://other.example')
self.mock_success_revoke.assert_called_once_with(self.tmp_cert_path)
@mock.patch('certbot._internal.main._delete_if_appropriate')
@mock.patch('certbot._internal.storage.RenewableCert')
@mock.patch('certbot._internal.storage.renewal_file_for_certname')
def test_revoke_by_certname_empty_server(self, unused_mock_renewal_file_for_certname,
@mock.patch('certbot._internal.client.acme_from_config_key')
@mock.patch('certbot._internal.cli.set_by_cli')
def test_revoke_by_certname_empty_server(self, mock_set_by_cli, mock_acme_from_config,
unused_mock_renewal_file_for_certname,
mock_cert, mock_delete_if_appropriate):
"""Revoking with --cert-name where the lineage server is empty shouldn't crash """
mock_cert.return_value = mock.MagicMock(cert_path=self.tmp_cert_path, server=None)
args = 'revoke --cert-name=example.com'.split()
mock_delete_if_appropriate.return_value = False
self._call(args)
self.mock_acme_client.assert_called_once_with(
mock.ANY, mock.ANY, constants.CLI_DEFAULTS['server'])
self.assertEqual(mock_acme_from_config.call_args_list[0][0][0].server,
constants.CLI_DEFAULTS['server'])
self.mock_success_revoke.assert_called_once_with(self.tmp_cert_path)
@mock.patch('certbot._internal.main._delete_if_appropriate')
@@ -1034,9 +1055,7 @@ class MainTest(test_util.ConfigTestCase):
plugins.visible().ifaces.assert_called_once_with(ifaces)
filtered = plugins.visible().ifaces()
self.assertEqual(filtered.init.call_count, 1)
filtered.verify.assert_called_once_with(ifaces)
verified = filtered.verify()
self.assertEqual(stdout.getvalue().strip(), str(verified))
self.assertEqual(stdout.getvalue().strip(), str(filtered))
@mock.patch('certbot._internal.main.plugins_disco')
@mock.patch('certbot._internal.main.cli.HelpfulArgumentParser.determine_help_topics')
@@ -1052,11 +1071,9 @@ class MainTest(test_util.ConfigTestCase):
plugins.visible().ifaces.assert_called_once_with(ifaces)
filtered = plugins.visible().ifaces()
self.assertEqual(filtered.init.call_count, 1)
filtered.verify.assert_called_once_with(ifaces)
verified = filtered.verify()
verified.prepare.assert_called_once_with()
verified.available.assert_called_once_with()
available = verified.available()
filtered.prepare.assert_called_once_with()
filtered.available.assert_called_once_with()
available = filtered.available()
self.assertEqual(stdout.getvalue().strip(), str(available))
def test_certonly_abspath(self):
@@ -1192,6 +1209,7 @@ class MainTest(test_util.ConfigTestCase):
mock_lineage.has_pending_deployment.return_value = False
mock_lineage.names.return_value = ['isnot.org']
mock_lineage.private_key_type = 'RSA'
mock_lineage.rsa_key_size = 2048
mock_certr = mock.MagicMock()
mock_key = mock.MagicMock(pem='pem_key')
mock_client = mock.MagicMock()
@@ -1566,11 +1584,12 @@ class MainTest(test_util.ConfigTestCase):
self._call_no_clientmock(['--cert-path', SS_CERT_PATH, '--key-path', RSA2048_KEY_PATH,
'--server', server, 'revoke'])
with open(RSA2048_KEY_PATH, 'rb') as f:
mock_acme_client.BackwardsCompatibleClientV2.assert_called_once_with(
mock.ANY, jose.JWK.load(f.read()), server)
self.assertEqual(mock_acme_client.ClientV2.call_count, 1)
self.assertEqual(mock_acme_client.ClientNetwork.call_args[0][0],
jose.JWK.load(f.read()))
with open(SS_CERT_PATH, 'rb') as f:
cert = crypto_util.pyopenssl_load_certificate(f.read())[0]
mock_revoke = mock_acme_client.BackwardsCompatibleClientV2().revoke
mock_revoke = mock_acme_client.ClientV2().revoke
mock_revoke.assert_called_once_with(
jose.ComparableX509(cert),
mock.ANY)
certbot/tests/plugins/disco_test.py
-32
View File
@@ -5,7 +5,6 @@ from typing import List
import unittest
import pkg_resources
import zope.interface
from certbot import errors
from certbot import interfaces
@@ -129,29 +128,6 @@ class PluginEntryPointTest(unittest.TestCase):
self.assertIs(self.plugin_ep.misconfigured, False)
self.assertIs(self.plugin_ep.available, False)
def test_verify(self):
iface1 = mock.MagicMock(__name__="iface1")
iface2 = mock.MagicMock(__name__="iface2")
iface3 = mock.MagicMock(__name__="iface3")
# pylint: disable=protected-access
self.plugin_ep._initialized = plugin = mock.MagicMock()
exceptions = zope.interface.exceptions
with mock.patch("certbot._internal.plugins.disco._verify") as mock_verify:
mock_verify.exceptions = exceptions
def verify_object(obj, cls, iface): # pylint: disable=missing-docstring
assert obj is plugin
assert iface is iface1 or iface is iface2 or iface is iface3
if iface is iface3:
return False
return True
mock_verify.side_effect = verify_object
self.assertTrue(self.plugin_ep.verify((iface1,)))
self.assertTrue(self.plugin_ep.verify((iface1, iface2)))
self.assertFalse(self.plugin_ep.verify((iface3,)))
self.assertFalse(self.plugin_ep.verify((iface1, iface3)))
def test_prepare(self):
config = mock.MagicMock()
self.plugin_ep.init(config=config)
@@ -264,14 +240,6 @@ class PluginsRegistryTest(unittest.TestCase):
self.plugin_ep.ifaces.return_value = False
self.assertEqual({}, self.reg.ifaces()._plugins)
def test_verify(self):
self.plugin_ep.verify.return_value = True
# pylint: disable=protected-access
self.assertEqual(
self.plugins, self.reg.verify(mock.MagicMock())._plugins)
self.plugin_ep.verify.return_value = False
self.assertEqual({}, self.reg.verify(mock.MagicMock())._plugins)
def test_prepare(self):
self.plugin_ep.prepare.return_value = "baz"
self.assertEqual(["baz"], self.reg.prepare())
certbot/tests/plugins/selection_test.py
+4 -4
View File
@@ -77,7 +77,7 @@ class PickPluginTest(unittest.TestCase):
plugin_ep.init.return_value = "foo"
plugin_ep.misconfigured = False
self.reg.visible().ifaces().verify().available.return_value = {
self.reg.visible().ifaces().available.return_value = {
"bar": plugin_ep}
self.assertEqual("foo", self._call())
@@ -86,14 +86,14 @@ class PickPluginTest(unittest.TestCase):
plugin_ep.init.return_value = "foo"
plugin_ep.misconfigured = True
self.reg.visible().ifaces().verify().available.return_value = {
self.reg.visible().ifaces().available.return_value = {
"bar": plugin_ep}
self.assertIsNone(self._call())
def test_multiple(self):
plugin_ep = mock.MagicMock()
plugin_ep.init.return_value = "foo"
self.reg.visible().ifaces().verify().available.return_value = {
self.reg.visible().ifaces().available.return_value = {
"bar": plugin_ep,
"baz": plugin_ep,
}
@@ -104,7 +104,7 @@ class PickPluginTest(unittest.TestCase):
[plugin_ep, plugin_ep], self.question)
def test_choose_plugin_none(self):
self.reg.visible().ifaces().verify().available.return_value = {
self.reg.visible().ifaces().available.return_value = {
"bar": None,
"baz": None,
}
certbot/tests/renewal_test.py
+39 -3
View File
@@ -55,7 +55,8 @@ class RenewalTest(test_util.ConfigTestCase):
self.assertEqual(self.config.webroot_map, {})
self.assertEqual(self.config.webroot_path, ['/var/www/test'])
def test_reuse_key_renewal_params(self):
@mock.patch('certbot._internal.renewal._avoid_reuse_key_conflicts')
def test_reuse_key_renewal_params(self, unused_mock_avoid_reuse_conflicts):
self.config.rsa_key_size = 'INVALID_VALUE'
self.config.reuse_key = True
self.config.dry_run = True
@@ -75,7 +76,8 @@ class RenewalTest(test_util.ConfigTestCase):
assert self.config.rsa_key_size == 2048
def test_reuse_ec_key_renewal_params(self):
@mock.patch('certbot._internal.renewal._avoid_reuse_key_conflicts')
def test_reuse_ec_key_renewal_params(self, unused_mock_avoid_reuse_conflicts):
self.config.elliptic_curve = 'INVALID_CURVE'
self.config.reuse_key = True
self.config.dry_run = True
@@ -99,7 +101,9 @@ class RenewalTest(test_util.ConfigTestCase):
assert self.config.elliptic_curve == 'secp256r1'
def test_new_key(self):
@mock.patch('certbot._internal.renewal.cli.set_by_cli')
def test_new_key(self, mock_set_by_cli):
mock_set_by_cli.return_value = False
# When renewing with both reuse_key and new_key, the key should be regenerated,
# the key type, key parameters and reuse_key should be kept.
self.config.reuse_key = True
@@ -125,6 +129,38 @@ class RenewalTest(test_util.ConfigTestCase):
# None is passed as the existing key, i.e. the key is not actually being reused.
le_client.obtain_certificate.assert_called_with(mock.ANY, None)
@mock.patch('certbot._internal.renewal.hooks.renew_hook')
@mock.patch('certbot._internal.renewal.cli.set_by_cli')
def test_reuse_key_conflicts(self, mock_set_by_cli, unused_mock_renew_hook):
mock_set_by_cli.return_value = False
# When renewing with reuse_key and a conflicting key parameter (size, curve)
# an error should be raised ...
self.config.reuse_key = True
self.config.key_type = "rsa"
self.config.rsa_key_size = 4096
self.config.dry_run = True
config = configuration.NamespaceConfig(self.config)
rc_path = test_util.make_lineage(
self.config.config_dir, 'sample-renewal.conf')
lineage = storage.RenewableCert(rc_path, config)
lineage.configuration["renewalparams"]["reuse_key"] = True
le_client = mock.MagicMock()
le_client.obtain_certificate.return_value = (None, None, None, None)
from certbot._internal import renewal
with self.assertRaisesRegex(errors.Error, "Unable to change the --rsa-key-type"):
renewal.renew_cert(self.config, None, le_client, lineage)
# ... unless --no-reuse-key is set
mock_set_by_cli.side_effect = lambda var: var == "reuse_key"
self.config.reuse_key = False
renewal.renew_cert(self.config, None, le_client, lineage)
@test_util.patch_display_util()
@mock.patch('certbot._internal.renewal.cli.set_by_cli')
def test_remove_deprecated_config_elements(self, mock_set_by_cli, unused_mock_get_utility):
certbot/tests/reporter_test.py
-89
View File
@@ -1,89 +0,0 @@
"""Tests for certbot._internal.reporter."""
import io
import sys
import unittest
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
class ReporterTest(unittest.TestCase):
"""Tests for certbot._internal.reporter.Reporter."""
def setUp(self):
from certbot._internal import reporter
self.reporter = reporter.Reporter(mock.MagicMock(quiet=False))
self.old_stdout = sys.stdout
sys.stdout = io.StringIO()
def tearDown(self):
sys.stdout = self.old_stdout
def test_multiline_message(self):
self.reporter.add_message("Line 1\nLine 2", self.reporter.LOW_PRIORITY)
self.reporter.print_messages()
output = sys.stdout.getvalue()
self.assertIn("Line 1\n", output)
self.assertIn("Line 2", output)
def test_tty_print_empty(self):
sys.stdout.isatty = lambda: True
self.test_no_tty_print_empty()
def test_no_tty_print_empty(self):
self.reporter.print_messages()
self.assertEqual(sys.stdout.getvalue(), "")
try:
raise ValueError
except ValueError:
self.reporter.print_messages()
self.assertEqual(sys.stdout.getvalue(), "")
def test_tty_successful_exit(self):
sys.stdout.isatty = lambda: True
self._successful_exit_common()
def test_no_tty_successful_exit(self):
self._successful_exit_common()
def test_tty_unsuccessful_exit(self):
sys.stdout.isatty = lambda: True
self._unsuccessful_exit_common()
def test_no_tty_unsuccessful_exit(self):
self._unsuccessful_exit_common()
def _successful_exit_common(self):
self._add_messages()
self.reporter.print_messages()
output = sys.stdout.getvalue()
self.assertIn("IMPORTANT NOTES:", output)
self.assertIn("High", output)
self.assertIn("Med", output)
self.assertIn("Low", output)
def _unsuccessful_exit_common(self):
self._add_messages()
try:
raise ValueError
except ValueError:
self.reporter.print_messages()
output = sys.stdout.getvalue()
self.assertIn("IMPORTANT NOTES:", output)
self.assertIn("High", output)
self.assertNotIn("Med", output)
self.assertNotIn("Low", output)
def _add_messages(self):
self.reporter.add_message("High", self.reporter.HIGH_PRIORITY)
self.reporter.add_message(
"Med", self.reporter.MEDIUM_PRIORITY, on_crash=False)
self.reporter.add_message(
"Low", self.reporter.LOW_PRIORITY, on_crash=False)
if __name__ == "__main__":
unittest.main() # pragma: no cover
certbot/tests/util_test.py
-13
View File
@@ -592,19 +592,6 @@ class OsInfoTest(unittest.TestCase):
self.assertEqual(cbutil.get_python_os_info(), ("testdist", "42"))
class GetStrictVersionTest(unittest.TestCase):
"""Test for certbot.util.get_strict_version."""
@classmethod
def _call(cls, *args, **kwargs):
from certbot.util import get_strict_version
return get_strict_version(*args, **kwargs)
def test_it(self):
with self.assertWarnsRegex(DeprecationWarning, "get_strict_version"):
self._call("1.2.3")
class AtexitRegisterTest(unittest.TestCase):
"""Tests for certbot.util.atexit_register."""
def setUp(self):