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

Miscellaneous notification fixes and tests

Browse Source
This commit is contained in:
Jasmine Schladen 2020-10-16 15:13:12 -07:00
parent 67c184a97c
commit 60bb0037f0
8 changed files with 157 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
lemur/notifications/messaging.py
View File

@ -101,6 +101,9 @@ def send_notification(event_type, data, targets, notification):
notification.plugin.send(event_type, data, targets, notification.options)
status = SUCCESS_METRIC_STATUS
except Exception as e:
current_app.logger.error(
"Unable to send notification to {}.".format(targets), exc_info=True
)
sentry.captureException()
metrics.send(
@ -190,13 +193,13 @@ def send_rotation_notification(certificate, notification_plugin=None):
status = FAILURE_METRIC_STATUS
if not notification_plugin:
notification_plugin = plugins.get(
current_app.config.get("LEMUR_DEFAULT_NOTIFICATION_PLUGIN")
current_app.config.get("LEMUR_DEFAULT_NOTIFICATION_PLUGIN", "email-notification")
)
data = certificate_notification_output_schema.dump(certificate).data
try:
notification_plugin.send("rotation", data, [data["owner"]])
notification_plugin.send("rotation", data, [data["owner"]], [])
status = SUCCESS_METRIC_STATUS
except Exception as e:
current_app.logger.error(
@ -290,7 +293,7 @@ def needs_notification(certificate):
for notification in certificate.notifications:
if not notification.active or not notification.options:
return
continue
interval = get_plugin_option("interval", notification.options)
unit = get_plugin_option("unit", notification.options)

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

@ -19,14 +19,16 @@ from lemur.plugins import lemur_email as email
from lemur.plugins.lemur_email.templates.config import env
def render_html(template_name, message):
def render_html(template_name, options, certificates):
"""
Renders the html for our email notification.
:param template_name:
:param message:
:param options:
:param certificates:
:return:
"""
message = {"options": options, "certificates": certificates}
template = env.get_template("{}.html".format(template_name))
return template.render(
dict(message=message, hostname=current_app.config.get("LEMUR_HOSTNAME"))
@ -100,8 +102,7 @@ class EmailNotificationPlugin(ExpirationNotificationPlugin):
subject = "Lemur: {0} Notification".format(notification_type.capitalize())
data = {"options": options, "certificates": message}
body = render_html(notification_type, data)
body = render_html(notification_type, options, message)
s_type = current_app.config.get("LEMUR_EMAIL_SENDER", "ses").lower()

18
lemur/plugins/lemur_email/templates/rotation.html
View File

@ -83,12 +83,12 @@
<td width="32px"></td>
<td width="16px"></td>
<td style="line-height:1.2">
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:20px;color:#202020">{{ certificate.name }}</span>
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:20px;color:#202020">{{ message.certificates.name }}</span>
<br>
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:13px;color:#727272">
<br>{{ certificate.owner }}
<br>{{ certificate.validityEnd | time }}
<a href="https://{{ hostname }}/#/certificates/{{ certificate.name }}" target="_blank">Details</a>
<br>{{ message.certificates.owner }}
<br>{{ message.certificates.validityEnd | time }}
<a href="https://{{ hostname }}/#/certificates/{{ message.certificates.name }}" target="_blank">Details</a>
</span>
</td>
</tr>
@ -110,12 +110,12 @@
<td width="32px"></td>
<td width="16px"></td>
<td style="line-height:1.2">
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:20px;color:#202020">{{ certificate.replacedBy[0].name }}</span>
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:20px;color:#202020">{{ message.certificates.name }}</span>
<br>
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:13px;color:#727272">
<br>{{ certificate.replacedBy[0].owner }}
<br>{{ certificate.replacedBy[0].validityEnd | time }}
<a href="https://{{ hostname }}/#/certificates/{{ certificate.replacedBy[0].name }}" target="_blank">Details</a>
<br>{{ message.certificates.owner }}
<br>{{ message.certificates.validityEnd | time }}
<a href="https://{{ hostname }}/#/certificates/{{ message.certificates.name }}" target="_blank">Details</a>
</span>
</td>
</tr>
@ -133,7 +133,7 @@
<table border="0" cellspacing="0" cellpadding="0"
style="margin-top:48px;margin-bottom:48px">
<tbody>
{% for endpoint in certificate.endpoints %}
{% for endpoint in message.certificates.endpoints %}
<tr valign="middle">
<td width="32px"></td>
<td width="16px"></td>

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

@ -1,36 +1,83 @@
import os
from lemur.plugins.lemur_email.templates.config import env
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
dir_path = os.path.dirname(os.path.realpath(__file__))
def test_render(certificate, endpoint):
from lemur.certificates.schemas import certificate_notification_output_schema
@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"},
]
def test_render_expiration(certificate, endpoint):
new_cert = CertificateFactory()
new_cert.replaces.append(certificate)
data = {
"certificates": [certificate_notification_output_schema.dump(certificate).data],
"options": [
{"name": "interval", "value": 10},
{"name": "unit", "value": "days"},
],
}
assert render_html("expiration", get_options(), [certificate_notification_output_schema.dump(certificate).data])
template = env.get_template("{}.html".format("expiration"))
body = template.render(dict(message=data, hostname="lemur.test.example.com"))
template = env.get_template("{}.html".format("rotation"))
def test_render_rotation(certificate, endpoint):
certificate.endpoints.append(endpoint)
body = template.render(
dict(
certificate=certificate_notification_output_schema.dump(certificate).data,
hostname="lemur.test.example.com",
)
)
assert render_html("rotation", get_options(), certificate_notification_output_schema.dump(certificate).data)
def test_render_rotation_failure(pending_certificate):
assert render_html("failed", get_options(), certificate_notification_output_schema.dump(pending_certificate).data)
@mock_ses
def test_send_expiration_notification():
from lemur.notifications.messaging import send_expiration_notifications
from lemur.tests.factories import CertificateFactory
from lemur.tests.factories import NotificationFactory
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()
notification = NotificationFactory(plugin_name="email-notification")
certificate.not_after = in_ten_days
certificate.notifications.append(notification)
certificate.notifications[0].options = get_options()
verify_sender_email()
assert send_expiration_notifications([]) == (2, 0)
@mock_ses
def test_send_rotation_notification(endpoint, source_plugin):
from lemur.notifications.messaging import send_rotation_notification
from lemur.deployment.service import rotate_certificate
new_certificate = CertificateFactory()
rotate_certificate(endpoint, new_certificate)
assert endpoint.certificate == new_certificate
verify_sender_email()
assert send_rotation_notification(new_certificate)
@mock_ses
def test_send_pending_failure_notification(user, pending_certificate, async_issuer_plugin):
from lemur.notifications.messaging import send_pending_failure_notification
verify_sender_email()
assert send_pending_failure_notification(pending_certificate)

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

@ -58,26 +58,19 @@ def create_rotation_attachments(certificate):
"title": certificate["name"],
"title_link": create_certificate_url(certificate["name"]),
"fields": [
{"title": "Owner", "value": certificate["owner"], "short": True},
{
{"title": "Owner", "value": certificate["owner"], "short": True},
{
"title": "Expires",
"value": arrow.get(certificate["validityEnd"]).format(
"dddd, MMMM D, YYYY"
),
"short": True,
},
{
"title": "Replaced By",
"value": len(certificate["replaced"][0]["name"]),
"short": True,
},
{
"title": "Endpoints Rotated",
"value": len(certificate["endpoints"]),
"short": True,
},
}
"title": "Expires",
"value": arrow.get(certificate["validityEnd"]).format(
"dddd, MMMM D, YYYY"
),
"short": True,
},
{
"title": "Endpoints Rotated",
"value": len(certificate["endpoints"]),
"short": True,
},
],
}

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

@ -21,3 +21,48 @@ def test_formatting(certificate):
}
assert attachment == create_expiration_attachments(data)[0]
def get_options():
return [
{"name": "interval", "value": 10},
{"name": "unit", "value": "days"},
]
# Currently disabled as we have no good way to mock Slack webhooks
# def test_send_expiration_notification():
# from lemur.notifications.messaging import send_expiration_notifications
# from lemur.tests.factories import CertificateFactory
#
# 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()
# notification = NotificationFactory(plugin_name="slack-notification")
#
# certificate.not_after = in_ten_days
# certificate.notifications.append(notification)
# certificate.notifications[0].options = get_options()
#
# assert send_expiration_notifications([]) == (2, 0)
# Currently disabled as we have no good way to mock Slack webhooks
# 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="slack-notification")
#
# new_certificate = CertificateFactory()
# rotate_certificate(endpoint, new_certificate)
# assert endpoint.certificate == new_certificate
#
# assert send_rotation_notification(new_certificate, notification_plugin=notification.plugin)
# Currently disabled as the Slack 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, notification_plugin=plugins.get("slack-notification"))

2
lemur/tests/conf.py
View File

@ -46,7 +46,7 @@ LEMUR_ALLOWED_DOMAINS = [
# Lemur currently only supports SES for sending email, this address
# needs to be verified
LEMUR_EMAIL = ""
LEMUR_EMAIL = "lemur@example.com"
LEMUR_SECURITY_TEAM_EMAIL = ["security@example.com"]
LEMUR_HOSTNAME = "lemur.example.com"

15
lemur/tests/test_messaging.py
View File

@ -1,8 +1,8 @@
from datetime import timedelta
import arrow
import pytest
from freezegun import freeze_time
from datetime import timedelta
import arrow
from moto import mock_ses
@ -105,4 +105,11 @@ def test_send_expiration_notification_with_no_notifications(
def test_send_rotation_notification(notification_plugin, certificate):
from lemur.notifications.messaging import send_rotation_notification
send_rotation_notification(certificate, notification_plugin=notification_plugin)
assert send_rotation_notification(certificate, notification_plugin=notification_plugin)
@mock_ses
def test_send_pending_failure_notification(notification_plugin, async_issuer_plugin, pending_certificate):
from lemur.notifications.messaging import send_pending_failure_notification
assert send_pending_failure_notification(pending_certificate, notification_plugin=notification_plugin)