SNS subscription and handler for django

test_views.py

import re

from django.core.urlresolvers import reverse

import responses

AWS_URL = re.compile('.*amazonaws.com.*')


class TestSNSSubscription(object):
    @responses.activate
    def test_receive_subscribe_request_confirm_it_successfully(self, client):
        responses.add(responses.GET, AWS_URL, status=200)
        res = subscribe_request(client)

        assert res.status_code == 200

    @responses.activate
    def test_receive_subscribe_request_confirm_it_unsuccessfully(self, client):
        responses.add(responses.GET, AWS_URL, status=403)
        res = subscribe_request(client)

        assert res.status_code == 400


def subscribe_request(client):
    payload = """{
    "Type" : "SubscriptionConfirmation",
    "MessageId" : "165545c9-2a5c-472c-8df2-7ff2be2b3b1b",
    "Token" : "2336412f37fb687f5d51e6e241d09c805a5a57b30d712f794cc5f6a988666d92768dd60a747ba6f3beb71854e285d6ad02428b09ceece29417f1f02d609c582afbacc99c583a916b9981dd2728f4ae6fdb82efd087cc3b7849e05798d2d2785c03b0879594eeac82c01f235d0e717736",
    "TopicArn" : "arn:aws:sns:us-west-2:123456789012:MyTopic",
    "Message" : "You have chosen to subscribe to the topic arn:aws:sns:us-west-2:123456789012:MyTopic.\\nTo confirm the subscription, visit the SubscribeURL included in this message.",
    "SubscribeURL" : "https://sns.us-west-2.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:us-west-2:123456789012:MyTopic&Token=2336412f37fb687f5d51e6e241d09c805a5a57b30d712f794cc5f6a988666d92768dd60a747ba6f3beb71854e285d6ad02428b09ceece29417f1f02d609c582afbacc99c583a916b9981dd2728f4ae6fdb82efd087cc3b7849e05798d2d2785c03b0879594eeac82c01f235d0e717736",
    "Timestamp" : "2012-04-26T20:45:04.751Z",
    "SignatureVersion" : "1",
    "Signature" : "EXAMPLEpH+DcEwjAPg8O9mY8dReBSwksfg2S7WKQcikcNKWLQjwu6A4VbeS0QHVCkhRS7fUQvi2egU3N858fiTDN6bkkOxYDVrY0Ad8L10Hs3zH81mtnPk5uvvolIC1CXGu43obcgFxeL3khZl8IKvO61GWB6jI9b5+gLPoBc1Q=",
    "SigningCertURL" : "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-f3ecfb7224c7233fe7bb5f59f96de52f.pem"
    }
    """  # noqa

    return client.post(
        reverse('rpm:integration.sns'),
        payload,
        content_type='text/plain',
        HTTP_X_AMZ_SNS_MESSAGE_TYPE='SubscriptionConfirmation',
        HTTP_X_AMZ_SNS_MESSAGE_ID='165545c9-2a5c-472c-8df2-7ff2be2b3b1b',
        HTTP_X_AMZ_SNS_TOPIC_ARN='arn:aws:sns:us-west-2:123456789012:MyTopic',
    )

views.py

import json
import logging

import requests
from braces.views import CsrfExemptMixin
from django.http import HttpResponse
from django.views.generic import View

logger = logging.getLogger(__file__)


class SNSSubscription(CsrfExemptMixin, View):
    message_type_header = 'HTTP_X_AMZ_SNS_MESSAGE_TYPE'

    def post(self, request):
        if self.message_type_header in request.META:
            payload = json.loads(request.body.decode('utf-8'))

            message_type = request.META[self.message_type_header]
            if message_type == 'SubscriptionConfirmation':
                subscribe_url = payload.get('SubscribeURL')
                res = requests.get(subscribe_url)
                if res.status_code != 200:
                    logger.error('Failed to verify SNS Subscription', extra={
                        'verification_reponse': res.content,
                        'sns_payload': request.body,
                    })

                    return HttpResponse(
                        'Invalid verification:\n{0}'.format(res.content),
                        status=400
                    )
            else:
                # The actual body is in the 'Message' key of the
                # notification, which in turn is also a json encoded
                # string. Which needs to be parsed as json before
                # actually being useful.
                message = json.loads(payload.get('Message'))
                logger.info('SNS Notification received', extra=dict(
                    message_type=message_type,
                    sns_payload=request.body,
                    payload_message=message,
                ))

                return self.handle_sns_message(message)

        return HttpResponse('OK')

    def handle_sns_message(self, message):
        return HttpResponse('OK')