6、Python RESTful API 开发
1、RESTful API 概述
1-1、展示微博开放平台的 RESTfulAPI
介绍微博开放平台
在linux或mac中，$ curl 请求URL?请求参数
上
通过HTTP请求，请求到JSON的过程，就是RESTful的调用
1-2、RESTful 设计理念
REST - Representational State Transfer 表现层状态转化
资源– resources – 网络上的具体信息，如文本，图片
URI – 统一资源标识符，用来唯一的标识一个资源
URL – 统一资源定位器，用来定位某个特定的资源，如一个网址
表现层（Representation）- 把资源具体呈现出来的形式
如：纯文本，HTML，JSON
状态转移 - State Transfer
HTTP协议，是一个无状态的协议，所有的状态都保存在服务器端
GET– 获取资源
POST – 新建资源
PUT– 更新资源
DELETE – 删除资源
一个RESTful请求：$ curl –X GET/2/users/
REST构架设计6原则
UniformInferface
Stateless
Cacheable
Client-Server
Layered System
Code on Demand
1-3、Python微型Web框架Flask简介
Flasks是一个基于Werkzeug，Jinja2以及”善意”构建的Python微型Web框架，并且基于BSD 开源证书！
1-4、一个例程和总结
Debug模式：
If __name__ == ’__main__’:
(debug=True)
Debug模式下，可以看到代码，可以直接调试
在浏览器中，使用POST方法较麻烦，但是可以在命令行中进行
$ curl –X POST 127.0.0.1:5000/index/yx
2、Python RESTful API开发工具介绍及应用
2-1 Chrome 开发者工具介绍
Network
Elements
Console
使用()方法，创建一个表格
使用()方法，统计一个操作所花费的时间
实例：
进行移动端的开发，点击手机图标
2-2 Python HTTP 库 Requests 介绍
/zh_CN/latest/user/
实例：
[root@localhost ~]# python2.7
Python 2.7.11 (default, Apr 6 2016, 00:07:19)
[GCC 4.1.2 20080704 (Red Hat 4.1.2-55)] onlinux2
Type "help","copyright", "credits" or "license" for moreinformation.
>>> import requests
>>> r =('/') //使用requests发送网络请求
/usr/local/lib/python2.7/site-packages/requests/packages/urllib3/util/ssl_.py:315:SNIMissingWarning: An HTTPS request has been made, but the SNI (Subject NameIndication) extension to TLS is not available on this platform. This may causethe server to present an incorrect TLS certificate, which can cause validationfailures. For more information, seehttps:///en/latest/#snimissingwarning.
SNIMissingWarning
>>>
u'{"message":"Hello there,wayfaring stranger. If you\u2019re reading this then you probably didn\u2019tsee our blog post a couple of years back announcing that this API would goaway: /17AROg Fear not, you should be able to get what you needfrom the shiny new Events API instead.","documentation_url":"/v3/activity/events/#list-public-events"}'
>>> payload = {'key1': 'value1','key2': 'value2'} //为URL传递参数
>>> r = ("/get",params=payload)
>>> //读取服务器响应的内容，unicode字符集
u'{\n "args": {\n "key1": "value1", \n "key2": "value2"\n }, \n "headers":{\n "Accept":"*/*", \n "Accept-Encoding": "gzip, deflate", \n "Host": "",\n "User-Agent":"python-requests/2.9.1"\n },\n "origin": "114.255.40.54",\n "url":"/get?key2=value2&key1=value1"\n}\n'
>>> //访问URL
u'/get?key2=value2&key1=value1'
>>> () //返回服务器JSON格式的内容
{u'origin': u'114.255.40.54', u'headers':{u'Host': u'', u'Accept-Encoding': u'gzip, deflate', u'Accept':u'*/*', u'User-Agent': u'python-requests/2.9.1'}, u'args': {u'key2': u'value2',u'key1': u'value1'}, u'url':u'/get?key2=value2&key1=value1'}
>>> type()
<type 'unicode'>
>>> type(())
<type 'dict'>
>>>
C:\Users\Administrator>python
Python 2.7.11 (v2.7.11:6d1b6a68f775,Dec 5 2015, 20:32:19) [MSC v.1500 32 bit(
Intel)] on win32
Type "help","copyright", "credits" or "license" for moreinformation.
>>> from PIL import Image //以请求返回的二进制数据创建一张图片
>>> from StringIO import StringIO
>>> import requests
>>> r =('/zh_CN/latest/_stat
ic/')
>>> i =(StringIO())
>>> ()
>>>
//定制请求头 – 模拟一个浏览器
# requests_t.py - Browser
import requests
hdr ={"User-Agent":"Mozilla/5.0 (Windows NT 6.1; WOW64)AppleWebKit/537.36 (KHTML, like Gecko) Chrome/49.0.2623.110Safari/537.36"}
r = ('http://127.0.0.1:5000/test1', headers = hdr)
# - Server
from flask import Flask,request
app = Flask(__name__)
@('/')
def index():
return 'Hello restful'
if __name__ == '__main__':
(host='0.0.0.0', port=5000, debug=True)
2-3 实践：动手编写一个验证登录的程序
token是一个加密的字符串，其中包含用户名，过期时间，以及一些随机信息
base64对token进行加密
代码：
#
import base64
import random
import time
from flask import Flask, request
app = Flask(__name__)
users = {
"magigo": ["123456"]
}
def gen_token(uid):
token = base64.b64encode(':'.join([str(uid), str(()), str(() + 7200)]))
users[uid].append(token)
return token
def verify_token(token):
_token = base64.b64decode(token)
if not (_token.split(':')[0])[-1] == token:
return -1
if float(_token.split(':')[-1]) >= ():
return 1
else:
return 0
@('/index', methods=['POST', 'GET'])
def index():
print
return 'hello'
@('/login', methods=['POST', 'GET'])
def login():
uid, pw = base64.b64decode(['Authorization'].split(' ')[-1]).split(':')
if (uid)[0] == pw:
return gen_token(uid)
else:
return 'error'
@('/test1', methods=['POST', 'GET'])
def test():
token = ('token')
if verify_token(token) == 1:
return 'data'
else:
return 'error'
if __name__ == '__main__':
(debug=True)
#requests_r.py
import requests
# r = ('http://127.0.0.1:5000/login', auth=('magigo', '123456'))
# print
token ='bWFnaWdvOjAuMzE4MTUxNTA1MjQ4OjE0MjU4MzkzMjMuODk='
r = ('http://127.0.0.1:5000/test1', params={'token': token})
print
# bWFnaWdvOjAuMzE4MTUxNTA1MjQ4OjE0MjU4MzkzMjMuODk=
3、OAuth 2.0介绍和实现（上）
3-1 OAuth 2.0的原理介绍
OAuth – 开放授权 – 一个正式的互联网标准协议
三方协作的过程：用户、网站、微博
Token – 令牌
OAuth概述 – 其中，B步骤是关键
OAuth四种授权模式
授权码模式
简化模式
密码模式
客户端模式
授权码模式
3-2 实现OAuth 2.0协议中的必选方法
1）实现重定向的机制
#
@('/client/login')
def client_login():
uri = 'http://localhost:5000/oauth'
return redirect(uri)
# request_t.py
r =('http://localhost:5000/client/login')
print '======='
print r.history //通过history属性，取得重定向情况
#
@('/oauth')
def oauth():
return 'Please login'
返回结果：
Please login
=======
[<Response [302]>] //表示重定向
2）实现授权码的机制
auth_code = {}
def gen_code(uri):
code = (0,10000)
auth_code[code] = uri
return code
3）实现token发放的机制
@('/oauth')
def oauth():
if ('code'):
if auth_code.get(int(('code'))) ==('redirect_uri'):
return gen_token(('client_id'))
return 'Pleaselogin'
3-3 编写OAuth授权服务器
知识点：
清除OAuth 2.0协议的内容
掌握Flask的重定向操作
掌握OAuth 2.0协议中的授权模式
代码：
#
import base64
import random
import time
from flask importFlask, request, redirect
app = Flask(__name__)
users = {
"yx": ["35"]
}
redirect_uri='http://localhost:5000/client/passport'
client_id = '35'
users[client_id] = []
auth_code = {}
oauth_redirect_uri = []
# 授权码生成器
def gen_auth_code(uri):
code = (0,10000)
auth_code[code] = uri
return code
def gen_token(uid):
token = base64.b64encode(':'.join([str(uid),str(()), str(() +7200)]))
users[uid].append(token)
return token
def verify_token(token):
_token = base64.b64decode(token)
if not (_token.split(':')[0])[-1] == token:
return -1
if float(_token.split(':')[-1]) >= ():
return 1
else:
return 0
@('/index',methods=['POST','GET'])
def index():
print
return 'hello'
@('/login',methods=['POST','GET'])
def login():
uid, pw =base64.b64decode(['Authorization'].split(' ')[-1]).split(':')
if (uid)[0] == pw:
return gen_token(uid)
else:
return 'error'
@('/oauth',methods=['POST','GET'])
def oauth():
# 登录
if ('user'):
if (('user'))[0] == ('pw')and oauth_redirect_uri:
uri = oauth_redirect_uri[0] + '?code=%s' % gen_auth_code(oauth_redirect_uri[0])
return redirect(uri)
# 验证授权码，发放token
if ('code'):
if auth_code.get(int(('code'))) == ('redirect_uri'):
return gen_token(('client_id'))
#
if ('redirect_uri'):
oauth_redirect_uri.append(('redirect_uri'))
return 'please login'
# 客户端
@('/client/login',methods=['POST','GET'])
def client_login():
uri = 'http://localhost:5000/oauth?response_type=code&client_id=%s&redirect_uri=%s'% (client_id, redirect_uri)
return redirect(uri)
@('/client/passport',methods=['POST','GET'])
def client_passport():
code = ('code')
uri = 'http://localhost:5000/oauth?grant_type=authorization_code&code=%s&redirect_uri=%s&client_id=%s'% (code, redirect_uri, client_id)
return redirect(uri)
@('/test1',methods=['POST','GET'])
def test():
token = ('token')
if verify_token(token) == 1:
return 'data'
else:
return 'error'
if __name__ == '__main__':
(debug=True)
#
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import requests
r = ('http://127.0.0.1:5000/client/login')
print
print '========='
print
print '========='
print
print '========='
uri_login = ('?')[0] +'?user=yx&pw=35'
r2 = (uri_login)
print
print
print '========='
r3 = ('http://127.0.0.1:5000/test1',params={'token': })
print
4、OAuth 2.0介绍和实现（下）
4-1 Flask渲染页面及Cookies
Flask中的表单
提交表单时，使用POST方法
/en/latest/#sessions
Flask响应对象
ResponseHeaders – 浏览器的响应头
/en/latest/#about-responses
Cookies简介 – 小型文本文件
1）分类：Cookies总是存在客户端中，按在客户端中的存储位置，可分为内存Cookies和硬盘Cookies
2）用途：Cookies可以用于购物车，以及免登陆
3）缺陷：Cookies增加了网络流量，造成安全问题，无法存储大量数据
/en/latest/#id11
4-2 Token的设计以及加密方法
知识点：
无需核对用户名密码的token验证机制 – 无法被伪造的token
hmac（哈希运算消息认证码）简介
Token的基本设计原则：
Token不携带用户敏感信息
无需查询数据库，Token可以进行自我验证
hmac简介
作用：验证消息是否被篡改
公式：
使用Python的hmac模块直接计算：(‘secret123’,value).digest()
缺陷：无法抵御重放攻击
验证token的有效性：真的，没过期，合法请求
验证验证码是否一致
验证是否过期 – 生存期10分钟左右
验证这个token是否属于被请求数据的用户
4-3 最终的编码
总结：
掌握Flask渲染页面和处理响应对象
了解hmac加密验证的内容
安全建议：使用HTTPS传输，使用请求头而不是参数传递Token
实战：制作一个基于OAuth 2.0的验证登录服务
在Chrome开发者工具中，resources-> Cookies->localhost中，可以看到用户名和密码
代码：
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import base64
import random
import time
import json
import hmac
from datetime importdatetime,timedelta
from flask importFlask, request, redirect, make_response
app = Flask(__name__)
users = {
"yx": ["35"]
}
redirect_uri='http://localhost:5000/client/passport'
client_id = '123456'
users[client_id] = []
auth_code = {}
oauth_redirect_uri = []
TIMEOUT = 3600 * 2
# 新版本的token生成器
def gen_token(data):
'''
:param data: dict type
:return: base64 str
'''
data = ()
if "salt" not in data:
data["salt"] = unicode(()).decode("ascii")
if "expires"not in data:
data["expires"] = () + TIMEOUT
payload = (data).encode("utf8")
# 生成签名
sig =_get_signature(payload)
return encode_token_bytes(payload + sig)
# 授权码生成器
def gen_auth_code(uri, user_id):
code = (0,10000)
auth_code[code] = [uri, user_id]
return code
# 新版本的token验证
def verify_token(token):
'''
:param token: base64 str
:return: dict type
'''
decoded_token =decode_token_bytes(str(token))
payload = decoded_token[:-16]
sig = decoded_token[-16:]
# 生成签名
expected_sig =_get_signature(payload)
if sig != expected_sig:
return {}
data = (("utf8"))
if ('expires') >=():
return data
return 0
# 使用hmac为消息生成签名
def _get_signature(value):
"""Calculatethe HMAC signature for the given value."""
return ('secret123456', value).digest()
# 下面两个函数将base64编码和解码单独封装
def encode_token_bytes(data):
return base64.urlsafe_b64encode(data)
def decode_token_bytes(data):
return base64.urlsafe_b64decode(data)
# 验证服务器端
@('/index',methods=['POST','GET'])
def index():
return 'hello'
@('/login',methods=['POST','GET'])
def login():
uid, pw =base64.b64decode(['Authorization'].split(' ')[-1]).split(':')
if (uid)[0] == pw:
return gen_token(dict(user=uid,pw=pw))
else:
return 'error'
@('/oauth',methods=['POST','GET'])
def oauth():
# 处理表单登录， 同时设置Cookie
if =='POST' and ['user']:
u = ['user']
p = ['pw']
if (u)[0] == pand oauth_redirect_uri:
uri = oauth_redirect_uri[0] + '?code=%s' % gen_auth_code(oauth_redirect_uri[0], u)
expire_date = ()+ timedelta(minutes=1)
resp =make_response(redirect(uri))
resp.set_cookie('login','_'.join([u, p]), expires=expire_date)
return resp
# 验证授权码，发放token
if ('code'):
auth_info = auth_code.get(int(('code')))
if auth_info[0] ==('redirect_uri'):
# 可以在授权码的auth_code中存储用户名，编进token中
return gen_token(dict(client_id=('client_id'),user_id=auth_info[1]))
# 如果登录用户有Cookie，则直接验证成功，否则需要填写登录表单
if ('redirect_uri'):
oauth_redirect_uri.append(('redirect_uri'))
if ('login'):
u, p = ('login').split('_')
if (u)[0] == p:
uri = oauth_redirect_uri[0] + '?code=%s' % gen_auth_code(oauth_redirect_uri[0], u)
return redirect(uri)
return '''
<form action=""method="post">
<p><inputtype=text name=user>
<p><inputtype=text name=pw>
<p><inputtype=submit value=Login>
</form>
'''
# 客户端
@('/client/login',methods=['POST','GET'])
def client_login():
uri = 'http://localhost:5000/oauth?response_type=code&client_id=%s&redirect_uri=%s'% (client_id, redirect_uri)
return redirect(uri)
@('/client/passport',methods=['POST','GET'])
def client_passport():
code = ('code')
uri = 'http://localhost:5000/oauth?grant_type=authorization_code&code=%s&redirect_uri=%s&client_id=%s'% (code, redirect_uri, client_id)
return redirect(uri)
# 资源服务器端
@('/test1',methods=['POST','GET'])
def test():
token = ('token')
ret = verify_token(token)
if ret:
return (ret)
else:
return 'error'
if __name__ == '__main__':
(debug=True)
5、Flask-RESTful插件介绍及应用
5-1 Flask-RESTful插件介绍
Flask-RESTful概述
Flask-RESTful 是一个Flask扩展，它添加了快速构建REST APIs的支持。它当然也是一个能够跟你现有的ORM/库协同工作的轻量级的扩展。Flask-RESTful鼓励以最小设置的最佳实践。如果你熟悉 Flask 的话，Flask-RESTful 应该很容易上手。
安装：>pip install flask-restful
资源路由
Flask-restful不在使用装饰器作为路由，处理请求的部分是类，而不是flask的函数
可插拔视图：
/en/latest/
代码：
from flask import Flask
from import restful
app = Flask(__name__)
api = (app)
class HelloWorld():
def get(self):
return {'hello': 'world'}
api.add_resource(HelloWorld, '/')
if __name__ == '__main__':
(debug=True)
参数解析
代码：
# 1. 关于参数解析的部分
# 一组虚拟的数据
TODOS = {
'todo1': {'task': 1},
'todo2': {'task': 2},
'todo3': {'task': 3},
}
# 定义允许的参数为task，类型为int，以及错误时的提示
parser = ()
parser.add_argument('task', type=int, help='Please set a int task content!')
# 真正处理请求的地方
class TodoList(Resource):
def get(self):
return TODOS, 200, {'Etag': 'some-opaque-string'}
def post(self):
args = parser.parse_args()
todo_id = int(max(()).lstrip('todo')) + 1
todo_id = 'todo%i' % todo_id
TODOS[todo_id] = {'task': args['task']}
return TODOS[todo_id], 201
# 实际定义路由的地方
api.add_resource(TodoList, '/todos', '/all_tasks')
Linux下测试：
% curl http://localhost:5000/todos -d “task=4” –X POST
% curl http://localhost:5000/todos
响应域
代码：
# 2. 关于响应域的部分
# ORM的数据模型
class TodoDao(object):
def __init__(self, todo_id, task):
self.todo_id = todo_id
= task
# 这个域不会被返回
= 'active'
# marshal-蒙版
resource_fields = {
'task': ,
'uri': ('todo_ep')
}
# 真正处理请求的地方
class Todo(Resource):
# 蒙版
@marshal_with(resource_fields)
def get(self, todo_id):
return TodoDao(todo_id=todo_id, task='Remember the milk'), 200
# 实际定义路由的地方
api.add_resource(Todo, '/todos/<todo_id>', endpoint='todo_ep')
Linux下测试：
% curl http://localhost:5000/todos/4
结果：
{
"task": "Remember themilk",
"uri": "/todos/4"
}
5-2 Flask-RESTful请求解析
知识点：
必选参数与多值参数的设定
指定获取参数的位置
代码：
from flask import Flask
from importreqparse, Api,Resource
app = Flask(__name__)
api = Api(app)
USERS = {
'row1': {'name':'jilu', 'rate': [70,65]},
'row2': {'name':'bob', 'rate': [80,90, 68]},
'row3': {'name':'tim', 'rate': [90,80]},
}
parser = ()
parser.add_argument('name', type=str, required=True) //必选参数
parser.add_argument('rate', type=int, help='rate is a number',action='append')
parser.add_argument('User-Agent',type=str, location='headers') //指定获取参数的位置
# 真正处理请求的地方
class UserInfo(Resource):
def get(self):
return USERS, 200
def post(self):
args = parser.parse_args()
user_id = int(max(()).lstrip('row')) + 1
user_id = 'row%i' % user_id
USERS[user_id] = {'name': args['name'], 'rate': args['rate']}
USERS[user_id]['ua'] = ('User-Agent')
return USERS[user_id], 201
api.add_resource(UserInfo, '/')
if __name__ == '__main__':
(debug=True)
参数解析器的继承
代码：
# 解析器继承
parser_copy = ()
parser_copy.add_argument('bar', type=int)
parser_copy.replace_argument('bar', type=str, required=True) // 覆盖原先已经设定好的参数
parser_copy.remove_argument('User-Agent') //删除原先已有的参数
5-3 Flask-RESTful的响应域
Flask-RESTful提供的一种控制响应对象的方法，可以和各种ORM结合，过滤掉不想对用户暴露的内部数据结构。
知识点：
重命名域名和设置默认域名
定制响应域的字段
生成复杂、嵌套结构的响应
代码：
import json
from datetime import datetime
from import Resource,fields, marshal_with, marshal
# 基本例子
resource_fields = {
'name': ,
'address': ,
'date_updated': (dt_format='rfc822'),
}
class UserInfo(object):
def __init__(self, name, address, date_updated=()):
= name
= address
self.date_updated = date_updated
print (marshal(UserInfo('magi', 'beijing'),resource_fields))//使用marshal()而不使用marshal_with()
# class Todo(Resource):
# @marshal_with(resource_fields, envelope='resource')
# def get(self, **kwargs):
# return UserInfo('magi', 'beijing')
# 输出域别名
resource_fields2 = {
'open_name': (attribute='name'),
'address': ,
'date_updated': (dt_format='rfc822'),
}
print (marshal(UserInfo('magi','beijing'), resource_fields2))
# 输出域默认值
class UserInfo2(object):
def __init__(self, address):
= address
resource_fields3 = {
'open_name': (default='add_magi'),
'address': ,
'date_updated': (dt_format='rfc822', default=str(())),
}
(marshal(UserInfo2(address='beijing'), resource_fields3))
# 自定义输出域
class UserInfo2(object):
def __init__(self, address, flag, date_updated=()):
= address
self.date_updated = date_updated
= flag
class UrgentItem():
def format(self, value):
return "Urgent" if value & 0x01 else "Normal"
resource_fields4 = {
'open_name': (default='add_magi'),
'address': ,
'date_updated': (dt_format='rfc822'),
'priority': UrgentItem(attribute='flags'),
}
(marshal(UserInfo2(address='beijing', flag=1), resource_fields4))
# 复杂结构
resource_fields = {'name': }
resource_fields['address'] = {}
resource_fields['address']['line 1'] =(attribute='addr1')
resource_fields['address']['line 2'] =(attribute='addr2')
resource_fields['address']['city'] =
resource_fields['address']['state'] =
resource_fields['address']['zip'] =
data = {'name': 'bob', 'addr1': '123 fakestreet', 'addr2': '', 'city': 'New York', 'state': 'NY', 'zip': '10468'}
print (marshal(data,resource_fields))
# 嵌套域
address_fields = {}
address_fields['line 1'] =(attribute='addr1')
address_fields['line 2'] =(attribute='addr2')
address_fields['city'] =(attribute='city')
address_fields['state'] =(attribute='state')
address_fields['zip'] =(attribute='zip')
resource_fields = {}
resource_fields['name'] =
resource_fields['billing_address'] = (address_fields)
resource_fields['shipping_address'] =(address_fields)
address1 = {'addr1': '123 fake street','city': 'New York', 'state': 'NY', 'zip': '10468'}
address2 = {'addr1': '555 nowhere', 'city':'New York', 'state': 'NY', 'zip': '10468'}
data = { 'name': 'bob', 'billing_address':address1, 'shipping_address': address2}
print (marshal(data,resource_fields))
5-4 重构程序
对资源服务器进行：
添加数据模型
使用marshal
将flask函数替换为restful形式的类
from import restful
from importfields,marshal_with
app = Flask(__name__)
api = (app)
# 资源服务器端
# 数据模型
class Test1Data(object):
def __init__(self, client_id, expires, salt, user_id):
self.client_id = client_id
self.expires = expires
self.salt = salt
self.user_id = user_id
# marshal-蒙版
resource_fields = {
'client_id': (default=''),
'expires': (default=0.0),
#'salt':(default=0.0),
'user_id': (default=''),
#'date':(default=str(()))
}
# 新的资源服务器
class Test1():
@marshal_with(resource_fields)
def get(self):
token = ('token')
ret = verify_token(token)
if ret:
return ret
else:
return 'error'
api.add_resource(Test1, '/test1')
知识点总结：
掌握Flask-RESTful的基本使用
掌握Flask-RESTful的如何解析请求
掌握Flask-RESTful的如何处理响应域
6、HTTPs以及Flask-OAuthlib插件使用
1 HTTPs介绍及搭建 Flask HTTPs 环境
HTTPs介绍
HTTPS（全称：Hyper Text Transfer Protocol over Secure Socket Layer），是以安全为目标的HTTP通道，简单讲是HTTP的安全版。是超文本传输协议和SSL/TLS的组合，用以提供加密通讯及对网络服务器身份的鉴定。HTTPS连接经常被用于万维网上的交易支付和企业信息系统中敏感信息的传输。这个系统的最初研发由网景公司(Netscape)进行，并内置于其浏览器NetscapeNavigator中，提供了身份验证与加密通讯方法。
HTTP访问
HTTPs访问
证书
HTTPs握手过程
在Linux或mac下，搭建Flask HTTPs开发环境
需要的软件：
OpenSSL
stunnel
$ openssl req -new -x509 -days 365 -nodes-out -keyout
$ chmod 600
$ vim https_st
代码：
1 pid =
2cert = /home/yuxiang/
3debug = 7
4foreground = yes
5
6[https]
7accept = 443
8connect = 5000
$ sudo stunnel https_st
然后启动程序，输入https://localhost:5000/client/login，会出现安全连接问题
2 使用Flask-OAuthlib插件搭建OAuth 2Server
Flask-OAuthlib的基本使用
/en/latest/
安装：
pipinstall Flask-OAuthlib
原理：
# – 客户端（8000）
1） 将用户重定向到oauth授权服务器
2） 使用得到的授权码去换取token
# - 授权服务器（5000）
使用Flask-OAuthlib搭建OAuth2 Server
代码：
#
# coding: utf-8
from flask import Flask, url_for, session, request,jsonify
from flask_oauthlib.client import OAuth
CLIENT_ID= 'LyofOAKrBZnSW5GQlp7xcg9DtbgK8lo6p641lY8t'
CLIENT_SECRET ='uCysaCWYh4aGUPIE19zMstcom9kYVUz9oXIrNwmMuyU1Y6hKl6'
app = Flask(__name__)
= True
app.secret_key = 'secret'
oauth = OAuth(app)
remote= oauth.remote_app(
'remote',
consumer_key=CLIENT_ID,
consumer_secret=CLIENT_SECRET,
request_token_params={'scope':'email'},
base_url='http://127.0.0.1:5000/api/',
request_token_url=None,
access_token_url='http://127.0.0.1:5000/oauth/token',
authorize_url='http://127.0.0.1:5000/oauth/authorize'
)
# 相当于/client/login，用于重定向用户登录
@('/')
def index():
if 'remote_oauth' in session:
resp = ('me')
return jsonify()
next_url = ('next')or or None
return (
callback=url_for('authorized',next=next_url, _external=True)
)
# 相当于/client/passport，用于获取token，并存储在Session中
@('/authorized')
def authorized():
resp = remote.authorized_response()
if resp is None:
return 'Access denied: reason=%serror=%s' % (
['error_reason'],
['error_description']
)
print resp
session['remote_oauth'] =(resp['access_token'], '')
returnjsonify(oauth_token=resp['access_token'])
@
def get_oauth_token():
return ('remote_oauth')
if __name__ == '__main__':
import os
['DEBUG'] = 'true'
['OAUTHLIB_INSECURE_TRANSPORT'] = 'true'
(host='localhost', port=8000)
#
# coding: utf-8
from datetime import datetime,timedelta
from flask importFlask
from flask importsession, request
from flask importrender_template, redirect, jsonify
from flask_sqlalchemy importSQLAlchemy
from importgen_salt
from flask_oauthlib.providerimport OAuth2Provider
from importrestful
from importfields,marshal_with
app = Flask(__name__, template_folder='templates')
= True
app.secret_key = 'secret'
({
'SQLALCHEMY_DATABASE_URI': 'sqlite:///',
})
db = SQLAlchemy(app)
oauth = OAuth2Provider(app)
# 相当于/
@('/',methods=('GET','POST'))
def home():
if == 'POST':
username = ('username')
user = .filter_by(username=username).first()
if not user:
user = User(username=username)
(user)
()
session['id'] =
return redirect('/')
user = current_user()
return render_template('',user=user)
@('/client') // 生成client_id 和 client_secret
def client():
'''
为登录用户注册一个新的客户端
:return:
'''
user =current_user()
if not user:
return redirect('/')
item = Client(
client_id=gen_salt(40),
client_secret=gen_salt(50),
_redirect_uris=' '.join([
'http://localhost:8000/authorized',
'http://127.0.0.1:8000/authorized',
'http://127.0.1:8000/authorized',
'http://127.1:8000/authorized',
]),
_default_scopes='email',
user_id=,
)
(item)
()
return jsonify(
client_id=item.client_id,
client_secret=item.client_secret,
)
# 相当于oauth
@('/oauth/token',methods=['GET','POST'])
@oauth.token_handler
def access_token():
return None
# 相当于login
@('/oauth/authorize',methods=['GET','POST'])
@oauth.authorize_handler
def authorize(*args, **kwargs):
user = current_user()
if not user:
return redirect('/')
if == 'GET':
client_id = ('client_id')
client = .filter_by(client_id=client_id).first()
kwargs['client'] = client
kwargs['user'] = user
return render_template('', **kwargs)
confirm = ('confirm','no')
return confirm == 'yes'
# 老式的资源服务器的写法
@(‘/api/me’)
@oauth.require_oauth()
def me():
user =
return jsonify(username=)
if __name__ == '__main__':
db.create_all()
()
3 使用Flask-OAuthlib提供的装饰器保护资源服务器
ORM - 对象关系映射 – 其实是创建了一个可在编程语言使用的“虚拟对象数据库”
属性装饰器
@property
代码：
1）
# 存储用户信息的ORM
class User():
id = (, primary_key=True)
username = ((40), unique=True)
# 存储客户端信息的ORM
class Client():
client_id = ((40), primary_key=True)
client_secret = ((55), nullable=False)
user_id = ((''))
user = ('User')
_redirect_uris = ()
_default_scopes = ()
@property
def client_type(self):
return 'pub lic'
@property
def redirect_uris(self):
if self._redirect_uris:
return self._redirect_uris.split()
return []
@property
def default_redirect_uri(self):
return self.redirect_uris[0]
@property
def default_scopes(self):
if self._default_scopes:
return self._default_scopes.split()
return []
# 存储授权码信息的ORM
class Grant():
id = (, primary_key=True)
user_id = (
, ('',ondelete='CASCADE')
)
user = ('User')
client_id = (
(40), ('client.client_id'),
nullable=False,
)
client = ('Client')
code = ((255), index=True, nullable=False)
redirect_uri = ((255))
expires = ()
_scopes = ()
def delete(self):
(self)
()
return self
@property
def scopes(self):
if self._scopes:
return self._scopes.split()
return []
# 存储token信息的ORM
class Token():
id = (, primary_key=True)
client_id = (
(40), ('client.client_id'),
nullable=False,
)
client = ('Client')
user_id = (
, ('')
)
user = ('User')
# currently onlybearer is supported
token_type =((40))
access_token = ((255), unique=True)
refresh_token = ((255), unique=True)
expires = ()
_scopes = ()
@property
def scopes(self):
if self._scopes:
return self._scopes.split()
return []
2）
@
def load_client(client_id):
return .filter_by(client_id=client_id).first()
@
def load_grant(client_id, code):
return .filter_by(client_id=client_id,code=code).first()
@
def save_grant(client_id, code, request, *args, **kwargs):
# decide theexpires time yourself
expires =() + timedelta(seconds=100)
grant = Grant(
client_id=client_id,
code=code['code'],
redirect_uri=request.redirect_uri,
_scopes=' '.join(),
user=current_user(),
expires=expires
)
(grant)
()
return grant
@
def load_token(access_token=None, refresh_token=None):
if access_token:
return .filter_by(access_token=access_token).first()
elif refresh_token:
return .filter_by(refresh_token=refresh_token).first()
@
def save_token(token, request, *args, **kwargs):
toks = .filter_by(
client_id=.client_id,
user_id=
)
# make sure thatevery client has only one token connected to a user
for t intoks:
(t)
expires_in = ('expires_in')
expires = () +timedelta(seconds=expires_in)
tok = Token(
access_token=token['access_token'],
refresh_token=token['refresh_token'],
token_type=token['token_type'],
_scopes=token['scope'],
expires=expires,
client_id=.client_id,
user_id=,
)
(tok)
()
return tok
3）
def current_user():
if 'id' in session:
uid = session['id']
return (uid)
return None
4 重构程序
代码：
# 新的资源服务器
api = (app, decorators=[oauth.require_oauth('email')])
resource_fields = {
'username': (),
'date': (default=str(())),
'id': ()
}
class ApiMe():
@marshal_with(resource_fields)
def get(self):
user =
return user
api.add_resource(ApiMe, '/api/me')
总结：
掌握HTTPs的基本概念以及搭建Flask开发环境
掌握使用Flask-OAuthlib搭建OAuth2 Server
掌握使用Flask-OAuthlib保护资源服务器