mirror of
https://github.com/GSA/notifications-api.git
synced 2026-08-15 15:28:02 -04:00
fixing tests
This commit is contained in:
jimmoffet
@@ -11,6 +11,7 @@ from sqlalchemy.exc import SQLAlchemyError
|
||||
|
||||
from app import notify_celery, statsd_client, zendesk_client
|
||||
from app.aws import s3
|
||||
from app.celery.process_ses_receipts_tasks import check_and_queue_callback_task
|
||||
from app.config import QueueNames
|
||||
from app.cronitor import cronitor
|
||||
from app.dao.fact_processing_time_dao import insert_update_processing_time
|
||||
@@ -37,9 +38,6 @@ from app.models import (
|
||||
FactProcessingTime,
|
||||
Notification,
|
||||
)
|
||||
from app.notifications.notifications_ses_callback import (
|
||||
check_and_queue_callback_task,
|
||||
)
|
||||
from app.utils import get_london_midnight_in_utc
|
||||
|
||||
|
||||
|
||||
@@ -6,64 +6,26 @@ from json import decoder
|
||||
import iso8601
|
||||
import requests
|
||||
from celery.exceptions import Retry
|
||||
from flask import Blueprint, current_app, json, jsonify, request
|
||||
from flask import current_app, json
|
||||
from sqlalchemy.orm.exc import NoResultFound
|
||||
|
||||
from app import notify_celery, statsd_client
|
||||
from app.celery.validate_sns_message import sns_notification_handler
|
||||
from app.celery.service_callback_tasks import (
|
||||
create_complaint_callback_data,
|
||||
create_delivery_status_callback_data,
|
||||
send_complaint_to_service,
|
||||
send_delivery_status_to_service,
|
||||
)
|
||||
from app.config import QueueNames
|
||||
from app.dao import notifications_dao
|
||||
from app.errors import InvalidRequest, register_errors
|
||||
from app.models import NOTIFICATION_PENDING, NOTIFICATION_SENDING
|
||||
from app.notifications.notifications_ses_callback import (
|
||||
_check_and_queue_complaint_callback_task,
|
||||
check_and_queue_callback_task,
|
||||
determine_notification_bounce_type,
|
||||
get_aws_responses,
|
||||
handle_complaint,
|
||||
from app.dao.complaint_dao import save_complaint
|
||||
from app.dao.notifications_dao import dao_get_notification_history_by_reference
|
||||
from app.dao.service_callback_api_dao import (
|
||||
get_service_complaint_callback_api_for_service,
|
||||
get_service_delivery_status_callback_api_for_service,
|
||||
)
|
||||
|
||||
ses_callback_blueprint = Blueprint('notifications_ses_callback', __name__)
|
||||
DEFAULT_MAX_AGE = timedelta(days=10000)
|
||||
|
||||
register_errors(ses_callback_blueprint)
|
||||
class SNSMessageType(enum.Enum):
|
||||
SubscriptionConfirmation = 'SubscriptionConfirmation'
|
||||
Notification = 'Notification'
|
||||
UnsubscribeConfirmation = 'UnsubscribeConfirmation'
|
||||
|
||||
|
||||
class InvalidMessageTypeException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def verify_message_type(message_type: str):
|
||||
try:
|
||||
SNSMessageType(message_type)
|
||||
except ValueError:
|
||||
raise InvalidMessageTypeException(f'{message_type} is not a valid message type.')
|
||||
|
||||
|
||||
# 400 counts as a permanent failure so SNS will not retry.
|
||||
# 500 counts as a failed delivery attempt so SNS will retry.
|
||||
# See https://docs.aws.amazon.com/sns/latest/dg/DeliveryPolicies.html#DeliveryPolicies
|
||||
# This should not be here, it used to be in notifications/notifications_ses_callback. It then
|
||||
# got refactored into a task, which is fine, but it created a circular dependency. Will need
|
||||
# to investigate why GDS extracted this into a lambda
|
||||
@ses_callback_blueprint.route('/notifications/email/ses', methods=['POST'])
|
||||
def email_ses_callback_handler():
|
||||
try:
|
||||
data = sns_notification_handler(request.data, request.headers)
|
||||
except Exception as e:
|
||||
raise InvalidRequest("SES-SNS callback failed: invalid message type", 400)
|
||||
|
||||
message = data.get("Message")
|
||||
if "mail" in message:
|
||||
process_ses_results.apply_async([{"Message": message}], queue=QueueNames.NOTIFY)
|
||||
|
||||
return jsonify(
|
||||
result="success", message="SES-SNS callback succeeded"
|
||||
), 200
|
||||
from app.models import NOTIFICATION_PENDING, NOTIFICATION_SENDING, Complaint
|
||||
from app.notifications.callbacks import create_complaint_callback_data
|
||||
|
||||
|
||||
@notify_celery.task(bind=True, name="process-ses-result", max_retries=5, default_retry_delay=300)
|
||||
@@ -145,3 +107,123 @@ def process_ses_results(self, response):
|
||||
current_app.logger.exception("Error processing SES results: {}".format(type(e)))
|
||||
self.retry(queue=QueueNames.RETRY)
|
||||
|
||||
def determine_notification_bounce_type(ses_message):
|
||||
notification_type = ses_message["notificationType"]
|
||||
if notification_type in ["Delivery", "Complaint"]:
|
||||
return notification_type
|
||||
|
||||
if notification_type != "Bounce":
|
||||
raise KeyError(f"Unhandled notification type {notification_type}")
|
||||
|
||||
remove_emails_from_bounce(ses_message)
|
||||
current_app.logger.info("SES bounce dict: {}".format(json.dumps(ses_message).replace("{", "(").replace("}", ")")))
|
||||
if ses_message["bounce"]["bounceType"] == "Permanent":
|
||||
return "Permanent"
|
||||
return "Temporary"
|
||||
|
||||
|
||||
def _determine_provider_response(ses_message):
|
||||
if ses_message["notificationType"] != "Bounce":
|
||||
return None
|
||||
|
||||
bounce_type = ses_message["bounce"]["bounceType"]
|
||||
bounce_subtype = ses_message["bounce"]["bounceSubType"]
|
||||
|
||||
# See https://docs.aws.amazon.com/ses/latest/DeveloperGuide/event-publishing-retrieving-sns-contents.html
|
||||
if bounce_type == "Permanent" and bounce_subtype == "Suppressed":
|
||||
return "The email address is on our email provider suppression list"
|
||||
elif bounce_type == "Permanent" and bounce_subtype == "OnAccountSuppressionList":
|
||||
return "The email address is on the GC Notify suppression list"
|
||||
elif bounce_type == "Transient" and bounce_subtype == "AttachmentRejected":
|
||||
return "The email was rejected because of its attachments"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def get_aws_responses(ses_message):
|
||||
status = determine_notification_bounce_type(ses_message)
|
||||
|
||||
base = {
|
||||
"Permanent": {
|
||||
"message": "Hard bounced",
|
||||
"success": False,
|
||||
"notification_status": "permanent-failure",
|
||||
},
|
||||
"Temporary": {
|
||||
"message": "Soft bounced",
|
||||
"success": False,
|
||||
"notification_status": "temporary-failure",
|
||||
},
|
||||
"Delivery": {
|
||||
"message": "Delivered",
|
||||
"success": True,
|
||||
"notification_status": "delivered",
|
||||
},
|
||||
"Complaint": {
|
||||
"message": "Complaint",
|
||||
"success": True,
|
||||
"notification_status": "delivered",
|
||||
},
|
||||
}[status]
|
||||
|
||||
base["provider_response"] = _determine_provider_response(ses_message)
|
||||
|
||||
return base
|
||||
|
||||
|
||||
def handle_complaint(ses_message):
|
||||
recipient_email = remove_emails_from_complaint(ses_message)[0]
|
||||
current_app.logger.info("Complaint from SES: \n{}".format(json.dumps(ses_message).replace("{", "(").replace("}", ")")))
|
||||
try:
|
||||
reference = ses_message["mail"]["messageId"]
|
||||
except KeyError as e:
|
||||
current_app.logger.exception("Complaint from SES failed to get reference from message", e)
|
||||
return
|
||||
notification = dao_get_notification_history_by_reference(reference)
|
||||
ses_complaint = ses_message.get("complaint", None)
|
||||
|
||||
complaint = Complaint(
|
||||
notification_id=notification.id,
|
||||
service_id=notification.service_id,
|
||||
ses_feedback_id=ses_complaint.get("feedbackId", None) if ses_complaint else None,
|
||||
complaint_type=ses_complaint.get("complaintFeedbackType", None) if ses_complaint else None,
|
||||
complaint_date=ses_complaint.get("timestamp", None) if ses_complaint else None,
|
||||
)
|
||||
save_complaint(complaint)
|
||||
return complaint, notification, recipient_email
|
||||
|
||||
|
||||
def remove_mail_headers(dict_to_edit):
|
||||
if dict_to_edit["mail"].get("headers"):
|
||||
dict_to_edit["mail"].pop("headers")
|
||||
if dict_to_edit["mail"].get("commonHeaders"):
|
||||
dict_to_edit["mail"].pop("commonHeaders")
|
||||
|
||||
|
||||
def remove_emails_from_bounce(bounce_dict):
|
||||
remove_mail_headers(bounce_dict)
|
||||
bounce_dict["mail"].pop("destination", None)
|
||||
bounce_dict["bounce"].pop("bouncedRecipients", None)
|
||||
|
||||
|
||||
def remove_emails_from_complaint(complaint_dict):
|
||||
remove_mail_headers(complaint_dict)
|
||||
complaint_dict["complaint"].pop("complainedRecipients")
|
||||
return complaint_dict["mail"].pop("destination")
|
||||
|
||||
|
||||
def check_and_queue_callback_task(notification):
|
||||
# queue callback task only if the service_callback_api exists
|
||||
service_callback_api = get_service_delivery_status_callback_api_for_service(service_id=notification.service_id)
|
||||
if service_callback_api:
|
||||
notification_data = create_delivery_status_callback_data(notification, service_callback_api)
|
||||
send_delivery_status_to_service.apply_async([str(notification.id), notification_data], queue=QueueNames.CALLBACKS)
|
||||
|
||||
|
||||
def _check_and_queue_complaint_callback_task(complaint, notification, recipient):
|
||||
# queue callback task only if the service_callback_api exists
|
||||
service_callback_api = get_service_complaint_callback_api_for_service(service_id=notification.service_id)
|
||||
if service_callback_api:
|
||||
complaint_data = create_complaint_callback_data(complaint, notification, service_callback_api, recipient)
|
||||
send_complaint_to_service.apply_async([complaint_data], queue=QueueNames.CALLBACKS)
|
||||
|
||||
@@ -6,13 +6,11 @@ from flask import current_app
|
||||
from notifications_utils.template import SMSMessageTemplate
|
||||
|
||||
from app import notify_celery, statsd_client
|
||||
from app.celery.process_ses_receipts_tasks import check_and_queue_callback_task
|
||||
from app.clients import ClientException
|
||||
from app.dao import notifications_dao
|
||||
from app.dao.templates_dao import dao_get_template_by_id
|
||||
from app.models import NOTIFICATION_PENDING
|
||||
from app.notifications.notifications_ses_callback import (
|
||||
check_and_queue_callback_task,
|
||||
)
|
||||
|
||||
sms_response_mapper = {
|
||||
# 'MMG': get_mmg_responses,
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
import base64
|
||||
import re
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import oscrypto.asymmetric
|
||||
import oscrypto.errors
|
||||
import requests
|
||||
import six
|
||||
|
||||
from app import redis_store
|
||||
from app.config import Config
|
||||
|
||||
USE_CACHE = True
|
||||
VALIDATE_ARN = True
|
||||
VALID_SNS_TOPICS = Config.VALID_SNS_TOPICS
|
||||
|
||||
|
||||
_signing_cert_cache = {}
|
||||
_cert_url_re = re.compile(
|
||||
r'sns\.([a-z]{1,3}-[a-z]+-[0-9]{1,2})\.amazonaws\.com',
|
||||
)
|
||||
|
||||
class ValidationError(Exception):
|
||||
"""
|
||||
ValidationError. Raised when a message fails integrity checks.
|
||||
"""
|
||||
|
||||
|
||||
def get_certificate(url):
|
||||
if USE_CACHE:
|
||||
res = redis_store.get(url)
|
||||
if res is not None:
|
||||
return res
|
||||
res = requests.get(url).text
|
||||
redis_store.set(url, res, ex=60 * 60) # 60 minutes
|
||||
return res
|
||||
else:
|
||||
return requests.get(url).text
|
||||
|
||||
|
||||
def validate_arn(sns_payload):
|
||||
if VALIDATE_ARN:
|
||||
arn = sns_payload.get('TopicArn')
|
||||
topic_name = arn.split(':')[5]
|
||||
if topic_name not in VALID_SNS_TOPICS:
|
||||
raise ValidationError("Invalid Topic Name")
|
||||
|
||||
|
||||
def get_string_to_sign(sns_payload):
|
||||
payload_type = sns_payload.get('Type')
|
||||
if payload_type in ['SubscriptionConfirmation', 'UnsubscribeConfirmation']:
|
||||
fields = ['Message', 'MessageId', 'SubscribeURL', 'Timestamp', 'Token', 'TopicArn', 'Type']
|
||||
elif payload_type == 'Notification':
|
||||
fields = ['Message', 'MessageId', 'Subject', 'Timestamp', 'TopicArn', 'Type']
|
||||
else:
|
||||
raise ValidationError("Unexpected Message Type")
|
||||
|
||||
string_to_sign = ''
|
||||
for field in fields:
|
||||
field_value = sns_payload.get(field)
|
||||
if not isinstance(field_value, str):
|
||||
if field == 'Subject' and field_value == None:
|
||||
continue
|
||||
raise ValidationError(f"In {field}, found non-string value: {field_value}")
|
||||
string_to_sign += field + '\n' + field_value + '\n'
|
||||
if isinstance(string_to_sign, six.text_type):
|
||||
string_to_sign = string_to_sign.encode()
|
||||
return string_to_sign
|
||||
|
||||
|
||||
def validate_sns_cert(sns_payload):
|
||||
"""
|
||||
Adapted from the solution posted at
|
||||
https://github.com/boto/boto3/issues/2508#issuecomment-992931814
|
||||
Modified to swap m2crypto for oscrypto
|
||||
"""
|
||||
if not isinstance(sns_payload, dict):
|
||||
raise ValidationError("Unexpected message type {!r}".format(type(sns_payload).__name__))
|
||||
|
||||
# Amazon SNS currently supports signature version 1.
|
||||
if sns_payload.get('SignatureVersion') != '1':
|
||||
raise ValidationError("Wrong Signature Version (expected 1)")
|
||||
|
||||
validate_arn(sns_payload)
|
||||
|
||||
string_to_sign = get_string_to_sign(sns_payload)
|
||||
|
||||
# Key signing cert url via Lambda and via webhook are slightly different
|
||||
signing_cert_url = sns_payload.get('SigningCertUrl') if 'SigningCertUrl' in sns_payload else sns_payload.get('SigningCertURL')
|
||||
if not isinstance(signing_cert_url, str):
|
||||
raise ValidationError("Signing cert url must be a string")
|
||||
cert_scheme, cert_netloc, *_ = urlparse(signing_cert_url)
|
||||
if cert_scheme != 'https' or not re.match(_cert_url_re, cert_netloc):
|
||||
raise ValidationError("Cert does not appear to be from AWS")
|
||||
|
||||
certificate = _signing_cert_cache.get(signing_cert_url)
|
||||
if certificate is None:
|
||||
certificate = get_certificate(signing_cert_url)
|
||||
if isinstance(certificate, six.text_type):
|
||||
certificate = certificate.encode()
|
||||
|
||||
signature = base64.b64decode(sns_payload["Signature"])
|
||||
|
||||
try:
|
||||
oscrypto.asymmetric.rsa_pkcs1v15_verify(
|
||||
oscrypto.asymmetric.load_certificate(certificate),
|
||||
signature,
|
||||
string_to_sign,
|
||||
"sha1"
|
||||
)
|
||||
return True
|
||||
except oscrypto.errors.SignatureError:
|
||||
raise ValidationError("Invalid signature")
|
||||
@@ -1,66 +0,0 @@
|
||||
import enum
|
||||
from datetime import timedelta
|
||||
from json import decoder
|
||||
|
||||
import requests
|
||||
from flask import current_app, json
|
||||
|
||||
from app.celery.validate_sns_cert import validate_sns_cert
|
||||
from app.errors import InvalidRequest
|
||||
|
||||
|
||||
DEFAULT_MAX_AGE = timedelta(days=10000)
|
||||
|
||||
|
||||
class SNSMessageType(enum.Enum):
|
||||
SubscriptionConfirmation = 'SubscriptionConfirmation'
|
||||
Notification = 'Notification'
|
||||
UnsubscribeConfirmation = 'UnsubscribeConfirmation'
|
||||
|
||||
|
||||
class InvalidMessageTypeException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def verify_message_type(message_type: str):
|
||||
try:
|
||||
SNSMessageType(message_type)
|
||||
except ValueError:
|
||||
raise InvalidRequest("SES-SNS callback failed: invalid message type", 400)
|
||||
|
||||
|
||||
def sns_notification_handler(data, headers):
|
||||
message_type = headers.get('x-amz-sns-message-type')
|
||||
try:
|
||||
verify_message_type(message_type)
|
||||
except InvalidMessageTypeException:
|
||||
current_app.logger.exception(f"Response headers: {headers}\nResponse data: {data}")
|
||||
raise InvalidRequest("SES-SNS callback failed: invalid message type", 400)
|
||||
|
||||
try:
|
||||
message = json.loads(data.decode('utf-8'))
|
||||
except decoder.JSONDecodeError:
|
||||
current_app.logger.exception(f"Response headers: {headers}\nResponse data: {data}")
|
||||
raise InvalidRequest("SES-SNS callback failed: invalid JSON given", 400)
|
||||
|
||||
try:
|
||||
validate_sns_cert(message)
|
||||
except Exception as e:
|
||||
current_app.logger.error(f"SES-SNS callback failed: validation failed with error: Signature validation failed with error {e}")
|
||||
raise InvalidRequest("SES-SNS callback failed: validation failed", 400)
|
||||
|
||||
if message.get('Type') == 'SubscriptionConfirmation':
|
||||
url = message.get('SubscribeUrl') if 'SubscribeUrl' in message else message.get('SubscribeURL')
|
||||
response = requests.get(url)
|
||||
try:
|
||||
response.raise_for_status()
|
||||
except Exception as e:
|
||||
current_app.logger.warning(f"Attempt to raise_for_status()SubscriptionConfirmation Type message files for response: {response.text} with error {e}")
|
||||
raise InvalidRequest("SES-SNS callback failed: attempt to raise_for_status()SubscriptionConfirmation Type message failed", 400)
|
||||
current_app.logger.info("SES-SNS auto-confirm subscription callback succeeded")
|
||||
return message
|
||||
|
||||
# TODO remove after smoke testing on prod is implemented
|
||||
current_app.logger.info(f"SNS message: {message} is a valid message. Attempting to process it now.")
|
||||
|
||||
return message
|
||||
Reference in New Issue
Block a user