Infra
/
lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'master' into send-single-email

This commit is contained in:
Hossein Shafagh 2020-11-11 16:54:40 -08:00 committed by GitHub
parent 1a65e09a99 95854c6f68
commit f4e0b3a90f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 1600 additions and 716 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
lemur/dns_providers/cli.py
View File

@ -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))

521
lemur/plugins/lemur_acme/acme_handlers.py Normal file
View File

@ -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

260
lemur/plugins/lemur_acme/challenge_types.py Normal file
View File

@ -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)

646
lemur/plugins/lemur_acme/plugin.py
View File

@ -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)

228
lemur/plugins/lemur_acme/tests/test_acme.py → lemur/plugins/lemur_acme/tests/test_acme_dns.py
View File

@ -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

112
lemur/plugins/lemur_acme/tests/test_acme_handler.py Normal file
View File

@ -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"]
)

171
lemur/plugins/lemur_acme/tests/test_acme_http.py Normal file
View File

@ -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)

200
lemur/plugins/lemur_sftp/plugin.py
View File

@ -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:
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")
# 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
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)
)
export_format = self.get_option("exportFormat", options)
# upload certificate files to the sftp destination
for filename, data in files.items():
# 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))

144
lemur/plugins/lemur_sftp/tests/test_sftp.py Normal file
View File

@ -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')

8
lemur/static/app/angular/authorities/authority/authority.js vendored
View File

@ -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;

17
lemur/static/app/angular/authorities/authority/options.tpl.html
View File

@ -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
View File

@ -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',