tornado | Odoo Connects Everything

tornado http basic and digest auth handler

from tornado.web import RequestHandler
from tornado.escape import utf8
from hashlib import md5

class BasicAuthHandler(RequestHandler):
    def get(self):
        realm = 'renjie'
        username = 'foo'
        password = 'bar'
        # Authorization: Basic base64("user:passwd")
        auth_header = self.request.headers.get('Authorization', None)
        if auth_header is not None:
            # Basic Zm9vOmJhcg==
            auth_mode, auth_base64 = auth_header.split(' ', 1)
            assert auth_mode == 'Basic'
            # 'Zm9vOmJhcg==' == base64("foo:bar")
            auth_username, auth_password = auth_base64.decode('base64').split(':', 1)
            if auth_username == username or auth_password == password:
                self.write('ok')
            else:
                self.write('fail')
        else:
            '''
            HTTP/1.1 401 Unauthorized
            WWW-Authenticate: Basic realm="renjie"
            '''
            self.set_status(401)
            self.set_header('WWW-Authenticate', 'Basic realm="%s"' % realm)

class DigestAuthHandler(RequestHandler):
    def get(self):
        realm = 'test'
        opaque = 'asdf'
        # Real implementations would use a random nonce.
        nonce = "1234"
        username = 'foo'
        password = 'bar'
        '''
        Authorization: Digest username="foo", 
                              realm="test", 
                              nonce="1234", 
                              uri="/auth/digest", 
                              response="e839337ef079c93238a4bf4f1ae712b3", 
                              opaque="asdf"
        '''
        auth_header = self.request.headers.get('Authorization', None)
        if auth_header is not None:
            auth_mode, params = auth_header.split(' ', 1)
            assert auth_mode == 'Digest'
            param_dict = {}
            for pair in params.split(','):
                k, v = pair.strip().split('=', 1)
                if v[0] == '"' and v[-1] == '"':
                    v = v[1:-1]
                param_dict[k] = v
            assert param_dict['realm'] == realm
            assert param_dict['opaque'] == opaque
            assert param_dict['nonce'] == nonce
            assert param_dict['username'] == username
            assert param_dict['uri'] == self.request.path
            h1 = md5(utf8('%s:%s:%s' % (username, realm, password))).hexdigest()
            h2 = md5(utf8('%s:%s' % (self.request.method,
                                     self.request.path))).hexdigest()
            digest = md5(utf8('%s:%s:%s' % (h1, nonce, h2))).hexdigest()
            if digest == param_dict['response']:
                self.write('ok')
            else:
                self.write('fail')
        else:
            self.set_status(401)
            # WWW-Authenticate: Digest realm="test", nonce="1234", opaque="asdf"
            self.set_header('WWW-Authenticate',
                            'Digest realm="%s", nonce="%s", opaque="%s"' %
                            (realm, nonce, opaque))

from tornado.web import RequestHandler

from tornado.escape import utf8

from hashlib import md5

class BasicAuthHandler(RequestHandler):

def get(self):

realm = 'renjie'

username = 'foo'

password = 'bar'

# Authorization: Basic base64("user:passwd")

auth_header = self.request.headers.get('Authorization', None)

if auth_header is not None:

# Basic Zm9vOmJhcg==

auth_mode, auth_base64 = auth_header.split(' ', 1)

assert auth_mode == 'Basic'

# 'Zm9vOmJhcg==' == base64("foo:bar")

auth_username, auth_password = auth_base64.decode('base64').split(':', 1)

if auth_username == username or auth_password == password:

self.write('ok')

else:

self.write('fail')

else:

'''

HTTP/1.1 401 Unauthorized

WWW-Authenticate: Basic realm="renjie"

'''

self.set_status(401)

self.set_header('WWW-Authenticate', 'Basic realm="%s"' % realm)

class DigestAuthHandler(RequestHandler):

def get(self):

realm = 'test'

opaque = 'asdf'

# Real implementations would use a random nonce.

nonce = "1234"

username = 'foo'

password = 'bar'

'''

Authorization: Digest username="foo",

realm="test",

nonce="1234",

uri="/auth/digest",

response="e839337ef079c93238a4bf4f1ae712b3",

opaque="asdf"

'''

auth_header = self.request.headers.get('Authorization', None)

if auth_header is not None:

auth_mode, params = auth_header.split(' ', 1)

assert auth_mode == 'Digest'

param_dict = {}

for pair in params.split(','):

k, v = pair.strip().split('=', 1)

if v[0] == '"' and v[-1] == '"':

v = v[1:-1]

param_dict[k] = v

assert param_dict['realm'] == realm

assert param_dict['opaque'] == opaque

assert param_dict['nonce'] == nonce

assert param_dict['username'] == username

assert param_dict['uri'] == self.request.path

h1 = md5(utf8('%s:%s:%s' % (username, realm, password))).hexdigest()

h2 = md5(utf8('%s:%s' % (self.request.method,

self.request.path))).hexdigest()

digest = md5(utf8('%s:%s:%s' % (h1, nonce, h2))).hexdigest()

if digest == param_dict['response']:

self.write('ok')

else:

self.write('fail')

else:

self.set_status(401)

# WWW-Authenticate: Digest realm="test", nonce="1234", opaque="asdf"

self.set_header('WWW-Authenticate',

'Digest realm="%s", nonce="%s", opaque="%s"' %

(realm, nonce, opaque))

python tornado微信公众号开发者认证示例代码

import tornado.web
import hashlib

class WechatHandler(tornado.web.RequestHandler):
    def check(self):
        #微信公众平台设置
        token = "renjie"
        #微信加密签名，signature结合了开发者填写的token参数和请求中的timestamp参数、nonce参数。
        signature = self.get_argument("signature", None)
        #时间戳
        timestamp = self.get_argument("timestamp", None)
        #随机数
        nonce = self.get_argument("nonce", None)
        #随机字符串
        echostr = self.get_argument("echostr", None)
        if signature and timestamp and nonce:
            #将token、timestamp、nonce三个参数进行字典序排序
            param = [token, timestamp, nonce]
            param.sort()
            #将三个参数字符串拼接成一个字符串进行sha1加密
            sha = hashlib.sha1("%s%s%s" % tuple(param)).hexdigest()
            #开发者获得加密后的字符串可与signature对比，标识该请求来源于微信
            if sha == signature:
                if echostr:
                    #请原样返回echostr参数内容，则接入生效，成为开发者成功，否则接入失败。
                    return echostr
                else:
                    return True
        return False

    def get(self):
        self.write(str(self.check()))

import tornado.web

import hashlib

class WechatHandler(tornado.web.RequestHandler):

def check(self):

#微信公众平台设置

token = "renjie"

#微信加密签名，signature结合了开发者填写的token参数和请求中的timestamp参数、nonce参数。

signature = self.get_argument("signature", None)

#时间戳

timestamp = self.get_argument("timestamp", None)

#随机数

nonce = self.get_argument("nonce", None)

#随机字符串

echostr = self.get_argument("echostr", None)

if signature and timestamp and nonce:

#将token、timestamp、nonce三个参数进行字典序排序

param = [token, timestamp, nonce]

param.sort()

#将三个参数字符串拼接成一个字符串进行sha1加密

sha = hashlib.sha1("%s%s%s" % tuple(param)).hexdigest()

#开发者获得加密后的字符串可与signature对比，标识该请求来源于微信

if sha == signature:

if echostr:

#请原样返回echostr参数内容，则接入生效，成为开发者成功，否则接入失败。

return echostr

else:

return True

return False

def get(self):

self.write(str(self.check()))

urllib2 vs tornado.httpclient and proxy vs auth

import urllib2

# set up authentication info
authinfo = urllib2.HTTPBasicAuthHandler()
authinfo.add_password(realm='PDQ Application',
                      uri='https://mahler:8092/site-updates.py',
                      user='klem',
                      passwd='geheim$parole')

proxy_support = urllib2.ProxyHandler({"http" : "http://ahad-haam:3128"})

# build a new opener that adds authentication and caching FTP handlers
opener = urllib2.build_opener(proxy_support, authinfo, urllib2.CacheFTPHandler)

# install it
urllib2.install_opener(opener)

f = urllib2.urlopen('http://www.python.org/')

import urllib2

# set up authentication info

authinfo = urllib2.HTTPBasicAuthHandler()

authinfo.add_password(realm='PDQ Application',

uri='https://mahler:8092/site-updates.py',

user='klem',

passwd='geheim$parole')

proxy_support = urllib2.ProxyHandler({"http" : "http://ahad-haam:3128"})

# build a new opener that adds authentication and caching FTP handlers

opener = urllib2.build_opener(proxy_support, authinfo, urllib2.CacheFTPHandler)

# install it

urllib2.install_opener(opener)

f = urllib2.urlopen('http://www.python.org/')

import tornado.httpclient

param = {
    "url" : "http://www.renjie.me/example",
    "proxy_host" : "proxy.renjie.me",
    "proxy_port" : 8080,
    "proxy_username" : "test",
    "proxy_password" : "welcome",
    "auth_mode" : "basic",
    "auth_username" : "renjie",
    "auth_password" : "CASwW8iJ"
}

http_request = tornado.httpclient.HTTPRequest(**param)
http_client = tornado.httpclient.HTTPClient()
try:
    response = http_client.fetch(http_request)
    print response.body
except tornado.httpclient.HTTPError as e:
    print "Error:", e
http_client.close()

import tornado.httpclient

param = {

"url" : "http://www.renjie.me/example",

"proxy_host" : "proxy.renjie.me",

"proxy_port" : 8080,

"proxy_username" : "test",

"proxy_password" : "welcome",

"auth_mode" : "basic",

"auth_username" : "renjie",

"auth_password" : "CASwW8iJ"

}

http_request = tornado.httpclient.HTTPRequest(**param)

http_client = tornado.httpclient.HTTPClient()

try:

response = http_client.fetch(http_request)

print response.body

except tornado.httpclient.HTTPError as e:

print "Error:", e

http_client.close()