mirror of
https://github.com/certbot/certbot.git
synced 2026-07-27 16:29:55 +02:00
Add CLI flag --ip-address (#10495)
Co-authored-by: ohemorange <ebportnoy@gmail.com> Co-authored-by: Brad Warren <bmw@users.noreply.github.com>
This commit is contained in:
co-authored by
ohemorange
Brad Warren
parent
1ea35193ab
commit
58724f68ec
@@ -2442,7 +2442,7 @@ class ApacheConfigurator(common.Configurator):
|
||||
###########################################################################
|
||||
# Challenges Section
|
||||
###########################################################################
|
||||
def get_chall_pref(self, unused_domain: str) -> Sequence[type[challenges.HTTP01]]:
|
||||
def get_chall_pref(self, unused_identifier: str) -> Sequence[type[challenges.HTTP01]]:
|
||||
"""Return list of challenge preferences."""
|
||||
return [challenges.HTTP01]
|
||||
|
||||
|
||||
@@ -24,6 +24,7 @@ class IntegrationTestsContext:
|
||||
|
||||
self.directory_url = acme_xdist['directory_url']
|
||||
self.https_port = acme_xdist['https_port'][self.worker_id]
|
||||
self.local_ip = str(acme_xdist['local_ip'][self.worker_id])
|
||||
self.http_01_port = acme_xdist['http_port'][self.worker_id]
|
||||
self.other_port = acme_xdist['other_port'][self.worker_id]
|
||||
# Challtestsrv REST API, that exposes entrypoints to register new DNS entries,
|
||||
|
||||
@@ -6,6 +6,7 @@ from os.path import join
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import Generator
|
||||
|
||||
from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurve
|
||||
@@ -32,7 +33,7 @@ from certbot_integration_tests.certbot_tests.assertions import assert_world_no_p
|
||||
from certbot_integration_tests.certbot_tests.assertions import assert_world_read_permissions
|
||||
from certbot_integration_tests.certbot_tests.assertions import EVERYBODY_SID
|
||||
from certbot_integration_tests.certbot_tests.context import IntegrationTestsContext
|
||||
from certbot_integration_tests.utils import misc
|
||||
from certbot_integration_tests.utils import misc, constants
|
||||
|
||||
|
||||
@pytest.fixture(name='context')
|
||||
@@ -111,6 +112,59 @@ def test_http_01(context: IntegrationTestsContext) -> None:
|
||||
assert_saved_lineage_option(context.config_dir, certname, 'key_type', 'ecdsa')
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform == 'darwin',
|
||||
reason='macOS has one IPv4 loopback address by default')
|
||||
def test_ipv4_address_standalone(context: IntegrationTestsContext) -> None:
|
||||
"""Test the HTTP-01 challenge with an IPv4 address using standalone authenticator.
|
||||
|
||||
While Pebble will offer both HTTP-01 and TLS-ALPN-01 challenges, we will
|
||||
only select HTTP-01 because TLS-ALPN-01 is not supported by the standalone
|
||||
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.
|
||||
"""
|
||||
context.certbot([
|
||||
'certonly', '--ip-address', context.local_ip, '--standalone',
|
||||
])
|
||||
assert_cert_count_for_lineage(context.config_dir, context.local_ip, 1)
|
||||
|
||||
|
||||
def test_ipv6_address_standalone(context: IntegrationTestsContext) -> None:
|
||||
"""Test the HTTP-01 challenge with an IPv6 address using standalone authenticator.
|
||||
|
||||
This test relies on some tricks. Pebble is configured to do validations on port 5002.
|
||||
Since multiple integration tests want to handle validation requests, and may run
|
||||
concurrently, proxy.py handles HTTP traffic for port 5002 and sends it to the appropriate
|
||||
integration test runner. However, it binds that port for IPv4 only. That is,
|
||||
GracefulTCPServer doesn't specify `address_family = AF_INET6` when subclassing
|
||||
socketserver.TCPServer.
|
||||
|
||||
For IPv4 integration tests (above), we simply assign each integration test runner a unique
|
||||
IP address under 127.0.0.0/8, and the proxy knows how to route those IP addresses. Pebble's
|
||||
validation connects to the proxy because all of 127.0.0.0/8 is defined to be loopback.
|
||||
|
||||
However, under IPv6 there is exactly one loopback address, so we can't use the same trick.
|
||||
Instead, we ensure that this is the only test that cares about IPv6, and bind [::1]:5002
|
||||
for IPv6 (via `--http-01-address`). When Pebble reaches out to validate `::1`, it reaches
|
||||
this test runner rather than the proxy.
|
||||
|
||||
Note: This works because this is the only test case that binds [::1]:5002. If additional
|
||||
IPv6 tests are added they could conflict. In that case we might try using the
|
||||
@pytest.mark.xdist_group annotation along with --dist loadgroup.
|
||||
https://pytest-xdist.readthedocs.io/en/stable/distribution.html
|
||||
"""
|
||||
context.certbot([
|
||||
'certonly', '--ip-address', '::1', '--standalone',
|
||||
'--http-01-address', '::1',
|
||||
'--http-01-port', str(constants.DEFAULT_HTTP_01_PORT),
|
||||
])
|
||||
assert_cert_count_for_lineage(context.config_dir, '::1', 1)
|
||||
|
||||
|
||||
def test_manual_http_auth(context: IntegrationTestsContext) -> None:
|
||||
"""Test the HTTP-01 challenge using manual plugin."""
|
||||
with misc.create_http_server(context.http_01_port) as webroot,\
|
||||
|
||||
@@ -115,6 +115,13 @@ class ACMEServer:
|
||||
acme_xdist['directory_url'] = PEBBLE_DIRECTORY_URL
|
||||
acme_xdist['challtestsrv_url'] = PEBBLE_CHALLTESTSRV_URL
|
||||
acme_xdist['http_port'] = dict(zip(nodes, range(5200, 5200 + len(nodes))))
|
||||
# When testing the standalone plugin with IP address certificates, we need a way for
|
||||
# the proxy to map incoming requests to workers. Since on Linux all 127.* addresses are
|
||||
# loopback, we give each worker a 127.0.0.n address. The proxy will route all requests for
|
||||
# that IP address to the http_port assigned for that worker.
|
||||
acme_xdist['local_ip'] = {}
|
||||
for i, node in enumerate(nodes):
|
||||
acme_xdist['local_ip'][node] = f"127.0.0.{i+2}"
|
||||
acme_xdist['https_port'] = dict(zip(nodes, range(5100, 5100 + len(nodes))))
|
||||
acme_xdist['other_port'] = dict(zip(nodes, range(5300, 5300 + len(nodes))))
|
||||
|
||||
@@ -161,6 +168,14 @@ class ACMEServer:
|
||||
http_port_map = cast(dict[str, int], self.acme_xdist['http_port'])
|
||||
mapping = {r'.+\.{0}\.wtf'.format(node): 'http://127.0.0.1:{0}'.format(port)
|
||||
for node, port in http_port_map.items()}
|
||||
|
||||
# Each worker gets a specific loopback IP address. All traffic to that IP address gets
|
||||
# routed to that worker's http_port.
|
||||
local_ip_map = cast(dict[str, str], self.acme_xdist['local_ip'])
|
||||
for node, local_ip in local_ip_map.items():
|
||||
mapping[r'{0}:.+'.format(local_ip)] = \
|
||||
'http://{0}:{1}'.format(local_ip, http_port_map[node])
|
||||
|
||||
command = [sys.executable, proxy.__file__, str(self._http_01_port), json.dumps(mapping)]
|
||||
self._launch_process(command)
|
||||
print('=> Finished configuring the HTTP proxy.')
|
||||
|
||||
@@ -12,6 +12,7 @@ import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import socket
|
||||
import socketserver
|
||||
import stat
|
||||
import sys
|
||||
@@ -74,6 +75,10 @@ class GracefulTCPServer(socketserver.TCPServer):
|
||||
just been released by another instance of TCPServer.
|
||||
"""
|
||||
allow_reuse_address = True
|
||||
# AF_INET is the default, but make it explicit. We don't want to conflict with
|
||||
# test_ipv6_address_standalone in the integration tests, which listens on [::1]:5002
|
||||
# (that is, AF_INET6).
|
||||
address_family = socket.AF_INET
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
|
||||
@@ -11,22 +11,30 @@ import requests
|
||||
|
||||
from certbot_integration_tests.utils.misc import GracefulTCPServer
|
||||
|
||||
|
||||
def _create_proxy(mapping: Mapping[str, str]) -> type[BaseHTTPServer.BaseHTTPRequestHandler]:
|
||||
# pylint: disable=missing-function-docstring
|
||||
class ProxyHandler(BaseHTTPServer.BaseHTTPRequestHandler):
|
||||
# pylint: disable=missing-class-docstring
|
||||
def do_GET(self) -> None:
|
||||
headers = {key.lower(): value for key, value in self.headers.items()}
|
||||
backend = [backend for pattern, backend in mapping.items()
|
||||
if re.match(pattern, headers['host'])][0]
|
||||
response = requests.get(backend + self.path, headers=headers, timeout=10)
|
||||
host = headers['host']
|
||||
for pattern, backend in mapping.items():
|
||||
if re.match(pattern, host):
|
||||
response = requests.get(backend + self.path, headers=headers, timeout=10)
|
||||
|
||||
self.send_response(response.status_code)
|
||||
for key, value in response.headers.items():
|
||||
self.send_header(key, value)
|
||||
self.send_response(response.status_code)
|
||||
for key, value in response.headers.items():
|
||||
self.send_header(key, value)
|
||||
self.end_headers()
|
||||
self.wfile.write(response.content)
|
||||
return
|
||||
|
||||
# We should never hit this if the tests are written correctly, but if we do, this may
|
||||
# be helpful debugging output.
|
||||
print(f"proxy.py: do_GET for {host}: No backend")
|
||||
self.send_response(200, "No backend")
|
||||
self.end_headers()
|
||||
self.wfile.write(response.content)
|
||||
self.wfile.write(bytes(f"No backend found for {host}\n", 'utf-8'))
|
||||
|
||||
return ProxyHandler
|
||||
|
||||
|
||||
@@ -124,8 +124,8 @@ class Proxy(interfaces.ConfiguratorProxy):
|
||||
def get_all_names(self) -> Iterable[str]:
|
||||
return self._configurator.get_all_names()
|
||||
|
||||
def get_chall_pref(self, domain: str) -> Iterable[type[Challenge]]:
|
||||
return self._configurator.get_chall_pref(domain)
|
||||
def get_chall_pref(self, identifier: str) -> Iterable[type[Challenge]]:
|
||||
return self._configurator.get_chall_pref(identifier)
|
||||
|
||||
@classmethod
|
||||
def inject_parser_options(cls, parser: argparse.ArgumentParser, name: str) -> None:
|
||||
|
||||
@@ -61,7 +61,7 @@ class Authenticator(common.Plugin, interfaces.Authenticator):
|
||||
def prepare(self) -> None:
|
||||
pass
|
||||
|
||||
def get_chall_pref(self, unused_domain: str) -> Iterable[type[challenges.Challenge]]:
|
||||
def get_chall_pref(self, unused_identifier: str) -> Iterable[type[challenges.Challenge]]:
|
||||
return [challenges.DNS01]
|
||||
|
||||
def perform(self, achalls: list[AnnotatedChallenge]) -> list[challenges.ChallengeResponse]:
|
||||
|
||||
@@ -1204,7 +1204,7 @@ class NginxConfigurator(common.Configurator):
|
||||
###########################################################################
|
||||
# Challenges Section for Authenticator
|
||||
###########################################################################
|
||||
def get_chall_pref(self, unused_domain: str) -> list[type[challenges.Challenge]]:
|
||||
def get_chall_pref(self, unused_identifier: str) -> list[type[challenges.Challenge]]:
|
||||
"""Return list of challenge preferences."""
|
||||
return [challenges.HTTP01]
|
||||
|
||||
|
||||
@@ -269,7 +269,7 @@ challenges: HTTP and DNS, represented by classes in `acme.challenges`.
|
||||
An authenticator plugin should implement support for at least one challenge type.
|
||||
|
||||
An Authenticator indicates which challenges it supports by implementing
|
||||
`get_chall_pref(domain)` to return a sorted list of challenge types in
|
||||
`get_chall_pref(identifier)` to return a sorted list of challenge types in
|
||||
preference order.
|
||||
|
||||
An Authenticator must also implement `perform(achalls)`, which "performs" a list
|
||||
|
||||
@@ -237,15 +237,19 @@ class AuthHandler:
|
||||
|
||||
return achalls
|
||||
|
||||
def _get_chall_pref(self, domain: str) -> list[type[challenges.Challenge]]:
|
||||
def _get_chall_pref(self, identifier: str) -> list[type[challenges.Challenge]]:
|
||||
"""Return list of challenge preferences.
|
||||
|
||||
:param str domain: domain for which you are requesting preferences
|
||||
|
||||
"""
|
||||
chall_prefs = []
|
||||
# Make sure to make a copy...
|
||||
plugin_pref = self.auth.get_chall_pref(domain)
|
||||
# The 'identifier' parameter of `get_chall_pref` used to be called `domain`.
|
||||
# There may be plugins in the wild that name their parameter `domain`. To keep
|
||||
# working with those plugins, make sure to continue to pass `identifier` as a
|
||||
# positional parameter rather than a kwarg.
|
||||
# Also, make sure to make a copy...
|
||||
plugin_pref = self.auth.get_chall_pref(identifier)
|
||||
if self.pref_challs:
|
||||
plugin_pref_types = {chall.typ for chall in plugin_pref}
|
||||
for typ in self.pref_challs:
|
||||
|
||||
@@ -252,7 +252,7 @@ def human_readable_cert_info(config: configuration.NamespaceConfig, cert: storag
|
||||
certinfo = []
|
||||
checker = ocsp.RevocationChecker()
|
||||
|
||||
config_sans = set(config.domains)
|
||||
config_sans = set(config.domains + config.ip_addresses)
|
||||
|
||||
if config.certname and cert.lineagename != config.certname and not skip_filter_checks:
|
||||
return None
|
||||
@@ -358,8 +358,10 @@ def _describe_certs(config: configuration.NamespaceConfig,
|
||||
notify("No certificates found.")
|
||||
else:
|
||||
if parsed_certs:
|
||||
match = "matching " if config.certname or config.domains else ""
|
||||
notify("Found the following {0}certs:".format(match))
|
||||
if config.certname or config.domains or config.ip_addresses:
|
||||
notify("Found the following matching certs:")
|
||||
else:
|
||||
notify("Found the following certs:")
|
||||
notify(_report_human_readable(config, parsed_certs))
|
||||
if parse_failures:
|
||||
notify("\nThe following renewal configurations "
|
||||
|
||||
@@ -32,6 +32,7 @@ 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 flag_default
|
||||
from certbot._internal.cli.cli_utils import HelpfulArgumentGroup
|
||||
from certbot._internal.cli.cli_utils import nonnegative_int
|
||||
@@ -123,6 +124,14 @@ def prepare_and_parse_args(plugins: plugins_disco.PluginsRegistry, args: list[st
|
||||
"will be used as the certificate name, unless otherwise specified or if you "
|
||||
"already have a certificate with the same name. In the case of a name conflict, "
|
||||
"a number like -0001 will be appended to the certificate name. (default: Ask)")
|
||||
helpful.add(
|
||||
[None, "certonly", "certificates"],
|
||||
"--ip-address", dest="ip_addresses",
|
||||
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 "
|
||||
"on the certificate.")
|
||||
helpful.add(
|
||||
[None, "run", "certonly", "register"],
|
||||
"--eab-kid", dest="eab_kid",
|
||||
|
||||
@@ -96,10 +96,15 @@ class DomainsAction(argparse.Action):
|
||||
"""Action class for parsing domains."""
|
||||
|
||||
def __call__(self, parser: argparse.ArgumentParser, namespace: argparse.Namespace,
|
||||
domain: Union[str, Sequence[Any], None],
|
||||
option_string: Optional[str] = None) -> None:
|
||||
values: str | Sequence[Any] | None,
|
||||
option_string: str | None = None) -> None:
|
||||
"""Just wrap add_domains in argparseese."""
|
||||
add_domains(namespace, str(domain) if domain is not None else None)
|
||||
match values:
|
||||
case str():
|
||||
add_domains(namespace, str(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_domains(args_or_config: Union[argparse.Namespace, configuration.NamespaceConfig],
|
||||
@@ -131,6 +136,21 @@ def add_domains(args_or_config: Union[argparse.Namespace, configuration.Namespac
|
||||
return validated_domains
|
||||
|
||||
|
||||
class _IPAddressAction(argparse.Action):
|
||||
"""Action class for parsing IP addresses."""
|
||||
|
||||
def __call__(self, parser: argparse.ArgumentParser, namespace: argparse.Namespace,
|
||||
values: str | Sequence[Any] | None,
|
||||
option_string: Optional[str] = None) -> None:
|
||||
match values:
|
||||
case str():
|
||||
# This will throw an exception if the IP address doesn't parse.
|
||||
namespace.ip_addresses.append(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")
|
||||
|
||||
|
||||
class CaseInsensitiveList(list):
|
||||
"""A list that will ignore case when searching.
|
||||
|
||||
|
||||
@@ -332,26 +332,26 @@ class HelpfulArgumentParser:
|
||||
csrfile, contents = config.csr[0:2]
|
||||
typ, util_csr, _ = crypto_util.import_csr_file(csrfile, contents)
|
||||
x509_req = x509.load_pem_x509_csr(util_csr.data)
|
||||
domains, _ = san.from_x509(x509_req.subject, x509_req.extensions)
|
||||
domains, ip_addresses = san.from_x509(x509_req.subject, x509_req.extensions)
|
||||
|
||||
# The SANs from the CSR are added to the domains from command line flags as this config
|
||||
# setting is where main.certonly gets the list of identifiers to request.
|
||||
config.domains.extend(domains)
|
||||
|
||||
if not domains:
|
||||
if not domains + ip_addresses:
|
||||
# TODO: add CN to domains instead:
|
||||
raise errors.Error(
|
||||
"Unfortunately, your CSR %s needs to have a SubjectAltName for every domain"
|
||||
% config.csr[0])
|
||||
f"Unfortunately, your CSR {config.csr[0]} needs to have a SubjectAltName for " +
|
||||
"every domain or IP address")
|
||||
|
||||
config.actual_csr = (util_csr, typ)
|
||||
|
||||
# Check that the original values for --domain set by the user were
|
||||
# Check that the original values for --domain and --ip-address set by the user were
|
||||
# a subset of the domains listed in the CSR.
|
||||
if set(config.domains) != set(domains):
|
||||
raise errors.ConfigurationError(
|
||||
"Inconsistent requests:\nFrom the CSR: {0}\nFrom command line/config: {1}"
|
||||
.format(san.display(domains),
|
||||
.format(san.display(san.join(domains, ip_addresses)),
|
||||
san.display(config.domains)))
|
||||
|
||||
|
||||
|
||||
@@ -399,8 +399,9 @@ class Client:
|
||||
elif self.config.rsa_key_size and self.config.key_type.lower() == 'rsa':
|
||||
key_size = self.config.rsa_key_size
|
||||
|
||||
domains, _ = san.split(sans)
|
||||
domains, ip_addresses = san.split(sans)
|
||||
domains_str = [d.dns_name for d in domains]
|
||||
ip_addresses_typed = [i.ip_address for i in ip_addresses]
|
||||
|
||||
# Create CSR from names
|
||||
if self.config.dry_run:
|
||||
@@ -416,7 +417,7 @@ class Client:
|
||||
csr = util.CSR(file=None, form="pem",
|
||||
data=acme_crypto_util.make_csr(
|
||||
key.pem, domains_str, self.config.must_staple,
|
||||
))
|
||||
ipaddrs=ip_addresses_typed))
|
||||
else:
|
||||
key = key or crypto_util.generate_key(
|
||||
key_size=key_size,
|
||||
@@ -426,7 +427,8 @@ class Client:
|
||||
strict_permissions=self.config.strict_permissions,
|
||||
)
|
||||
csr = crypto_util.generate_csr(
|
||||
key, domains_str, None, self.config.must_staple, self.config.strict_permissions)
|
||||
key, domains_str, None, self.config.must_staple, self.config.strict_permissions,
|
||||
ipaddrs=ip_addresses_typed)
|
||||
|
||||
try:
|
||||
orderr = self._get_order_and_authorizations(csr.data, self.config.allow_subset_of_names)
|
||||
|
||||
@@ -32,6 +32,7 @@ CLI_DEFAULTS: dict[str, Any] = dict( # pylint: disable=use-dict-literal
|
||||
noninteractive_mode=False,
|
||||
force_interactive=False,
|
||||
domains=[],
|
||||
ip_addresses=[],
|
||||
certname=None,
|
||||
dry_run=False,
|
||||
register_unsafely_without_email=False,
|
||||
|
||||
@@ -500,12 +500,17 @@ def _find_sans_or_certname(config: configuration.NamespaceConfig,
|
||||
:raises errors.Error: Usage message, if parameters are not used correctly
|
||||
|
||||
"""
|
||||
domains: list[san.SAN] = []
|
||||
ip_addresses: list[san.SAN] = []
|
||||
certname = config.certname
|
||||
sans: Optional[list[san.SAN]] = None
|
||||
|
||||
# first, try to get domains from the config
|
||||
if config.domains:
|
||||
sans = config.domains
|
||||
domains = config.domains
|
||||
if config.ip_addresses:
|
||||
ip_addresses = config.ip_addresses
|
||||
|
||||
sans: Optional[list[san.SAN]] = domains + ip_addresses
|
||||
|
||||
# if we can't do that but we have a certname, get the sans
|
||||
# by loading the latest certificate with that certname
|
||||
@@ -518,7 +523,7 @@ def _find_sans_or_certname(config: configuration.NamespaceConfig,
|
||||
sans = san.guess(display_ops.choose_names(installer, question))
|
||||
|
||||
if not sans:
|
||||
raise errors.Error("Please specify --domains, or --installer that "
|
||||
raise errors.Error("Please specify --domains, --ip-address, or --installer that "
|
||||
"will help in domain names autodiscovery, or "
|
||||
"--cert-name for an existing certificate name.")
|
||||
|
||||
|
||||
@@ -163,7 +163,7 @@ permitted by DNS standards.)
|
||||
)
|
||||
)
|
||||
|
||||
def get_chall_pref(self, domain: str) -> Iterable[type[challenges.Challenge]]:
|
||||
def get_chall_pref(self, identifier: str) -> Iterable[type[challenges.Challenge]]:
|
||||
# pylint: disable=unused-argument,missing-function-docstring
|
||||
return [challenges.HTTP01, challenges.DNS01]
|
||||
|
||||
|
||||
@@ -133,7 +133,7 @@ running. HTTP challenge only (wildcards not supported)."""
|
||||
def prepare(self) -> None: # pylint: disable=missing-function-docstring
|
||||
pass
|
||||
|
||||
def get_chall_pref(self, domain: str) -> Iterable[type[challenges.Challenge]]:
|
||||
def get_chall_pref(self, identifier: str) -> Iterable[type[challenges.Challenge]]:
|
||||
# pylint: disable=unused-argument,missing-function-docstring
|
||||
return [challenges.HTTP01]
|
||||
|
||||
|
||||
@@ -89,7 +89,7 @@ to serve all files under specified web root ({0})."""
|
||||
"the provided --webroot-path/-w and that files created there can be downloaded "
|
||||
"from the internet.")
|
||||
|
||||
def get_chall_pref(self, domain: str) -> Iterable[type[challenges.Challenge]]:
|
||||
def get_chall_pref(self, identifier: str) -> Iterable[type[challenges.Challenge]]:
|
||||
# pylint: disable=unused-argument,missing-function-docstring
|
||||
return [challenges.HTTP01]
|
||||
|
||||
|
||||
@@ -148,8 +148,9 @@ def reconstitute(config: configuration.NamespaceConfig,
|
||||
return None
|
||||
|
||||
try:
|
||||
domains, _ = san.split(renewal_candidate.sans())
|
||||
domains, ip_addresses = san.split(renewal_candidate.sans())
|
||||
config.domains = domains
|
||||
config.ip_addresses = ip_addresses
|
||||
except errors.ConfigurationError as error:
|
||||
logger.error("Renewal configuration file %s references a certificate "
|
||||
"that contains an invalid domain name. The problem "
|
||||
|
||||
@@ -225,6 +225,7 @@ class CertificatesTest(BaseCertManagerTest):
|
||||
|
||||
mock_config = mock.MagicMock(certname=None, lineagename=None)
|
||||
mock_config.domains = []
|
||||
mock_config.ip_addresses = []
|
||||
# pylint: disable=protected-access
|
||||
|
||||
# pylint: disable=protected-access
|
||||
|
||||
@@ -380,7 +380,7 @@ class ClientTest(ClientTestCommon):
|
||||
strict_permissions=True,
|
||||
)
|
||||
mock_crypto_util.generate_csr.assert_called_once_with(
|
||||
mock.sentinel.key, self.eg_domains, None, False, True)
|
||||
mock.sentinel.key, self.eg_domains, None, False, True, ipaddrs=[])
|
||||
mock_crypto_util.cert_and_chain_from_fullchain.assert_called_once_with(
|
||||
self.eg_order.fullchain_pem)
|
||||
|
||||
@@ -518,8 +518,8 @@ class ClientTest(ClientTestCommon):
|
||||
assert mock_crypto_util.generate_key.call_count == 2
|
||||
successful_domains = [d for d in self.eg_domains if d != 'example.com']
|
||||
mock_crypto_util.generate_csr.assert_has_calls([
|
||||
mock.call(key, self.eg_domains, None, self.config.must_staple, self.config.strict_permissions),
|
||||
mock.call(key, successful_domains, None, self.config.must_staple, self.config.strict_permissions)])
|
||||
mock.call(key, self.eg_domains, None, self.config.must_staple, self.config.strict_permissions, ipaddrs=[]),
|
||||
mock.call(key, successful_domains, None, self.config.must_staple, self.config.strict_permissions, ipaddrs=[])])
|
||||
assert mock_crypto_util.cert_and_chain_from_fullchain.call_count == 1
|
||||
|
||||
@mock.patch("certbot._internal.client.crypto_util")
|
||||
@@ -610,8 +610,8 @@ class ClientTest(ClientTestCommon):
|
||||
successful_domains = [d for d in self.eg_domains if d != 'example.com']
|
||||
assert mock_crypto_util.generate_key.call_count == 2
|
||||
mock_crypto_util.generate_csr.assert_has_calls([
|
||||
mock.call(key, self.eg_domains, None, self.config.must_staple, self.config.strict_permissions),
|
||||
mock.call(key, successful_domains, None, self.config.must_staple, self.config.strict_permissions)])
|
||||
mock.call(key, self.eg_domains, None, self.config.must_staple, self.config.strict_permissions, ipaddrs=[]),
|
||||
mock.call(key, successful_domains, None, self.config.must_staple, self.config.strict_permissions, ipaddrs=[])])
|
||||
assert mock_crypto_util.cert_and_chain_from_fullchain.call_count == 1
|
||||
|
||||
@mock.patch("certbot._internal.client.crypto_util")
|
||||
@@ -690,7 +690,7 @@ class ClientTest(ClientTestCommon):
|
||||
# Assumes all of eg_sans are DNSNames.
|
||||
eg_domains = list(map(str, self.eg_sans))
|
||||
mock_acme_crypto.make_csr.assert_called_once_with(
|
||||
mock.sentinel.key_pem, eg_domains, self.config.must_staple)
|
||||
mock.sentinel.key_pem, eg_domains, self.config.must_staple, ipaddrs=[])
|
||||
mock_crypto.generate_key.assert_not_called()
|
||||
mock_crypto.generate_csr.assert_not_called()
|
||||
assert mock_crypto.cert_and_chain_from_fullchain.call_count == 1
|
||||
|
||||
@@ -372,7 +372,7 @@ class FindSansOrCertnameTest(unittest.TestCase):
|
||||
|
||||
@mock.patch('certbot.display.ops.choose_names')
|
||||
def test_display_ops(self, mock_choose_names):
|
||||
mock_config = mock.Mock(domains=None, certname=None)
|
||||
mock_config = mock.Mock(domains=None, ip_addresses=None, certname=None)
|
||||
mock_choose_names.return_value = ["example.com"]
|
||||
# pylint: disable=protected-access
|
||||
assert main._find_sans_or_certname(mock_config, None) == ([san.DNSName("example.com")], None)
|
||||
|
||||
@@ -86,7 +86,7 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
assert isinstance(self.auth.more_info(), str)
|
||||
|
||||
def test_get_chall_pref(self):
|
||||
assert self.auth.get_chall_pref(domain=None) == \
|
||||
assert self.auth.get_chall_pref(identifier=None) == \
|
||||
[challenges.HTTP01]
|
||||
|
||||
def test_perform(self):
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
"""
|
||||
import datetime
|
||||
import hashlib
|
||||
import ipaddress
|
||||
import logging
|
||||
import re
|
||||
from typing import Optional
|
||||
@@ -98,7 +99,9 @@ def generate_key(key_size: int, key_dir: Optional[str], key_type: str = "rsa",
|
||||
|
||||
|
||||
def generate_csr(privkey: util.Key, names: Union[list[str], set[str]], path: Optional[str],
|
||||
must_staple: bool = False, strict_permissions: bool = True) -> util.CSR:
|
||||
must_staple: bool = False, strict_permissions: bool = True,
|
||||
ipaddrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] | None = None,
|
||||
) -> util.CSR:
|
||||
"""Initialize a CSR with the given private key.
|
||||
|
||||
:param privkey: Key to include in the CSR
|
||||
@@ -114,7 +117,7 @@ def generate_csr(privkey: util.Key, names: Union[list[str], set[str]], path: Opt
|
||||
|
||||
"""
|
||||
csr_pem = acme_crypto_util.make_csr(
|
||||
privkey.pem, names, must_staple=must_staple)
|
||||
privkey.pem, names, must_staple=must_staple, ipaddrs=ipaddrs)
|
||||
|
||||
# Save CSR, if requested
|
||||
csr_filename = None
|
||||
|
||||
@@ -162,10 +162,10 @@ class Authenticator(Plugin):
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get_chall_pref(self, domain: str) -> Iterable[type[Challenge]]:
|
||||
def get_chall_pref(self, identifier: str) -> Iterable[type[Challenge]]:
|
||||
"""Return `collections.Iterable` of challenge preferences.
|
||||
|
||||
:param str domain: Domain for which challenge preferences are sought.
|
||||
:param str identifier: Domain or IP address for which challenge preferences are sought.
|
||||
|
||||
:returns: `collections.Iterable` of challenge types (subclasses of
|
||||
:class:`acme.challenges.Challenge`) with the most
|
||||
|
||||
@@ -54,7 +54,7 @@ class DNSAuthenticator(common.Plugin, interfaces.Authenticator, metaclass=abc.AB
|
||||
.format(name=self.name, secs=delay, suffix='s' if delay != 1 else '')
|
||||
)
|
||||
|
||||
def get_chall_pref(self, unused_domain: str) -> Iterable[type[challenges.Challenge]]: # pylint: disable=missing-function-docstring
|
||||
def get_chall_pref(self, unused_identifier: str) -> Iterable[type[challenges.Challenge]]: # pylint: disable=missing-function-docstring
|
||||
return [challenges.DNS01]
|
||||
|
||||
def prepare(self) -> None: # pylint: disable=missing-function-docstring
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
A new command line flag, --ip-address, has been added. This requests certificates with IP address SANs when using the standalone plugin. Note that for Let's Encrypt's implementation of IP address certificates, you'll also need to pass `--preferred-profile shortlived`.
|
||||
@@ -0,0 +1 @@
|
||||
Authenticator.get_chall_pref's argument has been renamed from `domain` to `identifier`, and can now receive string-formatted IP addresses in addition to domain names.
|
||||
Reference in New Issue
Block a user