# -*- coding: utf-8 -*- # Copyright (c) 2024 Felix Fontein # GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) # SPDX-License-Identifier: GPL-3.0-or-later from __future__ import absolute_import, division, print_function __metaclass__ = type import os from ansible_collections.community.crypto.plugins.module_utils.acme.account import ( ACMEAccount, ) from ansible_collections.community.crypto.plugins.module_utils.acme.acme import ( ACMEClient, ) from ansible_collections.community.crypto.plugins.module_utils.acme.certificates import ( CertificateChain, Criterium, ) from ansible_collections.community.crypto.plugins.module_utils.acme.challenges import ( Authorization, wait_for_validation, ) from ansible_collections.community.crypto.plugins.module_utils.acme.errors import ( ModuleFailException, ) from ansible_collections.community.crypto.plugins.module_utils.acme.io import write_file from ansible_collections.community.crypto.plugins.module_utils.acme.orders import Order from ansible_collections.community.crypto.plugins.module_utils.acme.utils import ( pem_to_der, ) class ACMECertificateClient(object): """ ACME v2 client class. Uses an ACME account object and a CSR to start and validate ACME challenges and download the respective certificates. """ def __init__(self, module, backend, client=None, account=None): self.module = module self.version = module.params["acme_version"] self.csr = module.params.get("csr") self.csr_content = module.params.get("csr_content") if client is None: client = ACMEClient(module, backend) self.client = client if account is None: account = ACMEAccount(self.client) self.account = account self.order_uri = module.params.get("order_uri") self.order_creation_error_strategy = module.params.get( "order_creation_error_strategy", "auto" ) self.order_creation_max_retries = module.params.get( "order_creation_max_retries", 3 ) # Make sure account exists dummy, account_data = self.account.setup_account(allow_creation=False) if account_data is None: raise ModuleFailException(msg="Account does not exist or is deactivated.") if self.csr is not None and not os.path.exists(self.csr): raise ModuleFailException("CSR %s not found" % (self.csr)) # Extract list of identifiers from CSR if self.csr is not None or self.csr_content is not None: self.identifiers = self.client.backend.get_ordered_csr_identifiers( csr_filename=self.csr, csr_content=self.csr_content ) else: self.identifiers = None def parse_select_chain(self, select_chain): select_chain_matcher = [] if select_chain: for criterium_idx, criterium in enumerate(select_chain): try: select_chain_matcher.append( self.client.backend.create_chain_matcher( Criterium(criterium, index=criterium_idx) ) ) except ValueError as exc: self.module.warn( "Error while parsing criterium: {error}. Ignoring criterium.".format( error=exc ) ) return select_chain_matcher def load_order(self): if not self.order_uri: raise ModuleFailException("The order URI has not been provided") order = Order.from_url(self.client, self.order_uri) order.load_authorizations(self.client) return order def create_order(self, replaces_cert_id=None, profile=None): """ Create a new order. """ if self.identifiers is None: raise ModuleFailException("No identifiers have been provided") order = Order.create_with_error_handling( self.client, self.identifiers, error_strategy=self.order_creation_error_strategy, error_max_retries=self.order_creation_max_retries, replaces_cert_id=replaces_cert_id, profile=profile, message_callback=self.module.warn, ) self.order_uri = order.url order.load_authorizations(self.client) return order def get_challenges_data(self, order): """ Get challenge details. Return a tuple of generic challenge details, and specialized DNS challenge details. """ # Get general challenge data data = [] for authz in order.authorizations.values(): # Skip valid authentications: their challenges are already valid # and do not need to be returned if authz.status == "valid": continue data.append( dict( identifier=authz.identifier, identifier_type=authz.identifier_type, challenges=authz.get_challenge_data(self.client), ) ) # Get DNS challenge data data_dns = {} dns_challenge_type = "dns-01" for entry in data: dns_challenge = entry["challenges"].get(dns_challenge_type) if dns_challenge: values = data_dns.get(dns_challenge["record"]) if values is None: values = [] data_dns[dns_challenge["record"]] = values values.append(dns_challenge["resource_value"]) return data, data_dns def check_that_authorizations_can_be_used(self, order): bad_authzs = [] for authz in order.authorizations.values(): if authz.status not in ("valid", "pending"): bad_authzs.append( "{authz} (status={status!r})".format( authz=authz.combined_identifier, status=authz.status, ) ) if bad_authzs: raise ModuleFailException( "Some of the authorizations for the order are in a bad state, so the order" " can no longer be satisfied: {bad_authzs}".format( bad_authzs=", ".join(sorted(bad_authzs)), ), ) def collect_invalid_authzs(self, order): return [ authz for authz in order.authorizations.values() if authz.status == "invalid" ] def collect_pending_authzs(self, order): return [ authz for authz in order.authorizations.values() if authz.status == "pending" ] def call_validate(self, pending_authzs, get_challenge, wait=True): authzs_with_challenges_to_wait_for = [] for authz in pending_authzs: challenge_type = get_challenge(authz) authz.call_validate(self.client, challenge_type, wait=wait) authzs_with_challenges_to_wait_for.append( (authz, challenge_type, authz.find_challenge(challenge_type)) ) return authzs_with_challenges_to_wait_for def wait_for_validation(self, authzs_to_wait_for): wait_for_validation(authzs_to_wait_for, self.client) def _download_alternate_chains(self, cert): alternate_chains = [] for alternate in cert.alternates: try: alt_cert = CertificateChain.download(self.client, alternate) except ModuleFailException as e: self.module.warn( "Error while downloading alternative certificate {0}: {1}".format( alternate, e ) ) continue if alt_cert.cert is not None: alternate_chains.append(alt_cert) else: self.module.warn( "Error while downloading alternative certificate {0}: no certificate found".format( alternate ) ) return alternate_chains def download_certificate(self, order, download_all_chains=True): """ Download certificate from a valid oder. """ if order.status != "valid": raise ModuleFailException( "The order must be valid, but has state {state!r}!".format( state=order.state ) ) if not order.certificate_uri: raise ModuleFailException( "Order's crtificate URL {url!r} is empty!".format( url=order.certificate_uri ) ) cert = CertificateChain.download(self.client, order.certificate_uri) if cert.cert is None: raise ModuleFailException( "Certificate at {url} is empty!".format(url=order.certificate_uri) ) alternate_chains = None if download_all_chains: alternate_chains = self._download_alternate_chains(cert) return cert, alternate_chains def get_certificate(self, order, download_all_chains=True): """ Request a new certificate and downloads it, and optionally all certificate chains. First verifies whether all authorizations are valid; if not, aborts with an error. """ if self.csr is None and self.csr_content is None: raise ModuleFailException("No CSR has been provided") for identifier, authz in order.authorizations.items(): if authz.status != "valid": authz.raise_error( 'Status is {status!r} and not "valid"'.format(status=authz.status), module=self.module, ) order.finalize(self.client, pem_to_der(self.csr, self.csr_content)) return self.download_certificate(order, download_all_chains=download_all_chains) def find_matching_chain(self, chains, select_chain_matcher): for criterium_idx, matcher in enumerate(select_chain_matcher): for chain in chains: if matcher.match(chain): self.module.debug( "Found matching chain for criterium {0}".format(criterium_idx) ) return chain return None def write_cert_chain( self, cert, cert_dest=None, fullchain_dest=None, chain_dest=None ): changed = False if cert_dest and write_file(self.module, cert_dest, cert.cert.encode("utf8")): changed = True if fullchain_dest and write_file( self.module, fullchain_dest, (cert.cert + "\n".join(cert.chain)).encode("utf8"), ): changed = True if chain_dest and write_file( self.module, chain_dest, ("\n".join(cert.chain)).encode("utf8") ): changed = True return changed def deactivate_authzs(self, order): """ Deactivates all valid authz's. Does not raise exceptions. https://community.letsencrypt.org/t/authorization-deactivation/19860/2 https://tools.ietf.org/html/rfc8555#section-7.5.2 """ if len(order.authorization_uris) > len(order.authorizations): for authz_uri in order.authorization_uris: authz = None try: authz = Authorization.deactivate_url(self.client, authz_uri) except Exception: # ignore errors pass if authz is None or authz.status != "deactivated": self.module.warn( warning="Could not deactivate authz object {0}.".format( authz_uri ) ) else: for authz in order.authorizations.values(): try: authz.deactivate(self.client) except Exception: # ignore errors pass if authz.status != "deactivated": self.module.warn( warning="Could not deactivate authz object {0}.".format( authz.url ) )