Merge pull request #3167 from unic/feature/acme-http-challenge
Add support for acme http-01 challenge
This commit is contained in:
Hossein Shafagh
GitHub
commit
95854c6f68
|
|
@ -3,9 +3,9 @@ from flask_script import Manager
|
|||
import sys
|
||||
|
||||
from lemur.constants import SUCCESS_METRIC_STATUS
|
||||
from lemur.plugins.lemur_acme.acme_handlers import AcmeDnsHandler
|
||||
from lemur.dns_providers.service import get_all_dns_providers, set_domains
|
||||
from lemur.extensions import metrics, sentry
|
||||
from lemur.plugins.base import plugins
|
||||
|
||||
manager = Manager(
|
||||
usage="Iterates through all DNS providers and sets DNS zones in the database."
|
||||
|
|
@ -19,7 +19,7 @@ def get_all_zones():
|
|||
"""
|
||||
print("[+] Starting dns provider zone lookup and configuration.")
|
||||
dns_providers = get_all_dns_providers()
|
||||
acme_plugin = plugins.get("acme-issuer")
|
||||
acme_dns_handler = AcmeDnsHandler()
|
||||
|
||||
function = f"{__name__}.{sys._getframe().f_code.co_name}"
|
||||
log_data = {
|
||||
|
|
@ -29,7 +29,7 @@ def get_all_zones():
|
|||
|
||||
for dns_provider in dns_providers:
|
||||
try:
|
||||
zones = acme_plugin.get_all_zones(dns_provider)
|
||||
zones = acme_dns_handler.get_all_zones(dns_provider)
|
||||
set_domains(dns_provider, zones)
|
||||
except Exception as e:
|
||||
print("[+] Error with DNS Provider {}: {}".format(dns_provider.name, e))
|
||||
|
|
|
|||
|
|
@ -0,0 +1,521 @@
|
|||
"""
|
||||
.. module: lemur.plugins.lemur_acme.plugin
|
||||
:platform: Unix
|
||||
:synopsis: This module contains handlers for certain acme related tasks. It needed to be refactored to avoid circular imports
|
||||
:copyright: (c) 2018 by Netflix Inc., see AUTHORS for more
|
||||
:license: Apache, see LICENSE for more details.
|
||||
|
||||
Snippets from https://raw.githubusercontent.com/alex/letsencrypt-aws/master/letsencrypt-aws.py
|
||||
|
||||
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
|
||||
.. moduleauthor:: Mikhail Khodorovskiy <mikhail.khodorovskiy@jivesoftware.com>
|
||||
.. moduleauthor:: Curtis Castrapel <ccastrapel@netflix.com>
|
||||
.. moduleauthor:: Mathias Petermann <mathias.petermann@projektfokus.ch>
|
||||
"""
|
||||
import datetime
|
||||
import json
|
||||
import time
|
||||
|
||||
import OpenSSL.crypto
|
||||
import josepy as jose
|
||||
import dns.resolver
|
||||
from acme import challenges, errors, messages
|
||||
from acme.client import BackwardsCompatibleClientV2, ClientNetwork
|
||||
from acme.errors import TimeoutError
|
||||
from acme.messages import Error as AcmeError
|
||||
from flask import current_app
|
||||
|
||||
from lemur.common.utils import generate_private_key
|
||||
from lemur.dns_providers import service as dns_provider_service
|
||||
from lemur.exceptions import InvalidAuthority, UnknownProvider, InvalidConfiguration
|
||||
from lemur.extensions import metrics, sentry
|
||||
|
||||
from lemur.plugins.lemur_acme import cloudflare, dyn, route53, ultradns, powerdns
|
||||
from lemur.authorities import service as authorities_service
|
||||
from retrying import retry
|
||||
|
||||
|
||||
class AuthorizationRecord(object):
|
||||
def __init__(self, domain, target_domain, authz, dns_challenge, change_id):
|
||||
self.domain = domain
|
||||
self.target_domain = target_domain
|
||||
self.authz = authz
|
||||
self.dns_challenge = dns_challenge
|
||||
self.change_id = change_id
|
||||
|
||||
|
||||
class AcmeHandler(object):
|
||||
|
||||
def reuse_account(self, authority):
|
||||
if not authority.options:
|
||||
raise InvalidAuthority("Invalid authority. Options not set")
|
||||
existing_key = False
|
||||
existing_regr = False
|
||||
|
||||
for option in json.loads(authority.options):
|
||||
if option["name"] == "acme_private_key" and option["value"]:
|
||||
existing_key = True
|
||||
if option["name"] == "acme_regr" and option["value"]:
|
||||
existing_regr = True
|
||||
|
||||
if not existing_key and current_app.config.get("ACME_PRIVATE_KEY"):
|
||||
existing_key = True
|
||||
|
||||
if not existing_regr and current_app.config.get("ACME_REGR"):
|
||||
existing_regr = True
|
||||
|
||||
if existing_key and existing_regr:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def strip_wildcard(self, host):
|
||||
"""Removes the leading *. and returns Host and whether it was removed or not (True/False)"""
|
||||
prefix = "*."
|
||||
if host.startswith(prefix):
|
||||
return host[len(prefix):], True
|
||||
return host, False
|
||||
|
||||
def maybe_add_extension(self, host, dns_provider_options):
|
||||
if dns_provider_options and dns_provider_options.get(
|
||||
"acme_challenge_extension"
|
||||
):
|
||||
host = host + dns_provider_options.get("acme_challenge_extension")
|
||||
return host
|
||||
|
||||
def request_certificate(self, acme_client, authorizations, order):
|
||||
for authorization in authorizations:
|
||||
for authz in authorization.authz:
|
||||
authorization_resource, _ = acme_client.poll(authz)
|
||||
|
||||
deadline = datetime.datetime.now() + datetime.timedelta(seconds=360)
|
||||
|
||||
try:
|
||||
orderr = acme_client.poll_and_finalize(order, deadline)
|
||||
|
||||
except (AcmeError, TimeoutError):
|
||||
sentry.captureException(extra={"order_url": str(order.uri)})
|
||||
metrics.send("request_certificate_error", "counter", 1, metric_tags={"uri": order.uri})
|
||||
current_app.logger.error(
|
||||
f"Unable to resolve Acme order: {order.uri}", exc_info=True
|
||||
)
|
||||
raise
|
||||
except errors.ValidationError:
|
||||
if order.fullchain_pem:
|
||||
orderr = order
|
||||
else:
|
||||
raise
|
||||
|
||||
metrics.send("request_certificate_success", "counter", 1, metric_tags={"uri": order.uri})
|
||||
current_app.logger.info(
|
||||
f"Successfully resolved Acme order: {order.uri}", exc_info=True
|
||||
)
|
||||
|
||||
pem_certificate, pem_certificate_chain = self.extract_cert_and_chain(orderr.fullchain_pem)
|
||||
|
||||
current_app.logger.debug(
|
||||
"{0} {1}".format(type(pem_certificate), type(pem_certificate_chain))
|
||||
)
|
||||
return pem_certificate, pem_certificate_chain
|
||||
|
||||
def extract_cert_and_chain(self, fullchain_pem):
|
||||
pem_certificate = OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM,
|
||||
OpenSSL.crypto.load_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM, fullchain_pem
|
||||
),
|
||||
).decode()
|
||||
|
||||
if current_app.config.get("IDENTRUST_CROSS_SIGNED_LE_ICA", False) \
|
||||
and datetime.datetime.now() < datetime.datetime.strptime(
|
||||
current_app.config.get("IDENTRUST_CROSS_SIGNED_LE_ICA_EXPIRATION_DATE", "17/03/21"), '%d/%m/%y'):
|
||||
pem_certificate_chain = current_app.config.get("IDENTRUST_CROSS_SIGNED_LE_ICA")
|
||||
else:
|
||||
pem_certificate_chain = fullchain_pem[len(pem_certificate):].lstrip()
|
||||
|
||||
return pem_certificate, pem_certificate_chain
|
||||
|
||||
@retry(stop_max_attempt_number=5, wait_fixed=5000)
|
||||
def setup_acme_client(self, authority):
|
||||
if not authority.options:
|
||||
raise InvalidAuthority("Invalid authority. Options not set")
|
||||
options = {}
|
||||
|
||||
for option in json.loads(authority.options):
|
||||
options[option["name"]] = option.get("value")
|
||||
email = options.get("email", current_app.config.get("ACME_EMAIL"))
|
||||
tel = options.get("telephone", current_app.config.get("ACME_TEL"))
|
||||
directory_url = options.get(
|
||||
"acme_url", current_app.config.get("ACME_DIRECTORY_URL")
|
||||
)
|
||||
|
||||
existing_key = options.get(
|
||||
"acme_private_key", current_app.config.get("ACME_PRIVATE_KEY")
|
||||
)
|
||||
existing_regr = options.get("acme_regr", current_app.config.get("ACME_REGR"))
|
||||
|
||||
if existing_key and existing_regr:
|
||||
current_app.logger.debug("Reusing existing ACME account")
|
||||
# Reuse the same account for each certificate issuance
|
||||
key = jose.JWK.json_loads(existing_key)
|
||||
regr = messages.RegistrationResource.json_loads(existing_regr)
|
||||
current_app.logger.debug(
|
||||
"Connecting with directory at {0}".format(directory_url)
|
||||
)
|
||||
net = ClientNetwork(key, account=regr)
|
||||
client = BackwardsCompatibleClientV2(net, key, directory_url)
|
||||
return client, {}
|
||||
else:
|
||||
# Create an account for each certificate issuance
|
||||
key = jose.JWKRSA(key=generate_private_key("RSA2048"))
|
||||
|
||||
current_app.logger.debug("Creating a new ACME account")
|
||||
current_app.logger.debug(
|
||||
"Connecting with directory at {0}".format(directory_url)
|
||||
)
|
||||
|
||||
net = ClientNetwork(key, account=None, timeout=3600)
|
||||
client = BackwardsCompatibleClientV2(net, key, directory_url)
|
||||
registration = client.new_account_and_tos(
|
||||
messages.NewRegistration.from_data(email=email)
|
||||
)
|
||||
|
||||
# if store_account is checked, add the private_key and registration resources to the options
|
||||
if options['store_account']:
|
||||
new_options = json.loads(authority.options)
|
||||
# the key returned by fields_to_partial_json is missing the key type, so we add it manually
|
||||
key_dict = key.fields_to_partial_json()
|
||||
key_dict["kty"] = "RSA"
|
||||
acme_private_key = {
|
||||
"name": "acme_private_key",
|
||||
"value": json.dumps(key_dict)
|
||||
}
|
||||
new_options.append(acme_private_key)
|
||||
|
||||
acme_regr = {
|
||||
"name": "acme_regr",
|
||||
"value": json.dumps({"body": {}, "uri": registration.uri})
|
||||
}
|
||||
new_options.append(acme_regr)
|
||||
|
||||
authorities_service.update_options(authority.id, options=json.dumps(new_options))
|
||||
|
||||
current_app.logger.debug("Connected: {0}".format(registration.uri))
|
||||
|
||||
return client, registration
|
||||
|
||||
def get_domains(self, options):
|
||||
"""
|
||||
Fetches all domains currently requested
|
||||
:param options:
|
||||
:return:
|
||||
"""
|
||||
current_app.logger.debug("Fetching domains")
|
||||
|
||||
domains = [options["common_name"]]
|
||||
if options.get("extensions"):
|
||||
for dns_name in options["extensions"]["sub_alt_names"]["names"]:
|
||||
if dns_name.value not in domains:
|
||||
domains.append(dns_name.value)
|
||||
|
||||
current_app.logger.debug("Got these domains: {0}".format(domains))
|
||||
return domains
|
||||
|
||||
def revoke_certificate(self, certificate):
|
||||
if not self.reuse_account(certificate.authority):
|
||||
raise InvalidConfiguration("There is no ACME account saved, unable to revoke the certificate.")
|
||||
acme_client, _ = self.acme.setup_acme_client(certificate.authority)
|
||||
|
||||
fullchain_com = jose.ComparableX509(
|
||||
OpenSSL.crypto.load_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM, certificate.body))
|
||||
|
||||
try:
|
||||
acme_client.revoke(fullchain_com, 0) # revocation reason = 0
|
||||
except (errors.ConflictError, errors.ClientError, errors.Error) as e:
|
||||
# Certificate already revoked.
|
||||
current_app.logger.error("Certificate revocation failed with message: " + e.detail)
|
||||
metrics.send("acme_revoke_certificate_failure", "counter", 1)
|
||||
return False
|
||||
|
||||
current_app.logger.warning("Certificate succesfully revoked: " + certificate.name)
|
||||
metrics.send("acme_revoke_certificate_success", "counter", 1)
|
||||
return True
|
||||
|
||||
|
||||
class AcmeDnsHandler(AcmeHandler):
|
||||
|
||||
def __init__(self):
|
||||
self.dns_providers_for_domain = {}
|
||||
try:
|
||||
self.all_dns_providers = dns_provider_service.get_all_dns_providers()
|
||||
except Exception as e:
|
||||
metrics.send("AcmeHandler_init_error", "counter", 1)
|
||||
sentry.captureException()
|
||||
current_app.logger.error(f"Unable to fetch DNS Providers: {e}")
|
||||
self.all_dns_providers = []
|
||||
|
||||
def get_all_zones(self, dns_provider):
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
dns_provider_plugin = self.get_dns_provider(dns_provider.provider_type)
|
||||
return dns_provider_plugin.get_zones(account_number=account_number)
|
||||
|
||||
def get_dns_challenges(self, host, authorizations):
|
||||
"""Get dns challenges for provided domain"""
|
||||
|
||||
domain_to_validate, is_wildcard = self.strip_wildcard(host)
|
||||
dns_challenges = []
|
||||
for authz in authorizations:
|
||||
if not authz.body.identifier.value.lower() == domain_to_validate.lower():
|
||||
continue
|
||||
if is_wildcard and not authz.body.wildcard:
|
||||
continue
|
||||
if not is_wildcard and authz.body.wildcard:
|
||||
continue
|
||||
for combo in authz.body.challenges:
|
||||
if isinstance(combo.chall, challenges.DNS01):
|
||||
dns_challenges.append(combo)
|
||||
|
||||
return dns_challenges
|
||||
|
||||
def get_dns_provider(self, type):
|
||||
provider_types = {
|
||||
"cloudflare": cloudflare,
|
||||
"dyn": dyn,
|
||||
"route53": route53,
|
||||
"ultradns": ultradns,
|
||||
"powerdns": powerdns
|
||||
}
|
||||
provider = provider_types.get(type)
|
||||
if not provider:
|
||||
raise UnknownProvider("No such DNS provider: {}".format(type))
|
||||
return provider
|
||||
|
||||
def start_dns_challenge(
|
||||
self,
|
||||
acme_client,
|
||||
account_number,
|
||||
domain,
|
||||
target_domain,
|
||||
dns_provider,
|
||||
order,
|
||||
dns_provider_options,
|
||||
):
|
||||
current_app.logger.debug(f"Starting DNS challenge for {domain} using target domain {target_domain}.")
|
||||
|
||||
change_ids = []
|
||||
dns_challenges = self.get_dns_challenges(domain, order.authorizations)
|
||||
host_to_validate, _ = self.strip_wildcard(target_domain)
|
||||
host_to_validate = self.maybe_add_extension(host_to_validate, dns_provider_options)
|
||||
|
||||
if not dns_challenges:
|
||||
sentry.captureException()
|
||||
metrics.send("start_dns_challenge_error_no_dns_challenges", "counter", 1)
|
||||
raise Exception("Unable to determine DNS challenges from authorizations")
|
||||
|
||||
for dns_challenge in dns_challenges:
|
||||
|
||||
# Only prepend '_acme-challenge' if not using CNAME redirection
|
||||
if domain == target_domain:
|
||||
host_to_validate = dns_challenge.validation_domain_name(host_to_validate)
|
||||
|
||||
change_id = dns_provider.create_txt_record(
|
||||
host_to_validate,
|
||||
dns_challenge.validation(acme_client.client.net.key),
|
||||
account_number,
|
||||
)
|
||||
change_ids.append(change_id)
|
||||
|
||||
return AuthorizationRecord(
|
||||
domain, target_domain, order.authorizations, dns_challenges, change_ids
|
||||
)
|
||||
|
||||
def complete_dns_challenge(self, acme_client, authz_record):
|
||||
current_app.logger.debug(
|
||||
"Finalizing DNS challenge for {0}".format(
|
||||
authz_record.authz[0].body.identifier.value
|
||||
)
|
||||
)
|
||||
dns_providers = self.dns_providers_for_domain.get(authz_record.target_domain)
|
||||
if not dns_providers:
|
||||
metrics.send("complete_dns_challenge_error_no_dnsproviders", "counter", 1)
|
||||
raise Exception(
|
||||
"No DNS providers found for domain: {}".format(authz_record.target_domain)
|
||||
)
|
||||
|
||||
for dns_provider in dns_providers:
|
||||
# Grab account number (For Route53)
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
dns_provider_plugin = self.get_dns_provider(dns_provider.provider_type)
|
||||
for change_id in authz_record.change_id:
|
||||
try:
|
||||
dns_provider_plugin.wait_for_dns_change(
|
||||
change_id, account_number=account_number
|
||||
)
|
||||
except Exception:
|
||||
metrics.send("complete_dns_challenge_error", "counter", 1)
|
||||
sentry.captureException()
|
||||
current_app.logger.debug(
|
||||
f"Unable to resolve DNS challenge for change_id: {change_id}, account_id: "
|
||||
f"{account_number}",
|
||||
exc_info=True,
|
||||
)
|
||||
raise
|
||||
|
||||
for dns_challenge in authz_record.dns_challenge:
|
||||
response = dns_challenge.response(acme_client.client.net.key)
|
||||
|
||||
verified = response.simple_verify(
|
||||
dns_challenge.chall,
|
||||
authz_record.target_domain,
|
||||
acme_client.client.net.key.public_key(),
|
||||
)
|
||||
|
||||
if not verified:
|
||||
metrics.send("complete_dns_challenge_verification_error", "counter", 1)
|
||||
raise ValueError("Failed verification")
|
||||
|
||||
time.sleep(5)
|
||||
res = acme_client.answer_challenge(dns_challenge, response)
|
||||
current_app.logger.debug(f"answer_challenge response: {res}")
|
||||
|
||||
def get_authorizations(self, acme_client, order, order_info):
|
||||
authorizations = []
|
||||
|
||||
for domain in order_info.domains:
|
||||
|
||||
# If CNAME exists, set host to the target address
|
||||
target_domain = domain
|
||||
if current_app.config.get("ACME_ENABLE_DELEGATED_CNAME", False):
|
||||
cname_result, _ = self.strip_wildcard(domain)
|
||||
cname_result = challenges.DNS01().validation_domain_name(cname_result)
|
||||
cname_result = self.get_cname(cname_result)
|
||||
if cname_result:
|
||||
target_domain = cname_result
|
||||
self.autodetect_dns_providers(target_domain)
|
||||
|
||||
if not self.dns_providers_for_domain.get(target_domain):
|
||||
metrics.send(
|
||||
"get_authorizations_no_dns_provider_for_domain", "counter", 1
|
||||
)
|
||||
raise Exception("No DNS providers found for domain: {}".format(target_domain))
|
||||
|
||||
for dns_provider in self.dns_providers_for_domain[target_domain]:
|
||||
dns_provider_plugin = self.get_dns_provider(dns_provider.provider_type)
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
authz_record = self.start_dns_challenge(
|
||||
acme_client,
|
||||
account_number,
|
||||
domain,
|
||||
target_domain,
|
||||
dns_provider_plugin,
|
||||
order,
|
||||
dns_provider.options,
|
||||
)
|
||||
authorizations.append(authz_record)
|
||||
return authorizations
|
||||
|
||||
def autodetect_dns_providers(self, domain):
|
||||
"""
|
||||
Get DNS providers associated with a domain when it has not been provided for certificate creation.
|
||||
:param domain:
|
||||
:return: dns_providers: List of DNS providers that have the correct zone.
|
||||
"""
|
||||
self.dns_providers_for_domain[domain] = []
|
||||
match_length = 0
|
||||
for dns_provider in self.all_dns_providers:
|
||||
if not dns_provider.domains:
|
||||
continue
|
||||
for name in dns_provider.domains:
|
||||
if name == domain or domain.endswith("." + name):
|
||||
if len(name) > match_length:
|
||||
self.dns_providers_for_domain[domain] = [dns_provider]
|
||||
match_length = len(name)
|
||||
elif len(name) == match_length:
|
||||
self.dns_providers_for_domain[domain].append(dns_provider)
|
||||
|
||||
return self.dns_providers_for_domain
|
||||
|
||||
def finalize_authorizations(self, acme_client, authorizations):
|
||||
for authz_record in authorizations:
|
||||
self.complete_dns_challenge(acme_client, authz_record)
|
||||
for authz_record in authorizations:
|
||||
dns_challenges = authz_record.dns_challenge
|
||||
for dns_challenge in dns_challenges:
|
||||
dns_providers = self.dns_providers_for_domain.get(authz_record.target_domain)
|
||||
for dns_provider in dns_providers:
|
||||
# Grab account number (For Route53)
|
||||
dns_provider_plugin = self.get_dns_provider(
|
||||
dns_provider.provider_type
|
||||
)
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
host_to_validate, _ = self.strip_wildcard(authz_record.target_domain)
|
||||
host_to_validate = self.maybe_add_extension(host_to_validate, dns_provider_options)
|
||||
if authz_record.domain == authz_record.target_domain:
|
||||
host_to_validate = challenges.DNS01().validation_domain_name(host_to_validate)
|
||||
dns_provider_plugin.delete_txt_record(
|
||||
authz_record.change_id,
|
||||
account_number,
|
||||
host_to_validate,
|
||||
dns_challenge.validation(acme_client.client.net.key),
|
||||
)
|
||||
|
||||
return authorizations
|
||||
|
||||
def cleanup_dns_challenges(self, acme_client, authorizations):
|
||||
"""
|
||||
Best effort attempt to delete DNS challenges that may not have been deleted previously. This is usually called
|
||||
on an exception
|
||||
|
||||
:param acme_client:
|
||||
:param account_number:
|
||||
:param dns_provider:
|
||||
:param authorizations:
|
||||
:param dns_provider_options:
|
||||
:return:
|
||||
"""
|
||||
for authz_record in authorizations:
|
||||
dns_providers = self.dns_providers_for_domain.get(authz_record.target_domain)
|
||||
for dns_provider in dns_providers:
|
||||
# Grab account number (For Route53)
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
dns_challenges = authz_record.dns_challenge
|
||||
host_to_validate, _ = self.strip_wildcard(authz_record.target_domain)
|
||||
host_to_validate = self.maybe_add_extension(
|
||||
host_to_validate, dns_provider_options
|
||||
)
|
||||
|
||||
dns_provider_plugin = self.get_dns_provider(dns_provider.provider_type)
|
||||
for dns_challenge in dns_challenges:
|
||||
if authz_record.domain == authz_record.target_domain:
|
||||
host_to_validate = dns_challenge.validation_domain_name(host_to_validate)
|
||||
try:
|
||||
dns_provider_plugin.delete_txt_record(
|
||||
authz_record.change_id,
|
||||
account_number,
|
||||
host_to_validate,
|
||||
dns_challenge.validation(acme_client.client.net.key),
|
||||
)
|
||||
except Exception as e:
|
||||
# If this fails, it's most likely because the record doesn't exist (It was already cleaned up)
|
||||
# or we're not authorized to modify it.
|
||||
metrics.send("cleanup_dns_challenges_error", "counter", 1)
|
||||
sentry.captureException()
|
||||
pass
|
||||
|
||||
def get_cname(self, domain):
|
||||
"""
|
||||
:param domain: Domain name to look up a CNAME for.
|
||||
:return: First CNAME target or False if no CNAME record exists.
|
||||
"""
|
||||
try:
|
||||
result = dns.resolver.query(domain, 'CNAME')
|
||||
if len(result) > 0:
|
||||
return str(result[0].target).rstrip('.')
|
||||
except dns.exception.DNSException:
|
||||
return False
|
||||
|
|
@ -0,0 +1,260 @@
|
|||
"""
|
||||
.. module: lemur.plugins.lemur_acme.plugin
|
||||
:platform: Unix
|
||||
:synopsis: This module contains the different challenge types for ACME implementations
|
||||
:copyright: (c) 2018 by Netflix Inc., see AUTHORS for more
|
||||
:license: Apache, see LICENSE for more details.
|
||||
|
||||
.. moduleauthor:: Mathias Petermann <mathias.petermann@projektfokus.ch>
|
||||
"""
|
||||
import datetime
|
||||
import json
|
||||
|
||||
from acme import challenges
|
||||
from acme.messages import errors, STATUS_VALID, ERROR_CODES
|
||||
from flask import current_app
|
||||
|
||||
from lemur.authorizations import service as authorization_service
|
||||
from lemur.exceptions import LemurException, InvalidConfiguration
|
||||
from lemur.plugins.base import plugins
|
||||
from lemur.destinations import service as destination_service
|
||||
from lemur.plugins.lemur_acme.acme_handlers import AcmeHandler, AcmeDnsHandler
|
||||
|
||||
|
||||
class AcmeChallengeMissmatchError(LemurException):
|
||||
pass
|
||||
|
||||
|
||||
class AcmeChallenge(object):
|
||||
"""
|
||||
This is the base class, all ACME challenges will need to extend, allowing for future extendability
|
||||
"""
|
||||
|
||||
def create_certificate(self, csr, issuer_options):
|
||||
"""
|
||||
Create the new certificate, using the provided CSR and issuer_options.
|
||||
Right now this is basically a copy of the create_certificate methods in the AcmeHandlers, but should be cleaned
|
||||
and tried to make use of the deploy and cleanup methods
|
||||
|
||||
:param csr:
|
||||
:param issuer_options:
|
||||
:return:
|
||||
"""
|
||||
pass
|
||||
|
||||
def deploy(self, challenge, acme_client, validation_target):
|
||||
"""
|
||||
In here the challenge validation is fetched and deployed somewhere that it can be validated by the provider
|
||||
|
||||
:param self:
|
||||
:param challenge: the challenge object, must match for the challenge implementation
|
||||
:param acme_client: an already bootstrapped acme_client, to avoid passing all issuer_options and so on
|
||||
:param validation_target: an identifier for the validation target, e.g. the name of a DNS provider
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def cleanup(self, challenge, acme_client, validation_target):
|
||||
"""
|
||||
Ideally the challenge should be cleaned up, after the validation is done
|
||||
:param challenge: Needed to identify the challenge to be removed
|
||||
:param acme_client: an already bootstrapped acme_client, to avoid passing all issuer_options and so on
|
||||
:param validation_target: Needed to remove the validation
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class AcmeHttpChallenge(AcmeChallenge):
|
||||
challengeType = challenges.HTTP01
|
||||
|
||||
def create_certificate(self, csr, issuer_options):
|
||||
"""
|
||||
Creates an ACME certificate using the HTTP-01 challenge.
|
||||
|
||||
:param csr:
|
||||
:param issuer_options:
|
||||
:return: :raise Exception:
|
||||
"""
|
||||
self.acme = AcmeHandler()
|
||||
authority = issuer_options.get("authority")
|
||||
acme_client, registration = self.acme.setup_acme_client(authority)
|
||||
|
||||
orderr = acme_client.new_order(csr)
|
||||
|
||||
chall = []
|
||||
deployed_challenges = []
|
||||
all_pre_validated = True
|
||||
for authz in orderr.authorizations:
|
||||
# Choosing challenge.
|
||||
# check if authorizations is already in a valid state
|
||||
if authz.body.status != STATUS_VALID:
|
||||
all_pre_validated = False
|
||||
# authz.body.challenges is a set of ChallengeBody objects.
|
||||
for i in authz.body.challenges:
|
||||
# Find the supported challenge.
|
||||
if isinstance(i.chall, challenges.HTTP01):
|
||||
chall.append(i)
|
||||
else:
|
||||
current_app.logger.info("{} already validated, skipping".format(authz.body.identifier.value))
|
||||
|
||||
if len(chall) == 0 and not all_pre_validated:
|
||||
raise Exception('HTTP-01 challenge was not offered by the CA server at {}'.format(orderr.uri))
|
||||
elif not all_pre_validated:
|
||||
validation_target = None
|
||||
for option in json.loads(issuer_options["authority"].options):
|
||||
if option["name"] == "tokenDestination":
|
||||
validation_target = option["value"]
|
||||
|
||||
if validation_target is None:
|
||||
raise Exception('No token_destination configured for this authority. Cant complete HTTP-01 challenge')
|
||||
|
||||
for challenge in chall:
|
||||
try:
|
||||
response = self.deploy(challenge, acme_client, validation_target)
|
||||
deployed_challenges.append(challenge.chall.path)
|
||||
acme_client.answer_challenge(challenge, response)
|
||||
except Exception as e:
|
||||
current_app.logger.error(e)
|
||||
raise Exception('Failure while trying to deploy token to configure destination. See logs for more information')
|
||||
|
||||
current_app.logger.info("Uploaded HTTP-01 challenge tokens, trying to poll and finalize the order")
|
||||
|
||||
try:
|
||||
finalized_orderr = acme_client.poll_and_finalize(orderr,
|
||||
datetime.datetime.now() + datetime.timedelta(seconds=90))
|
||||
except errors.ValidationError as validationError:
|
||||
for authz in validationError.failed_authzrs:
|
||||
for chall in authz.body.challenges:
|
||||
if chall.error:
|
||||
current_app.logger.error(
|
||||
"ValidationError occured of type {}, with message {}".format(chall.error.typ,
|
||||
ERROR_CODES[chall.error.code]))
|
||||
raise Exception('Validation error occured, can\'t complete challenges. See logs for more information.')
|
||||
|
||||
pem_certificate, pem_certificate_chain = self.acme.extract_cert_and_chain(finalized_orderr.fullchain_pem)
|
||||
|
||||
if len(deployed_challenges) != 0:
|
||||
for token_path in deployed_challenges:
|
||||
self.cleanup(token_path, validation_target)
|
||||
|
||||
# validation is a random string, we use it as external id, to make it possible to implement revoke_certificate
|
||||
return pem_certificate, pem_certificate_chain, None
|
||||
|
||||
def deploy(self, challenge, acme_client, validation_target):
|
||||
|
||||
if not isinstance(challenge.chall, challenges.HTTP01):
|
||||
raise AcmeChallengeMissmatchError(
|
||||
'The provided challenge is not of type HTTP01, but instead of type {}'.format(
|
||||
challenge.__class__.__name__))
|
||||
|
||||
destination = destination_service.get(validation_target)
|
||||
|
||||
if destination is None:
|
||||
raise Exception(
|
||||
'Couldn\'t find the destination with name {}. Cant complete HTTP01 challenge'.format(validation_target))
|
||||
|
||||
destination_plugin = plugins.get(destination.plugin_name)
|
||||
|
||||
response, validation = challenge.response_and_validation(acme_client.net.key)
|
||||
|
||||
destination_plugin.upload_acme_token(challenge.chall.path, validation, destination.options)
|
||||
current_app.logger.info("Uploaded HTTP-01 challenge token.")
|
||||
|
||||
return response
|
||||
|
||||
def cleanup(self, token_path, validation_target):
|
||||
destination = destination_service.get(validation_target)
|
||||
|
||||
if destination is None:
|
||||
current_app.logger.info(
|
||||
'Couldn\'t find the destination with name {}, won\'t cleanup the challenge'.format(validation_target))
|
||||
|
||||
destination_plugin = plugins.get(destination.plugin_name)
|
||||
|
||||
destination_plugin.delete_acme_token(token_path, destination.options)
|
||||
current_app.logger.info("Cleaned up HTTP-01 challenge token.")
|
||||
|
||||
|
||||
class AcmeDnsChallenge(AcmeChallenge):
|
||||
challengeType = challenges.DNS01
|
||||
|
||||
def create_certificate(self, csr, issuer_options):
|
||||
"""
|
||||
Creates an ACME certificate.
|
||||
|
||||
:param csr:
|
||||
:param issuer_options:
|
||||
:return: :raise Exception:
|
||||
"""
|
||||
self.acme = AcmeDnsHandler()
|
||||
authority = issuer_options.get("authority")
|
||||
create_immediately = issuer_options.get("create_immediately", False)
|
||||
acme_client, registration = self.acme.setup_acme_client(authority)
|
||||
dns_provider = issuer_options.get("dns_provider", {})
|
||||
|
||||
if dns_provider:
|
||||
dns_provider_options = dns_provider.options
|
||||
credentials = json.loads(dns_provider.credentials)
|
||||
current_app.logger.debug(
|
||||
"Using DNS provider: {0}".format(dns_provider.provider_type)
|
||||
)
|
||||
dns_provider_plugin = __import__(
|
||||
dns_provider.provider_type, globals(), locals(), [], 1
|
||||
)
|
||||
account_number = credentials.get("account_id")
|
||||
provider_type = dns_provider.provider_type
|
||||
if provider_type == "route53" and not account_number:
|
||||
error = "Route53 DNS Provider {} does not have an account number configured.".format(
|
||||
dns_provider.name
|
||||
)
|
||||
current_app.logger.error(error)
|
||||
raise InvalidConfiguration(error)
|
||||
else:
|
||||
dns_provider = {}
|
||||
dns_provider_options = None
|
||||
account_number = None
|
||||
provider_type = None
|
||||
|
||||
domains = self.acme.get_domains(issuer_options)
|
||||
if not create_immediately:
|
||||
# Create pending authorizations that we'll need to do the creation
|
||||
dns_authorization = authorization_service.create(
|
||||
account_number, domains, provider_type
|
||||
)
|
||||
# Return id of the DNS Authorization
|
||||
return None, None, dns_authorization.id
|
||||
|
||||
authorizations = self.acme.get_authorizations(
|
||||
acme_client,
|
||||
account_number,
|
||||
domains,
|
||||
dns_provider_plugin,
|
||||
dns_provider_options,
|
||||
)
|
||||
self.acme.finalize_authorizations(
|
||||
acme_client,
|
||||
account_number,
|
||||
dns_provider_plugin,
|
||||
authorizations,
|
||||
dns_provider_options,
|
||||
)
|
||||
pem_certificate, pem_certificate_chain = self.acme.request_certificate(
|
||||
acme_client, authorizations, csr
|
||||
)
|
||||
# TODO add external ID (if possible)
|
||||
return pem_certificate, pem_certificate_chain, None
|
||||
|
||||
def deploy(self, challenge, acme_client, validation_target):
|
||||
pass
|
||||
|
||||
def cleanup(self, authorizations, acme_client, validation_target):
|
||||
"""
|
||||
Best effort attempt to delete DNS challenges that may not have been deleted previously. This is usually called
|
||||
on an exception
|
||||
|
||||
:param authorizations: all the authorizations to be cleaned up
|
||||
:param acme_client: an already bootstrapped acme_client, to avoid passing all issuer_options and so on
|
||||
:param validation_target: Unused right now
|
||||
:return:
|
||||
"""
|
||||
acme = AcmeDnsHandler()
|
||||
acme.cleanup_dns_challenges(acme_client, authorizations)
|
||||
|
|
@ -11,465 +11,27 @@
|
|||
.. moduleauthor:: Mikhail Khodorovskiy <mikhail.khodorovskiy@jivesoftware.com>
|
||||
.. moduleauthor:: Curtis Castrapel <ccastrapel@netflix.com>
|
||||
"""
|
||||
import datetime
|
||||
import json
|
||||
import time
|
||||
|
||||
import OpenSSL.crypto
|
||||
import dns.resolver
|
||||
import josepy as jose
|
||||
from acme import challenges, errors, messages
|
||||
from acme.client import BackwardsCompatibleClientV2, ClientNetwork
|
||||
from acme.errors import PollError, TimeoutError, WildcardUnsupportedError
|
||||
from acme.errors import PollError, WildcardUnsupportedError
|
||||
from acme.messages import Error as AcmeError
|
||||
from botocore.exceptions import ClientError
|
||||
from flask import current_app
|
||||
from lemur.authorizations import service as authorization_service
|
||||
from lemur.common.utils import generate_private_key
|
||||
from lemur.dns_providers import service as dns_provider_service
|
||||
from lemur.exceptions import InvalidAuthority, InvalidConfiguration, UnknownProvider
|
||||
from lemur.exceptions import InvalidConfiguration
|
||||
from lemur.extensions import metrics, sentry
|
||||
|
||||
from lemur.plugins import lemur_acme as acme
|
||||
from lemur.plugins.bases import IssuerPlugin
|
||||
from lemur.plugins.lemur_acme import cloudflare, dyn, route53, ultradns, powerdns
|
||||
from lemur.authorities import service as authorities_service
|
||||
from retrying import retry
|
||||
|
||||
|
||||
class AuthorizationRecord(object):
|
||||
def __init__(self, domain, target_domain, authz, dns_challenge, change_id):
|
||||
self.domain = domain
|
||||
self.target_domain = target_domain
|
||||
self.authz = authz
|
||||
self.dns_challenge = dns_challenge
|
||||
self.change_id = change_id
|
||||
|
||||
|
||||
class AcmeHandler(object):
|
||||
def __init__(self):
|
||||
self.dns_providers_for_domain = {}
|
||||
try:
|
||||
self.all_dns_providers = dns_provider_service.get_all_dns_providers()
|
||||
except Exception as e:
|
||||
metrics.send("AcmeHandler_init_error", "counter", 1)
|
||||
sentry.captureException()
|
||||
current_app.logger.error(f"Unable to fetch DNS Providers: {e}")
|
||||
self.all_dns_providers = []
|
||||
|
||||
def get_dns_challenges(self, host, authorizations):
|
||||
"""Get dns challenges for provided domain"""
|
||||
|
||||
domain_to_validate, is_wildcard = self.strip_wildcard(host)
|
||||
dns_challenges = []
|
||||
for authz in authorizations:
|
||||
if not authz.body.identifier.value.lower() == domain_to_validate.lower():
|
||||
continue
|
||||
if is_wildcard and not authz.body.wildcard:
|
||||
continue
|
||||
if not is_wildcard and authz.body.wildcard:
|
||||
continue
|
||||
for combo in authz.body.challenges:
|
||||
if isinstance(combo.chall, challenges.DNS01):
|
||||
dns_challenges.append(combo)
|
||||
|
||||
return dns_challenges
|
||||
|
||||
def strip_wildcard(self, host):
|
||||
"""Removes the leading *. and returns Host and whether it was removed or not (True/False)"""
|
||||
prefix = "*."
|
||||
if host.startswith(prefix):
|
||||
return host[len(prefix):], True
|
||||
return host, False
|
||||
|
||||
def maybe_add_extension(self, host, dns_provider_options):
|
||||
if dns_provider_options and dns_provider_options.get(
|
||||
"acme_challenge_extension"
|
||||
):
|
||||
host = host + dns_provider_options.get("acme_challenge_extension")
|
||||
return host
|
||||
|
||||
def start_dns_challenge(
|
||||
self,
|
||||
acme_client,
|
||||
account_number,
|
||||
domain,
|
||||
target_domain,
|
||||
dns_provider,
|
||||
order,
|
||||
dns_provider_options,
|
||||
):
|
||||
current_app.logger.debug(f"Starting DNS challenge for {domain} using target domain {target_domain}.")
|
||||
|
||||
change_ids = []
|
||||
dns_challenges = self.get_dns_challenges(domain, order.authorizations)
|
||||
host_to_validate, _ = self.strip_wildcard(target_domain)
|
||||
host_to_validate = self.maybe_add_extension(host_to_validate, dns_provider_options)
|
||||
|
||||
if not dns_challenges:
|
||||
sentry.captureException()
|
||||
metrics.send("start_dns_challenge_error_no_dns_challenges", "counter", 1)
|
||||
raise Exception("Unable to determine DNS challenges from authorizations")
|
||||
|
||||
for dns_challenge in dns_challenges:
|
||||
|
||||
# Only prepend '_acme-challenge' if not using CNAME redirection
|
||||
if domain == target_domain:
|
||||
host_to_validate = dns_challenge.validation_domain_name(host_to_validate)
|
||||
|
||||
change_id = dns_provider.create_txt_record(
|
||||
host_to_validate,
|
||||
dns_challenge.validation(acme_client.client.net.key),
|
||||
account_number,
|
||||
)
|
||||
change_ids.append(change_id)
|
||||
|
||||
return AuthorizationRecord(
|
||||
domain, target_domain, order.authorizations, dns_challenges, change_ids
|
||||
)
|
||||
|
||||
def complete_dns_challenge(self, acme_client, authz_record):
|
||||
current_app.logger.debug(
|
||||
"Finalizing DNS challenge for {0}".format(
|
||||
authz_record.authz[0].body.identifier.value
|
||||
)
|
||||
)
|
||||
dns_providers = self.dns_providers_for_domain.get(authz_record.target_domain)
|
||||
if not dns_providers:
|
||||
metrics.send("complete_dns_challenge_error_no_dnsproviders", "counter", 1)
|
||||
raise Exception(
|
||||
"No DNS providers found for domain: {}".format(authz_record.target_domain)
|
||||
)
|
||||
|
||||
for dns_provider in dns_providers:
|
||||
# Grab account number (For Route53)
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
dns_provider_plugin = self.get_dns_provider(dns_provider.provider_type)
|
||||
for change_id in authz_record.change_id:
|
||||
try:
|
||||
dns_provider_plugin.wait_for_dns_change(
|
||||
change_id, account_number=account_number
|
||||
)
|
||||
except Exception:
|
||||
metrics.send("complete_dns_challenge_error", "counter", 1)
|
||||
sentry.captureException()
|
||||
current_app.logger.debug(
|
||||
f"Unable to resolve DNS challenge for change_id: {change_id}, account_id: "
|
||||
f"{account_number}",
|
||||
exc_info=True,
|
||||
)
|
||||
raise
|
||||
|
||||
for dns_challenge in authz_record.dns_challenge:
|
||||
response = dns_challenge.response(acme_client.client.net.key)
|
||||
|
||||
verified = response.simple_verify(
|
||||
dns_challenge.chall,
|
||||
authz_record.target_domain,
|
||||
acme_client.client.net.key.public_key(),
|
||||
)
|
||||
|
||||
if not verified:
|
||||
metrics.send("complete_dns_challenge_verification_error", "counter", 1)
|
||||
raise ValueError("Failed verification")
|
||||
|
||||
time.sleep(5)
|
||||
res = acme_client.answer_challenge(dns_challenge, response)
|
||||
current_app.logger.debug(f"answer_challenge response: {res}")
|
||||
|
||||
def request_certificate(self, acme_client, authorizations, order):
|
||||
for authorization in authorizations:
|
||||
for authz in authorization.authz:
|
||||
authorization_resource, _ = acme_client.poll(authz)
|
||||
|
||||
deadline = datetime.datetime.now() + datetime.timedelta(seconds=360)
|
||||
|
||||
try:
|
||||
orderr = acme_client.poll_and_finalize(order, deadline)
|
||||
|
||||
except (AcmeError, TimeoutError):
|
||||
sentry.captureException(extra={"order_url": str(order.uri)})
|
||||
metrics.send("request_certificate_error", "counter", 1, metric_tags={"uri": order.uri})
|
||||
current_app.logger.error(
|
||||
f"Unable to resolve Acme order: {order.uri}", exc_info=True
|
||||
)
|
||||
raise
|
||||
except errors.ValidationError:
|
||||
if order.fullchain_pem:
|
||||
orderr = order
|
||||
else:
|
||||
raise
|
||||
|
||||
metrics.send("request_certificate_success", "counter", 1, metric_tags={"uri": order.uri})
|
||||
current_app.logger.info(
|
||||
f"Successfully resolved Acme order: {order.uri}", exc_info=True
|
||||
)
|
||||
|
||||
pem_certificate = OpenSSL.crypto.dump_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM,
|
||||
OpenSSL.crypto.load_certificate(
|
||||
OpenSSL.crypto.FILETYPE_PEM, orderr.fullchain_pem
|
||||
),
|
||||
).decode()
|
||||
|
||||
if current_app.config.get("IDENTRUST_CROSS_SIGNED_LE_ICA", False) \
|
||||
and datetime.datetime.now() < datetime.datetime.strptime(
|
||||
current_app.config.get("IDENTRUST_CROSS_SIGNED_LE_ICA_EXPIRATION_DATE", "17/03/21"), '%d/%m/%y'):
|
||||
pem_certificate_chain = current_app.config.get("IDENTRUST_CROSS_SIGNED_LE_ICA")
|
||||
else:
|
||||
pem_certificate_chain = orderr.fullchain_pem[
|
||||
len(pem_certificate) : # noqa
|
||||
].lstrip()
|
||||
|
||||
current_app.logger.debug(
|
||||
"{0} {1}".format(type(pem_certificate), type(pem_certificate_chain))
|
||||
)
|
||||
return pem_certificate, pem_certificate_chain
|
||||
|
||||
@retry(stop_max_attempt_number=5, wait_fixed=5000)
|
||||
def setup_acme_client(self, authority):
|
||||
if not authority.options:
|
||||
raise InvalidAuthority("Invalid authority. Options not set")
|
||||
options = {}
|
||||
|
||||
for option in json.loads(authority.options):
|
||||
options[option["name"]] = option.get("value")
|
||||
email = options.get("email", current_app.config.get("ACME_EMAIL"))
|
||||
tel = options.get("telephone", current_app.config.get("ACME_TEL"))
|
||||
directory_url = options.get(
|
||||
"acme_url", current_app.config.get("ACME_DIRECTORY_URL")
|
||||
)
|
||||
|
||||
existing_key = options.get(
|
||||
"acme_private_key", current_app.config.get("ACME_PRIVATE_KEY")
|
||||
)
|
||||
existing_regr = options.get("acme_regr", current_app.config.get("ACME_REGR"))
|
||||
|
||||
if existing_key and existing_regr:
|
||||
current_app.logger.debug("Reusing existing ACME account")
|
||||
# Reuse the same account for each certificate issuance
|
||||
key = jose.JWK.json_loads(existing_key)
|
||||
regr = messages.RegistrationResource.json_loads(existing_regr)
|
||||
current_app.logger.debug(
|
||||
"Connecting with directory at {0}".format(directory_url)
|
||||
)
|
||||
net = ClientNetwork(key, account=regr)
|
||||
client = BackwardsCompatibleClientV2(net, key, directory_url)
|
||||
return client, {}
|
||||
else:
|
||||
# Create an account for each certificate issuance
|
||||
key = jose.JWKRSA(key=generate_private_key("RSA2048"))
|
||||
|
||||
current_app.logger.debug("Creating a new ACME account")
|
||||
current_app.logger.debug(
|
||||
"Connecting with directory at {0}".format(directory_url)
|
||||
)
|
||||
|
||||
net = ClientNetwork(key, account=None, timeout=3600)
|
||||
client = BackwardsCompatibleClientV2(net, key, directory_url)
|
||||
registration = client.new_account_and_tos(
|
||||
messages.NewRegistration.from_data(email=email)
|
||||
)
|
||||
|
||||
# if store_account is checked, add the private_key and registration resources to the options
|
||||
if options['store_account']:
|
||||
new_options = json.loads(authority.options)
|
||||
# the key returned by fields_to_partial_json is missing the key type, so we add it manually
|
||||
key_dict = key.fields_to_partial_json()
|
||||
key_dict["kty"] = "RSA"
|
||||
acme_private_key = {
|
||||
"name": "acme_private_key",
|
||||
"value": json.dumps(key_dict)
|
||||
}
|
||||
new_options.append(acme_private_key)
|
||||
|
||||
acme_regr = {
|
||||
"name": "acme_regr",
|
||||
"value": json.dumps({"body": {}, "uri": registration.uri})
|
||||
}
|
||||
new_options.append(acme_regr)
|
||||
|
||||
authorities_service.update_options(authority.id, options=json.dumps(new_options))
|
||||
|
||||
current_app.logger.debug("Connected: {0}".format(registration.uri))
|
||||
|
||||
return client, registration
|
||||
|
||||
def get_domains(self, options):
|
||||
"""
|
||||
Fetches all domains currently requested
|
||||
:param options:
|
||||
:return:
|
||||
"""
|
||||
current_app.logger.debug("Fetching domains")
|
||||
|
||||
domains = [options["common_name"]]
|
||||
if options.get("extensions"):
|
||||
for dns_name in options["extensions"]["sub_alt_names"]["names"]:
|
||||
if dns_name.value not in domains:
|
||||
domains.append(dns_name.value)
|
||||
|
||||
current_app.logger.debug("Got these domains: {0}".format(domains))
|
||||
return domains
|
||||
|
||||
def get_authorizations(self, acme_client, order, order_info):
|
||||
authorizations = []
|
||||
|
||||
for domain in order_info.domains:
|
||||
|
||||
# If CNAME exists, set host to the target address
|
||||
target_domain = domain
|
||||
if current_app.config.get("ACME_ENABLE_DELEGATED_CNAME", False):
|
||||
cname_result, _ = self.strip_wildcard(domain)
|
||||
cname_result = challenges.DNS01().validation_domain_name(cname_result)
|
||||
cname_result = self.get_cname(cname_result)
|
||||
if cname_result:
|
||||
target_domain = cname_result
|
||||
self.autodetect_dns_providers(target_domain)
|
||||
|
||||
if not self.dns_providers_for_domain.get(target_domain):
|
||||
metrics.send(
|
||||
"get_authorizations_no_dns_provider_for_domain", "counter", 1
|
||||
)
|
||||
raise Exception("No DNS providers found for domain: {}".format(target_domain))
|
||||
|
||||
for dns_provider in self.dns_providers_for_domain[target_domain]:
|
||||
dns_provider_plugin = self.get_dns_provider(dns_provider.provider_type)
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
authz_record = self.start_dns_challenge(
|
||||
acme_client,
|
||||
account_number,
|
||||
domain,
|
||||
target_domain,
|
||||
dns_provider_plugin,
|
||||
order,
|
||||
dns_provider.options,
|
||||
)
|
||||
authorizations.append(authz_record)
|
||||
return authorizations
|
||||
|
||||
def autodetect_dns_providers(self, domain):
|
||||
"""
|
||||
Get DNS providers associated with a domain when it has not been provided for certificate creation.
|
||||
:param domain:
|
||||
:return: dns_providers: List of DNS providers that have the correct zone.
|
||||
"""
|
||||
self.dns_providers_for_domain[domain] = []
|
||||
match_length = 0
|
||||
for dns_provider in self.all_dns_providers:
|
||||
if not dns_provider.domains:
|
||||
continue
|
||||
for name in dns_provider.domains:
|
||||
if name == domain or domain.endswith("." + name):
|
||||
if len(name) > match_length:
|
||||
self.dns_providers_for_domain[domain] = [dns_provider]
|
||||
match_length = len(name)
|
||||
elif len(name) == match_length:
|
||||
self.dns_providers_for_domain[domain].append(dns_provider)
|
||||
|
||||
return self.dns_providers_for_domain
|
||||
|
||||
def finalize_authorizations(self, acme_client, authorizations):
|
||||
for authz_record in authorizations:
|
||||
self.complete_dns_challenge(acme_client, authz_record)
|
||||
for authz_record in authorizations:
|
||||
dns_challenges = authz_record.dns_challenge
|
||||
for dns_challenge in dns_challenges:
|
||||
dns_providers = self.dns_providers_for_domain.get(authz_record.target_domain)
|
||||
for dns_provider in dns_providers:
|
||||
# Grab account number (For Route53)
|
||||
dns_provider_plugin = self.get_dns_provider(
|
||||
dns_provider.provider_type
|
||||
)
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
host_to_validate, _ = self.strip_wildcard(authz_record.target_domain)
|
||||
host_to_validate = self.maybe_add_extension(host_to_validate, dns_provider_options)
|
||||
if authz_record.domain == authz_record.target_domain:
|
||||
host_to_validate = challenges.DNS01().validation_domain_name(host_to_validate)
|
||||
dns_provider_plugin.delete_txt_record(
|
||||
authz_record.change_id,
|
||||
account_number,
|
||||
host_to_validate,
|
||||
dns_challenge.validation(acme_client.client.net.key),
|
||||
)
|
||||
|
||||
return authorizations
|
||||
|
||||
def cleanup_dns_challenges(self, acme_client, authorizations):
|
||||
"""
|
||||
Best effort attempt to delete DNS challenges that may not have been deleted previously. This is usually called
|
||||
on an exception
|
||||
|
||||
:param acme_client:
|
||||
:param account_number:
|
||||
:param dns_provider:
|
||||
:param authorizations:
|
||||
:param dns_provider_options:
|
||||
:return:
|
||||
"""
|
||||
for authz_record in authorizations:
|
||||
dns_providers = self.dns_providers_for_domain.get(authz_record.target_domain)
|
||||
for dns_provider in dns_providers:
|
||||
# Grab account number (For Route53)
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
dns_challenges = authz_record.dns_challenge
|
||||
host_to_validate, _ = self.strip_wildcard(authz_record.target_domain)
|
||||
host_to_validate = self.maybe_add_extension(
|
||||
host_to_validate, dns_provider_options
|
||||
)
|
||||
|
||||
dns_provider_plugin = self.get_dns_provider(dns_provider.provider_type)
|
||||
for dns_challenge in dns_challenges:
|
||||
if authz_record.domain == authz_record.target_domain:
|
||||
host_to_validate = dns_challenge.validation_domain_name(host_to_validate)
|
||||
try:
|
||||
dns_provider_plugin.delete_txt_record(
|
||||
authz_record.change_id,
|
||||
account_number,
|
||||
host_to_validate,
|
||||
dns_challenge.validation(acme_client.client.net.key),
|
||||
)
|
||||
except Exception as e:
|
||||
# If this fails, it's most likely because the record doesn't exist (It was already cleaned up)
|
||||
# or we're not authorized to modify it.
|
||||
metrics.send("cleanup_dns_challenges_error", "counter", 1)
|
||||
sentry.captureException()
|
||||
pass
|
||||
|
||||
def get_dns_provider(self, type):
|
||||
provider_types = {
|
||||
"cloudflare": cloudflare,
|
||||
"dyn": dyn,
|
||||
"route53": route53,
|
||||
"ultradns": ultradns,
|
||||
"powerdns": powerdns
|
||||
}
|
||||
provider = provider_types.get(type)
|
||||
if not provider:
|
||||
raise UnknownProvider("No such DNS provider: {}".format(type))
|
||||
return provider
|
||||
|
||||
def get_cname(self, domain):
|
||||
"""
|
||||
:param domain: Domain name to look up a CNAME for.
|
||||
:return: First CNAME target or False if no CNAME record exists.
|
||||
"""
|
||||
try:
|
||||
result = dns.resolver.query(domain, 'CNAME')
|
||||
if len(result) > 0:
|
||||
return str(result[0].target).rstrip('.')
|
||||
except dns.exception.DNSException:
|
||||
return False
|
||||
from lemur.plugins.lemur_acme.acme_handlers import AcmeHandler, AcmeDnsHandler
|
||||
from lemur.plugins.lemur_acme.challenge_types import AcmeHttpChallenge, AcmeDnsChallenge
|
||||
|
||||
|
||||
class ACMEIssuerPlugin(IssuerPlugin):
|
||||
title = "Acme"
|
||||
slug = "acme-issuer"
|
||||
description = (
|
||||
"Enables the creation of certificates via ACME CAs (including Let's Encrypt)"
|
||||
"Enables the creation of certificates via ACME CAs (including Let's Encrypt), using the DNS-01 challenge"
|
||||
)
|
||||
version = acme.VERSION
|
||||
|
||||
|
|
@ -516,30 +78,8 @@ class ACMEIssuerPlugin(IssuerPlugin):
|
|||
def __init__(self, *args, **kwargs):
|
||||
super(ACMEIssuerPlugin, self).__init__(*args, **kwargs)
|
||||
|
||||
def get_dns_provider(self, type):
|
||||
self.acme = AcmeHandler()
|
||||
|
||||
provider_types = {
|
||||
"cloudflare": cloudflare,
|
||||
"dyn": dyn,
|
||||
"route53": route53,
|
||||
"ultradns": ultradns,
|
||||
"powerdns": powerdns
|
||||
}
|
||||
provider = provider_types.get(type)
|
||||
if not provider:
|
||||
raise UnknownProvider("No such DNS provider: {}".format(type))
|
||||
return provider
|
||||
|
||||
def get_all_zones(self, dns_provider):
|
||||
self.acme = AcmeHandler()
|
||||
dns_provider_options = json.loads(dns_provider.credentials)
|
||||
account_number = dns_provider_options.get("account_id")
|
||||
dns_provider_plugin = self.get_dns_provider(dns_provider.provider_type)
|
||||
return dns_provider_plugin.get_zones(account_number=account_number)
|
||||
|
||||
def get_ordered_certificate(self, pending_cert):
|
||||
self.acme = AcmeHandler()
|
||||
self.acme = AcmeDnsHandler()
|
||||
acme_client, registration = self.acme.setup_acme_client(pending_cert.authority)
|
||||
order_info = authorization_service.get(pending_cert.external_id)
|
||||
if pending_cert.dns_provider_id:
|
||||
|
|
@ -585,7 +125,8 @@ class ACMEIssuerPlugin(IssuerPlugin):
|
|||
return cert
|
||||
|
||||
def get_ordered_certificates(self, pending_certs):
|
||||
self.acme = AcmeHandler()
|
||||
self.acme = AcmeDnsHandler()
|
||||
self.acme_dns_challenge = AcmeDnsChallenge()
|
||||
pending = []
|
||||
certs = []
|
||||
for pending_cert in pending_certs:
|
||||
|
|
@ -682,76 +223,22 @@ class ACMEIssuerPlugin(IssuerPlugin):
|
|||
}
|
||||
)
|
||||
# Ensure DNS records get deleted
|
||||
self.acme.cleanup_dns_challenges(
|
||||
entry["acme_client"], entry["authorizations"]
|
||||
self.acme_dns_challenge.cleanup(
|
||||
entry["authorizations"], entry["acme_client"]
|
||||
)
|
||||
return certs
|
||||
|
||||
def create_certificate(self, csr, issuer_options):
|
||||
"""
|
||||
Creates an ACME certificate.
|
||||
Creates an ACME certificate using the DNS-01 challenge.
|
||||
|
||||
:param csr:
|
||||
:param issuer_options:
|
||||
:return: :raise Exception:
|
||||
"""
|
||||
self.acme = AcmeHandler()
|
||||
authority = issuer_options.get("authority")
|
||||
create_immediately = issuer_options.get("create_immediately", False)
|
||||
acme_client, registration = self.acme.setup_acme_client(authority)
|
||||
dns_provider = issuer_options.get("dns_provider", {})
|
||||
acme_dns_challenge = AcmeDnsChallenge()
|
||||
|
||||
if dns_provider:
|
||||
dns_provider_options = dns_provider.options
|
||||
credentials = json.loads(dns_provider.credentials)
|
||||
current_app.logger.debug(
|
||||
"Using DNS provider: {0}".format(dns_provider.provider_type)
|
||||
)
|
||||
dns_provider_plugin = __import__(
|
||||
dns_provider.provider_type, globals(), locals(), [], 1
|
||||
)
|
||||
account_number = credentials.get("account_id")
|
||||
provider_type = dns_provider.provider_type
|
||||
if provider_type == "route53" and not account_number:
|
||||
error = "Route53 DNS Provider {} does not have an account number configured.".format(
|
||||
dns_provider.name
|
||||
)
|
||||
current_app.logger.error(error)
|
||||
raise InvalidConfiguration(error)
|
||||
else:
|
||||
dns_provider = {}
|
||||
dns_provider_options = None
|
||||
account_number = None
|
||||
provider_type = None
|
||||
|
||||
domains = self.acme.get_domains(issuer_options)
|
||||
if not create_immediately:
|
||||
# Create pending authorizations that we'll need to do the creation
|
||||
dns_authorization = authorization_service.create(
|
||||
account_number, domains, provider_type
|
||||
)
|
||||
# Return id of the DNS Authorization
|
||||
return None, None, dns_authorization.id
|
||||
|
||||
authorizations = self.acme.get_authorizations(
|
||||
acme_client,
|
||||
account_number,
|
||||
domains,
|
||||
dns_provider_plugin,
|
||||
dns_provider_options,
|
||||
)
|
||||
self.acme.finalize_authorizations(
|
||||
acme_client,
|
||||
account_number,
|
||||
dns_provider_plugin,
|
||||
authorizations,
|
||||
dns_provider_options,
|
||||
)
|
||||
pem_certificate, pem_certificate_chain = self.acme.request_certificate(
|
||||
acme_client, authorizations, csr
|
||||
)
|
||||
# TODO add external ID (if possible)
|
||||
return pem_certificate, pem_certificate_chain, None
|
||||
return acme_dns_challenge.create_certificate(csr, issuer_options)
|
||||
|
||||
@staticmethod
|
||||
def create_authority(options):
|
||||
|
|
@ -779,3 +266,108 @@ class ACMEIssuerPlugin(IssuerPlugin):
|
|||
def cancel_ordered_certificate(self, pending_cert, **kwargs):
|
||||
# Needed to override issuer function.
|
||||
pass
|
||||
|
||||
def revoke_certificate(self, certificate, comments):
|
||||
self.acme = AcmeDnsHandler()
|
||||
return self.acme.revoke_certificate(certificate)
|
||||
|
||||
|
||||
class ACMEHttpIssuerPlugin(IssuerPlugin):
|
||||
title = "Acme HTTP-01"
|
||||
slug = "acme-http-issuer"
|
||||
description = (
|
||||
"Enables the creation of certificates via ACME CAs (including Let's Encrypt), using the HTTP-01 challenge"
|
||||
)
|
||||
version = acme.VERSION
|
||||
|
||||
author = "Netflix"
|
||||
author_url = "https://github.com/netflix/lemur.git"
|
||||
|
||||
options = [
|
||||
{
|
||||
"name": "acme_url",
|
||||
"type": "str",
|
||||
"required": True,
|
||||
"validation": r"/^http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+$/",
|
||||
"helpMessage": "Must be a valid web url starting with http[s]://",
|
||||
},
|
||||
{
|
||||
"name": "telephone",
|
||||
"type": "str",
|
||||
"default": "",
|
||||
"helpMessage": "Telephone to use",
|
||||
},
|
||||
{
|
||||
"name": "email",
|
||||
"type": "str",
|
||||
"default": "",
|
||||
"validation": r"/^?([-a-zA-Z0-9.`?{}]+@\w+\.\w+)$/",
|
||||
"helpMessage": "Email to use",
|
||||
},
|
||||
{
|
||||
"name": "certificate",
|
||||
"type": "textarea",
|
||||
"default": "",
|
||||
"validation": "/^-----BEGIN CERTIFICATE-----/",
|
||||
"helpMessage": "Certificate to use",
|
||||
},
|
||||
{
|
||||
"name": "store_account",
|
||||
"type": "bool",
|
||||
"required": False,
|
||||
"helpMessage": "Disable to create a new account for each ACME request",
|
||||
"default": False,
|
||||
},
|
||||
{
|
||||
"name": "tokenDestination",
|
||||
"type": "destinationSelect",
|
||||
"required": True,
|
||||
"helpMessage": "The destination to use to deploy the token.",
|
||||
},
|
||||
]
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(ACMEHttpIssuerPlugin, self).__init__(*args, **kwargs)
|
||||
|
||||
def create_certificate(self, csr, issuer_options):
|
||||
"""
|
||||
Creates an ACME certificate using the HTTP-01 challenge.
|
||||
|
||||
:param csr:
|
||||
:param issuer_options:
|
||||
:return: :raise Exception:
|
||||
"""
|
||||
acme_http_challenge = AcmeHttpChallenge()
|
||||
|
||||
return acme_http_challenge.create_certificate(csr, issuer_options)
|
||||
|
||||
@staticmethod
|
||||
def create_authority(options):
|
||||
"""
|
||||
Creates an authority, this authority is then used by Lemur to allow a user
|
||||
to specify which Certificate Authority they want to sign their certificate.
|
||||
|
||||
:param options:
|
||||
:return:
|
||||
"""
|
||||
role = {"username": "", "password": "", "name": "acme"}
|
||||
plugin_options = options.get("plugin", {}).get("plugin_options")
|
||||
if not plugin_options:
|
||||
error = "Invalid options for lemur_acme plugin: {}".format(options)
|
||||
current_app.logger.error(error)
|
||||
raise InvalidConfiguration(error)
|
||||
# Define static acme_root based off configuration variable by default. However, if user has passed a
|
||||
# certificate, use this certificate as the root.
|
||||
acme_root = current_app.config.get("ACME_ROOT")
|
||||
for option in plugin_options:
|
||||
if option.get("name") == "certificate":
|
||||
acme_root = option.get("value")
|
||||
return acme_root, "", [role]
|
||||
|
||||
def cancel_ordered_certificate(self, pending_cert, **kwargs):
|
||||
# Needed to override issuer function.
|
||||
pass
|
||||
|
||||
def revoke_certificate(self, certificate, comments):
|
||||
self.acme = AcmeHandler()
|
||||
return self.acme.revoke_certificate(certificate)
|
||||
|
|
|
|||
|
|
@ -5,15 +5,16 @@ import josepy as jose
|
|||
from cryptography.x509 import DNSName
|
||||
from flask import Flask
|
||||
from lemur.plugins.lemur_acme import plugin
|
||||
from lemur.plugins.lemur_acme.acme_handlers import AuthorizationRecord
|
||||
from lemur.common.utils import generate_private_key
|
||||
from mock import MagicMock
|
||||
|
||||
|
||||
class TestAcme(unittest.TestCase):
|
||||
@patch("lemur.plugins.lemur_acme.plugin.dns_provider_service")
|
||||
class TestAcmeDns(unittest.TestCase):
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.dns_provider_service")
|
||||
def setUp(self, mock_dns_provider_service):
|
||||
self.ACMEIssuerPlugin = plugin.ACMEIssuerPlugin()
|
||||
self.acme = plugin.AcmeHandler()
|
||||
self.acme = plugin.AcmeDnsHandler()
|
||||
mock_dns_provider = Mock()
|
||||
mock_dns_provider.name = "cloudflare"
|
||||
mock_dns_provider.credentials = "{}"
|
||||
|
|
@ -50,36 +51,19 @@ class TestAcme(unittest.TestCase):
|
|||
result = yield self.acme.get_dns_challenges(host, mock_authz)
|
||||
self.assertEqual(result, mock_entry)
|
||||
|
||||
def test_strip_wildcard(self):
|
||||
expected = ("example.com", False)
|
||||
result = self.acme.strip_wildcard("example.com")
|
||||
self.assertEqual(expected, result)
|
||||
|
||||
expected = ("example.com", True)
|
||||
result = self.acme.strip_wildcard("*.example.com")
|
||||
self.assertEqual(expected, result)
|
||||
|
||||
def test_authz_record(self):
|
||||
a = plugin.AuthorizationRecord("domain", "host", "authz", "challenge", "id")
|
||||
self.assertEqual(type(a), plugin.AuthorizationRecord)
|
||||
|
||||
@patch("acme.client.Client")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.len", return_value=1)
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.get_dns_challenges")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.get_dns_challenges")
|
||||
def test_start_dns_challenge(
|
||||
self, mock_get_dns_challenges, mock_len, mock_app, mock_acme
|
||||
self, mock_get_dns_challenges, mock_len, mock_acme
|
||||
):
|
||||
assert mock_len
|
||||
mock_order = Mock()
|
||||
mock_app.logger.debug = Mock()
|
||||
mock_authz = Mock()
|
||||
mock_authz.body.resolved_combinations = []
|
||||
mock_entry = MagicMock()
|
||||
from acme import challenges
|
||||
|
||||
c = challenges.DNS01()
|
||||
mock_entry.chall = TestAcme.test_complete_dns_challenge_fail
|
||||
mock_entry.chall = TestAcmeDns.test_complete_dns_challenge_fail
|
||||
mock_authz.body.resolved_combinations.append(mock_entry)
|
||||
mock_acme.request_domain_challenges = Mock(return_value=mock_authz)
|
||||
mock_dns_provider = Mock()
|
||||
|
|
@ -92,14 +76,13 @@ class TestAcme(unittest.TestCase):
|
|||
result = self.acme.start_dns_challenge(
|
||||
mock_acme, "accountid", "domain", "host", mock_dns_provider, mock_order, {}
|
||||
)
|
||||
self.assertEqual(type(result), plugin.AuthorizationRecord)
|
||||
self.assertEqual(type(result), AuthorizationRecord)
|
||||
|
||||
@patch("acme.client.Client")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.cloudflare.wait_for_dns_change")
|
||||
@patch("time.sleep")
|
||||
def test_complete_dns_challenge_success(
|
||||
self, mock_sleep, mock_wait_for_dns_change, mock_current_app, mock_acme
|
||||
self, mock_sleep, mock_wait_for_dns_change, mock_acme
|
||||
):
|
||||
mock_dns_provider = Mock()
|
||||
mock_dns_provider.wait_for_dns_change = Mock(return_value=True)
|
||||
|
|
@ -120,10 +103,9 @@ class TestAcme(unittest.TestCase):
|
|||
self.acme.complete_dns_challenge(mock_acme, mock_authz)
|
||||
|
||||
@patch("acme.client.Client")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.cloudflare.wait_for_dns_change")
|
||||
def test_complete_dns_challenge_fail(
|
||||
self, mock_wait_for_dns_change, mock_current_app, mock_acme
|
||||
self, mock_wait_for_dns_change, mock_acme
|
||||
):
|
||||
mock_dns_provider = Mock()
|
||||
mock_dns_provider.wait_for_dns_change = Mock(return_value=True)
|
||||
|
|
@ -150,11 +132,9 @@ class TestAcme(unittest.TestCase):
|
|||
@patch("acme.client.Client")
|
||||
@patch("OpenSSL.crypto", return_value="mock_cert")
|
||||
@patch("josepy.util.ComparableX509")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.get_dns_challenges")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.get_dns_challenges")
|
||||
def test_request_certificate(
|
||||
self,
|
||||
mock_current_app,
|
||||
mock_get_dns_challenges,
|
||||
mock_jose,
|
||||
mock_crypto,
|
||||
|
|
@ -171,7 +151,6 @@ class TestAcme(unittest.TestCase):
|
|||
mock_acme.fetch_chain = Mock(return_value="mock_chain")
|
||||
mock_crypto.dump_certificate = Mock(return_value=b"chain")
|
||||
mock_order = Mock()
|
||||
mock_current_app.config = {}
|
||||
self.acme.request_certificate(mock_acme, [], mock_order)
|
||||
|
||||
def test_setup_acme_client_fail(self):
|
||||
|
|
@ -180,10 +159,9 @@ class TestAcme(unittest.TestCase):
|
|||
with self.assertRaises(Exception):
|
||||
self.acme.setup_acme_client(mock_authority)
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.jose.JWK.json_loads")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.BackwardsCompatibleClientV2")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
def test_setup_acme_client_success_load_account_from_authority(self, mock_current_app, mock_acme, mock_key_json_load):
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.jose.JWK.json_loads")
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.BackwardsCompatibleClientV2")
|
||||
def test_setup_acme_client_success_load_account_from_authority(self, mock_acme, mock_key_json_load):
|
||||
mock_authority = Mock()
|
||||
mock_authority.id = 2
|
||||
mock_authority.options = '[{"name": "mock_name", "value": "mock_value"}, ' \
|
||||
|
|
@ -192,7 +170,6 @@ class TestAcme(unittest.TestCase):
|
|||
'{"name": "acme_regr", "value": "{\\"body\\": {}, \\"uri\\": \\"http://test.com\\"}"}]'
|
||||
mock_client = Mock()
|
||||
mock_acme.return_value = mock_client
|
||||
mock_current_app.config = {}
|
||||
|
||||
mock_key_json_load.return_value = jose.JWKRSA(key=generate_private_key("RSA2048"))
|
||||
|
||||
|
|
@ -202,11 +179,10 @@ class TestAcme(unittest.TestCase):
|
|||
assert result_client
|
||||
assert not result_registration
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.jose.JWKRSA.fields_to_partial_json")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.authorities_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.BackwardsCompatibleClientV2")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
def test_setup_acme_client_success_store_new_account(self, mock_current_app, mock_acme, mock_authorities_service,
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.jose.JWKRSA.fields_to_partial_json")
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.authorities_service")
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.BackwardsCompatibleClientV2")
|
||||
def test_setup_acme_client_success_store_new_account(self, mock_acme, mock_authorities_service,
|
||||
mock_key_generation):
|
||||
mock_authority = Mock()
|
||||
mock_authority.id = 2
|
||||
|
|
@ -219,7 +195,6 @@ class TestAcme(unittest.TestCase):
|
|||
mock_client.agree_to_tos = Mock(return_value=True)
|
||||
mock_client.new_account_and_tos.return_value = mock_registration
|
||||
mock_acme.return_value = mock_client
|
||||
mock_current_app.config = {}
|
||||
|
||||
mock_key_generation.return_value = {"n": "PwIOkViO"}
|
||||
|
||||
|
|
@ -232,10 +207,9 @@ class TestAcme(unittest.TestCase):
|
|||
'{"name": "acme_private_key", "value": "{\\"n\\": \\"PwIOkViO\\", \\"kty\\": \\"RSA\\"}"}, '
|
||||
'{"name": "acme_regr", "value": "{\\"body\\": {}, \\"uri\\": \\"http://test.com\\"}"}]')
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.authorities_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.BackwardsCompatibleClientV2")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
def test_setup_acme_client_success(self, mock_current_app, mock_acme, mock_authorities_service):
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.authorities_service")
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.BackwardsCompatibleClientV2")
|
||||
def test_setup_acme_client_success(self, mock_acme, mock_authorities_service):
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = '[{"name": "mock_name", "value": "mock_value"}, ' \
|
||||
'{"name": "store_account", "value": false}]'
|
||||
|
|
@ -245,20 +219,17 @@ class TestAcme(unittest.TestCase):
|
|||
mock_client.register = mock_registration
|
||||
mock_client.agree_to_tos = Mock(return_value=True)
|
||||
mock_acme.return_value = mock_client
|
||||
mock_current_app.config = {}
|
||||
result_client, result_registration = self.acme.setup_acme_client(mock_authority)
|
||||
mock_authorities_service.update_options.assert_not_called()
|
||||
assert result_client
|
||||
assert result_registration
|
||||
|
||||
@patch('lemur.plugins.lemur_acme.plugin.current_app')
|
||||
def test_get_domains_single(self, mock_current_app):
|
||||
def test_get_domains_single(self):
|
||||
options = {"common_name": "test.netflix.net"}
|
||||
result = self.acme.get_domains(options)
|
||||
self.assertEqual(result, [options["common_name"]])
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
def test_get_domains_multiple(self, mock_current_app):
|
||||
def test_get_domains_multiple(self):
|
||||
options = {
|
||||
"common_name": "test.netflix.net",
|
||||
"extensions": {
|
||||
|
|
@ -270,8 +241,7 @@ class TestAcme(unittest.TestCase):
|
|||
result, [options["common_name"], "test2.netflix.net", "test3.netflix.net"]
|
||||
)
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
def test_get_domains_san(self, mock_current_app):
|
||||
def test_get_domains_san(self):
|
||||
options = {
|
||||
"common_name": "test.netflix.net",
|
||||
"extensions": {
|
||||
|
|
@ -283,9 +253,63 @@ class TestAcme(unittest.TestCase):
|
|||
result, [options["common_name"], "test2.netflix.net"]
|
||||
)
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.start_dns_challenge", return_value="test")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app", return_value=False)
|
||||
def test_get_authorizations(self, mock_current_app, mock_start_dns_challenge):
|
||||
def test_create_authority(self):
|
||||
options = {
|
||||
"plugin": {"plugin_options": [{"name": "certificate", "value": "123"}]}
|
||||
}
|
||||
acme_root, b, role = self.ACMEIssuerPlugin.create_authority(options)
|
||||
self.assertEqual(acme_root, "123")
|
||||
self.assertEqual(b, "")
|
||||
self.assertEqual(role, [{"username": "", "password": "", "name": "acme"}])
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.dns_provider_service")
|
||||
def test_get_dns_provider(self, mock_dns_provider_service):
|
||||
provider = plugin.AcmeDnsHandler()
|
||||
route53 = provider.get_dns_provider("route53")
|
||||
assert route53
|
||||
cloudflare = provider.get_dns_provider("cloudflare")
|
||||
assert cloudflare
|
||||
dyn = provider.get_dns_provider("dyn")
|
||||
assert dyn
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.setup_acme_client")
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.dns_provider_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.get_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.finalize_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.request_certificate")
|
||||
@patch("lemur.plugins.lemur_acme.challenge_types.authorization_service")
|
||||
def test_create_certificate(
|
||||
self,
|
||||
mock_authorization_service,
|
||||
mock_request_certificate,
|
||||
mock_finalize_authorizations,
|
||||
mock_get_authorizations,
|
||||
mock_dns_provider_service,
|
||||
mock_acme,
|
||||
):
|
||||
provider = plugin.ACMEIssuerPlugin()
|
||||
mock_authority = Mock()
|
||||
|
||||
mock_client = Mock()
|
||||
mock_acme.return_value = (mock_client, "")
|
||||
|
||||
mock_dns_provider = Mock()
|
||||
mock_dns_provider.credentials = '{"account_id": 1}'
|
||||
mock_dns_provider.provider_type = "route53"
|
||||
mock_dns_provider_service.get.return_value = mock_dns_provider
|
||||
|
||||
issuer_options = {
|
||||
"authority": mock_authority,
|
||||
"dns_provider": mock_dns_provider,
|
||||
"common_name": "test.netflix.net",
|
||||
}
|
||||
csr = "123"
|
||||
mock_request_certificate.return_value = ("pem_certificate", "chain")
|
||||
result = provider.create_certificate(csr, issuer_options)
|
||||
assert result
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.start_dns_challenge", return_value="test")
|
||||
def test_get_authorizations(self, mock_start_dns_challenge):
|
||||
mock_order = Mock()
|
||||
mock_order.body.identifiers = []
|
||||
mock_domain = Mock()
|
||||
|
|
@ -299,7 +323,7 @@ class TestAcme(unittest.TestCase):
|
|||
self.assertEqual(result, ["test"])
|
||||
|
||||
@patch(
|
||||
"lemur.plugins.lemur_acme.plugin.AcmeHandler.complete_dns_challenge",
|
||||
"lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.complete_dns_challenge",
|
||||
return_value="test",
|
||||
)
|
||||
def test_finalize_authorizations(self, mock_complete_dns_challenge):
|
||||
|
|
@ -317,51 +341,21 @@ class TestAcme(unittest.TestCase):
|
|||
result = self.acme.finalize_authorizations(mock_acme_client, mock_authz)
|
||||
self.assertEqual(result, mock_authz)
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
def test_create_authority(self, mock_current_app):
|
||||
mock_current_app.config = Mock()
|
||||
options = {
|
||||
"plugin": {"plugin_options": [{"name": "certificate", "value": "123"}]}
|
||||
}
|
||||
acme_root, b, role = self.ACMEIssuerPlugin.create_authority(options)
|
||||
self.assertEqual(acme_root, "123")
|
||||
self.assertEqual(b, "")
|
||||
self.assertEqual(role, [{"username": "", "password": "", "name": "acme"}])
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.dyn.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.cloudflare.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.dns_provider_service")
|
||||
def test_get_dns_provider(
|
||||
self,
|
||||
mock_dns_provider_service,
|
||||
mock_current_app_cloudflare,
|
||||
mock_current_app_dyn,
|
||||
mock_current_app,
|
||||
):
|
||||
provider = plugin.ACMEIssuerPlugin()
|
||||
route53 = provider.get_dns_provider("route53")
|
||||
assert route53
|
||||
cloudflare = provider.get_dns_provider("cloudflare")
|
||||
assert cloudflare
|
||||
dyn = provider.get_dns_provider("dyn")
|
||||
assert dyn
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.setup_acme_client")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.authorization_service")
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.dns_provider_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.dns_provider_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.get_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.finalize_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.request_certificate")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.get_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.finalize_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.request_certificate")
|
||||
def test_get_ordered_certificate(
|
||||
self,
|
||||
mock_request_certificate,
|
||||
mock_finalize_authorizations,
|
||||
mock_get_authorizations,
|
||||
mock_dns_provider_service_p,
|
||||
mock_dns_provider_service,
|
||||
mock_authorization_service,
|
||||
mock_current_app,
|
||||
mock_acme,
|
||||
):
|
||||
mock_client = Mock()
|
||||
|
|
@ -379,20 +373,20 @@ class TestAcme(unittest.TestCase):
|
|||
)
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.setup_acme_client")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.authorization_service")
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.dns_provider_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.dns_provider_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.get_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.finalize_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.request_certificate")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.get_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.finalize_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeDnsHandler.request_certificate")
|
||||
def test_get_ordered_certificates(
|
||||
self,
|
||||
mock_request_certificate,
|
||||
mock_finalize_authorizations,
|
||||
mock_get_authorizations,
|
||||
mock_dns_provider_service,
|
||||
mock_dns_provider_service_p,
|
||||
mock_authorization_service,
|
||||
mock_current_app,
|
||||
mock_acme,
|
||||
):
|
||||
mock_client = Mock()
|
||||
|
|
@ -417,41 +411,3 @@ class TestAcme(unittest.TestCase):
|
|||
result[1]["cert"],
|
||||
{"body": "pem_certificate", "chain": "chain", "external_id": "2"},
|
||||
)
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.setup_acme_client")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.dns_provider_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.current_app")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.get_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.finalize_authorizations")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.request_certificate")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.authorization_service")
|
||||
def test_create_certificate(
|
||||
self,
|
||||
mock_authorization_service,
|
||||
mock_request_certificate,
|
||||
mock_finalize_authorizations,
|
||||
mock_get_authorizations,
|
||||
mock_current_app,
|
||||
mock_dns_provider_service,
|
||||
mock_acme,
|
||||
):
|
||||
provider = plugin.ACMEIssuerPlugin()
|
||||
mock_authority = Mock()
|
||||
|
||||
mock_client = Mock()
|
||||
mock_acme.return_value = (mock_client, "")
|
||||
|
||||
mock_dns_provider = Mock()
|
||||
mock_dns_provider.credentials = '{"account_id": 1}'
|
||||
mock_dns_provider.provider_type = "route53"
|
||||
mock_dns_provider_service.get.return_value = mock_dns_provider
|
||||
|
||||
issuer_options = {
|
||||
"authority": mock_authority,
|
||||
"dns_provider": mock_dns_provider,
|
||||
"common_name": "test.netflix.net",
|
||||
}
|
||||
csr = "123"
|
||||
mock_request_certificate.return_value = ("pem_certificate", "chain")
|
||||
result = provider.create_certificate(csr, issuer_options)
|
||||
assert result
|
||||
|
|
@ -0,0 +1,112 @@
|
|||
import unittest
|
||||
from unittest.mock import patch, Mock
|
||||
|
||||
from flask import Flask
|
||||
from cryptography.x509 import DNSName
|
||||
from lemur.plugins.lemur_acme import acme_handlers
|
||||
|
||||
|
||||
class TestAcmeHandler(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.acme = acme_handlers.AcmeHandler()
|
||||
|
||||
# Creates a new Flask application for a test duration. In python 3.8, manual push of application context is
|
||||
# needed to run tests in dev environment without getting error 'Working outside of application context'.
|
||||
_app = Flask('lemur_test_acme')
|
||||
self.ctx = _app.app_context()
|
||||
assert self.ctx
|
||||
self.ctx.push()
|
||||
|
||||
def tearDown(self):
|
||||
self.ctx.pop()
|
||||
|
||||
def test_strip_wildcard(self):
|
||||
expected = ("example.com", False)
|
||||
result = self.acme.strip_wildcard("example.com")
|
||||
self.assertEqual(expected, result)
|
||||
|
||||
expected = ("example.com", True)
|
||||
result = self.acme.strip_wildcard("*.example.com")
|
||||
self.assertEqual(expected, result)
|
||||
|
||||
def test_authz_record(self):
|
||||
a = acme_handlers.AuthorizationRecord("domain", "host", "authz", "challenge", "id")
|
||||
self.assertEqual(type(a), acme_handlers.AuthorizationRecord)
|
||||
|
||||
def test_setup_acme_client_fail(self):
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = []
|
||||
with self.assertRaises(Exception):
|
||||
self.acme.setup_acme_client(mock_authority)
|
||||
|
||||
def test_reuse_account_not_defined(self):
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = []
|
||||
with self.assertRaises(Exception):
|
||||
self.acme.reuse_account(mock_authority)
|
||||
|
||||
def test_reuse_account_from_authority(self):
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = '[{"name": "acme_private_key", "value": "PRIVATE_KEY"}, {"name": "acme_regr", "value": "ACME_REGR"}]'
|
||||
|
||||
self.assertTrue(self.acme.reuse_account(mock_authority))
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.current_app")
|
||||
def test_reuse_account_from_config(self, mock_current_app):
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = '[{"name": "mock_name", "value": "mock_value"}]'
|
||||
mock_current_app.config = {"ACME_PRIVATE_KEY": "PRIVATE_KEY", "ACME_REGR": "ACME_REGR"}
|
||||
|
||||
self.assertTrue(self.acme.reuse_account(mock_authority))
|
||||
|
||||
def test_reuse_account_no_configuration(self):
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = '[{"name": "mock_name", "value": "mock_value"}]'
|
||||
|
||||
self.assertFalse(self.acme.reuse_account(mock_authority))
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.authorities_service")
|
||||
@patch("lemur.plugins.lemur_acme.acme_handlers.BackwardsCompatibleClientV2")
|
||||
def test_setup_acme_client_success(self, mock_acme, mock_authorities_service):
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = '[{"name": "mock_name", "value": "mock_value"}, ' \
|
||||
'{"name": "store_account", "value": false}]'
|
||||
mock_client = Mock()
|
||||
mock_registration = Mock()
|
||||
mock_registration.uri = "http://test.com"
|
||||
mock_client.register = mock_registration
|
||||
mock_client.agree_to_tos = Mock(return_value=True)
|
||||
mock_acme.return_value = mock_client
|
||||
result_client, result_registration = self.acme.setup_acme_client(mock_authority)
|
||||
mock_authorities_service.update_options.assert_not_called()
|
||||
assert result_client
|
||||
assert result_registration
|
||||
|
||||
def test_get_domains_single(self):
|
||||
options = {"common_name": "test.netflix.net"}
|
||||
result = self.acme.get_domains(options)
|
||||
self.assertEqual(result, [options["common_name"]])
|
||||
|
||||
def test_get_domains_multiple(self):
|
||||
options = {
|
||||
"common_name": "test.netflix.net",
|
||||
"extensions": {
|
||||
"sub_alt_names": {"names": [DNSName("test2.netflix.net"), DNSName("test3.netflix.net")]}
|
||||
},
|
||||
}
|
||||
result = self.acme.get_domains(options)
|
||||
self.assertEqual(
|
||||
result, [options["common_name"], "test2.netflix.net", "test3.netflix.net"]
|
||||
)
|
||||
|
||||
def test_get_domains_san(self):
|
||||
options = {
|
||||
"common_name": "test.netflix.net",
|
||||
"extensions": {
|
||||
"sub_alt_names": {"names": [DNSName("test.netflix.net"), DNSName("test2.netflix.net")]}
|
||||
},
|
||||
}
|
||||
result = self.acme.get_domains(options)
|
||||
self.assertEqual(
|
||||
result, [options["common_name"], "test2.netflix.net"]
|
||||
)
|
||||
|
|
@ -0,0 +1,171 @@
|
|||
import unittest
|
||||
from unittest.mock import patch, Mock
|
||||
|
||||
from flask import Flask
|
||||
from acme import challenges
|
||||
from lemur.plugins.lemur_acme import plugin
|
||||
|
||||
|
||||
class TestAcmeHttp(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.ACMEHttpIssuerPlugin = plugin.ACMEHttpIssuerPlugin()
|
||||
self.acme = plugin.AcmeHandler()
|
||||
|
||||
# Creates a new Flask application for a test duration. In python 3.8, manual push of application context is
|
||||
# needed to run tests in dev environment without getting error 'Working outside of application context'.
|
||||
_app = Flask('lemur_test_acme')
|
||||
self.ctx = _app.app_context()
|
||||
assert self.ctx
|
||||
self.ctx.push()
|
||||
|
||||
def tearDown(self):
|
||||
self.ctx.pop()
|
||||
|
||||
def test_create_authority(self):
|
||||
options = {
|
||||
"plugin": {"plugin_options": [{"name": "certificate", "value": "123"}]}
|
||||
}
|
||||
acme_root, b, role = self.ACMEHttpIssuerPlugin.create_authority(options)
|
||||
self.assertEqual(acme_root, "123")
|
||||
self.assertEqual(b, "")
|
||||
self.assertEqual(role, [{"username": "", "password": "", "name": "acme"}])
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.setup_acme_client")
|
||||
@patch("lemur.plugins.base.manager.PluginManager.get")
|
||||
@patch("lemur.plugins.lemur_acme.challenge_types.destination_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.request_certificate")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.authorization_service")
|
||||
def test_create_certificate(
|
||||
self,
|
||||
mock_authorization_service,
|
||||
mock_request_certificate,
|
||||
mock_destination_service,
|
||||
mock_plugin_manager_get,
|
||||
mock_acme,
|
||||
):
|
||||
provider = plugin.ACMEHttpIssuerPlugin()
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = '[{"name": "tokenDestination", "value": "mock-sftp-destination"}]'
|
||||
|
||||
mock_order_resource = Mock()
|
||||
mock_order_resource.authorizations = [Mock()]
|
||||
mock_order_resource.authorizations[0].body.challenges = [Mock()]
|
||||
mock_order_resource.authorizations[0].body.challenges[0].response_and_validation.return_value = (Mock(), "Anything-goes")
|
||||
mock_order_resource.authorizations[0].body.challenges[0].chall = challenges.HTTP01(
|
||||
token=b'\x0f\x1c\xbe#od\xd1\x9c\xa6j\\\xa4\r\xed\xe5\xbf0pz\xeaxnl)\xea[i\xbc\x95\x08\x96\x1f')
|
||||
|
||||
mock_client = Mock()
|
||||
mock_client.new_order.return_value = mock_order_resource
|
||||
mock_client.answer_challenge.return_value = True
|
||||
|
||||
mock_finalized_order = Mock()
|
||||
mock_finalized_order.fullchain_pem = "-----BEGIN CERTIFICATE-----\nMIIEqzCCApOgAwIBAgIRAIvhKg5ZRO08VGQx8JdhT+UwDQYJKoZIhvcNAQELBQAw\nGjEYMBYGA1UEAwwPRmFrZSBMRSBSb290IFgxMB4XDTE2MDUyMzIyMDc1OVoXDTM2\nMDUyMzIyMDc1OVowIjEgMB4GA1UEAwwXRmFrZSBMRSBJbnRlcm1lZGlhdGUgWDEw\nggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDtWKySDn7rWZc5ggjz3ZB0\n8jO4xti3uzINfD5sQ7Lj7hzetUT+wQob+iXSZkhnvx+IvdbXF5/yt8aWPpUKnPym\noLxsYiI5gQBLxNDzIec0OIaflWqAr29m7J8+NNtApEN8nZFnf3bhehZW7AxmS1m0\nZnSsdHw0Fw+bgixPg2MQ9k9oefFeqa+7Kqdlz5bbrUYV2volxhDFtnI4Mh8BiWCN\nxDH1Hizq+GKCcHsinDZWurCqder/afJBnQs+SBSL6MVApHt+d35zjBD92fO2Je56\ndhMfzCgOKXeJ340WhW3TjD1zqLZXeaCyUNRnfOmWZV8nEhtHOFbUCU7r/KkjMZO9\nAgMBAAGjgeMwgeAwDgYDVR0PAQH/BAQDAgGGMBIGA1UdEwEB/wQIMAYBAf8CAQAw\nHQYDVR0OBBYEFMDMA0a5WCDMXHJw8+EuyyCm9Wg6MHoGCCsGAQUFBwEBBG4wbDA0\nBggrBgEFBQcwAYYoaHR0cDovL29jc3Auc3RnLXJvb3QteDEubGV0c2VuY3J5cHQu\nb3JnLzA0BggrBgEFBQcwAoYoaHR0cDovL2NlcnQuc3RnLXJvb3QteDEubGV0c2Vu\nY3J5cHQub3JnLzAfBgNVHSMEGDAWgBTBJnSkikSg5vogKNhcI5pFiBh54DANBgkq\nhkiG9w0BAQsFAAOCAgEABYSu4Il+fI0MYU42OTmEj+1HqQ5DvyAeyCA6sGuZdwjF\nUGeVOv3NnLyfofuUOjEbY5irFCDtnv+0ckukUZN9lz4Q2YjWGUpW4TTu3ieTsaC9\nAFvCSgNHJyWSVtWvB5XDxsqawl1KzHzzwr132bF2rtGtazSqVqK9E07sGHMCf+zp\nDQVDVVGtqZPHwX3KqUtefE621b8RI6VCl4oD30Olf8pjuzG4JKBFRFclzLRjo/h7\nIkkfjZ8wDa7faOjVXx6n+eUQ29cIMCzr8/rNWHS9pYGGQKJiY2xmVC9h12H99Xyf\nzWE9vb5zKP3MVG6neX1hSdo7PEAb9fqRhHkqVsqUvJlIRmvXvVKTwNCP3eCjRCCI\nPTAvjV+4ni786iXwwFYNz8l3PmPLCyQXWGohnJ8iBm+5nk7O2ynaPVW0U2W+pt2w\nSVuvdDM5zGv2f9ltNWUiYZHJ1mmO97jSY/6YfdOUH66iRtQtDkHBRdkNBsMbD+Em\n2TgBldtHNSJBfB3pm9FblgOcJ0FSWcUDWJ7vO0+NTXlgrRofRT6pVywzxVo6dND0\nWzYlTWeUVsO40xJqhgUQRER9YLOLxJ0O6C8i0xFxAMKOtSdodMB3RIwt7RFQ0uyt\nn5Z5MqkYhlMI3J1tPRTp1nEt9fyGspBOO05gi148Qasp+3N+svqKomoQglNoAxU=\n-----END CERTIFICATE-----\n-----BEGIN CERTIFICATE-----\nMIIEqzCCApOgAwIBAgIRAIvhKg5ZRO08VGQx8JdhT+UwDQYJKoZIhvcNAQELBQAw\nGjEYMBYGA1UEAwwPRmFrZSBMRSBSb290IFgxMB4XDTE2MDUyMzIyMDc1OVoXDTM2\nMDUyMzIyMDc1OVowIjEgMB4GA1UEAwwXRmFrZSBMRSBJbnRlcm1lZGlhdGUgWDEw\nggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDtWKySDn7rWZc5ggjz3ZB0\n8jO4xti3uzINfD5sQ7Lj7hzetUT+wQob+iXSZkhnvx+IvdbXF5/yt8aWPpUKnPym\noLxsYiI5gQBLxNDzIec0OIaflWqAr29m7J8+NNtApEN8nZFnf3bhehZW7AxmS1m0\nZnSsdHw0Fw+bgixPg2MQ9k9oefFeqa+7Kqdlz5bbrUYV2volxhDFtnI4Mh8BiWCN\nxDH1Hizq+GKCcHsinDZWurCqder/afJBnQs+SBSL6MVApHt+d35zjBD92fO2Je56\ndhMfzCgOKXeJ340WhW3TjD1zqLZXeaCyUNRnfOmWZV8nEhtHOFbUCU7r/KkjMZO9\nAgMBAAGjgeMwgeAwDgYDVR0PAQH/BAQDAgGGMBIGA1UdEwEB/wQIMAYBAf8CAQAw\nHQYDVR0OBBYEFMDMA0a5WCDMXHJw8+EuyyCm9Wg6MHoGCCsGAQUFBwEBBG4wbDA0\nBggrBgEFBQcwAYYoaHR0cDovL29jc3Auc3RnLXJvb3QteDEubGV0c2VuY3J5cHQu\nb3JnLzA0BggrBgEFBQcwAoYoaHR0cDovL2NlcnQuc3RnLXJvb3QteDEubGV0c2Vu\nY3J5cHQub3JnLzAfBgNVHSMEGDAWgBTBJnSkikSg5vogKNhcI5pFiBh54DANBgkq\nhkiG9w0BAQsFAAOCAgEABYSu4Il+fI0MYU42OTmEj+1HqQ5DvyAeyCA6sGuZdwjF\nUGeVOv3NnLyfofuUOjEbY5irFCDtnv+0ckukUZN9lz4Q2YjWGUpW4TTu3ieTsaC9\nAFvCSgNHJyWSVtWvB5XDxsqawl1KzHzzwr132bF2rtGtazSqVqK9E07sGHMCf+zp\nDQVDVVGtqZPHwX3KqUtefE621b8RI6VCl4oD30Olf8pjuzG4JKBFRFclzLRjo/h7\nIkkfjZ8wDa7faOjVXx6n+eUQ29cIMCzr8/rNWHS9pYGGQKJiY2xmVC9h12H99Xyf\nzWE9vb5zKP3MVG6neX1hSdo7PEAb9fqRhHkqVsqUvJlIRmvXvVKTwNCP3eCjRCCI\nPTAvjV+4ni786iXwwFYNz8l3PmPLCyQXWGohnJ8iBm+5nk7O2ynaPVW0U2W+pt2w\nSVuvdDM5zGv2f9ltNWUiYZHJ1mmO97jSY/6YfdOUH66iRtQtDkHBRdkNBsMbD+Em\n2TgBldtHNSJBfB3pm9FblgOcJ0FSWcUDWJ7vO0+NTXlgrRofRT6pVywzxVo6dND0\nWzYlTWeUVsO40xJqhgUQRER9YLOLxJ0O6C8i0xFxAMKOtSdodMB3RIwt7RFQ0uyt\nn5Z5MqkYhlMI3J1tPRTp1nEt9fyGspBOO05gi148Qasp+3N+svqKomoQglNoAxU=\n-----END CERTIFICATE-----\n"
|
||||
mock_client.poll_and_finalize.return_value = mock_finalized_order
|
||||
|
||||
mock_acme.return_value = (mock_client, "")
|
||||
|
||||
mock_destination = Mock()
|
||||
mock_destination.label = "mock-sftp-destination"
|
||||
mock_destination.plugin_name = "SFTPDestinationPlugin"
|
||||
mock_destination_service.get.return_value = mock_destination
|
||||
|
||||
mock_destination_plugin = Mock()
|
||||
mock_destination_plugin.upload_acme_token.return_value = True
|
||||
mock_plugin_manager_get.return_value = mock_destination_plugin
|
||||
|
||||
issuer_options = {
|
||||
"authority": mock_authority,
|
||||
"tokenDestination": "mock-sftp-destination",
|
||||
"common_name": "test.netflix.net",
|
||||
}
|
||||
csr = "123"
|
||||
mock_request_certificate.return_value = ("pem_certificate", "chain")
|
||||
pem_certificate, pem_certificate_chain, _ = provider.create_certificate(csr, issuer_options)
|
||||
|
||||
self.assertEqual(pem_certificate, "-----BEGIN CERTIFICATE-----\nMIIEqzCCApOgAwIBAgIRAIvhKg5ZRO08VGQx8JdhT+UwDQYJKoZIhvcNAQELBQAw\nGjEYMBYGA1UEAwwPRmFrZSBMRSBSb290IFgxMB4XDTE2MDUyMzIyMDc1OVoXDTM2\nMDUyMzIyMDc1OVowIjEgMB4GA1UEAwwXRmFrZSBMRSBJbnRlcm1lZGlhdGUgWDEw\nggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDtWKySDn7rWZc5ggjz3ZB0\n8jO4xti3uzINfD5sQ7Lj7hzetUT+wQob+iXSZkhnvx+IvdbXF5/yt8aWPpUKnPym\noLxsYiI5gQBLxNDzIec0OIaflWqAr29m7J8+NNtApEN8nZFnf3bhehZW7AxmS1m0\nZnSsdHw0Fw+bgixPg2MQ9k9oefFeqa+7Kqdlz5bbrUYV2volxhDFtnI4Mh8BiWCN\nxDH1Hizq+GKCcHsinDZWurCqder/afJBnQs+SBSL6MVApHt+d35zjBD92fO2Je56\ndhMfzCgOKXeJ340WhW3TjD1zqLZXeaCyUNRnfOmWZV8nEhtHOFbUCU7r/KkjMZO9\nAgMBAAGjgeMwgeAwDgYDVR0PAQH/BAQDAgGGMBIGA1UdEwEB/wQIMAYBAf8CAQAw\nHQYDVR0OBBYEFMDMA0a5WCDMXHJw8+EuyyCm9Wg6MHoGCCsGAQUFBwEBBG4wbDA0\nBggrBgEFBQcwAYYoaHR0cDovL29jc3Auc3RnLXJvb3QteDEubGV0c2VuY3J5cHQu\nb3JnLzA0BggrBgEFBQcwAoYoaHR0cDovL2NlcnQuc3RnLXJvb3QteDEubGV0c2Vu\nY3J5cHQub3JnLzAfBgNVHSMEGDAWgBTBJnSkikSg5vogKNhcI5pFiBh54DANBgkq\nhkiG9w0BAQsFAAOCAgEABYSu4Il+fI0MYU42OTmEj+1HqQ5DvyAeyCA6sGuZdwjF\nUGeVOv3NnLyfofuUOjEbY5irFCDtnv+0ckukUZN9lz4Q2YjWGUpW4TTu3ieTsaC9\nAFvCSgNHJyWSVtWvB5XDxsqawl1KzHzzwr132bF2rtGtazSqVqK9E07sGHMCf+zp\nDQVDVVGtqZPHwX3KqUtefE621b8RI6VCl4oD30Olf8pjuzG4JKBFRFclzLRjo/h7\nIkkfjZ8wDa7faOjVXx6n+eUQ29cIMCzr8/rNWHS9pYGGQKJiY2xmVC9h12H99Xyf\nzWE9vb5zKP3MVG6neX1hSdo7PEAb9fqRhHkqVsqUvJlIRmvXvVKTwNCP3eCjRCCI\nPTAvjV+4ni786iXwwFYNz8l3PmPLCyQXWGohnJ8iBm+5nk7O2ynaPVW0U2W+pt2w\nSVuvdDM5zGv2f9ltNWUiYZHJ1mmO97jSY/6YfdOUH66iRtQtDkHBRdkNBsMbD+Em\n2TgBldtHNSJBfB3pm9FblgOcJ0FSWcUDWJ7vO0+NTXlgrRofRT6pVywzxVo6dND0\nWzYlTWeUVsO40xJqhgUQRER9YLOLxJ0O6C8i0xFxAMKOtSdodMB3RIwt7RFQ0uyt\nn5Z5MqkYhlMI3J1tPRTp1nEt9fyGspBOO05gi148Qasp+3N+svqKomoQglNoAxU=\n-----END CERTIFICATE-----\n")
|
||||
self.assertEqual(pem_certificate_chain,
|
||||
"-----BEGIN CERTIFICATE-----\nMIIEqzCCApOgAwIBAgIRAIvhKg5ZRO08VGQx8JdhT+UwDQYJKoZIhvcNAQELBQAw\nGjEYMBYGA1UEAwwPRmFrZSBMRSBSb290IFgxMB4XDTE2MDUyMzIyMDc1OVoXDTM2\nMDUyMzIyMDc1OVowIjEgMB4GA1UEAwwXRmFrZSBMRSBJbnRlcm1lZGlhdGUgWDEw\nggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDtWKySDn7rWZc5ggjz3ZB0\n8jO4xti3uzINfD5sQ7Lj7hzetUT+wQob+iXSZkhnvx+IvdbXF5/yt8aWPpUKnPym\noLxsYiI5gQBLxNDzIec0OIaflWqAr29m7J8+NNtApEN8nZFnf3bhehZW7AxmS1m0\nZnSsdHw0Fw+bgixPg2MQ9k9oefFeqa+7Kqdlz5bbrUYV2volxhDFtnI4Mh8BiWCN\nxDH1Hizq+GKCcHsinDZWurCqder/afJBnQs+SBSL6MVApHt+d35zjBD92fO2Je56\ndhMfzCgOKXeJ340WhW3TjD1zqLZXeaCyUNRnfOmWZV8nEhtHOFbUCU7r/KkjMZO9\nAgMBAAGjgeMwgeAwDgYDVR0PAQH/BAQDAgGGMBIGA1UdEwEB/wQIMAYBAf8CAQAw\nHQYDVR0OBBYEFMDMA0a5WCDMXHJw8+EuyyCm9Wg6MHoGCCsGAQUFBwEBBG4wbDA0\nBggrBgEFBQcwAYYoaHR0cDovL29jc3Auc3RnLXJvb3QteDEubGV0c2VuY3J5cHQu\nb3JnLzA0BggrBgEFBQcwAoYoaHR0cDovL2NlcnQuc3RnLXJvb3QteDEubGV0c2Vu\nY3J5cHQub3JnLzAfBgNVHSMEGDAWgBTBJnSkikSg5vogKNhcI5pFiBh54DANBgkq\nhkiG9w0BAQsFAAOCAgEABYSu4Il+fI0MYU42OTmEj+1HqQ5DvyAeyCA6sGuZdwjF\nUGeVOv3NnLyfofuUOjEbY5irFCDtnv+0ckukUZN9lz4Q2YjWGUpW4TTu3ieTsaC9\nAFvCSgNHJyWSVtWvB5XDxsqawl1KzHzzwr132bF2rtGtazSqVqK9E07sGHMCf+zp\nDQVDVVGtqZPHwX3KqUtefE621b8RI6VCl4oD30Olf8pjuzG4JKBFRFclzLRjo/h7\nIkkfjZ8wDa7faOjVXx6n+eUQ29cIMCzr8/rNWHS9pYGGQKJiY2xmVC9h12H99Xyf\nzWE9vb5zKP3MVG6neX1hSdo7PEAb9fqRhHkqVsqUvJlIRmvXvVKTwNCP3eCjRCCI\nPTAvjV+4ni786iXwwFYNz8l3PmPLCyQXWGohnJ8iBm+5nk7O2ynaPVW0U2W+pt2w\nSVuvdDM5zGv2f9ltNWUiYZHJ1mmO97jSY/6YfdOUH66iRtQtDkHBRdkNBsMbD+Em\n2TgBldtHNSJBfB3pm9FblgOcJ0FSWcUDWJ7vO0+NTXlgrRofRT6pVywzxVo6dND0\nWzYlTWeUVsO40xJqhgUQRER9YLOLxJ0O6C8i0xFxAMKOtSdodMB3RIwt7RFQ0uyt\nn5Z5MqkYhlMI3J1tPRTp1nEt9fyGspBOO05gi148Qasp+3N+svqKomoQglNoAxU=\n-----END CERTIFICATE-----\n")
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.setup_acme_client")
|
||||
@patch("lemur.plugins.base.manager.PluginManager.get")
|
||||
@patch("lemur.plugins.lemur_acme.challenge_types.destination_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.request_certificate")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.authorization_service")
|
||||
def test_create_certificate_missing_destination_token(
|
||||
self,
|
||||
mock_authorization_service,
|
||||
mock_request_certificate,
|
||||
mock_destination_service,
|
||||
mock_plugin_manager_get,
|
||||
mock_acme,
|
||||
):
|
||||
provider = plugin.ACMEHttpIssuerPlugin()
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = '[{"name": "mock_name", "value": "mock_value"}]'
|
||||
|
||||
mock_order_resource = Mock()
|
||||
mock_order_resource.authorizations = [Mock()]
|
||||
mock_order_resource.authorizations[0].body.challenges = [Mock()]
|
||||
mock_order_resource.authorizations[0].body.challenges[0].chall = challenges.HTTP01(
|
||||
token=b'\x0f\x1c\xbe#od\xd1\x9c\xa6j\\\xa4\r\xed\xe5\xbf0pz\xeaxnl)\xea[i\xbc\x95\x08\x96\x1f')
|
||||
|
||||
mock_client = Mock()
|
||||
mock_client.new_order.return_value = mock_order_resource
|
||||
mock_acme.return_value = (mock_client, "")
|
||||
|
||||
mock_destination = Mock()
|
||||
mock_destination.label = "mock-sftp-destination"
|
||||
mock_destination.plugin_name = "SFTPDestinationPlugin"
|
||||
mock_destination_service.get_by_label.return_value = mock_destination
|
||||
|
||||
mock_destination_plugin = Mock()
|
||||
mock_destination_plugin.upload_acme_token.return_value = True
|
||||
mock_plugin_manager_get.return_value = mock_destination_plugin
|
||||
|
||||
issuer_options = {
|
||||
"authority": mock_authority,
|
||||
"tokenDestination": "mock-sftp-destination",
|
||||
"common_name": "test.netflix.net",
|
||||
}
|
||||
csr = "123"
|
||||
mock_request_certificate.return_value = ("pem_certificate", "chain")
|
||||
with self.assertRaisesRegex(Exception, "No token_destination configured"):
|
||||
provider.create_certificate(csr, issuer_options)
|
||||
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.setup_acme_client")
|
||||
@patch("lemur.plugins.base.manager.PluginManager.get")
|
||||
@patch("lemur.plugins.lemur_acme.challenge_types.destination_service")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.AcmeHandler.request_certificate")
|
||||
@patch("lemur.plugins.lemur_acme.plugin.authorization_service")
|
||||
def test_create_certificate_missing_http_challenge(
|
||||
self,
|
||||
mock_authorization_service,
|
||||
mock_request_certificate,
|
||||
mock_destination_service,
|
||||
mock_plugin_manager_get,
|
||||
mock_acme,
|
||||
):
|
||||
provider = plugin.ACMEHttpIssuerPlugin()
|
||||
mock_authority = Mock()
|
||||
mock_authority.options = '[{"name": "tokenDestination", "value": "mock-sftp-destination"}]'
|
||||
|
||||
mock_order_resource = Mock()
|
||||
mock_order_resource.authorizations = [Mock()]
|
||||
mock_order_resource.authorizations[0].body.challenges = [Mock()]
|
||||
mock_order_resource.authorizations[0].body.challenges[0].chall = challenges.DNS01(
|
||||
token=b'\x0f\x1c\xbe#od\xd1\x9c\xa6j\\\xa4\r\xed\xe5\xbf0pz\xeaxnl)\xea[i\xbc\x95\x08\x96\x1f')
|
||||
|
||||
mock_client = Mock()
|
||||
mock_client.new_order.return_value = mock_order_resource
|
||||
mock_acme.return_value = (mock_client, "")
|
||||
|
||||
issuer_options = {
|
||||
"authority": mock_authority,
|
||||
"tokenDestination": "mock-sftp-destination",
|
||||
"common_name": "test.netflix.net",
|
||||
}
|
||||
csr = "123"
|
||||
mock_request_certificate.return_value = ("pem_certificate", "chain")
|
||||
with self.assertRaisesRegex(Exception, "HTTP-01 challenge was not offered"):
|
||||
provider.create_certificate(csr, issuer_options)
|
||||
|
|
@ -16,8 +16,10 @@
|
|||
|
||||
.. moduleauthor:: Dmitry Zykov https://github.com/DmitryZykov
|
||||
"""
|
||||
from os import path
|
||||
|
||||
import paramiko
|
||||
from paramiko.ssh_exception import AuthenticationException, NoValidConnectionsError
|
||||
|
||||
from flask import current_app
|
||||
from lemur.plugins import lemur_sftp
|
||||
|
|
@ -95,33 +97,15 @@ class SFTPDestinationPlugin(DestinationPlugin):
|
|||
},
|
||||
]
|
||||
|
||||
def upload(self, name, body, private_key, cert_chain, options, **kwargs):
|
||||
|
||||
current_app.logger.debug("SFTP destination plugin is started")
|
||||
|
||||
cn = common_name(parse_certificate(body))
|
||||
def open_sftp_connection(self, options):
|
||||
host = self.get_option("host", options)
|
||||
port = self.get_option("port", options)
|
||||
user = self.get_option("user", options)
|
||||
password = self.get_option("password", options)
|
||||
ssh_priv_key = self.get_option("privateKeyPath", options)
|
||||
ssh_priv_key_pass = self.get_option("privateKeyPass", options)
|
||||
dst_path = self.get_option("destinationPath", options)
|
||||
export_format = self.get_option("exportFormat", options)
|
||||
|
||||
# prepare files for upload
|
||||
files = {cn + ".key": private_key, cn + ".pem": body}
|
||||
|
||||
if cert_chain:
|
||||
if export_format == "NGINX":
|
||||
# assemble body + chain in the single file
|
||||
files[cn + ".pem"] += "\n" + cert_chain
|
||||
|
||||
elif export_format == "Apache":
|
||||
# store chain in the separate file
|
||||
files[cn + ".ca.bundle.pem"] = cert_chain
|
||||
|
||||
# upload files
|
||||
# delete files
|
||||
try:
|
||||
current_app.logger.debug(
|
||||
"Connecting to {0}@{1}:{2}".format(user, host, port)
|
||||
|
|
@ -145,50 +129,170 @@ class SFTPDestinationPlugin(DestinationPlugin):
|
|||
current_app.logger.error(
|
||||
"No password or private key provided. Can't proceed"
|
||||
)
|
||||
raise paramiko.ssh_exception.AuthenticationException
|
||||
raise AuthenticationException
|
||||
|
||||
# open the sftp session inside the ssh connection
|
||||
sftp = ssh.open_sftp()
|
||||
return ssh.open_sftp(), ssh
|
||||
|
||||
# make sure that the destination path exist
|
||||
try:
|
||||
current_app.logger.debug("Creating {0}".format(dst_path))
|
||||
sftp.mkdir(dst_path)
|
||||
except IOError:
|
||||
current_app.logger.debug("{0} already exist, resuming".format(dst_path))
|
||||
try:
|
||||
dst_path_cn = dst_path + "/" + cn
|
||||
current_app.logger.debug("Creating {0}".format(dst_path_cn))
|
||||
sftp.mkdir(dst_path_cn)
|
||||
except IOError:
|
||||
current_app.logger.debug(
|
||||
"{0} already exist, resuming".format(dst_path_cn)
|
||||
)
|
||||
except AuthenticationException as e:
|
||||
current_app.logger.error("ERROR in {0}: {1}".format(e.__class__, e))
|
||||
raise AuthenticationException("Couldn't connect to {0}, due to an Authentication exception.")
|
||||
except NoValidConnectionsError as e:
|
||||
current_app.logger.error("ERROR in {0}: {1}".format(e.__class__, e))
|
||||
raise NoValidConnectionsError("Couldn't connect to {0}, possible timeout or invalid hostname")
|
||||
|
||||
# upload certificate files to the sftp destination
|
||||
for filename, data in files.items():
|
||||
# this is called when using this as a default destination plugin
|
||||
def upload(self, name, body, private_key, cert_chain, options, **kwargs):
|
||||
|
||||
current_app.logger.debug("SFTP destination plugin is started")
|
||||
|
||||
cn = common_name(parse_certificate(body))
|
||||
dst_path = self.get_option("destinationPath", options)
|
||||
dst_path_cn = dst_path + "/" + cn
|
||||
export_format = self.get_option("exportFormat", options)
|
||||
|
||||
# prepare files for upload
|
||||
files = {cn + ".key": private_key, cn + ".pem": body}
|
||||
|
||||
if cert_chain:
|
||||
if export_format == "NGINX":
|
||||
# assemble body + chain in the single file
|
||||
files[cn + ".pem"] += "\n" + cert_chain
|
||||
|
||||
elif export_format == "Apache":
|
||||
# store chain in the separate file
|
||||
files[cn + ".ca.bundle.pem"] = cert_chain
|
||||
|
||||
self.upload_file(dst_path_cn, files, options)
|
||||
|
||||
# this is called from the acme http challenge
|
||||
def upload_acme_token(self, token_path, token, options, **kwargs):
|
||||
|
||||
current_app.logger.debug("SFTP destination plugin is started for HTTP-01 challenge")
|
||||
|
||||
dst_path = self.get_option("destinationPath", options)
|
||||
|
||||
_, filename = path.split(token_path)
|
||||
|
||||
# prepare files for upload
|
||||
files = {filename: token}
|
||||
|
||||
self.upload_file(dst_path, files, options)
|
||||
|
||||
# this is called from the acme http challenge
|
||||
def delete_acme_token(self, token_path, options, **kwargs):
|
||||
dst_path = self.get_option("destinationPath", options)
|
||||
|
||||
_, filename = path.split(token_path)
|
||||
|
||||
# prepare files for upload
|
||||
files = {filename: None}
|
||||
|
||||
self.delete_file(dst_path, files, options)
|
||||
|
||||
# here the file is deleted
|
||||
def delete_file(self, dst_path, files, options):
|
||||
|
||||
try:
|
||||
# open the ssh and sftp sessions
|
||||
sftp, ssh = self.open_sftp_connection(options)
|
||||
|
||||
# delete files
|
||||
for filename, _ in files.items():
|
||||
current_app.logger.debug(
|
||||
"Uploading {0} to {1}".format(filename, dst_path_cn)
|
||||
"Deleting {0} from {1}".format(filename, dst_path)
|
||||
)
|
||||
try:
|
||||
with sftp.open(dst_path_cn + "/" + filename, "w") as f:
|
||||
f.write(data)
|
||||
except (PermissionError) as permerror:
|
||||
sftp.remove(path.join(dst_path, filename))
|
||||
except PermissionError as permerror:
|
||||
if permerror.errno == 13:
|
||||
current_app.logger.debug(
|
||||
"Uploading {0} to {1} returned Permission Denied Error, making file writable and retrying".format(filename, dst_path_cn)
|
||||
"Deleting {0} from {1} returned Permission Denied Error, making file writable and retrying".format(
|
||||
filename, dst_path)
|
||||
)
|
||||
sftp.chmod(dst_path_cn + "/" + filename, 0o600)
|
||||
with sftp.open(dst_path_cn + "/" + filename, "w") as f:
|
||||
f.write(data)
|
||||
# read only for owner, -r--------
|
||||
sftp.chmod(dst_path_cn + "/" + filename, 0o400)
|
||||
sftp.chmod(path.join(dst_path, filename), 0o600)
|
||||
sftp.remove(path.join(dst_path, filename))
|
||||
|
||||
ssh.close()
|
||||
|
||||
except (AuthenticationException, NoValidConnectionsError) as e:
|
||||
raise e
|
||||
except Exception as e:
|
||||
current_app.logger.error("ERROR in {0}: {1}".format(e.__class__, e))
|
||||
try:
|
||||
ssh.close()
|
||||
except BaseException:
|
||||
pass
|
||||
|
||||
# here the file is uploaded for real, this helps to keep this class DRY
|
||||
def upload_file(self, dst_path, files, options):
|
||||
|
||||
try:
|
||||
# open the ssh and sftp sessions
|
||||
sftp, ssh = self.open_sftp_connection(options)
|
||||
|
||||
# split the path into it's segments, so we can create it recursively
|
||||
allparts = []
|
||||
path_copy = dst_path
|
||||
while True:
|
||||
parts = path.split(path_copy)
|
||||
if parts[0] == path_copy: # sentinel for absolute paths
|
||||
allparts.insert(0, parts[0])
|
||||
break
|
||||
elif parts[1] == path_copy: # sentinel for relative paths
|
||||
allparts.insert(0, parts[1])
|
||||
break
|
||||
else:
|
||||
path_copy = parts[0]
|
||||
allparts.insert(0, parts[1])
|
||||
|
||||
# make sure that the destination path exists, recursively
|
||||
remote_path = allparts[0]
|
||||
for part in allparts:
|
||||
try:
|
||||
if part != "/" and part != "":
|
||||
remote_path = path.join(remote_path, part)
|
||||
sftp.stat(remote_path)
|
||||
except IOError:
|
||||
current_app.logger.debug("{0} doesn't exist, trying to create it".format(remote_path))
|
||||
try:
|
||||
sftp.mkdir(remote_path)
|
||||
except IOError as ioerror:
|
||||
current_app.logger.debug(
|
||||
"Couldn't create {0}, error message: {1}".format(remote_path, ioerror))
|
||||
|
||||
# upload certificate files to the sftp destination
|
||||
for filename, data in files.items():
|
||||
current_app.logger.debug(
|
||||
"Uploading {0} to {1}".format(filename, dst_path)
|
||||
)
|
||||
try:
|
||||
with sftp.open(path.join(dst_path, filename), "w") as f:
|
||||
f.write(data)
|
||||
except PermissionError as permerror:
|
||||
if permerror.errno == 13:
|
||||
current_app.logger.debug(
|
||||
"Uploading {0} to {1} returned Permission Denied Error, making file writable and retrying".format(
|
||||
filename, dst_path)
|
||||
)
|
||||
sftp.chmod(path.join(dst_path, filename), 0o600)
|
||||
with sftp.open(path.join(dst_path, filename), "w") as f:
|
||||
f.write(data)
|
||||
# most likely the upload user isn't the webuser, -rw-r--r--
|
||||
sftp.chmod(path.join(dst_path, filename), 0o644)
|
||||
|
||||
ssh.close()
|
||||
|
||||
except (AuthenticationException, NoValidConnectionsError) as e:
|
||||
raise e
|
||||
except Exception as e:
|
||||
current_app.logger.error("ERROR in {0}: {1}".format(e.__class__, e))
|
||||
try:
|
||||
ssh.close()
|
||||
except BaseException:
|
||||
pass
|
||||
message = ''
|
||||
if hasattr(e, 'errors'):
|
||||
for _, error in e.errors.items():
|
||||
message = error.strerror
|
||||
raise Exception(
|
||||
'Couldn\'t upload file to {}, error message: {}'.format(self.get_option("host", options), message))
|
||||
|
|
|
|||
|
|
@ -0,0 +1,144 @@
|
|||
import unittest
|
||||
from unittest.mock import patch, Mock, MagicMock, mock_open
|
||||
|
||||
from flask import Flask
|
||||
from lemur.plugins.lemur_sftp import plugin
|
||||
from paramiko.ssh_exception import AuthenticationException
|
||||
|
||||
|
||||
class TestSftp(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.sftp_destination = plugin.SFTPDestinationPlugin()
|
||||
# Creates a new Flask application for a test duration. In python 3.8, manual push of application context is
|
||||
# needed to run tests in dev environment without getting error 'Working outside of application context'.
|
||||
_app = Flask('lemur_test_sftp')
|
||||
self.ctx = _app.app_context()
|
||||
assert self.ctx
|
||||
self.ctx.push()
|
||||
|
||||
def tearDown(self):
|
||||
self.ctx.pop()
|
||||
|
||||
def test_failing_ssh_connection(self):
|
||||
dst_path = '/var/non-existent'
|
||||
files = {'first-file': 'data'}
|
||||
options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
|
||||
{'name': 'user', 'value': 'test_acme'}]
|
||||
|
||||
with self.assertRaises(AuthenticationException):
|
||||
self.sftp_destination.upload_file(dst_path, files, options)
|
||||
|
||||
@patch("lemur.plugins.lemur_sftp.plugin.paramiko")
|
||||
def test_upload_file_single_with_password(self, mock_paramiko):
|
||||
dst_path = '/var/non-existent'
|
||||
files = {'first-file': 'data'}
|
||||
options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
|
||||
{'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'}]
|
||||
|
||||
mock_sftp = Mock()
|
||||
mock_sftp.open = mock_open()
|
||||
|
||||
mock_ssh = mock_paramiko.SSHClient.return_value
|
||||
mock_ssh.connect = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp
|
||||
|
||||
self.sftp_destination.upload_file(dst_path, files, options)
|
||||
|
||||
mock_sftp.open.assert_called_once_with('/var/non-existent/first-file', 'w')
|
||||
handle = mock_sftp.open()
|
||||
handle.write.assert_called_once_with('data')
|
||||
mock_ssh.close.assert_called_once()
|
||||
mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
|
||||
password='test_password')
|
||||
|
||||
@patch("lemur.plugins.lemur_sftp.plugin.paramiko")
|
||||
def test_upload_file_multiple_with_key(self, mock_paramiko):
|
||||
dst_path = '/var/non-existent'
|
||||
files = {'first-file': 'data', 'second-file': 'data2'}
|
||||
options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
|
||||
{'name': 'user', 'value': 'test_acme'}, {'name': 'privateKeyPath', 'value': '/var/id_rsa'},
|
||||
{'name': 'privateKeyPass', 'value': 'ssh-key-password'}]
|
||||
|
||||
mock_sftp = Mock()
|
||||
mock_sftp.open = mock_open()
|
||||
|
||||
mock_paramiko.RSAKey.from_private_key_file.return_value = 'ssh-rsa test-key'
|
||||
|
||||
mock_ssh = mock_paramiko.SSHClient.return_value
|
||||
mock_ssh.connect = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp
|
||||
|
||||
self.sftp_destination.upload_file(dst_path, files, options)
|
||||
|
||||
mock_sftp.open.assert_called_with('/var/non-existent/second-file', 'w')
|
||||
handle = mock_sftp.open()
|
||||
handle.write.assert_called_with('data2')
|
||||
mock_ssh.close.assert_called_once()
|
||||
|
||||
mock_paramiko.RSAKey.from_private_key_file.assert_called_with('/var/id_rsa', 'ssh-key-password')
|
||||
mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
|
||||
pkey='ssh-rsa test-key')
|
||||
|
||||
@patch("lemur.plugins.lemur_sftp.plugin.paramiko")
|
||||
def test_upload_acme_token(self, mock_paramiko):
|
||||
token_path = './well-known/acme-challenge/some-token-path'
|
||||
token = 'token-data'
|
||||
options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
|
||||
{'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'},
|
||||
{'name': 'destinationPath', 'value': '/var/destination-path'}]
|
||||
|
||||
mock_sftp = Mock()
|
||||
mock_sftp.open = mock_open()
|
||||
|
||||
mock_ssh = mock_paramiko.SSHClient.return_value
|
||||
mock_ssh.connect = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp
|
||||
|
||||
self.sftp_destination.upload_acme_token(token_path, token, options)
|
||||
|
||||
mock_sftp.open.assert_called_once_with('/var/destination-path/some-token-path', 'w')
|
||||
handle = mock_sftp.open()
|
||||
handle.write.assert_called_once_with('token-data')
|
||||
mock_ssh.close.assert_called_once()
|
||||
mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
|
||||
password='test_password')
|
||||
|
||||
@patch("lemur.plugins.lemur_sftp.plugin.paramiko")
|
||||
def test_delete_file_with_password(self, mock_paramiko):
|
||||
dst_path = '/var/non-existent'
|
||||
files = {'first-file': None}
|
||||
options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
|
||||
{'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'}]
|
||||
|
||||
mock_sftp = Mock()
|
||||
|
||||
mock_ssh = mock_paramiko.SSHClient.return_value
|
||||
mock_ssh.connect = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp
|
||||
|
||||
self.sftp_destination.delete_file(dst_path, files, options)
|
||||
|
||||
mock_sftp.remove.assert_called_once_with('/var/non-existent/first-file')
|
||||
mock_ssh.close.assert_called_once()
|
||||
mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
|
||||
password='test_password')
|
||||
|
||||
@patch("lemur.plugins.lemur_sftp.plugin.paramiko")
|
||||
def test_delete_acme_token(self, mock_paramiko):
|
||||
token_path = './well-known/acme-challenge/some-token-path'
|
||||
options = [{'name': 'host', 'value': 'non-existent'}, {'name': 'port', 'value': '22'},
|
||||
{'name': 'user', 'value': 'test_acme'}, {'name': 'password', 'value': 'test_password'},
|
||||
{'name': 'destinationPath', 'value': '/var/destination-path'}]
|
||||
|
||||
mock_sftp = Mock()
|
||||
|
||||
mock_ssh = mock_paramiko.SSHClient.return_value
|
||||
mock_ssh.connect = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp
|
||||
|
||||
self.sftp_destination.delete_acme_token(token_path, options)
|
||||
|
||||
mock_sftp.remove.assert_called_once_with('/var/destination-path/some-token-path')
|
||||
mock_ssh.close.assert_called_once()
|
||||
mock_ssh.connect.assert_called_with('non-existent', username='test_acme', port='22',
|
||||
password='test_password')
|
||||
|
|
@ -34,7 +34,7 @@ angular.module('lemur')
|
|||
};
|
||||
})
|
||||
|
||||
.controller('AuthorityCreateController', function ($scope, $uibModalInstance, AuthorityService, AuthorityApi, LemurRestangular, RoleService, PluginService, WizardHandler, toaster) {
|
||||
.controller('AuthorityCreateController', function ($scope, $uibModalInstance, AuthorityService, AuthorityApi, LemurRestangular, RoleService, PluginService, WizardHandler, toaster, DestinationService) {
|
||||
$scope.authority = LemurRestangular.restangularizeElement(null, {}, 'authorities');
|
||||
// set the defaults
|
||||
AuthorityService.getDefaults($scope.authority).then(function () {
|
||||
|
|
@ -52,6 +52,12 @@ angular.module('lemur')
|
|||
});
|
||||
});
|
||||
|
||||
$scope.getDestinations = function() {
|
||||
return DestinationService.findDestinationsByName('').then(function(destinations) {
|
||||
$scope.destinations = destinations;
|
||||
});
|
||||
};
|
||||
|
||||
$scope.getAuthoritiesByName = function (value) {
|
||||
return AuthorityService.findAuthorityByName(value).then(function (authorities) {
|
||||
$scope.authorities = authorities;
|
||||
|
|
|
|||
|
|
@ -66,11 +66,28 @@
|
|||
<div class="col-sm-10">
|
||||
<input name="sub" ng-if="item.type == 'int'" type="number" ng-pattern="item.validation?item.validation:'^[0-9]+$'"
|
||||
class="form-control" ng-model="item.value"/>
|
||||
|
||||
<select name="sub" ng-if="item.type == 'select'" class="form-control" ng-options="i for i in item.available"
|
||||
ng-model="item.value"></select>
|
||||
|
||||
<!-- DestSelect options -->
|
||||
<ui-select class="input-md" ng-model="item.value" theme="bootstrap" title="choose a destination" ng-if="item.type == 'destinationSelect'">
|
||||
<ui-select-match placeholder="select an destination...">{{$select.selected.label}}</ui-select-match>
|
||||
<ui-select-choices class="form-control"
|
||||
refresh="getDestinations()"
|
||||
refresh-delay="300"
|
||||
repeat="destination.id as destination in destinations | filter: $select.search">
|
||||
<div ng-bind-html="destination.label | highlight: $select.search"></div>
|
||||
<small>
|
||||
<span ng-bind-html="''+destination.description | highlight: $select.search"></span>
|
||||
</small>
|
||||
</ui-select-choices>
|
||||
</ui-select>
|
||||
|
||||
<input name="sub" ng-if="item.type == 'bool'" class="form-control" type="checkbox" ng-model="item.value">
|
||||
<input name="sub" ng-if="item.type == 'str'" type="text" class="form-control" ng-model="item.value"/>
|
||||
<textarea name="sub" ng-if="item.type == 'textarea'" class="form-control" ng-model="item.value"></textarea>
|
||||
|
||||
<div ng-if="item.type == 'export-plugin'">
|
||||
<form name="exportForm" class="form-horizontal" role="form" novalidate>
|
||||
<select class="form-control" ng-model="item.value"
|
||||
|
|
|
|||
1
setup.py
1
setup.py
|
|
@ -132,6 +132,7 @@ setup(
|
|||
'lemur.plugins': [
|
||||
'verisign_issuer = lemur.plugins.lemur_verisign.plugin:VerisignIssuerPlugin',
|
||||
'acme_issuer = lemur.plugins.lemur_acme.plugin:ACMEIssuerPlugin',
|
||||
'acme_http_issuer = lemur.plugins.lemur_acme.plugin:ACMEHttpIssuerPlugin',
|
||||
'aws_destination = lemur.plugins.lemur_aws.plugin:AWSDestinationPlugin',
|
||||
'aws_source = lemur.plugins.lemur_aws.plugin:AWSSourcePlugin',
|
||||
'aws_s3 = lemur.plugins.lemur_aws.plugin:S3DestinationPlugin',
|
||||
|
|
|
|||
Loading…
Reference in New Issue