Add identifier field to AnnotatedChallenge subclasses (#10491)

This field is optional to maintain backwards compatibility. Note that
`AnnotatedChallenge` inherits from `jose.ImmutableMap`, which has a
[check in
__init__](https://github.com/certbot/josepy/blob/4b747476703fe4fff1aaccd76ebe570698bbf4f0/src/josepy/util.py#L125-L131)
that all slots are provided. That check would not allow us to do a
backwards-compatible addition, so I implemented an `__init__` for each
of these subclasses that fills the fields without calling the parent
`__init__`, and so doesn't hit an error when `identifier` is absent.

I chose to use `acme.messages.Identifier` rather than
`certbot._internal.san.SAN` here because these are wrapped ACME types,
so they should use the ACME representation. Also, `AnnotatedChallenge`
is passed to plugins, so we need to pass a type that the plugins can
understand.

Additionally, `domain` is marked as deprecated.

Part of #10346

/cc @bmw, who noticed the issue with `AnnotatedChallenge`
[here](https://github.com/certbot/certbot/pull/10468#issuecomment-3403294394)
and provided additional feedback
[here](https://github.com/jsha/certbot/pull/2#issuecomment-3534895793).
Note that there's still some work to do to finish excising `domain`
assumptions from this portion of the code.

---------

Co-authored-by: ohemorange <ebportnoy@gmail.com>
This commit is contained in:
Jacob Hoffman-Andrews
2025-12-05 13:44:04 -08:00
committed by GitHub
co-authored by ohemorange
parent 9e7a98f4cd
commit b1cf53ff6b
25 changed files with 218 additions and 110 deletions
@@ -4,6 +4,7 @@ import logging
from typing import TYPE_CHECKING
from acme.challenges import KeyAuthorizationChallengeResponse
from acme import messages
from certbot import errors
from certbot.achallenges import KeyAuthorizationAnnotatedChallenge
from certbot.compat import filesystem
@@ -52,6 +53,9 @@ class ApacheHttp01(common.ChallengePerformer):
"""Perform all HTTP-01 challenges."""
if not self.achalls:
return []
if any(achall.identifier.typ == messages.IDENTIFIER_IP for achall in self.achalls):
raise errors.ConfigurationError(
"Apache authenticator not supported for IP address certificates")
# Save any changes to the configuration as a precaution
# About to make temporary changes to the config
self.configurator.save("Changes before challenge setup", True)
@@ -82,7 +86,7 @@ class ApacheHttp01(common.ChallengePerformer):
# Search for VirtualHosts matching by name
for chall in self.achalls:
selected_vhosts += self._matching_vhosts(chall.domain)
selected_vhosts += self._matching_vhosts(chall.identifier.value)
# Ensure that we have one or more VirtualHosts that we can continue
# with. (one that listens to port configured with --http-01-port)
@@ -9,7 +9,7 @@ from unittest import mock
import pytest
from acme import challenges
from acme import challenges, messages
from certbot import achallenges
from certbot import crypto_util
from certbot import errors
@@ -1193,17 +1193,20 @@ class MultipleVhostsTest(util.ApacheTest):
challenges.HTTP01(
token=b"jIq_Xy1mXGN37tb4L6Xj_es58fW571ZNyXekdZzhh7Q"),
"pending"),
domain="encryption-example.demo", account_key=account_key)
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="encryption-example.demo"),
account_key=account_key)
achall2 = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(
token=b"uqnaPzxtrndteOqtrXb0Asl5gOJfWAnnx6QJyvcmlDU"),
"pending"),
domain="certbot.demo", account_key=account_key)
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="certbot.demo"),
account_key=account_key)
achall3 = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=(b'x' * 16)), "pending"),
domain="example.org", account_key=account_key)
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="example.org"),
account_key=account_key)
return account_key, (achall1, achall2, achall3)
@@ -5,7 +5,7 @@ from unittest import mock
import pytest
from acme import challenges
from acme import challenges, messages
from certbot import achallenges
from certbot import errors
from certbot.compat import filesystem
@@ -37,7 +37,8 @@ class ApacheHttp01Test(util.ApacheTest):
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=((chr(ord('a') + i).encode() * 16))),
"pending"),
domain=self.vhosts[i].name, account_key=self.account_key))
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value=self.vhosts[i].name),
account_key=self.account_key))
modules = ["ssl", "rewrite", "authz_core", "authz_host"]
for mod in modules:
@@ -50,6 +51,16 @@ class ApacheHttp01Test(util.ApacheTest):
def test_empty_perform(self):
assert len(self.http.perform()) == 0
def test_ip_address_perform(self):
self.http.achalls = [achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=((b'a' * 16))),
"pending"),
identifier=messages.Identifier(typ=messages.IDENTIFIER_IP, value="127.0.0.1"),
account_key=self.account_key)]
with pytest.raises(errors.ConfigurationError):
self.http.perform()
@mock.patch("certbot_apache._internal.configurator.ApacheConfigurator.enable_mod")
def test_enable_modules_apache_2_4(self, mock_enmod):
del self.config.parser.modules["authz_core_module"]
@@ -83,12 +94,14 @@ class ApacheHttp01Test(util.ApacheTest):
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=((b'a' * 16))),
"pending"),
domain=vhost.name, account_key=self.account_key),
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value=vhost.name),
account_key=self.account_key),
achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=((b'b' * 16))),
"pending"),
domain=next(iter(vhost.aliases)), account_key=self.account_key)
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value=next(iter(vhost.aliases))),
account_key=self.account_key)
]
self.common_perform_test(achalls, [vhost])
@@ -99,7 +112,8 @@ class ApacheHttp01Test(util.ApacheTest):
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=((b'a' * 16))),
"pending"),
domain="something.nonexistent", account_key=self.account_key)]
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="something.nonexistent"),
account_key=self.account_key)]
self.common_perform_test(achalls, vhosts)
def test_configure_multiple_vhosts(self):
@@ -110,7 +124,8 @@ class ApacheHttp01Test(util.ApacheTest):
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=((b'a' * 16))),
"pending"),
domain="duplicate.example.com", account_key=self.account_key)]
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="duplicate.example.com"),
account_key=self.account_key)]
self.common_perform_test(achalls, vhosts)
def test_configure_name_and_blank(self):
@@ -121,7 +136,8 @@ class ApacheHttp01Test(util.ApacheTest):
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=((b'a' * 16))),
"pending"),
domain=domain, account_key=self.account_key),
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value=domain),
account_key=self.account_key),
]
self.common_perform_test(achalls, vhosts)
@@ -148,7 +164,8 @@ class ApacheHttp01Test(util.ApacheTest):
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=((b'a' * 16))),
"pending"),
domain="certbot.demo", account_key=self.account_key)]
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="certbot.demo"),
account_key=self.account_key)]
vhosts[0].enabled = False
self.common_perform_test(achalls, vhosts)
matches = self.config.parser.find_dir(
@@ -65,22 +65,22 @@ def test_authenticator(plugin: common.Proxy, config: str, temp_dir: str) -> bool
if not response:
logger.error(
"Plugin failed to complete %s for %s in %s",
type(achall), achall.domain, config)
type(achall), achall.identifier.value, config)
success = False
elif isinstance(response, challenges.HTTP01Response):
# We fake the DNS resolution to ensure that any domain is resolved
# to the local HTTP server setup for the compatibility tests
with _fake_dns_resolution("127.0.0.1"):
verified = response.simple_verify(
achall.chall, achall.domain,
achall.chall, achall.identifier.value,
util.JWK.public_key(), port=plugin.http_port)
if verified:
logger.info(
"http-01 verification for %s succeeded", achall.domain)
"http-01 verification for %s succeeded", achall.identifier.value)
else:
logger.error(
"**** http-01 verification for %s in %s failed",
achall.domain, config)
achall.identifier.value, config)
success = False
if success:
@@ -70,7 +70,7 @@ class Authenticator(common.Plugin, interfaces.Authenticator):
try:
change_ids = [
self._change_txt_record("UPSERT",
achall.validation_domain_name(achall.domain),
achall.validation_domain_name(achall.identifier.value),
achall.validation(achall.account_key))
for achall in achalls
]
@@ -85,7 +85,7 @@ class Authenticator(common.Plugin, interfaces.Authenticator):
def cleanup(self, achalls: list[achallenges.AnnotatedChallenge]) -> None:
if self._attempt_cleanup:
for achall in achalls:
domain = achall.domain
domain = achall.identifier.value
validation_domain_name = achall.validation_domain_name(domain)
validation = achall.validation(achall.account_key)
@@ -9,7 +9,7 @@ from botocore.exceptions import NoCredentialsError
import josepy as jose
import pytest
from acme import challenges
from acme import challenges, messages
from certbot import achallenges
from certbot import errors
from certbot.compat import os
@@ -24,7 +24,9 @@ class AuthenticatorTest(unittest.TestCase):
# pylint: disable=protected-access
achall = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.DNS01, domain=DOMAIN, account_key=KEY)
challb=acme_util.DNS01,
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value=DOMAIN),
account_key=KEY)
def setUp(self):
from certbot_dns_route53._internal.dns_route53 import Authenticator
@@ -5,7 +5,7 @@ from typing import Any
from typing import Optional
from typing import TYPE_CHECKING
from acme import challenges
from acme import challenges, messages
from acme.challenges import KeyAuthorizationChallengeResponse
from certbot import errors
from certbot.achallenges import KeyAuthorizationAnnotatedChallenge
@@ -55,6 +55,9 @@ class NginxHttp01(common.ChallengePerformer):
"""
if not self.achalls:
return []
if any(achall.identifier.typ == messages.IDENTIFIER_IP for achall in self.achalls):
raise errors.ConfigurationError(
"nginx authenticator not supported for IP address certificates")
responses = [x.response(x.account_key) for x in self.achalls]
@@ -190,7 +193,7 @@ class NginxHttp01(common.ChallengePerformer):
document_root = os.path.join(
self.configurator.config.work_dir, "http_01_nonexistent")
block.extend([['server_name', ' ', achall.domain],
block.extend([['server_name', ' ', achall.identifier.value],
['root', ' ', document_root],
self._location_directive_for_achall(achall)
])
@@ -219,7 +222,7 @@ class NginxHttp01(common.ChallengePerformer):
:rtype: list
"""
http_vhosts, https_vhosts = self.configurator.choose_auth_vhosts(achall.domain)
http_vhosts, https_vhosts = self.configurator.choose_auth_vhosts(achall.identifier.value)
new_vhost: Optional[list[Any]] = None
if not http_vhosts:
@@ -375,7 +375,9 @@ class NginxConfiguratorTest(util.NginxTest):
chall=challenges.HTTP01(token=b"m8TdO1qik4JVFtgPPurJmg"),
uri="https://ca.org/chall1_uri",
status=messages.Status("pending"),
), domain="example.com", account_key=self.rsa512jwk)
),
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="example.com"),
account_key=self.rsa512jwk)
expected = [
achall.response(self.rsa512jwk),
@@ -5,8 +5,7 @@ from unittest import mock
import josepy as jose
import pytest
from acme import challenges
from acme import messages
from acme import challenges, messages
from certbot import achallenges
from certbot.tests import acme_util
from certbot.tests import util as test_util
@@ -24,29 +23,34 @@ class HttpPerformTest(util.NginxTest):
achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=b"kNdwjwOeX0I_A8DXt9Msmg"), messages.STATUS_PENDING),
domain="www.example.com", account_key=account_key),
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="www.example.com"),
account_key=account_key),
achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(
token=b"\xba\xa9\xda?<m\xaewmx\xea\xad\xadv\xf4\x02\xc9y"
b"\x80\xe2_X\t\xe7\xc7\xa4\t\xca\xf7&\x945"
), messages.STATUS_PENDING),
domain="ipv6.com", account_key=account_key),
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="ipv6.com"),
account_key=account_key),
achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(
token=b"\x8c\x8a\xbf_-f\\cw\xee\xd6\xf8/\xa5\xe3\xfd"
b"\xeb9\xf1\xf5\xb9\xefVM\xc9w\xa4u\x9c\xe1\x87\xb4"
), messages.STATUS_PENDING),
domain="www.example.org", account_key=account_key),
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="www.example.org"),
account_key=account_key),
achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=b"kNdwjxOeX0I_A8DXt9Msmg"), messages.STATUS_PENDING),
domain="migration.com", account_key=account_key),
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="migration.com"),
account_key=account_key),
achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=b"kNdwjxOeX0I_A8DXt9Msmg"), messages.STATUS_PENDING),
domain="ipv6ssl.com", account_key=account_key),
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="ipv6ssl.com"),
account_key=account_key),
]
def setUp(self):
@@ -138,7 +142,8 @@ class HttpPerformTest(util.NginxTest):
achall = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=b"kNdwjxOeX0I_A8DXt9Msmg"), messages.STATUS_PENDING),
domain="ssl.both.com", account_key=AUTH_KEY)
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="ssl.both.com"),
account_key=AUTH_KEY)
self.http01.add_chall(achall)
self.http01._mod_config() # pylint: disable=protected-access
+11 -12
View File
@@ -290,8 +290,7 @@ class AuthHandler:
for index in path:
challb = authzr.body.challenges[index]
achalls.append(challb_to_achall(
challb, self.account.key, authzr.body.identifier.value))
achalls.append(challb_to_achall(challb, self.account.key, authzr.body.identifier))
return achalls
@@ -300,7 +299,7 @@ class AuthHandler:
if not self.account:
raise errors.Error("Account is not set.")
problems: dict[str, list[achallenges.AnnotatedChallenge]] = {}
failed_achalls = [challb_to_achall(challb, self.account.key, authzr.body.identifier.value)
failed_achalls = [challb_to_achall(challb, self.account.key, authzr.body.identifier)
for authzr in failed_authzrs for challb in authzr.body.challenges
if challb.error]
@@ -338,11 +337,11 @@ class AuthHandler:
dns01_achalls = {}
for achall in achalls:
if isinstance(achall.chall, challenges.HTTP01):
http01_achalls[achall.chall.uri(achall.domain)] = (
http01_achalls[achall.chall.uri(achall.identifier.value)] = (
achall.validation(achall.account_key) + "\n"
)
if isinstance(achall.chall, challenges.DNS01):
dns01_achalls[achall.validation_domain_name(achall.domain)] = (
dns01_achalls[achall.validation_domain_name(achall.identifier.value)] = (
achall.validation(achall.account_key) + "\n"
)
if http01_achalls:
@@ -361,7 +360,7 @@ class AuthHandler:
def challb_to_achall(challb: messages.ChallengeBody, account_key: josepy.JWK,
domain: str) -> achallenges.AnnotatedChallenge:
identifier: messages.Identifier) -> achallenges.AnnotatedChallenge:
"""Converts a ChallengeBody object to an AnnotatedChallenge.
:param .ChallengeBody challb: ChallengeBody
@@ -373,15 +372,15 @@ def challb_to_achall(challb: messages.ChallengeBody, account_key: josepy.JWK,
"""
chall = challb.chall
logger.info("%s challenge for %s", chall.typ, domain)
logger.info("%s challenge for %s", chall.typ, identifier)
if isinstance(chall, challenges.KeyAuthorizationChallenge):
return achallenges.KeyAuthorizationAnnotatedChallenge(
challb=challb, domain=domain, account_key=account_key)
challb=challb, account_key=account_key, identifier=identifier)
elif isinstance(chall, challenges.DNS):
return achallenges.DNS(challb=challb, domain=domain)
return achallenges.DNS(challb=challb, identifier=identifier)
else:
return achallenges.Other(challb=challb, domain=domain)
return achallenges.Other(challb=challb, identifier=identifier)
def gen_challenge_path(challbs: list[messages.ChallengeBody],
@@ -473,7 +472,7 @@ def _generate_failed_chall_msg(failed_achalls: list[achallenges.AnnotatedChallen
msg = []
for achall in failed_achalls:
msg.append("\n Domain: %s\n Type: %s\n Detail: %s\n" % (
achall.domain, typ, achall.error.detail))
msg.append("\n Identifier: %s\n Type: %s\n Detail: %s\n" % (
achall.identifier.value, typ, achall.error.detail))
return "".join(msg)
+6 -1
View File
@@ -1064,6 +1064,9 @@ def _install_cert(config: configuration.NamespaceConfig, le_client: client.Clien
domains, ip_addresses = san.split(sans)
if len(ip_addresses) > 0:
# Our apache and nginx plugins are currently relying on this check for a user friendly error
# message about their lack of support for IP certificates. If you're removing this check,
# please check that the plugins can process IP addresses.
raise errors.ConfigurationError("Enhancements not supported for IP address certificates")
le_client.deploy_certificate(domains, path_provider.key_path, path_provider.cert_path,
@@ -1242,9 +1245,11 @@ def enhance(config: configuration.NamespaceConfig,
cert_sans = cert_manager.sans_for_certname(config, config.certname)
if cert_sans is None:
raise errors.Error("Could not find the list of domains for the given certificate name.")
cert_domains, ip_addresses = san.split(cert_sans)
if len(ip_addresses) > 0:
# Our apache and nginx plugins are currently relying on this check for a user friendly error
# message about their lack of support for IP certificates. If you're removing this check,
# please check that the plugins can process IP addresses.
raise errors.ConfigurationError("Enhancements not supported for IP address certificates")
if config.noninteractive_mode:
@@ -4,7 +4,7 @@ from typing import Any
from typing import Callable
from typing import Iterable
from acme import challenges
from acme import challenges, messages
from certbot import achallenges
from certbot import errors
from certbot import interfaces
@@ -184,10 +184,13 @@ permitted by DNS standards.)
def _perform_achall_with_script(self, achall: achallenges.AnnotatedChallenge,
achalls: list[achallenges.AnnotatedChallenge]) -> None:
if not achall.identifier.typ == messages.IDENTIFIER_FQDN:
raise errors.ConfigurationError("non-FQDN identifiers not yet supported")
domain = achall.identifier.value
env = {
"CERTBOT_DOMAIN": achall.domain,
"CERTBOT_DOMAIN": domain,
"CERTBOT_VALIDATION": achall.validation(achall.account_key),
"CERTBOT_ALL_DOMAINS": ','.join(one_achall.domain for one_achall in achalls),
"CERTBOT_ALL_DOMAINS": ','.join(one_achall.identifier.value for one_achall in achalls),
"CERTBOT_REMAINING_CHALLENGES": str(len(achalls) - achalls.index(achall) - 1),
}
if isinstance(achall.chall, challenges.HTTP01):
@@ -195,22 +198,25 @@ permitted by DNS standards.)
else:
os.environ.pop('CERTBOT_TOKEN', None)
os.environ.update(env)
_, out = self._execute_hook('auth-hook', achall.domain)
_, out = self._execute_hook('auth-hook', domain)
env['CERTBOT_AUTH_OUTPUT'] = out.strip()
self.env[achall] = env
def _perform_achall_manually(self, achall: achallenges.AnnotatedChallenge,
last_dns_achall: bool = False) -> None:
if not achall.identifier.typ == messages.IDENTIFIER_FQDN:
raise errors.ConfigurationError("non-FQDN identifiers not yet supported")
domain = achall.identifier.value
validation = achall.validation(achall.account_key)
if isinstance(achall.chall, challenges.HTTP01):
msg = self._HTTP_INSTRUCTIONS.format(
achall=achall, encoded_token=achall.chall.encode('token'),
port=self.config.http01_port,
uri=achall.chall.uri(achall.domain), validation=validation)
uri=achall.chall.uri(domain), validation=validation)
else:
assert isinstance(achall.chall, challenges.DNS01)
msg = self._DNS_INSTRUCTIONS.format(
domain=achall.validation_domain_name(achall.domain),
domain=achall.validation_domain_name(domain),
validation=validation)
if isinstance(achall.chall, challenges.DNS01):
if self.subsequent_dns_challenge:
@@ -224,7 +230,7 @@ permitted by DNS standards.)
if last_dns_achall:
# last dns-01 challenge
msg += self._DNS_VERIFY_INSTRUCTIONS.format(
domain=achall.validation_domain_name(achall.domain))
domain=achall.validation_domain_name(domain))
elif self.subsequent_any_challenge:
# 2nd or later challenge of another type
msg += self._SUBSEQUENT_CHALLENGE_INSTRUCTIONS
@@ -238,7 +244,7 @@ permitted by DNS standards.)
if 'CERTBOT_TOKEN' not in env:
os.environ.pop('CERTBOT_TOKEN', None)
os.environ.update(env)
self._execute_hook('cleanup-hook', achall.domain)
self._execute_hook('cleanup-hook', achall.identifier.value)
self.reverter.recovery_routine()
def _execute_hook(self, hook_name: str, achall_domain: str) -> tuple[str, str]:
@@ -10,7 +10,7 @@ from typing import Optional
from typing import Sequence
from typing import Union
from acme import challenges
from acme import challenges, messages
from certbot import crypto_util
from certbot import errors
from certbot import interfaces
@@ -106,6 +106,9 @@ 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,12 +121,12 @@ to serve all files under specified web root ({0})."""
logger.info("Using the webroot path %s for all unmatched domains.",
webroot_path)
for achall in achalls:
self.conf("map").setdefault(achall.domain, webroot_path)
self.conf("map").setdefault(achall.identifier.value, webroot_path)
else:
known_webroots = list(set(self.conf("map").values()))
for achall in achalls:
if achall.domain not in self.conf("map"):
new_webroot = self._prompt_for_webroot(achall.domain,
if achall.identifier.value not in self.conf("map"):
new_webroot = self._prompt_for_webroot(achall.identifier.value,
known_webroots)
# Put the most recently input
# webroot first for easy selection
@@ -132,7 +135,7 @@ to serve all files under specified web root ({0})."""
except ValueError:
pass
known_webroots.insert(0, new_webroot)
self.conf("map")[achall.domain] = new_webroot
self.conf("map")[achall.identifier.value] = new_webroot
def _prompt_for_webroot(self, domain: str, known_webroots: list[str]) -> Optional[str]:
webroot = None
@@ -238,7 +241,7 @@ to serve all files under specified web root ({0})."""
def _perform_single(self, achall: AnnotatedChallenge) -> challenges.ChallengeResponse:
response, validation = achall.response_and_validation()
root_path = self.full_roots[achall.domain]
root_path = self.full_roots[achall.identifier.value]
validation_path = self._get_validation_path(root_path, achall)
logger.debug("Attempting to save validation to %s", validation_path)
@@ -252,7 +255,7 @@ to serve all files under specified web root ({0})."""
def cleanup(self, achalls: list[AnnotatedChallenge]) -> None: # pylint: disable=missing-function-docstring
for achall in achalls:
root_path = self.full_roots.get(achall.domain, None)
root_path = self.full_roots.get(achall.identifier.value, None)
if root_path is not None:
validation_path = self._get_validation_path(root_path, achall)
logger.debug("Removing %s", validation_path)
@@ -430,13 +430,14 @@ class ChallbToAchallTest(unittest.TestCase):
def _call(self, challb):
from certbot._internal.auth_handler import challb_to_achall
return challb_to_achall(challb, "account_key", "domain")
ident = messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="domain")
return challb_to_achall(challb, "account_key", ident)
def test_it(self):
assert self._call(acme_util.HTTP01_P) == \
achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.HTTP01_P, account_key="account_key",
domain="domain")
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="domain"))
class GenChallengePathTest(unittest.TestCase):
@@ -517,11 +518,11 @@ class ReportFailedAuthzrsTest(unittest.TestCase):
'\n'
'Certbot failed to authenticate some domains (authenticator: buzz). '
'The Certificate Authority reported these problems:\n'
' Domain: example.com\n'
' Identifier: example.com\n'
' Type: tls\n'
' Detail: detail\n'
'\n'
' Domain: example.com\n'
' Identifier: example.com\n'
' Type: tls\n'
' Detail: detail\n'
'\nHint: the buzz hint\n'
@@ -536,15 +537,15 @@ class ReportFailedAuthzrsTest(unittest.TestCase):
'\n'
'Certbot failed to authenticate some domains (authenticator: quux). '
'The Certificate Authority reported these problems:\n'
' Domain: foo.bar\n'
' Identifier: foo.bar\n'
' Type: dnssec\n'
' Detail: detail\n'
'\n'
' Domain: example.com\n'
' Identifier: example.com\n'
' Type: tls\n'
' Detail: detail\n'
'\n'
' Domain: example.com\n'
' Identifier: example.com\n'
' Type: tls\n'
' Detail: detail\n'
'\nHint: quuuuuux\n'
@@ -565,7 +566,7 @@ class ReportFailedAuthzrsTest(unittest.TestCase):
def gen_auth_resp(chall_list):
"""Generate a dummy authorization response."""
return ["%s%s" % (chall.__class__.__name__, chall.domain)
return ["%s%s" % (chall.__class__.__name__, chall.identifier.value)
for chall in chall_list]
@@ -16,7 +16,8 @@ class FailedChallengesTest(unittest.TestCase):
def setUp(self):
from certbot.errors import FailedChallenges
self.error = FailedChallenges({achallenges.DNS(
domain="example.com", challb=messages.ChallengeBody(
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="example.com"),
challb=messages.ChallengeBody(
chall=acme_util.DNS01, uri=None,
error=messages.Error.with_code("tls", detail="detail")))})
@@ -29,7 +30,8 @@ class FailedChallengesTest(unittest.TestCase):
from certbot.errors import FailedChallenges
arabic_detail = u'\u0639\u062f\u0627\u0644\u0629'
arabic_error = FailedChallenges({achallenges.DNS(
domain="example.com", challb=messages.ChallengeBody(
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="example.com"),
challb=messages.ChallengeBody(
chall=acme_util.DNS01, uri=None,
error=messages.Error.with_code("tls", detail=arabic_detail)))})
@@ -22,7 +22,8 @@ AUTH_KEY = jose.JWKRSA.load(test_util.load_vector("rsa512_key.pem"))
ACHALL = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(challenges.HTTP01(token=b'token1'),
messages.STATUS_PENDING),
domain="encryption-example.demo", account_key=AUTH_KEY)
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="encryption-example.demo"),
account_key=AUTH_KEY)
class NamespaceFunctionsTest(unittest.TestCase):
@@ -73,14 +73,14 @@ class AuthenticatorTest(test_util.TempDirTestCase):
'print(os.environ.get(\'CERTBOT_REMAINING_CHALLENGES\'));"'
.format(sys.executable))
dns_expected = '{0}\n{1}\n{2}\n{3}\n{4}'.format(
self.dns_achall.domain, 'notoken',
self.dns_achall.identifier.value, 'notoken',
self.dns_achall.validation(self.dns_achall.account_key),
','.join(achall.domain for achall in self.achalls),
','.join(achall.identifier.value for achall in self.achalls),
len(self.achalls) - self.achalls.index(self.dns_achall) - 1)
http_expected = '{0}\n{1}\n{2}\n{3}\n{4}'.format(
self.http_achall.domain, self.http_achall.chall.encode('token'),
self.http_achall.identifier.value, self.http_achall.chall.encode('token'),
self.http_achall.validation(self.http_achall.account_key),
','.join(achall.domain for achall in self.achalls),
','.join(achall.identifier.value for achall in self.achalls),
len(self.achalls) - self.achalls.index(self.http_achall) - 1)
assert self.auth.perform(self.achalls) == \
@@ -115,7 +115,7 @@ class AuthenticatorTest(test_util.TempDirTestCase):
for achall in self.achalls:
self.auth.cleanup([achall])
assert os.environ['CERTBOT_AUTH_OUTPUT'] == 'foo'
assert os.environ['CERTBOT_DOMAIN'] == achall.domain
assert os.environ['CERTBOT_DOMAIN'] == achall.identifier.value
if isinstance(achall.chall, (challenges.HTTP01, challenges.DNS01)):
assert os.environ['CERTBOT_VALIDATION'] == \
achall.validation(achall.account_key)
@@ -8,7 +8,7 @@ from unittest import mock
import josepy as jose
import pytest
from acme import challenges
from acme import challenges, messages
from acme import standalone as acme_standalone
from certbot import achallenges
from certbot import errors
@@ -144,7 +144,9 @@ class AuthenticatorTest(unittest.TestCase):
domain = b'localhost'
key = jose.JWK.load(test_util.load_vector('rsa512_key.pem'))
http_01 = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.HTTP01_P, domain=domain, account_key=key)
challb=acme_util.HTTP01_P,
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value=domain),
account_key=key)
return [http_01]
@@ -14,7 +14,7 @@ from unittest import mock
import josepy as jose
import pytest
from acme import challenges
from acme import challenges, messages
from certbot import achallenges
from certbot import errors
from certbot.compat import filesystem
@@ -30,7 +30,9 @@ class AuthenticatorTest(unittest.TestCase):
"""Tests for certbot._internal.plugins.webroot.Authenticator."""
achall = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.HTTP01_P, domain="thing.com", account_key=KEY)
challb=acme_util.HTTP01_P,
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="thing.com"),
account_key=KEY)
def setUp(self):
from certbot._internal.plugins.webroot import Authenticator
@@ -79,11 +81,11 @@ class AuthenticatorTest(unittest.TestCase):
self.auth.perform([self.achall])
assert mock_display.menu.called
for call in mock_display.menu.call_args_list:
assert self.achall.domain in call[0][0]
assert self.achall.identifier.value in call[0][0]
assert all(
webroot in call[0][1]
for webroot in self.config.webroot_map.values())
assert self.config.webroot_map[self.achall.domain] == \
assert self.config.webroot_map[self.achall.identifier.value] == \
self.path
@unittest.skipIf(filesystem.POSIX_MODE, reason='Test specific to Windows')
@@ -120,7 +122,8 @@ class AuthenticatorTest(unittest.TestCase):
# Covers bug https://github.com/certbot/certbot/issues/9091
achall_2 = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(challenges.HTTP01(token=b"bingo"), "pending"),
domain="second-thing.com", account_key=KEY)
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="second-thing.com"),
account_key=KEY)
self.config.webroot_map["second-thing.com"] = self.path
challenge_path = os.path.join(self.path, ".well-known", "acme-challenge")
@@ -142,7 +145,7 @@ class AuthenticatorTest(unittest.TestCase):
self.auth.perform([self.achall])
assert mock_display.menu.called
for call in mock_display.menu.call_args_list:
assert self.achall.domain in call[0][0]
assert self.achall.identifier.value in call[0][0]
assert all(
webroot in call[0][1]
for webroot in self.config.webroot_map.values())
@@ -160,7 +163,7 @@ class AuthenticatorTest(unittest.TestCase):
self.auth.perform([self.achall])
assert self.config.webroot_map[self.achall.domain] == self.path
assert self.config.webroot_map[self.achall.identifier.value] == self.path
@test_util.patch_display_util()
def test_new_webroot_empty_map_cancel(self, mock_get_utility):
@@ -210,11 +213,13 @@ class AuthenticatorTest(unittest.TestCase):
mock_display.menu.side_effect = ((display_util.OK, 0),
(display_util.OK, new_webroot))
achall = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.HTTP01_P, domain="something.com", account_key=KEY)
challb=acme_util.HTTP01_P,
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="something.com"),
account_key=KEY)
with mock.patch('certbot.display.ops.validated_directory') as m:
m.return_value = (display_util.OK, new_webroot,)
self.auth.perform([achall])
assert self.config.webroot_map[achall.domain] == new_webroot
assert self.config.webroot_map[achall.identifier.value] == new_webroot
def test_perform_permissions(self):
self.auth.prepare()
@@ -262,7 +267,8 @@ class AuthenticatorTest(unittest.TestCase):
bingo_achall = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.chall_to_challb(
challenges.HTTP01(token=b"bingo"), "pending"),
domain="thing.com", account_key=KEY)
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="thing.com"),
account_key=KEY)
bingo_validation_path = "YmluZ28"
filesystem.mkdir(self.partial_root_challenge_path)
@@ -307,7 +313,9 @@ class WebrootActionTest(unittest.TestCase):
"""Tests for webroot argparse actions."""
achall = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.HTTP01_P, domain="thing.com", account_key=KEY)
challb=acme_util.HTTP01_P,
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value="thing.com"),
account_key=KEY)
def setUp(self):
from certbot._internal.plugins.webroot import Authenticator
@@ -324,9 +332,9 @@ class WebrootActionTest(unittest.TestCase):
def test_domain_before_webroot(self):
args = self.parser.parse_args(
"-d {0} -w {1}".format(self.achall.domain, self.path).split())
"-d {0} -w {1}".format(self.achall.identifier.value, self.path).split())
config = self._get_config_after_perform(args)
assert config.webroot_map[self.achall.domain] == self.path
assert config.webroot_map[self.achall.identifier.value] == self.path
def test_domain_before_webroot_error(self):
with pytest.raises(errors.PluginError):
@@ -336,10 +344,10 @@ class WebrootActionTest(unittest.TestCase):
def test_multiwebroot(self):
args = self.parser.parse_args("-w {0} -d {1} -w {2} -d bar".format(
self.path, self.achall.domain, tempfile.mkdtemp()).split())
assert args.webroot_map[self.achall.domain] == self.path
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.domain] == self.path
assert config.webroot_map[self.achall.identifier.value] == self.path
def test_webroot_map_partial_without_perform(self):
# This test acknowledges the fact that webroot_map content will be partial if webroot
@@ -350,8 +358,8 @@ class WebrootActionTest(unittest.TestCase):
# See https://github.com/certbot/certbot/pull/7095 for details.
other_webroot_path = tempfile.mkdtemp()
args = self.parser.parse_args("-w {0} -d {1} -w {2} -d bar".format(
self.path, self.achall.domain, other_webroot_path).split())
assert args.webroot_map == {self.achall.domain: self.path}
self.path, self.achall.identifier.value, other_webroot_path).split())
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):
+44 -5
View File
@@ -19,12 +19,15 @@ Note, that all annotated challenges act as a proxy objects::
"""
import logging
from typing import Any
import warnings
import josepy as jose
from acme import challenges
from acme import challenges, messages
from acme.challenges import Challenge
from certbot import errors
logger = logging.getLogger(__name__)
@@ -43,10 +46,47 @@ class AnnotatedChallenge(jose.ImmutableMap):
def __getattr__(self, name: str) -> Any:
return getattr(self.challb, name)
def __getattribute__(self, name: str) -> Any:
if name == 'domain':
warnings.warn("The domain attribute is deprecated and will be removed in "
"an upcoming release. Access the AnnotatedChallenge.identifier.value "
"attribute instead",
DeprecationWarning)
return super().__getattribute__(name)
def __hash__(self) -> int:
with warnings.catch_warnings():
warnings.filterwarnings('ignore', 'the domain attribute is deprecated')
return super().__hash__()
def __eq__(self, other: Any) -> bool:
with warnings.catch_warnings():
warnings.filterwarnings('ignore', 'the domain attribute is deprecated')
return super().__eq__(other)
def __init__(self, **kwargs: Any) -> None:
if 'domain' in kwargs:
if 'identifier' in kwargs:
raise errors.Error("AnnotatedChallenge takes either domain or identifier, not both")
warnings.warn("The domain attribute is deprecated and will be removed in "
"an upcoming release. Replace domain=<domain> with "
"identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, "
"value=<domain>)",
DeprecationWarning)
if 'identifier' not in kwargs:
kwargs['identifier'] = messages.Identifier(
typ=messages.IDENTIFIER_FQDN, value=kwargs['domain'])
if 'domain' not in kwargs:
if kwargs['identifier'].typ == messages.IDENTIFIER_FQDN:
kwargs['domain'] = kwargs['identifier'].value
else:
kwargs['domain'] = None
super().__init__(**kwargs)
class KeyAuthorizationAnnotatedChallenge(AnnotatedChallenge):
"""Client annotated `KeyAuthorizationChallenge` challenge."""
__slots__ = ('challb', 'domain', 'account_key') # pylint: disable=redefined-slots-in-subclass
__slots__ = ('challb', 'domain', 'account_key', 'identifier') # pylint: disable=redefined-slots-in-subclass
def response_and_validation(self, *args: Any, **kwargs: Any
) -> tuple['challenges.KeyAuthorizationChallengeResponse', Any]:
@@ -57,11 +97,10 @@ class KeyAuthorizationAnnotatedChallenge(AnnotatedChallenge):
class DNS(AnnotatedChallenge):
"""Client annotated "dns" ACME challenge."""
__slots__ = ('challb', 'domain') # pylint: disable=redefined-slots-in-subclass
__slots__ = ('challb', 'domain', 'identifier') # pylint: disable=redefined-slots-in-subclass
acme_type = challenges.DNS
class Other(AnnotatedChallenge):
"""Client annotated ACME challenge of an unknown type."""
__slots__ = ('challb', 'domain') # pylint: disable=redefined-slots-in-subclass
__slots__ = ('challb', 'domain', 'identifier') # pylint: disable=redefined-slots-in-subclass
acme_type = challenges.Challenge
+1 -1
View File
@@ -62,7 +62,7 @@ class FailedChallenges(AuthorizationError):
def __str__(self) -> str:
return "Failed authorization procedure. {0}".format(
", ".join(
"{0} ({1}): {2}".format(achall.domain, achall.typ, achall.error)
"{0} ({1}): {2}".format(achall.identifier.value, achall.typ, achall.error)
for achall in self.failed_achalls if achall.error is not None))
+2 -2
View File
@@ -71,7 +71,7 @@ class DNSAuthenticator(common.Plugin, interfaces.Authenticator, metaclass=abc.AB
responses = []
for achall in achalls:
domain = achall.domain
domain = achall.identifier.value
validation_domain_name = achall.validation_domain_name(domain)
validation = achall.validation(achall.account_key)
@@ -90,7 +90,7 @@ class DNSAuthenticator(common.Plugin, interfaces.Authenticator, metaclass=abc.AB
def cleanup(self, achalls: list[achallenges.AnnotatedChallenge]) -> None: # pylint: disable=missing-function-docstring
if self._attempt_cleanup:
for achall in achalls:
domain = achall.domain
domain = achall.identifier.value
validation_domain_name = achall.validation_domain_name(domain)
validation = achall.validation(achall.account_key)
@@ -7,7 +7,7 @@ from unittest import mock
import configobj
import josepy as jose
from acme import challenges
from acme import challenges, messages
from certbot import achallenges
from certbot.compat import filesystem
from certbot.plugins.dns_common import DNSAuthenticator
@@ -51,7 +51,9 @@ class BaseAuthenticatorTest:
"""
achall = achallenges.KeyAuthorizationAnnotatedChallenge(
challb=acme_util.DNS01, domain=DOMAIN, account_key=KEY)
challb=acme_util.DNS01,
identifier=messages.Identifier(typ=messages.IDENTIFIER_FQDN, value=DOMAIN),
account_key=KEY)
def test_more_info(self: _AuthenticatorCallableTestCase) -> None:
self.assertTrue(isinstance(self.auth.more_info(), str)) # pylint: disable=no-member
+6 -3
View File
@@ -45,9 +45,12 @@ CHALLENGES_P = [HTTP01_P, DNS01_P]
# AnnotatedChallenge objects
HTTP01_A = auth_handler.challb_to_achall(HTTP01_P, JWK, "example.com")
DNS01_A = auth_handler.challb_to_achall(DNS01_P, JWK, "example.org")
DNS01_A_2 = auth_handler.challb_to_achall(DNS01_P_2, JWK, "esimerkki.example.org")
HTTP01_A = auth_handler.challb_to_achall(HTTP01_P, JWK, messages.Identifier(
typ=messages.IDENTIFIER_FQDN, value="example.com"))
DNS01_A = auth_handler.challb_to_achall(DNS01_P, JWK, messages.Identifier(
typ=messages.IDENTIFIER_FQDN, value="example.org"))
DNS01_A_2 = auth_handler.challb_to_achall(DNS01_P_2, JWK, messages.Identifier(
typ=messages.IDENTIFIER_FQDN, value="esimerkki.example.org"))
ACHALLENGES = [HTTP01_A, DNS01_A]
+1
View File
@@ -0,0 +1 @@
`achallenges.KeyAuthorizationAnnotatedChallenge`, `achallenges.DNS`, and `achallenges.Other` have a new field `identifier`, of type `acme.messages.Identifier`. This should be used in place of the `domain` field, which is now deprecated both as an attribute and during object creation.