222      阿裏雲  技術社區[雲棲]

HttpEndpoint 示例代碼__Python SDK_SDK使用手冊_消息服務-阿裏雲

這裏僅展示HttpEndpoint 部分核心代碼，完整的代碼請參考 python sdk中simple_http_notify_endpoint.py。

class SimpleHttpNotifyEndpoint(BaseHTTPServer.BaseHTTPRequestHandler):
    server_version = "SimpleHttpNotifyEndpoint/" + __version__
    access_log_file = "access_log"
    msg_type = "XML"
    def do_POST(self):
        content_length = int(self.headers.getheader('content-length', 0))
        self.req_body = self.rfile.read(content_length)
        self.msg = NotifyMessage()
        logger.info("Headers:%snBody:%s" % (self.headers, self.req_body))
        if not self.authenticate():
            res_code = 403
            res_content = "Access Forbidden"
            logger.warning("Access Forbidden!nHeaders:%snReqBody:%sn" % (self.headers, self.req_body))
        elif not self.validateBody(self.req_body, self.msg, self.msg_type):
            res_code = 400
            res_content = "Invalid Notify Message"
            logger.warning("Invalid Notify Message!nHeaders:%snReqBody:%sn" % (self.headers, self.req_body))
        else:
            res_code = 201
            res_content = ""
            logger.info("Notify Message Succeed!n%s" % self.msg)
        self.access_log(res_code)
        self.response(res_code, res_content)
    def authenticate(self):
        #get string to signature
        service_str = "n".join(sorted(["%s:%s" % (k,v) for k,v in self.headers.items() if k.startswith("x-mns-")]))
        sign_header_list = []
        for key in ["content-md5", "content-type", "date"]:
            if key in self.headers.keys():
                sign_header_list.append(self.headers.getheader(key))
            else:
                sign_header_list.append("")
        str2sign = "%sn%sn%sn%s" % (self.command, "n".join(sign_header_list), service_str, self.path)
        #verify
        authorization = self.headers.getheader('Authorization')
        signature = base64.b64decode(authorization)
        cert_str = urllib2.urlopen(base64.b64decode(self.headers.getheader('x-mns-signing-cert-url'))).read()
        pubkey = M2Crypto.X509.load_cert_string(cert_str).get_pubkey()
        pubkey.reset_context(md='sha1')
        pubkey.verify_init()
        pubkey.verify_update(str2sign)
        return pubkey.verify_final(signature)
    def validateBody(self, data, msg, type):
        if type == "XML":
            return self.xml_decode(data, msg)
        else:
            msg.message = data
            return True
    def xml_decode(self, data, msg):
        if data == "":
            logger.error("Data is "".")
            return False
        try:
            dom = xml.dom.minidom.parseString(data)
        except Exception, e:
            logger.error("Parse string fail, exception:%s" % e)
            return False
        node_list = dom.getElementsByTagName("Notification")
        if not node_list:
            logger.error("Get node of "Notification" fail:%s" % e)
            return False
        data_dic = {}
        for node in node_list[0].childNodes:
            if node.nodeName != "#text" and node.childNodes != []:
                data_dic[node.nodeName] = str(node.childNodes[0].nodeValue.strip())
        key_list = ["TopicOwner", "TopicName", "Subscriber", "SubscriptionName", "MessageId", "MessageMD5", "Message", "PublishTime"]
        for key in key_list:
            if key not in data_dic.keys():
                logger.error("Check item fail. Need "%s"." % key)
                return False
        msg.topic_owner = data_dic["TopicOwner"]
        msg.topic_name = data_dic["TopicName"]
        msg.subscriber = data_dic["Subscriber"]
        msg.subscription_name = data_dic["SubscriptionName"]
        msg.message_id = data_dic["MessageId"]
        msg.message_md5 = data_dic["MessageMD5"]
        msg.message_tag = data_dic["MessageTag"] if data_dic.has_key("MessageTag") else ""
        msg.message = data_dic["Message"]
        msg.publish_time = data_dic["PublishTime"]
        return True