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

Merge branch 'master' into log_update

Browse Source
This commit is contained in:
charhate
2020-10-28 15:40:16 -07:00
committed by GitHub
parent e9824a6808 31b9e2cd20
commit c25782468b
15 changed files with 386 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
lemur/certificates/service.py
View File

@ -560,20 +560,21 @@ def query_common_name(common_name, args):
:return:
"""
owner = args.pop("owner")
if not owner:
owner = "%"
# only not expired certificates
current_time = arrow.utcnow()
result = (
Certificate.query.filter(Certificate.cn.ilike(common_name))
.filter(Certificate.owner.ilike(owner))
.filter(Certificate.not_after >= current_time.format("YYYY-MM-DD"))
.all()
)
query = Certificate.query.filter(Certificate.not_after >= current_time.format("YYYY-MM-DD"))\
.filter(not_(Certificate.revoked))\
.filter(not_(Certificate.replaced.any())) # ignore rotated certificates to avoid duplicates
return result
if owner:
query = query.filter(Certificate.owner.ilike(owner))
if common_name != "%":
# if common_name is a wildcard ('%'), no need to include it in the query
query = query.filter(Certificate.cn.ilike(common_name))
return query.all()
def create_csr(**csr_config):

157
lemur/notifications/messaging.py
View File

@ -30,7 +30,7 @@ from lemur.plugins.utils import get_plugin_option
def get_certificates(exclude=None):
"""
Finds all certificates that are eligible for notifications.
Finds all certificates that are eligible for expiration notifications.
:param exclude:
:return:
"""
@ -42,6 +42,7 @@ def get_certificates(exclude=None):
.filter(Certificate.not_after <= max)
.filter(Certificate.notify == True)
.filter(Certificate.expired == False)
.filter(Certificate.revoked == False)
) # noqa
exclude_conditions = []
@ -62,7 +63,8 @@ def get_certificates(exclude=None):
def get_eligible_certificates(exclude=None):
"""
Finds all certificates that are eligible for certificate expiration.
Finds all certificates that are eligible for certificate expiration notification.
Returns the set of all eligible certificates, grouped by owner, with a list of applicable notifications.
:param exclude:
:return:
"""
@ -87,29 +89,30 @@ def get_eligible_certificates(exclude=None):
return certificates
def send_notification(event_type, data, targets, notification):
def send_plugin_notification(event_type, data, recipients, notification):
"""
Executes the plugin and handles failure.
:param event_type:
:param data:
:param targets:
:param recipients:
:param notification:
:return:
"""
function = f"{__name__}.{sys._getframe().f_code.co_name}"
log_data = {
"function": function,
"message": f"Sending expiration notification for to targets {targets}",
"message": f"Sending expiration notification for to recipients {recipients}",
"notification_type": "expiration",
"certificate_targets": targets,
"certificate_targets": recipients,
}
status = FAILURE_METRIC_STATUS
try:
notification.plugin.send(event_type, data, targets, notification.options)
current_app.logger.debug(log_data)
notification.plugin.send(event_type, data, recipients, notification.options)
status = SUCCESS_METRIC_STATUS
except Exception as e:
log_data["message"] = f"Unable to send expiration notification to targets {targets}"
log_data["message"] = f"Unable to send {event_type} notification to recipients {recipients}"
current_app.logger.error(log_data, exc_info=True)
sentry.captureException()
@ -150,36 +153,27 @@ def send_expiration_notifications(exclude):
notification_data.append(cert_data)
security_data.append(cert_data)
if send_notification(
"expiration", notification_data, [owner], notification
if send_default_notification(
"expiration", notification_data, [owner], notification.options
):
success += 1
else:
failure += 1
notification_recipient = get_plugin_option(
"recipients", notification.options
)
if notification_recipient:
notification_recipient = notification_recipient.split(",")
# removing owner and security_email from notification_recipient
notification_recipient = [i for i in notification_recipient if i not in security_email and i != owner]
recipients = notification.plugin.filter_recipients(notification.options, security_email + [owner])
if (
notification_recipient
if send_plugin_notification(
"expiration",
notification_data,
recipients,
notification,
):
if send_notification(
"expiration",
notification_data,
notification_recipient,
notification,
):
success += 1
else:
failure += 1
success += 1
else:
failure += 1
if send_notification(
"expiration", security_data, security_email, notification
if send_default_notification(
"expiration", security_data, security_email, notification.options
):
success += 1
else:
@ -188,37 +182,36 @@ def send_expiration_notifications(exclude):
return success, failure
def send_rotation_notification(certificate, notification_plugin=None):
def send_default_notification(notification_type, data, targets, notification_options=None):
"""
Sends a report to certificate owners when their certificate has been
rotated.
Sends a report to the specified target via the default notification plugin. Applicable for any notification_type.
At present, "default" means email, as the other notification plugins do not support dynamically configured targets.
:param certificate:
:param notification_plugin:
:param notification_type:
:param data:
:param targets:
:param notification_options:
:return:
"""
function = f"{__name__}.{sys._getframe().f_code.co_name}"
log_data = {
"function": function,
"message": f"Sending rotation notification for certificate {certificate.name}",
"notification_type": "rotation",
"certificate_name": certificate.name,
"certificate_owner": certificate.owner,
"message": f"Sending notification for certificate data {data}",
"notification_type": notification_type,
}
status = FAILURE_METRIC_STATUS
if not notification_plugin:
notification_plugin = plugins.get(
current_app.config.get("LEMUR_DEFAULT_NOTIFICATION_PLUGIN", "email-notification")
)
data = certificate_notification_output_schema.dump(certificate).data
notification_plugin = plugins.get(
current_app.config.get("LEMUR_DEFAULT_NOTIFICATION_PLUGIN", "email-notification")
)
try:
notification_plugin.send("rotation", data, [data["owner"]], [])
current_app.logger.debug(log_data)
# we need the notification.options here because the email templates utilize the interval/unit info
notification_plugin.send(notification_type, data, targets, notification_options)
status = SUCCESS_METRIC_STATUS
except Exception as e:
log_data["message"] = f"Unable to send rotation notification for certificate {certificate.name} " \
f"to owner {data['owner']}"
log_data["message"] = f"Unable to send {notification_type} notification for certificate data {data} " \
f"to target {targets}"
current_app.logger.error(log_data, exc_info=True)
sentry.captureException()
@ -226,82 +219,49 @@ def send_rotation_notification(certificate, notification_plugin=None):
"notification",
"counter",
1,
metric_tags={"status": status, "event_type": "rotation"},
metric_tags={"status": status, "event_type": notification_type},
)
if status == SUCCESS_METRIC_STATUS:
return True
def send_rotation_notification(certificate):
data = certificate_notification_output_schema.dump(certificate).data
return send_default_notification("rotation", data, [data["owner"]])
def send_pending_failure_notification(
pending_cert, notify_owner=True, notify_security=True, notification_plugin=None
pending_cert, notify_owner=True, notify_security=True
):
"""
Sends a report to certificate owners when their pending certificate failed to be created.
:param pending_cert:
:param notification_plugin:
:param notify_owner:
:param notify_security:
:return:
"""
function = f"{__name__}.{sys._getframe().f_code.co_name}"
log_data = {
"function": function,
"message": f"Sending pending failure notification for pending certificate {pending_cert}",
"notification_type": "failed",
"certificate_name": pending_cert.name,
"certificate_owner": pending_cert.owner,
}
status = FAILURE_METRIC_STATUS
if not notification_plugin:
notification_plugin = plugins.get(
current_app.config.get(
"LEMUR_DEFAULT_NOTIFICATION_PLUGIN", "email-notification"
)
)
data = pending_certificate_output_schema.dump(pending_cert).data
data["security_email"] = current_app.config.get("LEMUR_SECURITY_TEAM_EMAIL")
notify_owner_success = False
if notify_owner:
try:
notification_plugin.send("failed", data, [data["owner"]], pending_cert)
status = SUCCESS_METRIC_STATUS
except Exception as e:
log_data["recipient"] = data["owner"]
log_data["message"] = f"Unable to send pending failure notification for certificate {pending_cert.name} " \
f"to owner {pending_cert.owner}"
current_app.logger.error(log_data, exc_info=True)
sentry.captureException()
notify_owner_success = send_default_notification("failed", data, [data["owner"]], pending_cert)
notify_security_success = False
if notify_security:
try:
notification_plugin.send(
"failed", data, data["security_email"], pending_cert
)
status = SUCCESS_METRIC_STATUS
except Exception as e:
log_data["recipient"] = data["security_email"]
log_data["message"] = f"Unable to send pending failure notification for certificate {pending_cert.name} " \
f"to security email {pending_cert.owner}"
current_app.logger.error(log_data, exc_info=True)
sentry.captureException()
notify_security_success = send_default_notification("failed", data, data["security_email"], pending_cert)
metrics.send(
"notification",
"counter",
1,
metric_tags={"status": status, "event_type": "failed"},
)
if status == SUCCESS_METRIC_STATUS:
return True
return notify_owner_success or notify_security_success
def needs_notification(certificate):
"""
Determine if notifications for a given certificate should
currently be sent
Determine if notifications for a given certificate should currently be sent.
For each notification configured for the cert, verifies it is active, properly configured,
and that the configured expiration period is currently met.
:param certificate:
:return:
@ -331,7 +291,6 @@ def needs_notification(certificate):
raise Exception(
f"Invalid base unit for expiration interval: {unit}"
)
if days == interval:
notifications.append(notification)
return notifications

11
lemur/plugins/bases/notification.py
View File

@ -20,6 +20,15 @@ class NotificationPlugin(Plugin):
def send(self, notification_type, message, targets, options, **kwargs):
raise NotImplementedError
def filter_recipients(self, options, excluded_recipients):
"""
Given a set of options (which should include configured recipient info), filters out recipients that
we do NOT want to notify.
For any notification types where recipients can't be dynamically modified, this returns an empty list.
"""
return []
class ExpirationNotificationPlugin(NotificationPlugin):
"""
@ -50,5 +59,5 @@ class ExpirationNotificationPlugin(NotificationPlugin):
def options(self):
return self.default_options + self.additional_options
def send(self, notification_type, message, targets, options, **kwargs):
def send(self, notification_type, message, excluded_targets, options, **kwargs):
raise NotImplementedError

55
lemur/plugins/lemur_aws/plugin.py
View File

@ -32,13 +32,14 @@
.. moduleauthor:: Mikhail Khodorovskiy <mikhail.khodorovskiy@jivesoftware.com>
.. moduleauthor:: Harm Weites <harm@weites.com>
"""
from acme.errors import ClientError
from flask import current_app
from lemur.extensions import sentry, metrics
from lemur.plugins import lemur_aws as aws
from lemur.extensions import sentry, metrics
from lemur.plugins import lemur_aws as aws, ExpirationNotificationPlugin
from lemur.plugins.bases import DestinationPlugin, ExportDestinationPlugin, SourcePlugin
from lemur.plugins.lemur_aws import iam, s3, elb, ec2
from lemur.plugins.lemur_aws import iam, s3, elb, ec2, sns
def get_region_from_dns(dns):
@ -406,3 +407,51 @@ class S3DestinationPlugin(ExportDestinationPlugin):
self.get_option("encrypt", options),
account_number=self.get_option("accountNumber", options),
)
class SNSNotificationPlugin(ExpirationNotificationPlugin):
title = "AWS SNS"
slug = "aws-sns"
description = "Sends notifications to AWS SNS"
version = aws.VERSION
author = "Jasmine Schladen <jschladen@netflix.com>"
author_url = "https://github.com/Netflix/lemur"
additional_options = [
{
"name": "accountNumber",
"type": "str",
"required": True,
"validation": "[0-9]{12}",
"helpMessage": "A valid AWS account number with permission to access the SNS topic",
},
{
"name": "region",
"type": "str",
"required": True,
"validation": "[0-9a-z\\-]{1,25}",
"helpMessage": "Region in which the SNS topic is located, e.g. \"us-east-1\"",
},
{
"name": "topicName",
"type": "str",
"required": True,
# base topic name is 1-256 characters (alphanumeric plus underscore and hyphen)
"validation": "^[a-zA-Z0-9_\\-]{1,256}$",
"helpMessage": "The name of the topic to use for expiration notifications",
}
]
def send(self, notification_type, message, excluded_targets, options, **kwargs):
"""
While we receive a `targets` parameter here, it is unused, as the SNS topic is pre-configured in the
plugin configuration, and can't reasonably be changed dynamically.
"""
topic_arn = f"arn:aws:sns:{self.get_option('region', options)}:" \
f"{self.get_option('accountNumber', options)}:" \
f"{self.get_option('topicName', options)}"
current_app.logger.info(f"Publishing {notification_type} notification to topic {topic_arn}")
sns.publish(topic_arn, message, notification_type, region_name=self.get_option("region", options))

55
lemur/plugins/lemur_aws/sns.py Normal file
View File

@ -0,0 +1,55 @@
"""
.. module: lemur.plugins.lemur_aws.sts
:platform: Unix
:copyright: (c) 2020 by Netflix Inc., see AUTHORS for more
:license: Apache, see LICENSE for more details.
.. moduleauthor:: Jasmine Schladen <jschladen@netflix.com>
"""
import json
import arrow
import boto3
from flask import current_app
def publish(topic_arn, certificates, notification_type, **kwargs):
sns_client = boto3.client("sns", **kwargs)
message_ids = {}
for certificate in certificates:
message_ids[certificate["name"]] = publish_single(sns_client, topic_arn, certificate, notification_type)
return message_ids
def publish_single(sns_client, topic_arn, certificate, notification_type):
response = sns_client.publish(
TopicArn=topic_arn,
Message=format_message(certificate, notification_type),
)
response_code = response["ResponseMetadata"]["HTTPStatusCode"]
if response_code != 200:
raise Exception(f"Failed to publish {notification_type} notification to SNS topic {topic_arn}. "
f"SNS response: {response_code} {response}")
current_app.logger.info(f"AWS SNS message published to topic [{topic_arn}] with message ID {response['MessageId']}")
current_app.logger.debug(f"AWS SNS message published to topic [{topic_arn}]: [{response}]")
return response["MessageId"]
def create_certificate_url(name):
return "https://{hostname}/#/certificates/{name}".format(
hostname=current_app.config.get("LEMUR_HOSTNAME"), name=name
)
def format_message(certificate, notification_type):
json_message = {
"notification_type": notification_type,
"certificate_name": certificate["name"],
"expires": arrow.get(certificate["validityEnd"]).format("YYYY-MM-ddTHH:mm:ss"), # 2047-12-T22:00:00
"endpoints_detected": len(certificate["endpoints"]),
"details": create_certificate_url(certificate["name"])
}
return json.dumps(json_message)

120
lemur/plugins/lemur_aws/tests/test_sns.py Normal file
View File

@ -0,0 +1,120 @@
import json
from datetime import timedelta
import arrow
import boto3
from moto import mock_sns, mock_sqs, mock_ses
from lemur.certificates.schemas import certificate_notification_output_schema
from lemur.plugins.lemur_aws.sns import format_message
from lemur.plugins.lemur_aws.sns import publish
from lemur.tests.factories import NotificationFactory, CertificateFactory
from lemur.tests.test_messaging import verify_sender_email
@mock_sns()
def test_format(certificate, endpoint):
data = [certificate_notification_output_schema.dump(certificate).data]
for certificate in data:
expected_message = {
"notification_type": "expiration",
"certificate_name": certificate["name"],
"expires": arrow.get(certificate["validityEnd"]).format("YYYY-MM-ddTHH:mm:ss"),
"endpoints_detected": 0,
"details": "https://lemur.example.com/#/certificates/{name}".format(name=certificate["name"])
}
assert expected_message == json.loads(format_message(certificate, "expiration"))
@mock_sns()
@mock_sqs()
def create_and_subscribe_to_topic():
sns_client = boto3.client("sns", region_name="us-east-1")
topic_arn = sns_client.create_topic(Name='lemursnstest')["TopicArn"]
sqs_client = boto3.client("sqs", region_name="us-east-1")
queue = sqs_client.create_queue(QueueName="lemursnstestqueue")
queue_url = queue["QueueUrl"]
queue_arn = sqs_client.get_queue_attributes(QueueUrl=queue_url)["Attributes"]["QueueArn"]
sns_client.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn)
return [topic_arn, sqs_client, queue_url]
@mock_sns()
@mock_sqs()
def test_publish(certificate, endpoint):
data = [certificate_notification_output_schema.dump(certificate).data]
topic_arn, sqs_client, queue_url = create_and_subscribe_to_topic()
message_ids = publish(topic_arn, data, "expiration", region_name="us-east-1")
assert len(message_ids) == len(data)
received_messages = sqs_client.receive_message(QueueUrl=queue_url)["Messages"]
for certificate in data:
expected_message_id = message_ids[certificate["name"]]
actual_message = next(
(m for m in received_messages if json.loads(m["Body"])["MessageId"] == expected_message_id), None)
assert json.loads(actual_message["Body"])["Message"] == format_message(certificate, "expiration")
def get_options():
return [
{"name": "interval", "value": 10},
{"name": "unit", "value": "days"},
{"name": "region", "value": "us-east-1"},
{"name": "accountNumber", "value": "123456789012"},
{"name": "topicName", "value": "lemursnstest"},
]
@mock_sns()
@mock_sqs()
@mock_ses() # because email notifications are also sent
def test_send_expiration_notification():
from lemur.notifications.messaging import send_expiration_notifications
verify_sender_email() # emails are sent to owner and security; SNS only used for configured notification
topic_arn, sqs_client, queue_url = create_and_subscribe_to_topic()
notification = NotificationFactory(plugin_name="aws-sns")
notification.options = get_options()
now = arrow.utcnow()
in_ten_days = now + timedelta(days=10, hours=1) # a bit more than 10 days since we'll check in the future
certificate = CertificateFactory()
certificate.not_after = in_ten_days
certificate.notifications.append(notification)
assert send_expiration_notifications([]) == (3, 0) # owner, SNS, and security
received_messages = sqs_client.receive_message(QueueUrl=queue_url)["Messages"]
assert len(received_messages) == 1
expected_message = format_message(certificate_notification_output_schema.dump(certificate).data, "expiration")
actual_message = json.loads(received_messages[0]["Body"])["Message"]
assert actual_message == expected_message
# Currently disabled as the SNS plugin doesn't support this type of notification
# def test_send_rotation_notification(endpoint, source_plugin):
# from lemur.notifications.messaging import send_rotation_notification
# from lemur.deployment.service import rotate_certificate
#
# notification = NotificationFactory(plugin_name="aws-sns")
# notification.options = get_options()
#
# new_certificate = CertificateFactory()
# rotate_certificate(endpoint, new_certificate)
# assert endpoint.certificate == new_certificate
#
# assert send_rotation_notification(new_certificate)
# Currently disabled as the SNS plugin doesn't support this type of notification
# def test_send_pending_failure_notification(user, pending_certificate, async_issuer_plugin):
# from lemur.notifications.messaging import send_pending_failure_notification
#
# assert send_pending_failure_notification(pending_certificate)

11
lemur/plugins/lemur_email/plugin.py
View File

@ -17,6 +17,7 @@ from lemur.plugins.bases import ExpirationNotificationPlugin
from lemur.plugins import lemur_email as email
from lemur.plugins.lemur_email.templates.config import env
from lemur.plugins.utils import get_plugin_option
def render_html(template_name, options, certificates):
@ -111,3 +112,13 @@ class EmailNotificationPlugin(ExpirationNotificationPlugin):
elif s_type == "smtp":
send_via_smtp(subject, body, targets)
@staticmethod
def filter_recipients(options, excluded_recipients, **kwargs):
notification_recipients = get_plugin_option("recipients", options)
if notification_recipients:
notification_recipients = notification_recipients.split(",")
# removing owner and security_email from notification_recipient
notification_recipients = [i for i in notification_recipients if i not in excluded_recipients]
return notification_recipients

23
lemur/plugins/lemur_email/tests/test_email.py
View File

@ -2,26 +2,21 @@ import os
from datetime import timedelta
import arrow
import boto3
from moto import mock_ses
from lemur.certificates.schemas import certificate_notification_output_schema
from lemur.plugins.lemur_email.plugin import render_html
from lemur.tests.factories import CertificateFactory
from lemur.tests.test_messaging import verify_sender_email
dir_path = os.path.dirname(os.path.realpath(__file__))
@mock_ses
def verify_sender_email():
ses_client = boto3.client("ses", region_name="us-east-1")
ses_client.verify_email_identity(EmailAddress="lemur@example.com")
def get_options():
return [
{"name": "interval", "value": 10},
{"name": "unit", "value": "days"},
{"name": "recipients", "value": "person1@example.com,person2@example.com"},
]
@ -59,7 +54,7 @@ def test_send_expiration_notification():
certificate.notifications[0].options = get_options()
verify_sender_email()
assert send_expiration_notifications([]) == (2, 0)
assert send_expiration_notifications([]) == (3, 0) # owner, recipients (only counted as 1), and security
@mock_ses
@ -81,3 +76,15 @@ def test_send_pending_failure_notification(user, pending_certificate, async_issu
verify_sender_email()
assert send_pending_failure_notification(pending_certificate)
def test_filter_recipients(certificate, endpoint):
from lemur.plugins.lemur_email.plugin import EmailNotificationPlugin
options = [{"name": "recipients", "value": "security@example.com,bob@example.com,joe@example.com"}]
assert EmailNotificationPlugin.filter_recipients(options, []) == ["security@example.com", "bob@example.com",
"joe@example.com"]
assert EmailNotificationPlugin.filter_recipients(options, ["security@example.com"]) == ["bob@example.com",
"joe@example.com"]
assert EmailNotificationPlugin.filter_recipients(options, ["security@example.com", "bob@example.com",
"joe@example.com"]) == []

11
lemur/plugins/lemur_slack/plugin.py
View File

@ -112,6 +112,9 @@ class SlackNotificationPlugin(ExpirationNotificationPlugin):
"""
A typical check can be performed using the notify command:
`lemur notify`
While we receive a `targets` parameter here, it is unused, as Slack webhooks do not allow
dynamic re-targeting of messages. The webhook itself specifies a channel.
"""
attachments = None
if notification_type == "expiration":
@ -124,7 +127,7 @@ class SlackNotificationPlugin(ExpirationNotificationPlugin):
raise Exception("Unable to create message attachments")
body = {
"text": "Lemur {0} Notification".format(notification_type.capitalize()),
"text": f"Lemur {notification_type.capitalize()} Notification",
"attachments": attachments,
"channel": self.get_option("recipients", options),
"username": self.get_option("username", options),
@ -133,8 +136,8 @@ class SlackNotificationPlugin(ExpirationNotificationPlugin):
r = requests.post(self.get_option("webhook", options), json.dumps(body))
if r.status_code not in [200]:
raise Exception("Failed to send message")
raise Exception(f"Failed to send message. Slack response: {r.status_code} {body}")
current_app.logger.error(
"Slack response: {0} Message Body: {1}".format(r.status_code, body)
current_app.logger.info(
f"Slack response: {r.status_code} Message Body: {body}"
)

7
lemur/plugins/lemur_slack/tests/test_slack.py
View File

@ -1,8 +1,10 @@
from datetime import timedelta
import arrow
from moto import mock_ses
from lemur.tests.factories import NotificationFactory, CertificateFactory
from lemur.tests.test_messaging import verify_sender_email
def test_formatting(certificate):
@ -38,9 +40,12 @@ def get_options():
]
@mock_ses() # because email notifications are also sent
def test_send_expiration_notification():
from lemur.notifications.messaging import send_expiration_notifications
verify_sender_email() # emails are sent to owner and security; Slack only used for configured notification
notification = NotificationFactory(plugin_name="slack-notification")
notification.options = get_options()
@ -51,7 +56,7 @@ def test_send_expiration_notification():
certificate.not_after = in_ten_days
certificate.notifications.append(notification)
assert send_expiration_notifications([]) == (2, 0)
assert send_expiration_notifications([]) == (3, 0) # owner, Slack, and security
# Currently disabled as the Slack plugin doesn't support this type of notification

18
lemur/tests/test_messaging.py
View File

@ -1,11 +1,18 @@
from datetime import timedelta
import arrow
import boto3
import pytest
from freezegun import freeze_time
from moto import mock_ses
@mock_ses
def verify_sender_email():
ses_client = boto3.client("ses", region_name="us-east-1")
ses_client.verify_email_identity(EmailAddress="lemur@example.com")
def test_needs_notification(app, certificate, notification):
from lemur.notifications.messaging import needs_notification
@ -78,6 +85,7 @@ def test_get_eligible_certificates(app, certificate, notification):
@mock_ses
def test_send_expiration_notification(certificate, notification, notification_plugin):
from lemur.notifications.messaging import send_expiration_notifications
verify_sender_email()
certificate.notifications.append(notification)
certificate.notifications[0].options = [
@ -87,7 +95,9 @@ def test_send_expiration_notification(certificate, notification, notification_pl
delta = certificate.not_after - timedelta(days=10)
with freeze_time(delta.datetime):
assert send_expiration_notifications([]) == (2, 0)
# this will only send owner and security emails (no additional recipients),
# but it executes 3 successful send attempts
assert send_expiration_notifications([]) == (3, 0)
@mock_ses
@ -104,12 +114,14 @@ def test_send_expiration_notification_with_no_notifications(
@mock_ses
def test_send_rotation_notification(notification_plugin, certificate):
from lemur.notifications.messaging import send_rotation_notification
verify_sender_email()
assert send_rotation_notification(certificate, notification_plugin=notification_plugin)
assert send_rotation_notification(certificate)
@mock_ses
def test_send_pending_failure_notification(notification_plugin, async_issuer_plugin, pending_certificate):
from lemur.notifications.messaging import send_pending_failure_notification
verify_sender_email()
assert send_pending_failure_notification(pending_certificate, notification_plugin=notification_plugin)
assert send_pending_failure_notification(pending_certificate)