"""
Confirmation handling API that helps you get the whole confirm/pending/verify
process correct. It doesn't implement any handlers, but what it does do is
provide the logic for doing the following:
* Take an email, put it in a "pending" queue, and then send out a confirm
email with a strong random id.
* Store the pending message ID and the random secret someplace for later
verification.
* Verify an incoming email against the expected ID, and get back the
original.
You then just work this into your project's state flow, write your own
templates, and possibly write your own storage.
"""
from email.utils import parseaddr
import uuid
from salmon import queue, view
[docs]class ConfirmationStorage:
"""
This is the basic confirmation storage. For simple testing purposes
you can just use the default hash db parameter. If you do a deployment
you can probably get away with a shelf hash instead.
You can write your own version of this and use it. The confirmation engine
only cares that it gets something that supports all of these methods.
"""
def __init__(self, db={}):
"""
Change the db parameter to a shelf to get persistent storage.
"""
self.confirmations = db
[docs] def clear(self):
"""
Used primarily in testing, this clears out all pending confirmations.
"""
self.confirmations.clear()
[docs] def key(self, target, from_address):
"""
Used internally to construct a string key, if you write
your own you don't need this.
NOTE: To support proper equality and shelve storage, this encodes the
key into ASCII. Make a different subclass if you need Unicode and your
storage supports it.
"""
key = target + ':' + from_address
return key.encode('ascii')
[docs] def get(self, target, from_address):
"""
Given a target and a from address, this returns a tuple of (expected_secret, pending_message_id).
If it doesn't find that target+from_address, then it should return a (None, None) tuple.
"""
return self.confirmations.get(self.key(target, from_address), (None, None))
[docs] def delete(self, target, from_address):
"""
Removes a target+from_address from the storage.
"""
try:
del self.confirmations[self.key(target, from_address)]
except KeyError:
pass
[docs] def store(self, target, from_address, expected_secret, pending_message_id):
"""
Given a target, from_address it will store the expected_secret and pending_message_id
of later verification. The target should be a string indicating what is being
confirmed. Like "subscribe", "post", etc.
When implementing your own you should *never* allow more than one target+from_address
combination.
"""
self.confirmations[self.key(target, from_address)] = (expected_secret,
pending_message_id)
[docs]class ConfirmationEngine:
"""
The confirmation engine is what does the work of sending a confirmation,
and verifying that it was confirmed properly. In order to use it you
have to construct the ConfirmationEngine (usually in settings module) and
you write your confirmation message templates for sending.
The primary methods you use are ConfirmationEngine.send and ConfirmationEngine.verify.
"""
def __init__(self, pending_queue, storage):
"""
The pending_queue should be a string with the path to the salmon.queue.Queue
that will store pending messages. These messages are the originals the user
sent when they tried to confirm.
Storage should be something that is like ConfirmationStorage so that this
can store things for later verification.
"""
self.pending = queue.Queue(pending_queue)
self.storage = storage
[docs] def get_pending(self, pending_id):
"""
Returns the pending message for the given ID.
"""
return self.pending.get(pending_id)
[docs] def push_pending(self, message):
"""
Puts a pending message into the pending queue.
"""
return self.pending.push(message)
[docs] def delete_pending(self, pending_id):
"""
Removes the pending message from the pending queue.
"""
self.pending.remove(pending_id)
[docs] def cancel(self, target, from_address, expect_secret):
"""
Used to cancel a pending confirmation.
"""
name, addr = parseaddr(from_address)
secret, pending_id = self.storage.get(target, addr)
if expect_secret and secret == expect_secret:
self.storage.delete(target, addr)
self.delete_pending(pending_id)
[docs] def make_random_secret(self):
"""
Generates a random uuid as the secret, in hex form.
"""
return uuid.uuid4().hex
[docs] def register(self, target, message):
"""
Don't call this directly unless you know what you are doing.
It does the job of registering the original message and the
expected confirmation into the storage.
"""
from_address = message.From
pending_id = self.push_pending(message)
secret = self.make_random_secret()
self.storage.store(target, from_address, secret, pending_id)
return "%s-confirm-%s" % (target, secret)
[docs] def verify(self, target, from_address, expect_secret):
"""
Given a target (i.e. "subscribe", "post", etc), a from_address
of someone trying to confirm, and the secret they should use, this
will try to verify their confirmation. If the verify works then
you'll get the original message back to do what you want with.
If the verification fails then you are given None.
The message is *not* deleted from the pending queue. You can do
that yourself with delete_pending.
"""
name, addr = parseaddr(from_address)
secret, pending_id = self.storage.get(target, addr)
if expect_secret and secret == expect_secret:
self.storage.delete(target, addr)
return self.get_pending(pending_id)
[docs] def send(self, relay, target, message, template, vars):
"""
This is the method you should use to send out confirmation messages.
You give it the relay, a target (i.e. "subscribe"), the message they
sent requesting the confirm, your confirmation template, and any
vars that template needs.
The result of calling this is that the template message gets sent through
the relay, the original message is stored in the pending queue, and
data is put into the storage for later calls to verify.
"""
confirm_address = self.register(target, message)
vars.update(locals())
msg = view.respond(vars, template, To=message['from'],
From="%(confirm_address)s@%(host)s",
Subject="Confirmation required")
msg['Reply-To'] = "%(confirm_address)s@%(host)s" % vars
relay.deliver(msg)
[docs] def clear(self):
"""
Used in testing to make sure there's nothing in the pending
queue or storage.
"""
self.pending.clear()
self.storage.clear()