Skip to main content

Python API

授权

验证访问令牌

安装依赖

把如下依赖添加到requirements.txt中:

requirements.txt
flask
python-dotenv
python-jose
flask-cors
six

创建 Flask 应用

创建server.js文件并初始化 Flask 应用:

server.py
import json
from six.moves.urllib.request import urlopen
from functools import wraps

from flask import Flask, request, jsonify, _request_ctx_stack
from flask_cors import cross_origin
from jose import jwt

AUTHOK_DOMAIN = 'YOUR_DOMAIN'
API_AUDIENCE = YOUR_API_AUDIENCE
ALGORITHMS = ["RS256"]

APP = Flask(__name__)

# Error handler
class AuthError(Exception):
def __init__(self, error, status_code):
self.error = error
self.status_code = status_code

@APP.errorhandler(AuthError)
def handle_auth_error(ex):
response = jsonify(ex.error)
response.status_code = ex.status_code
return response

创建 JWT 验证装饰器

server.py
# Format error response and append status code
def get_token_auth_header():
"""Obtains the Access Token from the Authorization Header
"""
auth = request.headers.get("Authorization", None)
if not auth:
raise AuthError({"code": "authorization_header_missing",
"description":
"Authorization header is expected"}, 401)

parts = auth.split()

if parts[0].lower() != "bearer":
raise AuthError({"code": "invalid_header",
"description":
"Authorization header must start with"
" Bearer"}, 401)
elif len(parts) == 1:
raise AuthError({"code": "invalid_header",
"description": "Token not found"}, 401)
elif len(parts) > 2:
raise AuthError({"code": "invalid_header",
"description":
"Authorization header must be"
" Bearer token"}, 401)

token = parts[1]
return token

def requires_auth(f):
"""Determines if the Access Token is valid
"""
@wraps(f)
def decorated(*args, **kwargs):
token = get_token_auth_header()
jsonurl = urlopen("https://"+AUTHOK_DOMAIN+"/.well-known/jwks.json")
jwks = json.loads(jsonurl.read())
unverified_header = jwt.get_unverified_header(token)
rsa_key = {}
for key in jwks["keys"]:
if key["kid"] == unverified_header["kid"]:
rsa_key = {
"kty": key["kty"],
"kid": key["kid"],
"use": key["use"],
"n": key["n"],
"e": key["e"]
}
if rsa_key:
try:
payload = jwt.decode(
token,
rsa_key,
algorithms=ALGORITHMS,
audience=API_AUDIENCE,
issuer="https://"+AUTHOK_DOMAIN+"/"
)
except jwt.ExpiredSignatureError:
raise AuthError({"code": "token_expired",
"description": "token is expired"}, 401)
except jwt.JWTClaimsError:
raise AuthError({"code": "invalid_claims",
"description":
"incorrect claims,"
"please check the audience and issuer"}, 401)
except Exception:
raise AuthError({"code": "invalid_header",
"description":
"Unable to parse authentication"
" token."}, 401)

_request_ctx_stack.top.current_user = payload
return f(*args, **kwargs)
raise AuthError({"code": "invalid_header",
"description": "Unable to find appropriate key"}, 401)
return decorated

验证 scope

def requires_scope(required_scope):
"""Determines if the required scope is present in the Access Token
Args:
required_scope (str): The scope required to access the resource
"""
token = get_token_auth_header()
unverified_claims = jwt.get_unverified_claims(token)
if unverified_claims.get("scope"):
token_scopes = unverified_claims["scope"].split()
for token_scope in token_scopes:
if token_scope == required_scope:
return True
return False

保护 API 端点

# Controllers API

# 无需授权
@APP.route("/api/public")
@cross_origin(headers=["Content-Type", "Authorization"])
def public():
response = "公开端点! 无需认证即可见."
return jsonify(message=response)

# This needs authentication
@APP.route("/api/private")
@cross_origin(headers=["Content-Type", "Authorization"])
@requires_auth
def private():
response = "私有端点! 经过认证后可见."
return jsonify(message=response)

# 需要授权
@APP.route("/api/private-scoped")
@cross_origin(headers=["Content-Type", "Authorization"])
@requires_auth
def private_scoped():
if requires_scope("read:messages"):
response = "私有端点! 经过认证且拥有 read:messages scope 后可见."
return jsonify(message=response)
raise AuthError({
"code": "Unauthorized",
"description": "您没有权限访问当前资源"
}, 403)

使用 API

在应用中调用API

curl --request GET \
--url http://localhost:3010/api/private \
--header 'Authorization: Bearer YOUR_ACCESS_TOKEN'

获取访问令牌

在单页应用 或 移动端/原生应用中, 在授权成功后,你需要获取 访问令牌. 如何获取令牌以及如何调用API将取决于您正在开发的应用程序类型和使用的框架.

更多信息请参考相关应用程序快速入门:

curl --request POST \
--url 'https://YOUR_DOMAIN/oauth/token' \
--header 'content-type: application/x-www-form-urlencoded' \
--data grant_type=client_credentials \
--data 'client_id=YOUR_CLIENT_ID' \
--data client_secret=YOUR_CLIENT_SECRET \
--data audience=YOUR_API_IDENTIFIER

测试API

1. 调用被保护端点

curl --request GET \
--url http://localhost:3010/api/private

以上调用会返回 401 HTTP (Unauthorized) 状态码.

携带 AccessToken 进行调用

curl --request GET \
--url http://localhost:3010/api/private \
--header 'Authorization: Bearer YOUR_ACCESS_TOKEN'

此时,会返回成功响应.

2. 调用被作用域保护的端点

curl --request GET \
--url http://localhost:3010/api/private-scoped \
--header 'Authorization: Bearer YOUR_ACCESS_TOKEN'