mirror of
https://github.com/certbot/certbot.git
synced 2026-08-03 08:03:10 +02:00
@@ -222,6 +222,24 @@ def test_certonly_webroot(context: IntegrationTestsContext) -> None:
|
||||
assert_cert_count_for_lineage(context.config_dir, certname, 1)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform == 'darwin',
|
||||
reason='macOS has one IPv4 loopback address by default')
|
||||
def test_certonly_webroot_ipv4(context: IntegrationTestsContext) -> None:
|
||||
"""Test the HTTP-01 challenge with an IPv4 address using webroot authenticator.
|
||||
|
||||
This test relies on proxy.py being able to forward requests for, e.g. `127.0.0.2`,
|
||||
(`local_ip`) to this test runner. That works on Linux because 127.0.0.0/8 all routes
|
||||
to localhost. However, on macOS by default only 127.0.0.1 is routed, so this test is
|
||||
skipped. If you want to run it, configure additional loopback addresses:
|
||||
for n in $(seq 2 127) ; do sudo ifconfig lo0 alias "127.0.0.${n}" up ; done.
|
||||
"""
|
||||
with misc.create_http_server(context.http_01_port) as webroot:
|
||||
context.certbot(['certonly', '-a', 'webroot', '--webroot-path', webroot,
|
||||
'--ip-address', context.local_ip])
|
||||
|
||||
assert_cert_count_for_lineage(context.config_dir, context.local_ip, 1)
|
||||
|
||||
|
||||
def test_auth_and_install_with_csr(context: IntegrationTestsContext) -> None:
|
||||
"""Test certificate issuance and install using an existing CSR."""
|
||||
certname = context.get_domain('le3')
|
||||
|
||||
@@ -25,12 +25,13 @@ from certbot._internal.cli.cli_constants import ZERO_ARG_ACTIONS
|
||||
from certbot._internal.cli.cli_utils import _EncodeReasonAction
|
||||
from certbot._internal.cli.cli_utils import _PrefChallAction
|
||||
from certbot._internal.cli.cli_utils import _user_agent_comment_type
|
||||
from certbot._internal.cli.cli_utils import add_domains
|
||||
from certbot._internal.cli.cli_utils import add_dns_name
|
||||
from certbot._internal.cli.cli_utils import add_ip_address
|
||||
from certbot._internal.cli.cli_utils import CaseInsensitiveList
|
||||
from certbot._internal.cli.cli_utils import config_help
|
||||
from certbot._internal.cli.cli_utils import CustomHelpFormatter
|
||||
from certbot._internal.cli.cli_utils import DomainsAction
|
||||
from certbot._internal.cli.cli_utils import _IPAddressAction
|
||||
from certbot._internal.cli.cli_utils import IPAddressAction
|
||||
from certbot._internal.cli.cli_utils import flag_default
|
||||
from certbot._internal.cli.cli_utils import HelpfulArgumentGroup
|
||||
from certbot._internal.cli.cli_utils import nonnegative_int
|
||||
@@ -125,7 +126,7 @@ def prepare_and_parse_args(plugins: plugins_disco.PluginsRegistry, args: list[st
|
||||
helpful.add(
|
||||
[None, "certonly", "certificates"],
|
||||
"--ip-address", dest="ip_addresses",
|
||||
action=_IPAddressAction,
|
||||
action=IPAddressAction,
|
||||
default=flag_default("ip_addresses"),
|
||||
help="IP addresses to include. For multiple IP addresses you can use multiple "
|
||||
"--ip-address flags. All IP addresses will be included as Subject Alternative Names "
|
||||
|
||||
@@ -98,45 +98,31 @@ class DomainsAction(argparse.Action):
|
||||
def __call__(self, parser: argparse.ArgumentParser, namespace: argparse.Namespace,
|
||||
values: str | Sequence[Any] | None,
|
||||
option_string: str | None = None) -> None:
|
||||
"""Just wrap add_domains in argparseese."""
|
||||
match values:
|
||||
case str():
|
||||
add_domains(namespace, str(values))
|
||||
for domain in values.split(","):
|
||||
add_dns_name(namespace, san.DNSName(domain.strip()))
|
||||
case _:
|
||||
# https://docs.python.org/3/library/argparse.html#nargs
|
||||
raise TypeError("shouldn't happen: non-str passed by argparse when nargs=None")
|
||||
|
||||
|
||||
def add_domains(args_or_config: Union[argparse.Namespace, configuration.NamespaceConfig],
|
||||
domains: Optional[str]) -> list[san.DNSName]:
|
||||
"""Registers new domains to be used during the current client run.
|
||||
def add_dns_name(args_or_config: Union[argparse.Namespace, configuration.NamespaceConfig],
|
||||
dns_name: san.DNSName) -> None:
|
||||
"""Registers a new domain to be used during the current client run.
|
||||
|
||||
Domains are not added to the list of requested domains if they have
|
||||
already been registered.
|
||||
The domain is not added if it has already been registered.
|
||||
|
||||
:param args_or_config: parsed command line arguments
|
||||
:type args_or_config: argparse.Namespace or
|
||||
configuration.NamespaceConfig
|
||||
:param str domain: one or more comma separated domains
|
||||
|
||||
:returns: domains after they have been normalized and validated
|
||||
:rtype: `list` of `str`
|
||||
|
||||
:param san.DNSName dns_name: a DNS name
|
||||
"""
|
||||
validated_domains: list[san.DNSName] = []
|
||||
if not domains:
|
||||
return validated_domains
|
||||
|
||||
for d in domains.split(","):
|
||||
domain = san.DNSName(d.strip())
|
||||
validated_domains.append(domain)
|
||||
if domain not in args_or_config.domains:
|
||||
args_or_config.domains.append(domain)
|
||||
|
||||
return validated_domains
|
||||
if dns_name not in args_or_config.domains:
|
||||
args_or_config.domains.append(dns_name)
|
||||
|
||||
|
||||
class _IPAddressAction(argparse.Action):
|
||||
class IPAddressAction(argparse.Action):
|
||||
"""Action class for parsing IP addresses."""
|
||||
|
||||
def __call__(self, parser: argparse.ArgumentParser, namespace: argparse.Namespace,
|
||||
@@ -145,12 +131,26 @@ class _IPAddressAction(argparse.Action):
|
||||
match values:
|
||||
case str():
|
||||
# This will throw an exception if the IP address doesn't parse.
|
||||
namespace.ip_addresses.append(san.IPAddress(values))
|
||||
add_ip_address(namespace, san.IPAddress(values))
|
||||
case _:
|
||||
# https://docs.python.org/3/library/argparse.html#nargs
|
||||
raise TypeError("shouldn't happen: non-str passed by argparse when nargs=None")
|
||||
|
||||
|
||||
def add_ip_address(args_or_config: Union[argparse.Namespace, configuration.NamespaceConfig],
|
||||
ip_address: san.IPAddress) -> None:
|
||||
"""Registers a new IP address to be used during the current client run.
|
||||
|
||||
The IP address is not added if it has already been registered.
|
||||
|
||||
:param args_or_config: parsed command line arguments
|
||||
:type args_or_config: argparse.Namespace or
|
||||
configuration.NamespaceConfig
|
||||
:param san.IPAddress ip_address: an IP address
|
||||
"""
|
||||
if ip_address not in args_or_config.ip_addresses:
|
||||
args_or_config.ip_addresses.append(ip_address)
|
||||
|
||||
class CaseInsensitiveList(list):
|
||||
"""A list that will ignore case when searching.
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ from certbot import crypto_util
|
||||
from certbot import errors
|
||||
from certbot import interfaces
|
||||
from certbot._internal import cli
|
||||
from certbot._internal import san
|
||||
from certbot.achallenges import AnnotatedChallenge
|
||||
from certbot.compat import filesystem
|
||||
from certbot.compat import os
|
||||
@@ -69,14 +70,14 @@ to serve all files under specified web root ({0})."""
|
||||
def add_parser_arguments(cls, add: Callable[..., None]) -> None:
|
||||
add("path", "-w", default=[], action=_WebrootPathAction,
|
||||
help="public_html / webroot path. This can be specified multiple "
|
||||
"times to handle different domains; each domain will have "
|
||||
"times to handle different identifiers; each identifier will have "
|
||||
"the webroot path that preceded it. For instance: `-w "
|
||||
"/var/www/example -d example.com -d www.example.com -w "
|
||||
"/var/www/thing -d thing.net -d m.thing.net` (default: Ask)")
|
||||
add("map", default={}, action=_WebrootMapAction,
|
||||
help="JSON dictionary mapping domains to webroot paths; this "
|
||||
"implies -d for each entry. You may need to escape this from "
|
||||
"your shell. E.g.: --webroot-map "
|
||||
help="JSON dictionary mapping identifiers to webroot paths; this "
|
||||
"implies -d or --ip-address for each entry. You may need to "
|
||||
" escape this from your shell. E.g.: --webroot-map "
|
||||
'\'{"eg1.is,m.eg1.is":"/www/eg1/", "eg2.is":"/www/eg2"}\' '
|
||||
"This option is merged with, but takes precedence over, -w / "
|
||||
"-d entries. At present, if you put webroot-map in a config "
|
||||
@@ -85,7 +86,7 @@ to serve all files under specified web root ({0})."""
|
||||
|
||||
def auth_hint(self, failed_achalls: list[AnnotatedChallenge]) -> str: # pragma: no cover
|
||||
return ("The Certificate Authority failed to download the temporary challenge files "
|
||||
"created by Certbot. Ensure that the listed domains serve their content from "
|
||||
"created by Certbot. Ensure that the listed identifiers serve their content from "
|
||||
"the provided --webroot-path/-w and that files created there can be downloaded "
|
||||
"from the internet.")
|
||||
|
||||
@@ -106,9 +107,6 @@ to serve all files under specified web root ({0})."""
|
||||
pass
|
||||
|
||||
def perform(self, achalls: list[AnnotatedChallenge]) -> list[challenges.ChallengeResponse]: # pylint: disable=missing-function-docstring
|
||||
if any(achall.identifier.typ == messages.IDENTIFIER_IP for achall in achalls):
|
||||
raise errors.ConfigurationError(
|
||||
"webroot authenticator not supported for IP address certificates")
|
||||
self._set_webroots(achalls)
|
||||
|
||||
self._create_challenge_dirs()
|
||||
@@ -118,7 +116,7 @@ to serve all files under specified web root ({0})."""
|
||||
def _set_webroots(self, achalls: Iterable[AnnotatedChallenge]) -> None:
|
||||
if self.conf("path"):
|
||||
webroot_path = self.conf("path")[-1]
|
||||
logger.info("Using the webroot path %s for all unmatched domains.",
|
||||
logger.info("Using the webroot path %s for all unmatched identifiers.",
|
||||
webroot_path)
|
||||
for achall in achalls:
|
||||
self.conf("map").setdefault(achall.identifier.value, webroot_path)
|
||||
@@ -126,7 +124,7 @@ to serve all files under specified web root ({0})."""
|
||||
known_webroots = list(set(self.conf("map").values()))
|
||||
for achall in achalls:
|
||||
if achall.identifier.value not in self.conf("map"):
|
||||
new_webroot = self._prompt_for_webroot(achall.identifier.value,
|
||||
new_webroot = self._prompt_for_webroot(achall.identifier,
|
||||
known_webroots)
|
||||
# Put the most recently input
|
||||
# webroot first for easy selection
|
||||
@@ -137,46 +135,48 @@ to serve all files under specified web root ({0})."""
|
||||
known_webroots.insert(0, new_webroot)
|
||||
self.conf("map")[achall.identifier.value] = new_webroot
|
||||
|
||||
def _prompt_for_webroot(self, domain: str, known_webroots: list[str]) -> Optional[str]:
|
||||
def _prompt_for_webroot(self, identifier: messages.Identifier,
|
||||
known_webroots: list[str]) -> Optional[str]:
|
||||
webroot = None
|
||||
|
||||
while webroot is None:
|
||||
if known_webroots:
|
||||
# Only show the menu if we have options for it
|
||||
webroot = self._prompt_with_webroot_list(domain, known_webroots)
|
||||
webroot = self._prompt_with_webroot_list(identifier, known_webroots)
|
||||
if webroot is None:
|
||||
webroot = self._prompt_for_new_webroot(domain)
|
||||
webroot = self._prompt_for_new_webroot(identifier)
|
||||
else:
|
||||
# Allow prompt to raise PluginError instead of looping forever
|
||||
webroot = self._prompt_for_new_webroot(domain, True)
|
||||
webroot = self._prompt_for_new_webroot(identifier, True)
|
||||
|
||||
return webroot
|
||||
|
||||
def _prompt_with_webroot_list(self, domain: str,
|
||||
def _prompt_with_webroot_list(self, identifier: messages.Identifier,
|
||||
known_webroots: list[str]) -> Optional[str]:
|
||||
path_flag = "--" + self.option_name("path")
|
||||
|
||||
while True:
|
||||
code, index = display_util.menu(
|
||||
"Select the webroot for {0}:".format(domain),
|
||||
"Select the webroot for {0}:".format(identifier.value),
|
||||
["Enter a new webroot"] + known_webroots,
|
||||
cli_flag=path_flag, force_interactive=True)
|
||||
if code == display_util.CANCEL:
|
||||
raise errors.PluginError(
|
||||
"Every requested domain must have a "
|
||||
"Every requested identifier must have a "
|
||||
"webroot when using the webroot plugin.")
|
||||
return None if index == 0 else known_webroots[index - 1] # code == display_util.OK
|
||||
|
||||
def _prompt_for_new_webroot(self, domain: str, allowraise: bool = False) -> Optional[str]:
|
||||
def _prompt_for_new_webroot(self, identifier: messages.Identifier,
|
||||
allowraise: bool = False) -> Optional[str]:
|
||||
code, webroot = ops.validated_directory(
|
||||
_validate_webroot,
|
||||
"Input the webroot for {0}:".format(domain),
|
||||
"Input the webroot for {0}:".format(identifier.value),
|
||||
force_interactive=True)
|
||||
if code == display_util.CANCEL:
|
||||
if not allowraise:
|
||||
return None
|
||||
raise errors.PluginError(
|
||||
"Every requested domain must have a "
|
||||
"Every requested identifier must have a "
|
||||
"webroot when using the webroot plugin.")
|
||||
return _validate_webroot(webroot) # code == display_util.OK
|
||||
|
||||
@@ -184,9 +184,10 @@ to serve all files under specified web root ({0})."""
|
||||
path_map = self.conf("map")
|
||||
if not path_map:
|
||||
raise errors.PluginError(
|
||||
"Missing parts of webroot configuration; please set either "
|
||||
"--webroot-path and --domains, or --webroot-map. Run with "
|
||||
" --help webroot for examples.")
|
||||
"Missing parts of webroot configuration; please set "
|
||||
"--webroot-path and --domains or --ip-address. "
|
||||
"Alternatively you may set --webroot-map. "
|
||||
"Run with --help webroot for examples.")
|
||||
for name, path in path_map.items():
|
||||
self.full_roots[name] = os.path.join(path, os.path.normcase(
|
||||
challenges.HTTP01.URI_ROOT_PATH))
|
||||
@@ -295,10 +296,16 @@ class _WebrootMapAction(argparse.Action):
|
||||
option_string: Optional[str] = None) -> None:
|
||||
if webroot_map is None:
|
||||
return
|
||||
for domains, webroot_path in json.loads(str(webroot_map)).items():
|
||||
for identlist, webroot_path in json.loads(str(webroot_map)).items():
|
||||
webroot_path = _validate_webroot(webroot_path)
|
||||
namespace.webroot_map.update(
|
||||
(d.dns_name, webroot_path) for d in cli.add_domains(namespace, domains))
|
||||
for s in san.guess(identlist.split(",")):
|
||||
match s:
|
||||
case san.IPAddress():
|
||||
cli.add_ip_address(namespace, s)
|
||||
case san.DNSName():
|
||||
cli.add_dns_name(namespace, s)
|
||||
|
||||
namespace.webroot_map[str(s)] = webroot_path
|
||||
|
||||
|
||||
class _WebrootPathAction(argparse.Action):
|
||||
@@ -306,17 +313,17 @@ class _WebrootPathAction(argparse.Action):
|
||||
|
||||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
self._domain_before_webroot = False
|
||||
self._ident_before_webroot = False
|
||||
|
||||
def __call__(self, parser: argparse.ArgumentParser, namespace: argparse.Namespace,
|
||||
webroot_path: Union[str, Sequence[Any], None],
|
||||
option_string: Optional[str] = None) -> None:
|
||||
if webroot_path is None:
|
||||
return
|
||||
if self._domain_before_webroot:
|
||||
if self._ident_before_webroot:
|
||||
raise errors.PluginError(
|
||||
"If you specify multiple webroot paths, "
|
||||
"one of them must precede all domain flags")
|
||||
"one of them must precede all --domain and --ip-address flags")
|
||||
|
||||
if namespace.webroot_path:
|
||||
# Apply previous webroot to all matched
|
||||
@@ -324,8 +331,10 @@ class _WebrootPathAction(argparse.Action):
|
||||
prev_webroot = namespace.webroot_path[-1]
|
||||
for domain in namespace.domains:
|
||||
namespace.webroot_map.setdefault(domain.dns_name, prev_webroot)
|
||||
elif namespace.domains:
|
||||
self._domain_before_webroot = True
|
||||
for ip_address in namespace.ip_addresses:
|
||||
namespace.webroot_map.setdefault(str(ip_address), prev_webroot)
|
||||
elif namespace.domains or namespace.ip_addresses:
|
||||
self._ident_before_webroot = True
|
||||
|
||||
namespace.webroot_path.append(_validate_webroot(str(webroot_path)))
|
||||
|
||||
|
||||
@@ -632,15 +632,17 @@ def _renew_describe_results(config: configuration.NamespaceConfig, renew_success
|
||||
def handle_renewal_request(config: configuration.NamespaceConfig) -> None:
|
||||
"""Examine each lineage; renew if due and report results"""
|
||||
|
||||
# This is trivially False if config.domains is empty
|
||||
if any(domain.dns_name not in config.webroot_map for domain in config.domains):
|
||||
# If more plugins start using cli.add_domains,
|
||||
sans: list[san.SAN] = config.domains + config.ip_addresses
|
||||
|
||||
# This is trivially False if sans is empty
|
||||
if any(str(san) not in config.webroot_map for san in sans):
|
||||
# If more plugins start using cli.add_domain / cli.add_ip_address,
|
||||
# we may want to only log a warning here
|
||||
raise errors.Error("Currently, the renew verb is capable of either "
|
||||
"renewing all installed certificates that are due "
|
||||
"to be renewed or renewing a single certificate specified "
|
||||
"by its name. If you would like to renew specific "
|
||||
"certificates by their domains, use the certonly command "
|
||||
"certificates by their identifiers, use the certonly command "
|
||||
"instead. The renew verb may provide other options "
|
||||
"for selecting certificates to renew in the future.")
|
||||
|
||||
|
||||
@@ -317,18 +317,33 @@ class WebrootActionTest(unittest.TestCase):
|
||||
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="thing.com"),
|
||||
account_key=KEY)
|
||||
|
||||
ipchall = achallenges.KeyAuthorizationAnnotatedChallenge(
|
||||
challb=acme_util.chall_to_challb(
|
||||
challenges.HTTP01(token=((b'a' * 16))),
|
||||
messages.STATUS_PENDING),
|
||||
identifier=messages.Identifier(typ=messages.IDENTIFIER_IP, value="1.2.3.4"),
|
||||
account_key=KEY)
|
||||
|
||||
def setUp(self):
|
||||
from certbot._internal.plugins.webroot import Authenticator
|
||||
self.path = tempfile.mkdtemp()
|
||||
self.parser = argparse.ArgumentParser()
|
||||
self.parser.ip_addresses = []
|
||||
self.parser.add_argument("-d", "--domains",
|
||||
action=cli_utils.DomainsAction, default=[])
|
||||
self.parser.add_argument("--ip-address",
|
||||
action=cli_utils.IPAddressAction,
|
||||
dest="ip_addresses",
|
||||
default=[])
|
||||
Authenticator.inject_parser_options(self.parser, "webroot")
|
||||
|
||||
def test_webroot_map_action(self):
|
||||
other_path = tempfile.mkdtemp()
|
||||
args = self.parser.parse_args(
|
||||
["--webroot-map", json.dumps({'thing.com': self.path})])
|
||||
["--webroot-map", json.dumps({'thing.com,thunk.com,9.8.7.6': self.path,'thunk.com': other_path})])
|
||||
assert args.webroot_map["thing.com"] == self.path
|
||||
assert args.webroot_map["9.8.7.6"] == self.path
|
||||
assert args.webroot_map["thunk.com"] == other_path
|
||||
|
||||
def test_domain_before_webroot(self):
|
||||
args = self.parser.parse_args(
|
||||
@@ -336,6 +351,15 @@ class WebrootActionTest(unittest.TestCase):
|
||||
config = self._get_config_after_perform(args)
|
||||
assert config.webroot_map[self.achall.identifier.value] == self.path
|
||||
|
||||
def test_multi_identifier(self):
|
||||
args = self.parser.parse_args(
|
||||
"-w {0} -d {1} --ip-address {2}".format(
|
||||
self.path, self.achall.identifier.value, self.ipchall.identifier.value).split())
|
||||
|
||||
config = self._get_config_after_perform(args, challs=[self.achall, self.ipchall])
|
||||
assert config.webroot_map[self.achall.identifier.value] == self.path
|
||||
assert config.webroot_map[self.ipchall.identifier.value] == self.path
|
||||
|
||||
def test_domain_before_webroot_error(self):
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.parser.parse_args("-d foo -w bar -w baz".split())
|
||||
@@ -343,11 +367,26 @@ class WebrootActionTest(unittest.TestCase):
|
||||
self.parser.parse_args("-d foo -w bar -d baz -w qux".split())
|
||||
|
||||
def test_multiwebroot(self):
|
||||
args = self.parser.parse_args("-w {0} -d {1} -w {2} -d bar".format(
|
||||
self.path, self.achall.identifier.value, tempfile.mkdtemp()).split())
|
||||
assert args.webroot_map[self.achall.identifier.value] == self.path
|
||||
config = self._get_config_after_perform(args)
|
||||
assert config.webroot_map[self.achall.identifier.value] == self.path
|
||||
ip = self.ipchall.identifier.value
|
||||
dns_name = self.achall.identifier.value
|
||||
|
||||
ip_path = tempfile.mkdtemp()
|
||||
dns_path = tempfile.mkdtemp()
|
||||
args = self.parser.parse_args(f"-w {dns_path} -d {dns_name} -w {ip_path} --ip-address {ip}".split())
|
||||
config = self._get_config_after_perform(args, challs=[self.achall, self.ipchall])
|
||||
assert config.webroot_map[dns_name] == dns_path
|
||||
assert config.webroot_map[ip] == ip_path
|
||||
|
||||
def test_multiwebroot_ip_first(self):
|
||||
ip = self.ipchall.identifier.value
|
||||
dns_name = self.achall.identifier.value
|
||||
|
||||
ip_path = tempfile.mkdtemp()
|
||||
dns_path = tempfile.mkdtemp()
|
||||
args = self.parser.parse_args(f"-w {ip_path} --ip-address {ip} -w {dns_path} -d {dns_name}".split())
|
||||
config = self._get_config_after_perform(args, challs=[self.achall, self.ipchall])
|
||||
assert config.webroot_map[dns_name] == dns_path
|
||||
assert config.webroot_map[ip] == ip_path
|
||||
|
||||
def test_webroot_map_partial_without_perform(self):
|
||||
# This test acknowledges the fact that webroot_map content will be partial if webroot
|
||||
@@ -362,10 +401,12 @@ class WebrootActionTest(unittest.TestCase):
|
||||
assert args.webroot_map == {self.achall.identifier.value: self.path}
|
||||
assert args.webroot_path == [self.path, other_webroot_path]
|
||||
|
||||
def _get_config_after_perform(self, config):
|
||||
def _get_config_after_perform(self, config, challs=None):
|
||||
if not challs:
|
||||
challs = [self.achall]
|
||||
from certbot._internal.plugins.webroot import Authenticator
|
||||
auth = Authenticator(config, "webroot")
|
||||
auth.perform([self.achall])
|
||||
auth.perform(challs)
|
||||
return auth.config
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
The webroot plugin now supports IP address issuance.
|
||||
Reference in New Issue
Block a user