Subscription Confirmation Support-3

This patch is the third part of subscription confirmation feature.
Support to send email to subscriber if confirmation is needed.

Change-Id: I230f5c7fbc9d19554bbcf34ce9b2f3b14230321b
Implements: blueprint subscription-confirmation-support

doc/source/subscription_confirm.rst

@@ -15,9 +15,13 @@
 The subscription Confirm Guide
 ==============================
 
-The subscription confirm feature now only support webhook with mongoDB backend.
+The subscription confirm feature now supports webhook and email with both
+mongoDB and redis backend.
 This guide shows how to use this feature:
 
+Webhook
+-------
+
 1. Set the config option "require_confirmation" and add the policy to the
 policy.json file. Then restart Zaqar-wsgi service::
 
@@ -202,3 +206,78 @@ The response::
 
 Then try to post a message. The subscriber will not receive the notification
 any more.
+
+Email
+-----
+
+1. For the email confirmation way, also need to set the config option
+"external_confirmation_url", "subscription_confirmation_email_template" and
+"unsubscribe_confirmation_email_template".
+The confirmation page url that will be used in email subscription confirmation
+before notification, this page is not hosted in Zaqar server, user should
+build their own web service to provide this web page.
+The subscription_confirmation_email_template let user to customize the 
+subscription confirmation email content, including topic, body and sender.
+The unsubscribe_confirmation_email_template let user to customize the 
+unsubscribe confirmation email content, including topic, body and sender too::
+
+    In the config file:
+    [notification]
+    require_confirmation = True
+    external_confirmation_url = http://web_service_url/
+    subscription_confirmation_email_template = topic:Zaqar Notification - Subscription Confirmation,\
+                                              body:'You have chosen to subscribe to the queue: {0}. This queue belongs to project: {1}. To confirm this subscription, click or visit this link below: {2}',\
+                                              sender:Zaqar Notifications <no-reply@openstack.org>
+    unsubscribe_confirmation_email_template = topic: Zaqar Notification - Unsubscribe Confirmation,\
+                                              body:'You have unsubscribed successfully to the queue: {0}. This queue belongs to project: {1}. To resubscribe this subscription, click or visit this link below: {2}',\
+                                              sender:Zaqar Notifications <no-reply@openstack.org>
+
+    In the policy.json file:
+    "subscription:confirm": "",
+
+2. Create a subscription.
+For email confirmation, you should create a subscription like this::
+
+    curl -i -X POST http://10.229.47.217:8888/v2/queues/test/subscriptions \
+    -H "Content-type: application/json" \
+    -H "Client-ID: de305d54-75b4-431b-adb2-eb6b9e546014" \
+    -H "X-Auth-Token: 440b677561454ea8a7f872201dd4e2c4" \
+    -d '{"subscriber":"your email address", "ttl":3600, "options":{}}'
+
+The response::
+
+    HTTP/1.1 201 Created
+    content-length: 47
+    content-type: application/json; charset=UTF-8
+    location: http://10.229.47.217:8888/v2/queues/test/subscriptions
+    Connection: close
+    {"subscription_id": "576256b03990b480617b4063"}
+
+After the subscription created, Zaqar will send a email to the email address
+of subscriber. The email specifies how to confirm the subscription.
+
+3. Click the confirmation page link in the email body
+
+4. The confirmation page will send the subscription confirmation request to
+Zaqar server automatically. User also can choose to unsubscribe by clicking
+the unsubscription link in this page, that will cause Zaqar to cancel this
+subscription and send another email to notify this unsubscription action.
+Zaqar providers two examples of those web pages that will help user to build
+their own pages::
+
+    zaqar/sample/html/subscriptionConfirmation.html
+    zaqar/sample/html/unsubscriptionConfirmation.html
+
+User can place those pages in web server like Apache to access them by browser,
+so the external_confirmation_url will be like this::
+    http://127.0.0.1:8080/subscriptionConfirmation.html
+For CORS, here used zaqar/samples/html/confirmation_web_service_sample.py
+be a simple web service for example, it will relay the confirmation request to
+Zaqar Server. So before Step 3, you should start the web service first.
+The service could be started simply by the command::
+
+    python zaqar/samples/html/confirmation_web_service_sample.py
+The service's default port is 5678. If you want to use a new port, the command
+will be like::
+
+    python zaqar/samples/html/confirmation_web_service_sample.py new_port_number

releasenotes/notes/subscription-confirmation-support-email-0c2a56cfedc5d1e2.yaml

@@ -0,0 +1,15 @@
+---
+features:
+  - This feature is the third part of subscription confirmation feature.
+    Support to send email to subscriber if confirmation is needed.
+    To use this feature, user need to set the config option
+    "external_confirmation_url", "subscription_confirmation_email_template"
+    and "unsubscribe_confirmation_email_template".
+    The confirmation page url that will be used in email subscription
+    confirmation before notification, this page is not hosted in Zaqar server,
+    user should build their own web service to provide this web page.
+    The subscription_confirmation_email_template let user to customize the
+    subscription confimation email content, including topic, body and
+    sender. The unsubscribe_confirmation_email_template let user to customize
+    the unsubscribe confimation email content, including topic, body and
+    sender too.

samples/html/confirmation_web_service_sample.py

@@ -0,0 +1,86 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License.  You may obtain a copy
+# of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+import json
+import logging
+import requests
+import sys
+import uuid
+
+try:
+    import SimpleHTTPServer
+    import SocketServer
+except Exception:
+    from http import server as SimpleHTTPServer
+    import socketserver as SocketServer
+
+
+if len(sys.argv) > 2:
+    PORT = int(sys.argv[2])
+elif len(sys.argv) > 1:
+    PORT = int(sys.argv[1])
+else:
+    PORT = 5678
+
+
+class ServerHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
+    """This is the sample service for email subscription confirmation.
+
+    """
+
+    def do_OPTIONS(self):
+        logging.warning('=================== OPTIONS =====================')
+        self.send_response(200)
+        self.send_header('Access-Control-Allow-Origin', self.headers['origin'])
+        self.send_header('Access-Control-Allow-Methods', 'PUT')
+        self.send_header('Access-Control-Allow-Headers',
+                         'client-id,confirmation-url,content-type,url-expires,'
+                         'url-methods,url-paths,url-signature,x-project-id,'
+                         'confirm')
+        self.end_headers()
+        logging.warning(self.headers)
+        return
+
+    def do_PUT(self):
+        logging.warning('=================== PUT =====================')
+        self._send_confirm_request()
+        self.send_response(200)
+        self.send_header('Access-Control-Allow-Origin', self.headers['origin'])
+        self.end_headers()
+        message = "{\"message\": \"ok\"}"
+        self.wfile.write(message)
+        logging.warning(self.headers)
+        return
+
+    def _send_confirm_request(self):
+        url = self.headers['confirmation-url']
+        confirmed_value = True
+        try:
+            if self.headers['confirm'] == "false":
+                confirmed_value = False
+        except KeyError:
+            pass
+        headers = {
+            'Accept': 'application/json',
+            'Content-Type': 'application/json',
+            'X-Project-ID': self.headers['x-project-id'],
+            'Client-ID': str(uuid.uuid4()),
+            'URL-Methods': self.headers['url-methods'],
+            'URL-Signature': self.headers['url-signature'],
+            'URL-Paths': self.headers['url-paths'],
+            'URL-Expires': self.headers['url-expires'],
+        }
+        data = {'confirmed': confirmed_value}
+        requests.put(url=url, data=json.dumps(data), headers=headers)
+
+Handler = ServerHandler
+httpd = SocketServer.TCPServer(("", PORT), Handler)
+httpd.serve_forever()

samples/html/subscriptionConfirmation.html

@@ -0,0 +1,148 @@
+<!DOCTYPE html>
+<html>
+<head>
+<script type="text/javascript" src="http://code.jquery.com/jquery-3.1.1.min.js" integrity="sha256-hVVnYaiADRTO2PzUGmuLJr8BLUSjGIZsDYGmIJLv2b8=" crossorigin="anonymous"></script>
+
+<style>
+html, body {
+    margin: 0; padding: 0;
+    background: #fff;
+    color: #333;
+    font: 12px/15px Verdana, sans-serif;
+}
+#container {
+    max-width: 520px;
+    margin: 30px;
+    padding: 0px;
+
+}
+#content h2 {
+    font: bold 16px/16px Verdana, sans-serif;
+    margin: 0;
+    padding: 0;
+    margin-bottom: 12px;
+}
+#header {
+    height: 40px;
+    padding-bottom: 0;
+    color: #e47911;
+    position:relative;
+    margin-bottom: 12px;
+}
+#header h1 {
+    font-size: 12px;
+    line-height: 12px;
+    margin: 0; padding: 0;
+    position: absolute;
+    bottom: 0;
+    right: 0;
+}
+#content {
+    padding: 12px;
+    background: #ecf5fb;
+    border: 1px solid #c9e1f4;
+    -moz-border-radius: 10px;
+    -webkit-border-radius: 10px;
+    border-radius: 10px;
+}
+p {
+    margin-top: 0;
+    margin-bottom: 12px;
+}
+.error {
+    color: #900;
+}
+.success {
+    color: #090;
+}
+code {
+    font-style: normal;
+    color: #000;
+}
+abbr {
+    font-weight: bold;
+}
+a:visited, a:hover {
+    color: #004b91;
+}
+</style>
+</head>
+<body>
+<div id="container">
+<div id="header">
+<h1>OpenStack Zaqar Service</h1>
+</div>
+<div id="content">
+<h2 id="status">Confirming subscription...</h2>
+<div id="progress">
+<noscript><p>Your browser has JavaScript disabled. <i>To confirm a subscription via this page, your browser must have JavaScript enabled.</i></p></noscript>
+</div>
+</div>
+</div>
+
+<script type="text/javascript">
+function getParameterByName( name )
+{
+  name = name.replace(/[\[]/,"\\\[").replace(/[\]]/,"\\\]");
+  var regexS = "[\\?&]"+name+"=([^&#]*)";
+  var regex = new RegExp( regexS );
+  var results = regex.exec( window.location.href );
+  if( results == null )
+    return "";
+  else
+    return results[1];
+}
+
+$(document).ready(function(){
+    var confirmationUrl = getParameterByName("Url");
+    var Signature = getParameterByName("Signature");
+    var Methods = getParameterByName("Methods");
+    var Paths = getParameterByName("Paths");
+    var Project = getParameterByName("Project");
+    var Expires = getParameterByName("Expires");
+    var Queue = getParameterByName("Queue");
+
+    var failureString = "<p>Your subscription could not be confirmed because of an error. To receive messages from the queue, please resubscribe your email address.</p>";
+
+    if (Queue == "") {
+        $("#status").html("Subscription <i>not</i> confirmed").addClass("error");
+        $("#progress").html("<p>Your subscription could not be confirmed because your queue is incomplete. Please make sure to use exactly the URL from the subscription confirmation message.</p>");
+    } else {
+        var response = $.ajax({ type: "PUT",
+                                url: "http://127.0.0.1:5678",
+                                dataType: "json",
+                                data: {'confirmed': true},
+                                beforeSend: function(request) {
+                                    request.setRequestHeader("Content-type", "application/json");
+                                    request.setRequestHeader("URL-Signature", Signature);
+                                    request.setRequestHeader("URL-Methods", Methods);
+                                    request.setRequestHeader("URL-Paths", Paths);
+                                    request.setRequestHeader("X-Project-ID", Project);
+                                    request.setRequestHeader("URL-Expires", Expires);
+                                    request.setRequestHeader("Confirmation-Url", confirmationUrl);
+                                },
+                                success: function(data, status, req){
+                                    $("#status").html("Subscription confirmed!").addClass("success");
+                                    $("#progress").html("<p>You have subscribed to the queue:<br /><abbr title=\""
+                                        + Queue
+                                        + "\">"
+                                        + Queue
+                                        + "</abbr>.</p><p>If it was not your intention to subscribe, <a href=\""
+                                        + "unsubscriptionConfirmation.html?Signature="
+                                        + Signature + "&Methods=" + Methods + "&Paths=" + Paths
+                                        + "&Project=" + Project + "&Expires=" + Expires + "&Queue=" + Queue
+                                        + "&Url=" + confirmationUrl + "&Confirm=false"
+                                        + "\">click here to unsubscribe</a>.</p>");
+                                    },
+                                error: function(req, status, error){
+                                    $("#status").html("Subscription <i>not</i> confirmed").addClass("error");
+                                    $("#progress").html(failureString);
+                                }
+                              });
+        }
+    }
+);
+</script>
+
+</body>
+</html>

samples/html/unsubscriptionConfirmation.html

@@ -0,0 +1,145 @@
+<!DOCTYPE html>
+<html>
+<head>
+<script type="text/javascript" src="http://code.jquery.com/jquery-3.1.1.min.js" integrity="sha256-hVVnYaiADRTO2PzUGmuLJr8BLUSjGIZsDYGmIJLv2b8=" crossorigin="anonymous"></script>
+
+<style>
+html, body {
+    margin: 0; padding: 0;
+    background: #fff;
+    color: #333;
+    font: 12px/15px Verdana, sans-serif;
+}
+#container {
+    max-width: 520px;
+    margin: 30px;
+    padding: 0px;
+
+}
+#content h2 {
+    font: bold 16px/16px Verdana, sans-serif;
+    margin: 0;
+    padding: 0;
+    margin-bottom: 12px;
+}
+#header {
+    height: 40px;
+    padding-bottom: 0;
+    color: #e47911;
+    position:relative;
+    margin-bottom: 12px;
+}
+#header h1 {
+    font-size: 12px;
+    line-height: 12px;
+    margin: 0; padding: 0;
+    position: absolute;
+    bottom: 0;
+    right: 0;
+}
+#content {
+    padding: 12px;
+    background: #ecf5fb;
+    border: 1px solid #c9e1f4;
+    -moz-border-radius: 10px;
+    -webkit-border-radius: 10px;
+    border-radius: 10px;
+}
+p {
+    margin-top: 0;
+    margin-bottom: 12px;
+}
+.error {
+    color: #900;
+}
+.success {
+    color: #090;
+}
+code {
+    font-style: normal;
+    color: #000;
+}
+abbr {
+    font-weight: bold;
+}
+a:visited, a:hover {
+    color: #004b91;
+}
+</style>
+</head>
+<body>
+<div id="container">
+<div id="header">
+<h1>OpenStack Zaqar Service</h1>
+</div>
+<div id="content">
+<h2 id="status">Removing subscription...</h2>
+<div id="progress">
+<noscript><p>Your browser has JavaScript disabled. <i>To confirm a subscription via this page, your browser must have JavaScript enabled.</i></p></noscript>
+</div>
+</div>
+</div>
+
+<script type="text/javascript">
+function getParameterByName( name )
+{
+  name = name.replace(/[\[]/,"\\\[").replace(/[\]]/,"\\\]");
+  var regexS = "[\\?&]"+name+"=([^&#]*)";
+  var regex = new RegExp( regexS );
+  var results = regex.exec( window.location.href );
+  if( results == null )
+    return "";
+  else
+    return results[1];
+}
+
+$(document).ready(function(){
+    var confirmationUrl = getParameterByName("Url");
+    var Signature = getParameterByName("Signature");
+    var Methods = getParameterByName("Methods");
+    var Paths = getParameterByName("Paths");
+    var Project = getParameterByName("Project");
+    var Expires = getParameterByName("Expires");
+    var Queue = getParameterByName("Queue");
+    var Confirmed = getParameterByName("Confirm");
+
+    var failureString = "<p>Your subscription could not be removed because of an error.</p>";
+
+    if (Queue == "") {
+        $("#status").html("Subscription <i>not</i> removed").addClass("error");
+        $("#progress").html("<p>Your subscription could not be removed because queue is missing. To unsubscribe, please use the full URL from the message you received.</p>");
+    } else {
+        var response = $.ajax({ type: "PUT",
+                                url: "http://127.0.0.1:5678",
+                                dataType: "json",
+                                data: {'confirmed': false},
+                                beforeSend: function(request) {
+                                    request.setRequestHeader("Content-type", "application/json");
+                                    request.setRequestHeader("URL-Signature", Signature);
+                                    request.setRequestHeader("URL-Methods", Methods);
+                                    request.setRequestHeader("URL-Paths", Paths);
+                                    request.setRequestHeader("X-Project-ID", Project);
+                                    request.setRequestHeader("URL-Expires", Expires);
+                                    request.setRequestHeader("Confirmation-Url", confirmationUrl);
+                                    request.setRequestHeader("Confirm", Confirmed);
+                                },
+                                success: function(data, status, req){
+                                    $("#status").html("Subscription removed!").addClass("success");
+                                    $("#progress").html("<p>You have removed subscription to the queue:<br /><abbr title=\""
+                                        + Queue
+                                        + "\">"
+                                        + Queue
+                                        + "</abbr>.</p>");
+                                    },
+                                error: function(req, status, error){
+                                    $("#status").html("Subscription <i>not</i> removed").addClass("error");
+                                    $("#progress").html(failureString);
+                                }
+                              });
+        }
+    }
+);
+</script>
+
+</body>
+</html>

zaqar/common/configs.py

@@ -67,6 +67,39 @@ _NOTIFICATION_OPTIONS = (
     cfg.BoolOpt('require_confirmation', default=False,
                 help='Whether the http/https/email subscription need to be '
                      'confirmed before notification.'),
+    cfg.StrOpt('external_confirmation_url',
+               help='The confirmation page url that will be used in email '
+                    'subscription confirmation before notification.'),
+    cfg.DictOpt("subscription_confirmation_email_template",
+                default={'topic': 'Zaqar Notification - Subscription '
+                                  'Confirmation',
+                         'body': 'You have chosen to subscribe to the '
+                                 'queue: {0}. This queue belongs to '
+                                 'project: {1}. '
+                                 'To confirm this subscription, '
+                                 'click or visit this link below: {2}',
+                         'sender': 'Zaqar Notifications '
+                                   '<no-reply@openstack.org>'},
+                help="Defines the set of subscription confirmation email "
+                     "content, including topic, body and sender. There is "
+                     "a mapping is {0} -> queue name, {1} ->project id, "
+                     "{2}-> confirm url in body string. User can use any of "
+                     "the three value. But they can't use more than three."),
+    cfg.DictOpt("unsubscribe_confirmation_email_template",
+                default={'topic': 'Zaqar Notification - '
+                                  'Unsubscribe Confirmation',
+                         'body': 'You have unsubscribed successfully to the '
+                                 'queue: {0}. This queue belongs to '
+                                 'project: {1}. '
+                                 'To resubscribe this subscription, '
+                                 'click or visit this link below: {2}',
+                         'sender': 'Zaqar Notifications '
+                                   '<no-reply@openstack.org>'},
+                help="Defines the set of unsubscribe confirmation email "
+                     "content, including topic, body and sender. There is "
+                     "a mapping is {0} -> queue name, {1} ->project id, "
+                     "{2}-> confirm url in body string. User can use any of "
+                     "the three value. But they can't use more than three."),
 )
 
 _NOTIFICATION_GROUP = 'notification'

zaqar/notification/notifier.py

@@ -81,7 +81,7 @@ class NotifierDriver(object):
 
     def send_confirm_notification(self, queue, subscription, conf,
                                   project=None, expires=None,
-                                  api_version=None):
+                                  api_version=None, is_unsubscribed=False):
         # NOTE(flwang): If the confirmation feature isn't enabled, just do
         # nothing. Here we're getting the require_confirmation from conf
         # object instead of using self.require_confirmation, because the
@@ -100,7 +100,15 @@ class NotifierDriver(object):
                                                           subscription['id'])
         pre_url = urls.create_signed_url(key, [url], project=project,
                                          expires=expires, methods=['PUT'])
-        message_type = MessageType.SubscriptionConfirmation.name
+        message = None
+        if is_unsubscribed:
+            message_type = MessageType.UnsubscribeConfirmation.name
+            message = ('You have unsubscribed successfully to the queue: %s, '
+                       'you can resubscribe it by using confirmed=True.'
+                       % queue)
+        else:
+            message_type = MessageType.SubscriptionConfirmation.name
+            message = 'You have chosen to subscribe to the queue: %s' % queue
 
         messages = {}
         endpoint_dict = auth.get_public_endpoint()
@@ -116,8 +124,7 @@ class NotifierDriver(object):
                     websocket_endpoint, url)
                 messages['WebSocketSubscribeURL'] = websocket_subscribe_url
         messages.update({'Message_Type': message_type,
-                         'Message': 'You have chosen to subscribe to the '
+                         'Message': message,
-                                    'queue: %s' % queue,
                          'URL-Signature': pre_url['signature'],
                          'URL-Methods': pre_url['methods'][0],
                          'URL-Paths': pre_url['paths'][0],
@@ -126,8 +133,8 @@ class NotifierDriver(object):
                          'SubscribeBody': {'confirmed': True},
                          'UnsubscribeBody': {'confirmed': False}})
         s_type = urllib_parse.urlparse(subscription['subscriber']).scheme
-        LOG.info(_LI('Begin to send %(type)s confirm notification. The request'
+        LOG.info(_LI('Begin to send %(type)s confirm/unsubscribe notification.'
-                     'body is %(messages)s'),
+                     ' The request body is %(messages)s'),
                  {'type': s_type, 'messages': messages})
 
         self._execute(s_type, subscription, [messages], conf)

zaqar/notification/tasks/mailto.py

@@ -20,32 +20,86 @@ import subprocess
 
 from oslo_log import log as logging
 
-from zaqar.i18n import _LE
+from zaqar.i18n import _, _LE
+from zaqar.notification.notifier import MessageType
 
 LOG = logging.getLogger(__name__)
 
 
 class MailtoTask(object):
 
+    def _make_confirm_string(self, conf_n, message, queue_name):
+        confirm_url = conf_n.external_confirmation_url
+        if confirm_url is None:
+            msg = _("Can't make confirmation email body, need a valid "
+                    "confirm url.")
+            LOG.error(msg)
+            raise Exception(msg)
+        param_string_signature = '?Signature=' + message.get('URL-Signature',
+                                                             '')
+        param_string_methods = '&Methods=' + message.get('URL-Methods', '')
+        param_string_paths = '&Paths=' + message.get('URL-Paths', '')
+        param_string_project = '&Project=' + message.get('X-Project-ID', '')
+        param_string_expires = '&Expires=' + message.get('URL-Expires', '')
+        param_string_confirm_url = '&Url=' + message.get('WSGISubscribeURL',
+                                                         '')
+        param_string_queue = '&Queue=' + queue_name
+        confirm_url_string = (confirm_url + param_string_signature +
+                              param_string_methods + param_string_paths +
+                              param_string_project + param_string_expires +
+                              param_string_confirm_url + param_string_queue)
+        return confirm_url_string
+
+    def _make_confirmation_email(self, body, subscription, message, conf_n):
+        queue_name = subscription['source']
+        confirm_url = self._make_confirm_string(conf_n, message,
+                                                queue_name)
+        email_body = ""
+        if body is not None:
+            email_body = body.format(queue_name, message['X-Project-ID'],
+                                     confirm_url)
+        return text.MIMEText(email_body)
+
     def execute(self, subscription, messages, **kwargs):
         subscriber = urllib_parse.urlparse(subscription['subscriber'])
         params = urllib_parse.parse_qs(subscriber.query)
         params = dict((k.lower(), v) for k, v in params.items())
-        conf = kwargs.get('conf')
+        conf_n = kwargs.get('conf').notification
         try:
             for message in messages:
-                p = subprocess.Popen(conf.notification.smtp_command.split(' '),
+                p = subprocess.Popen(conf_n.smtp_command.split(' '),
                                      stdin=subprocess.PIPE)
-                # NOTE(Eva-i): Unfortunately this will add 'queue_name' key to
+                # Send confirmation email to subscriber.
-                # our original messages(dicts) which will be later consumed in
+                if (message.get('Message_Type') ==
-                # the storage controller. It seems safe though.
+                        MessageType.SubscriptionConfirmation.name):
-                message['queue_name'] = subscription['source']
+                    content = conf_n.subscription_confirmation_email_template
-                msg = text.MIMEText(json.dumps(message))
+                    msg = self._make_confirmation_email(content['body'],
-                msg["to"] = subscriber.path
+                                                        subscription,
-                msg["from"] = subscription['options'].get('from', '')
+                                                        message, conf_n)
-                subject_opt = subscription['options'].get('subject', '')
+                    msg["to"] = subscriber.path
-                msg["subject"] = params.get('subject', subject_opt)
+                    msg["from"] = content['sender']
+                    msg["subject"] = content['topic']
+                elif (message.get('Message_Type') ==
+                        MessageType.UnsubscribeConfirmation.name):
+                    content = conf_n.unsubscribe_confirmation_email_template
+                    msg = self._make_confirmation_email(content['body'],
+                                                        subscription,
+                                                        message, conf_n)
+                    msg["to"] = subscriber.path
+                    msg["from"] = content['sender']
+                    msg["subject"] = content['topic']
+                else:
+                    # NOTE(Eva-i): Unfortunately this will add 'queue_name' key
+                    # to our original messages(dicts) which will be later
+                    # consumed in the storage controller. It seems safe though.
+                    message['queue_name'] = subscription['source']
+                    msg = text.MIMEText(json.dumps(message))
+                    msg["to"] = subscriber.path
+                    msg["from"] = subscription['options'].get('from', '')
+                    subject_opt = subscription['options'].get('subject', '')
+                    msg["subject"] = params.get('subject', subject_opt)
                 p.communicate(msg.as_string())
+                LOG.debug("Send mail successfully: %s", msg.as_string())
         except OSError as err:
             LOG.exception(_LE('Failed to create process for sendmail, '
                               'because %s.') % str(err))

zaqar/tests/unit/notification/test_notifier.py

@@ -16,6 +16,7 @@
 import json
 import uuid
 
+import ddt
 import mock
 
 from zaqar.common import urls
@@ -23,6 +24,7 @@ from zaqar.notification import notifier
 from zaqar import tests as testing
 
 
+@ddt.ddt
 class NotifierTest(testing.TestBase):
 
     def setUp(self):
@@ -314,3 +316,98 @@ class NotifierTest(testing.TestBase):
                                          str(self.project), self.api_version)
 
         self.assertFalse(mock_create_signed_url.called)
+
+    def _make_confirm_string(self, conf, message, queue_name):
+        confirmation_url = conf.notification.external_confirmation_url
+        param_string_signature = '?Signature=' + message.get('signature')
+        param_string_methods = '&Methods=' + message.get('methods')[0]
+        param_string_paths = '&Paths=' + message.get('paths')[0]
+        param_string_project = '&Project=' + message.get('project')
+        param_string_expires = '&Expires=' + message.get('expires')
+        param_string_confirm_url = '&Url=' + message.get('WSGISubscribeURL',
+                                                         '')
+        param_string_queue = '&Queue=' + queue_name
+        confirm_url_string = (confirmation_url + param_string_signature +
+                              param_string_methods + param_string_paths +
+                              param_string_project + param_string_expires +
+                              param_string_confirm_url + param_string_queue)
+        return confirm_url_string
+
+    @mock.patch('zaqar.common.urls.create_signed_url')
+    @mock.patch('subprocess.Popen')
+    def _send_confirm_notification_with_email(self, mock_popen,
+                                              mock_signed_url,
+                                              is_unsubscribed=False):
+        subscription = {'id': '5760c9fb3990b42e8b7c20bd',
+                        'subscriber': 'mailto:aaa@example.com',
+                        'source': 'test_queue',
+                        'options': {'subject': 'Hello',
+                                    'from': 'zaqar@example.com'}
+                        }
+        driver = notifier.NotifierDriver(require_confirmation=True)
+        self.conf.signed_url.secret_key = 'test_key'
+        self.conf.notification.external_confirmation_url = 'http://127.0.0.1'
+        self.conf.notification.require_confirmation = True
+
+        message = {'methods': ['PUT'],
+                   'paths': ['/v2/queues/test_queue/subscriptions/'
+                             '5760c9fb3990b42e8b7c20bd/confirm'],
+                   'project': str(self.project),
+                   'expires': '2016-12-20T02:01:23',
+                   'signature': 'e268676368c235dbe16e0e9ac40f2829a92c948288df'
+                                '36e1cbabd9de73f698df',
+                   }
+        confirm_url = self._make_confirm_string(self.conf, message,
+                                                'test_queue')
+        msg = ('Content-Type: text/plain; charset="us-ascii"\n'
+               'MIME-Version: 1.0\nContent-Transfer-Encoding: 7bit\nto:'
+               ' %(to)s\nfrom: %(from)s\nsubject: %(subject)s\n\n%(body)s')
+        if is_unsubscribed:
+            e = self.conf.notification.unsubscribe_confirmation_email_template
+            body = e['body']
+            topic = e['topic']
+            sender = e['sender']
+        else:
+            e = self.conf.notification.subscription_confirmation_email_template
+            body = e['body']
+            topic = e['topic']
+            sender = e['sender']
+        body = body.format(subscription['source'], str(self.project),
+                           confirm_url)
+        mail1 = msg % {'to': subscription['subscriber'][7:],
+                       'from': sender,
+                       'subject': topic,
+                       'body': body}
+
+        called = set()
+
+        def _communicate(msg):
+            called.add(msg)
+
+        mock_process = mock.Mock()
+        attrs = {'communicate': _communicate}
+        mock_process.configure_mock(**attrs)
+        mock_popen.return_value = mock_process
+        mock_signed_url.return_value = message
+        driver.send_confirm_notification('test_queue', subscription, self.conf,
+                                         str(self.project),
+                                         api_version=self.api_version,
+                                         is_unsubscribed=is_unsubscribed)
+        driver.executor.shutdown()
+
+        self.assertEqual(1, mock_popen.call_count)
+        options, body = mail1.split('\n\n')
+        expec_options = [options]
+        expect_body = [body]
+        called_options = []
+        called_bodies = []
+        for call in called:
+            options, body = call.split('\n\n')
+            called_options.append(options)
+            called_bodies.append(body)
+        self.assertEqual(expec_options, called_options)
+        self.assertEqual(expect_body, called_bodies)
+
+    @ddt.data(False, True)
+    def test_send_confirm_notification_with_email(self, is_unsub):
+        self._send_confirm_notification_with_email(is_unsubscribed=is_unsub)

zaqar/transport/wsgi/v2_0/__init__.py

@@ -114,7 +114,8 @@ def public_endpoints(driver, conf):
 
         ('/queues/{queue_name}/subscriptions/{subscription_id}/confirm',
          subscriptions.ConfirmResource(driver._validate,
-                                       subscription_controller)),
+                                       subscription_controller,
+                                       conf)),
 
         # Pre-Signed URL Endpoint
         ('/queues/{queue_name}/share', urls.Resource(driver)),

zaqar/transport/wsgi/v2_0/subscriptions.py

@@ -248,11 +248,14 @@ class CollectionResource(object):
 
 class ConfirmResource(object):
 
-    __slots__ = ('_subscription_controller', '_validate')
+    __slots__ = ('_subscription_controller', '_validate', '_notification',
+                 '_conf')
 
-    def __init__(self, validate, subscription_controller):
+    def __init__(self, validate, subscription_controller, conf):
         self._subscription_controller = subscription_controller
         self._validate = validate
+        self._notification = notifier.NotifierDriver()
+        self._conf = conf
 
     @decorators.TransportLog("Subscriptions confirmation item")
     @acl.enforce("subscription:confirm")
@@ -268,6 +271,22 @@ class ConfirmResource(object):
             self._subscription_controller.confirm(queue_name, subscription_id,
                                                   project=project_id,
                                                   confirmed=confirmed)
+            if confirmed is False:
+                now = timeutils.utcnow_ts()
+                now_dt = datetime.datetime.utcfromtimestamp(now)
+                ttl = self._conf.transport.default_subscription_ttl
+                expires = now_dt + datetime.timedelta(seconds=ttl)
+                api_version = req.path.split('/')[1]
+                sub = self._subscription_controller.get(queue_name,
+                                                        subscription_id,
+                                                        project=project_id)
+                self._notification.send_confirm_notification(queue_name,
+                                                             sub,
+                                                             self._conf,
+                                                             project_id,
+                                                             str(expires),
+                                                             api_version,
+                                                             True)
             resp.status = falcon.HTTP_204
             resp.location = req.path
         except storage_errors.SubscriptionDoesNotExist as ex: