-
OAuth2.0认证的四种模式
1.Authorization Code(授权码模式)
2.Resource Owner Password Credentials(密码模式)
3.Implicit(简化授权码模式)
4.Client Credentials(客户端模式)
-
Resource Owner Password Credentials(密码模式)
个人感觉密码模式按照使用场景可以分为2类：
- 第一类，第三方登录场景 比如A应用是我创建的应用，现在我们需要登录A应用，A应用提供了可以使用B应用的用户名密码登录的接口，步骤如下：
1.用户向A应用提供B应用的用户名和密码
2.A应用拿着B应用的用户名和密码向B应用的授权服务器（Authorization Server）请求令牌（Token）
3.A应用拿着令牌，就可以B应用中允许范围之内的信息了
假设微信能用QQ号登录
A应用就是微信
B应用就是QQ - 第二类，A应用和B应用属于是同一个应用的情况。认证服务器和资源服务器属于用一个服务器。
1.客户端发起一个post请求，请求的类型是默认类型，也就是contentType: 'application/x-www-form-urlencoded'，请求参数中必须包含grant_type,username,password，如下：
......
$.ajax({
url: $this.data('href'),
type: "POST",
//http的basic校验
headers: {Authorization: 'Basic ' + base64_str},
contentType: 'application/x-www-form-urlencoded',
data: {grant_type: 'password', username: 'andy', password: '12345678'},
success: function (data) {
access_token = data.access_token;
token_type = data.token_type;
console.log('access_token=' + access_token)
},
error: function (data) {
console.log('data=' + data);
}
});
2.授权服务器（Flask）收到请求Token的请求后，做如下操作：做Http的Basic校验，验证grant_type,username,password,根据username找到User对象，然后根据User对象生成access_token，使用Flask中的itsdangerous模块可以非常方便的生成带有效期的token（token的有效期一般为几个小时），返回给客户端
#根据授权请求，生成access_token
......
def post(self):
# 客户端http basic校验/
username = request.authorization.username
password = request.authorization.password
# 对客户端进行http basic认证
logging.debug('username=%s,password=%s', username, password)
# if (not (username == current_app.config['BASIC_HTTP_USERNAME'])
# or not (password == current_app.config['BASIC_HTTP_PASSWORD'])):
# return api_abort(code=401, message='Http basic认证失败')
grant_type = request.form.get('grant_type')
username = request.form.get('username')
password = request.form.get('password')
# 客户端的请求进行验证grant_type username password都需要验证
if grant_type is None or grant_type.lower() != 'password':
return api_abort(code=400, message='The grant type must be password.')
user = User.query.filter_by(username=username).first()
if not user or not user.validate_password(password):
return api_abort(400, message='用户名或者密码错误')
# 生成token，返回给客户端
token, expires_in = generate_token(user)
logging.debug('access_token=%s,expires_in=%s', token, expires_in)
response = jsonify({
'access_token': token,
'token_type': 'Bearer',
'expires_in': expires_in
})
# Cache-Control优先级低于Pragma，但Pragma已经逐步被抛弃
# 优先级 Pragma>Cache-Control>Expires
# 禁用客户端的缓存
response.headers['Cache-Control'] = 'no-store'
response.headers['Pragma'] = 'no-cache'
return response
......
# 注册视图
api_v1.add_url_rule('/oauth/token', view_func=AuthTokenAPI.as_view('token'), methods=['GET', 'POST'])
3.客户端根据收到的access_token，请求资源服务器上的资源。
......
$.ajax({
type: 'GET',
headers: {Authorization: token_type + " " + access_token},
url: $this.data('href'),
success: function (data) {
console.log('data=' + JSON.stringify(data));
}
});
4.资源服务器校验token，如何校验token？
- 获取token：
def get_token():
if 'Authorization' in request.headers:
try:
# 从header中按照空格截取，只截取一次
token_type, token = request.headers['Authorization'].split(None, 1)
logging.debug('token_type=%s,token=%s', token_type, token)
except ValueError:
token_type = token = None
else:
token_type = token = None
# 返回token和token的令牌（Bearer表示不记名令牌）
return token_type, token
- token校验
def validate_token(token):
s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
try:
data = s.loads(token)
except BadSignature as b:
logging.debug(b.message)
return False
except SignatureExpired as s:
logging.debug(s.message)
return False
# token中取出用户id，并进而得到用户对象
user = User.query.get(data['id'])
if not user:
# 没有查到用户，直接校验失败
return False
# 绑定全局变量
g.current_user = user
return True
- 使用装饰器，在每个视图函数或类上调用
def auth_required(f):
@functools.wraps(f)
def decorated(*args, **kwargs):
token_type, token = get_token()
if not token_type or token_type.lower() != 'bearer':
return api_abort(400, 'token的类型必须是bearer。')
if not token:
# 校验token丢失
return token_missing()
# token是否合法
if not validate_token(token):
return invalid_token()
return f(*args, **kwargs)
return decorated
......
# 模拟服务器的资源
class UserAPI(MethodView):
# 调用token校验的装饰器
decorators = [auth_required]
def get(self):
return jsonify(user_schema(g.current_user))
# 注册视图
api_v1.add_url_rule('/user', view_func=UserAPI.as_view('user'), methods=['GET'])