diff --git a/.travis.yml b/.travis.yml
index f38555a0..129d774b 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,5 +1,5 @@
 language: python
-dist: xenial
+dist: bionic
 
 node_js:
   - "6.2.0"
diff --git a/docs/administration.rst b/docs/administration.rst
index c2f20362..724b136f 100644
--- a/docs/administration.rst
+++ b/docs/administration.rst
@@ -269,7 +269,7 @@ Certificates marked as inactive will **not** be notified of upcoming expiration.
 silence the expiration. If a certificate is active and is expiring the above will be notified according to the `LEMUR_DEFAULT_EXPIRATION_NOTIFICATION_INTERVALS` or
 30, 15, 2 days before expiration if no intervals are set.
 
-Lemur supports sending certification expiration notifications through SES and SMTP.
+Lemur supports sending certificate expiration notifications through SES and SMTP.
 
 
 .. data:: LEMUR_EMAIL_SENDER
@@ -1436,7 +1436,7 @@ Slack
     Adds support for slack notifications.
 
 
-AWS
+AWS (Source)
 ----
 
 :Authors:
@@ -1449,7 +1449,7 @@ AWS
     Uses AWS IAM as a source of certificates to manage. Supports a multi-account deployment.
 
 
-AWS
+AWS (Destination)
 ----
 
 :Authors:
@@ -1462,6 +1462,19 @@ AWS
     Uses AWS IAM as a destination for Lemur generated certificates. Support a multi-account deployment.
 
 
+AWS (SNS Notification)
+-----
+
+:Authors:
+    Jasmine Schladen <jschladen@netflix.com>
+:Type:
+    Notification
+:Description:
+    Adds support for SNS notifications. SNS notifications (like other notification plugins) are currently only supported
+    for certificate expiration. Configuration requires a region, account number, and SNS topic name; these elements
+    are then combined to build the topic ARN. Lemur must have access to publish messages to the specified SNS topic.
+
+
 Kubernetes
 ----------
 
diff --git a/docs/developer/plugins/index.rst b/docs/developer/plugins/index.rst
index 8af5e1c8..c2a8c48a 100644
--- a/docs/developer/plugins/index.rst
+++ b/docs/developer/plugins/index.rst
@@ -215,18 +215,21 @@ Notification
 ------------
 
 Lemur includes the ability to create Email notifications by **default**. These notifications
-currently come in the form of expiration notices. Lemur periodically checks certifications expiration dates and
+currently come in the form of expiration and rotation notices. Lemur periodically checks certificate expiration dates and
 determines if a given certificate is eligible for notification. There are currently only two parameters used to
 determine if a certificate is eligible; validity expiration (date the certificate is no longer valid) and the number
 of days the current date (UTC) is from that expiration date.
 
-There are currently two objects that available for notification plugins the first is `NotficationPlugin`. This is the base object for
-any notification within Lemur. Currently the only support notification type is an certificate expiration notification. If you
+Expiration notifications can also be configured for Slack or AWS SNS. Rotation notifications are not configurable.
+Notifications sent to a certificate owner and security team (`LEMUR_SECURITY_TEAM_EMAIL`) can currently only be sent via email.
+
+There are currently two objects that are available for notification plugins. The first is `NotificationPlugin`, which is the base object for
+any notification within Lemur. Currently the only supported notification type is a certificate expiration notification. If you
 are trying to create a new notification type (audit, failed logins, etc.) this would be the object to base your plugin on.
 You would also then need to build additional code to trigger the new notification type.
 
-The second is `ExpirationNotificationPlugin`, this object inherits from `NotificationPlugin` object.
-You will most likely want to base your plugin on, if you want to add new channels for expiration notices (Slack, HipChat, Jira, etc.). It adds default options that are required by
+The second is `ExpirationNotificationPlugin`, which inherits from the `NotificationPlugin` object.
+You will most likely want to base your plugin on this object if you want to add new channels for expiration notices (HipChat, Jira, etc.). It adds default options that are required by
 all expiration notifications (interval, unit). This interface expects for the child to define the following function::
 
     def send(self, notification_type, message, targets, options, **kwargs):
diff --git a/lemur/certificates/service.py b/lemur/certificates/service.py
index b90d7e47..167425cc 100644
--- a/lemur/certificates/service.py
+++ b/lemur/certificates/service.py
@@ -561,20 +561,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):
diff --git a/lemur/notifications/messaging.py b/lemur/notifications/messaging.py
index ca955b69..3928689e 100644
--- a/lemur/notifications/messaging.py
+++ b/lemur/notifications/messaging.py
@@ -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
diff --git a/lemur/plugins/bases/notification.py b/lemur/plugins/bases/notification.py
index 730f68be..0da0dad2 100644
--- a/lemur/plugins/bases/notification.py
+++ b/lemur/plugins/bases/notification.py
@@ -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
diff --git a/lemur/plugins/lemur_aws/plugin.py b/lemur/plugins/lemur_aws/plugin.py
index 8692348a..1be641b0 100644
--- a/lemur/plugins/lemur_aws/plugin.py
+++ b/lemur/plugins/lemur_aws/plugin.py
@@ -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))
diff --git a/lemur/plugins/lemur_aws/sns.py b/lemur/plugins/lemur_aws/sns.py
new file mode 100644
index 00000000..c98bbc0c
--- /dev/null
+++ b/lemur/plugins/lemur_aws/sns.py
@@ -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)
diff --git a/lemur/plugins/lemur_aws/tests/test_sns.py b/lemur/plugins/lemur_aws/tests/test_sns.py
new file mode 100644
index 00000000..ce05c33c
--- /dev/null
+++ b/lemur/plugins/lemur_aws/tests/test_sns.py
@@ -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)
diff --git a/lemur/plugins/lemur_email/plugin.py b/lemur/plugins/lemur_email/plugin.py
index 08332ef1..5b9c188e 100644
--- a/lemur/plugins/lemur_email/plugin.py
+++ b/lemur/plugins/lemur_email/plugin.py
@@ -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
diff --git a/lemur/plugins/lemur_email/tests/test_email.py b/lemur/plugins/lemur_email/tests/test_email.py
index 4f1ea187..fd4dc575 100644
--- a/lemur/plugins/lemur_email/tests/test_email.py
+++ b/lemur/plugins/lemur_email/tests/test_email.py
@@ -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"]) == []
diff --git a/lemur/plugins/lemur_slack/plugin.py b/lemur/plugins/lemur_slack/plugin.py
index 67c3fd84..70d97aa5 100644
--- a/lemur/plugins/lemur_slack/plugin.py
+++ b/lemur/plugins/lemur_slack/plugin.py
@@ -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}"
         )
diff --git a/lemur/plugins/lemur_slack/tests/test_slack.py b/lemur/plugins/lemur_slack/tests/test_slack.py
index da232d61..2161b28b 100644
--- a/lemur/plugins/lemur_slack/tests/test_slack.py
+++ b/lemur/plugins/lemur_slack/tests/test_slack.py
@@ -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
diff --git a/lemur/tests/test_messaging.py b/lemur/tests/test_messaging.py
index dd8f339f..13b6c9b3 100644
--- a/lemur/tests/test_messaging.py
+++ b/lemur/tests/test_messaging.py
@@ -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)
diff --git a/setup.py b/setup.py
index 4da14c3d..58367f14 100644
--- a/setup.py
+++ b/setup.py
@@ -135,6 +135,7 @@ setup(
             'aws_destination = lemur.plugins.lemur_aws.plugin:AWSDestinationPlugin',
             'aws_source = lemur.plugins.lemur_aws.plugin:AWSSourcePlugin',
             'aws_s3 = lemur.plugins.lemur_aws.plugin:S3DestinationPlugin',
+            'aws_sns = lemur.plugins.lemur_aws.plugin:SNSNotificationPlugin',
             'email_notification = lemur.plugins.lemur_email.plugin:EmailNotificationPlugin',
             'slack_notification = lemur.plugins.lemur_slack.plugin:SlackNotificationPlugin',
             'java_truststore_export = lemur.plugins.lemur_jks.plugin:JavaTruststoreExportPlugin',