Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #560 from jstewmon/boto3-ses

Browse Source 
update SES backend to support domain identities and multiple recipients
This commit is contained in:
Steve Pulec 2016-03-10 08:57:29 -05:00
commit 66032ad37c
5 changed files with 236 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
moto/ses/exceptions.py Normal file
View File

@ -0,0 +1,10 @@
from __future__ import unicode_literals
from moto.core.exceptions import RESTError
class MessageRejectedError(RESTError):
code = 400
def __init__(self, message):
super(MessageRejectedError, self).__init__(
"MessageRejected", message)

80
moto/ses/models.py
View File

@ -1,70 +1,96 @@
from __future__ import unicode_literals
import email
from moto.core import BaseBackend
from .exceptions import MessageRejectedError
from .utils import get_random_message_id
RECIPIENT_LIMIT = 50
class Message(object):
def __init__(self, message_id, source, subject, body, destination):
def __init__(self, message_id):
self.id = message_id
self.source = source
self.subject = subject
self.body = body
self.destination = destination
class RawMessage(object):
def __init__(self, message_id, source, destination, raw_data):
def __init__(self, message_id):
self.id = message_id
self.source = source
self.destination = destination
self.raw_data = raw_data
class SESQuota(object):
def __init__(self, messages):
self.messages = messages
def __init__(self, sent):
self.sent = sent
@property
def sent_past_24(self):
return len(self.messages)
return self.sent
class SESBackend(BaseBackend):
def __init__(self):
self.addresses = []
self.sent_messages = []
self.domains = []
self.sent_message_count = 0
def _is_verified_address(self, address):
if address in self.addresses:
return True
user, host = address.split('@', 1)
return host in self.domains
def verify_email_identity(self, address):
self.addresses.append(address)
def verify_domain(self, domain):
self.addresses.append(domain)
self.domains.append(domain)
def list_identities(self):
return self.addresses
return self.domains + self.addresses
def delete_identity(self, identity):
self.addresses.remove(identity)
if '@' in identity:
self.addresses.remove(identity)
else:
self.domains.remove(identity)
def send_email(self, source, subject, body, destination):
if source not in self.addresses:
return False
def send_email(self, source, subject, body, destinations):
recipient_count = sum(map(len, destinations.values()))
if recipient_count > RECIPIENT_LIMIT:
raise MessageRejectedError('Too many recipients.')
if not self._is_verified_address(source):
raise MessageRejectedError(
"Email address not verified %s" % source
)
message_id = get_random_message_id()
message = Message(message_id, source, subject, body, destination)
self.sent_messages.append(message)
message = Message(message_id)
self.sent_message_count += recipient_count
return message
def send_raw_email(self, source, destination, raw_data):
def send_raw_email(self, source, destinations, raw_data):
if source not in self.addresses:
return False
raise MessageRejectedError(
"Did not have authority to send from email %s" % source
)
recipient_count = len(destinations)
message = email.message_from_string(raw_data)
for header in 'TO', 'CC', 'BCC':
recipient_count += sum(
d.strip() and 1 or 0
for d in message.get(header, '').split(',')
)
if recipient_count > RECIPIENT_LIMIT:
raise MessageRejectedError('Too many recipients.')
self.sent_message_count += recipient_count
message_id = get_random_message_id()
message = RawMessage(message_id, source, destination, raw_data)
self.sent_messages.append(message)
return message
return RawMessage(message_id)
def get_send_quota(self):
return SESQuota(self.sent_messages)
return SESQuota(self.sent_message_count)
ses_backend = SESBackend()

50
moto/ses/responses.py
View File

@ -1,4 +1,7 @@
from __future__ import unicode_literals
import base64
import six
from moto.core.responses import BaseResponse
from .models import ses_backend
@ -26,7 +29,7 @@ class EmailResponse(BaseResponse):
def verify_domain_identity(self):
domain = self.querystring.get('Domain')[0]
ses_backend.verify_domain(domain)
template = self.response_template(VERIFY_DOMAIN_DKIM_RESPONSE)
template = self.response_template(VERIFY_DOMAIN_IDENTITY_RESPONSE)
return template.render()
def delete_identity(self):
@ -42,21 +45,40 @@ class EmailResponse(BaseResponse):
body = self.querystring.get(bodydatakey)[0]
source = self.querystring.get('Source')[0]
subject = self.querystring.get('Message.Subject.Data')[0]
destination = self.querystring.get('Destination.ToAddresses.member.1')[0]
message = ses_backend.send_email(source, subject, body, destination)
if not message:
return "Did not have authority to send from email {0}".format(source), dict(status=400)
destinations = {
'ToAddresses': [],
'CcAddresses': [],
'BccAddresses': [],
}
for dest_type in destinations:
# consume up to 51 to allow exception
for i in six.moves.range(1, 52):
field = 'Destination.%s.member.%s' % (dest_type, i)
address = self.querystring.get(field)
if address is None:
break
destinations[dest_type].append(address[0])
message = ses_backend.send_email(source, subject, body, destinations)
template = self.response_template(SEND_EMAIL_RESPONSE)
return template.render(message=message)
def send_raw_email(self):
source = self.querystring.get('Source')[0]
destination = self.querystring.get('Destinations.member.1')[0]
raw_data = self.querystring.get('RawMessage.Data')[0]
raw_data = base64.b64decode(raw_data)
if six.PY3:
raw_data = raw_data.decode('utf-8')
destinations = []
# consume up to 51 to allow exception
for i in six.moves.range(1, 52):
field = 'Destinations.member.%s' % i
address = self.querystring.get(field)
if address is None:
break
destinations.append(address[0])
message = ses_backend.send_raw_email(source, destination, raw_data)
if not message:
return "Did not have authority to send from email {0}".format(source), dict(status=400)
message = ses_backend.send_raw_email(source, destinations, raw_data)
template = self.response_template(SEND_RAW_EMAIL_RESPONSE)
return template.render(message=message)
@ -99,6 +121,16 @@ VERIFY_DOMAIN_DKIM_RESPONSE = """<VerifyDomainDkimResponse xmlns="http://ses.ama
</ResponseMetadata>
</VerifyDomainDkimResponse>"""
VERIFY_DOMAIN_IDENTITY_RESPONSE = """\
<VerifyDomainIdentityResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
<VerifyDomainIdentityResult>
<VerificationToken>QTKknzFg2J4ygwa+XvHAxUl1hyHoY0gVfZdfjIedHZ0=</VerificationToken>
</VerifyDomainIdentityResult>
<ResponseMetadata>
<RequestId>94f6368e-9bf2-11e1-8ee7-c98a0037a2b6</RequestId>
</ResponseMetadata>
</VerifyDomainIdentityResponse>"""
DELETE_IDENTITY_RESPONSE = """<DeleteIdentityResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
<DeleteIdentityResult/>
<ResponseMetadata>

5
tests/test_ses/test_ses.py
View File

@ -75,11 +75,10 @@ def test_send_html_email():
def test_send_raw_email():
conn = boto.connect_ses('the_key', 'the_secret')
to = 'to@example.com'
message = email.mime.multipart.MIMEMultipart()
message['Subject'] = 'Test'
message['From'] = 'test@example.com'
message['To'] = to
message['To'] = 'to@example.com'
# Message body
part = email.mime.text.MIMEText('test file attached')
@ -93,14 +92,12 @@ def test_send_raw_email():
conn.send_raw_email.when.called_with(
source=message['From'],
raw_message=message.as_string(),
destinations=message['To']
).should.throw(BotoServerError)
conn.verify_email_identity("test@example.com")
conn.send_raw_email(
source=message['From'],
raw_message=message.as_string(),
destinations=message['To']
)
send_quota = conn.get_send_quota()

131
tests/test_ses/test_ses_boto3.py Normal file
View File

@ -0,0 +1,131 @@
from __future__ import unicode_literals
import boto3
from botocore.exceptions import ClientError
from six.moves.email_mime_multipart import MIMEMultipart
from six.moves.email_mime_text import MIMEText
import sure # noqa
from moto import mock_ses
@mock_ses
def test_verify_email_identity():
conn = boto3.client('ses', region_name='us-east-1')
conn.verify_email_identity(EmailAddress="test@example.com")
identities = conn.list_identities()
address = identities['Identities'][0]
address.should.equal('test@example.com')
@mock_ses
def test_domain_verify():
conn = boto3.client('ses', region_name='us-east-1')
conn.verify_domain_dkim(Domain="domain1.com")
conn.verify_domain_identity(Domain="domain2.com")
identities = conn.list_identities()
domains = list(identities['Identities'])
domains.should.equal(['domain1.com', 'domain2.com'])
@mock_ses
def test_delete_identity():
conn = boto3.client('ses', region_name='us-east-1')
conn.verify_email_identity(EmailAddress="test@example.com")
conn.list_identities()['Identities'].should.have.length_of(1)
conn.delete_identity(Identity="test@example.com")
conn.list_identities()['Identities'].should.have.length_of(0)
@mock_ses
def test_send_email():
conn = boto3.client('ses', region_name='us-east-1')
kwargs = dict(
Source="test@example.com",
Destination={
"ToAddresses": ["test_to@example.com"],
"CcAddresses": ["test_cc@example.com"],
"BccAddresses": ["test_bcc@example.com"],
},
Message={
"Subject": {"Data": "test subject"},
"Body": {"Text": {"Data": "test body"}}
}
)
conn.send_email.when.called_with(**kwargs).should.throw(ClientError)
conn.verify_domain_identity(Domain='example.com')
conn.send_email(**kwargs)
too_many_addresses = list('to%s@example.com' % i for i in range(51))
conn.send_email.when.called_with(
**dict(kwargs, Destination={'ToAddresses': too_many_addresses})
).should.throw(ClientError)
send_quota = conn.get_send_quota()
sent_count = int(send_quota['SentLast24Hours'])
sent_count.should.equal(3)
@mock_ses
def test_send_html_email():
conn = boto3.client('ses', region_name='us-east-1')
kwargs = dict(
Source="test@example.com",
Destination={
"ToAddresses": ["test_to@example.com"]
},
Message={
"Subject": {"Data": "test subject"},
"Body": {"Html": {"Data": "test body"}}
}
)
conn.send_email.when.called_with(**kwargs).should.throw(ClientError)
conn.verify_email_identity(EmailAddress="test@example.com")
conn.send_email(**kwargs)
send_quota = conn.get_send_quota()
sent_count = int(send_quota['SentLast24Hours'])
sent_count.should.equal(1)
@mock_ses
def test_send_raw_email():
conn = boto3.client('ses', region_name='us-east-1')
message = MIMEMultipart()
message['Subject'] = 'Test'
message['From'] = 'test@example.com'
message['To'] = 'to@example.com, foo@example.com'
# Message body
part = MIMEText('test file attached')
message.attach(part)
# Attachment
part = MIMEText('contents of test file here')
part.add_header('Content-Disposition', 'attachment; filename=test.txt')
message.attach(part)
kwargs = dict(
Source=message['From'],
RawMessage={'Data': message.as_string()},
)
conn.send_raw_email.when.called_with(**kwargs).should.throw(ClientError)
conn.verify_email_identity(EmailAddress="test@example.com")
conn.send_raw_email(**kwargs)
send_quota = conn.get_send_quota()
sent_count = int(send_quota['SentLast24Hours'])
sent_count.should.equal(2)