基于Flask开发企业级REST API应用（二）

阳仔

发布于 2019-08-06 10:25:17

9460

发布于 2019-08-06 10:25:17

文章被收录于专栏：终身开发者

关于我编程界的一名小小程序猿，目前在一个创业团队任team lead，技术栈涉及Android、Python、Java和Go，这个也是我们团队的主要技术栈。 Github：https://github.com/hylinux1024 微信公众号：angrycode

本节开始项目的编码实现。首先我们来实现登录注册模块的相关 API。本项目我们是使用前后端分离的模式，在实现登录注册功能之前，假设我们的接口是开放的，那么需要确定接口校验方案。

0x00 接口校验方案

我们的目标是接口不能被抓包重复访问，并且要对客户端的可靠性进行验证。

防重放攻击可使用参数有 timestamp、 nonce、 token 和 sign
支持可信任的客户端的请求则可以考虑添加 appkey 和 appsecret参数

公共参数

1、 timestamp 时间戳

单位毫秒，也可以是秒，与服务端保持一致；时间戳是无关时区的，所以客户端与服务端的时间戳是可以用来比较的；如果客户端与服务端时间戳相差比较大，则可以考虑使用服务端时间进行校准；时间戳的作用是，保证这个请求在一定时间内（例如60秒内）是有效的。有效期内的校验就需要 nonce参数

2、 nonce 随机数

由客户端产生的随机数，客户端每次接口请求时需要保证它是不一样的。 nonce的作用是保证 timestamp有效期内的请求是否是合法的。 服务端接收到这个参数后，会将其保存在某个集合中。服务端会检测这个 nonce是否在该集合中出现过，如果出现过说明该请求是不合法的。每个 nonce的有效期设置跟 timestamp参数有关，例如可以设置为60秒。

3、 token 登录态

需要登录的接口则需要一个 token参数。服务端生成的 token在有效期内有效，如果 token过期则需要提示客户端重新登录。 token的生成规则可使用随机数

token = md5(1024位的随机数)

4、 sign 签名或校验参数

msg = 除了timestamp、nonce、token、sign参数之外的其它排序后的参数列表和值列表 = sort(参数1=值1&参数2=值2&参数3=值3...)

sign = md5(msg+token+timestamp+nonce+salt) 

salt = 客户端与服务端约定字符串

5、 appkey 和 appsecret

服务端为可信任的客户端分配 appkey和 appsecret参数。可由随机数或自定义的规则生成，要保证 appkey和 appsecret是对应的。客户端需保证 appsecret不被泄露。客户端接口请求时只需带上 appkey参数。 appsecret则添加到 sign校验参数的计算中

sign = md5(token+msg+timestamp+nonce+appsecret)

结合上面的参数，一个接口请求应该类似这样

http://api.example.com/v1/login?phone=13499990000&timestamp=1564486841415&nonce=34C2AF&sign=e10adc3949ba59abbe56e057f20f883e&appkey=A23CE80D

服务端程序接收到请求后验证流程应该是这样的

通过 appkey查询到 appsecret，如果查不到则返回出错信息，否则继续；
通过 timestamp检查 nonce是否在有效时间内是的重复请求，如果是多次重复请求，则返回出错信息，否则继续；
通过请求参数构造 msg并计算 sign，将此参数与请求中获取到的参数进行对比，验证成功后才开始我们的业务逻辑。

这样我们的一个简单实用的接口验证方案就出来了，当然可能还有其它一些好的想法，欢迎留言一起探讨学习。

0x01 show me the code

现在开始实现登录注册功能，相信这个模块走通了，之后其它模块也是依样画葫芦。

先看下模块

├── api
│   ├── __init__.py
│   └── auth.py
├── app.py
├── config.ini
├── datingtoday.sql
├── models.py
├── requirements.txt
├── test
└── venv

增加了一个 api相关的文件包。还有一个 config.ini，主要用于配置数据库等信息，而 models.py文件是定义实体类的地方。

`api/init.py`

from flask import jsonify

def make_response_ok(data=None):
    resp = {'code': 0, 'msg': 'success'}
    if data:
        resp['data'] = data
    return jsonify(resp)

def make_response_error(code, msg):
    resp = {'code': code, 'msg': msg}
    return jsonify(resp)

def validsign(func):
    """
    验证签名
    :param func:
    :return:
    """

    def decorator():
        params = request.form
        appkey = params.get('appkey')
        sign = params.get('sign')
        csign = signature(params)
        if not appkey:
            return make_response_error(300, 'appkey is none.')
        if csign != sign:
            return make_response_error(500, 'signature is error.')
        return func()

    return decorator

在 __init__.py中首先定义了两个封装统一的 json数据结构的的方法，主要是用到 flask中的 jsonify函数，它可以把一个对象转成 json。

在前面我们讲了接口的验证逻辑，这一部分对参数的校验功能其实是可以通用的，所以对这个逻辑也进行了封装成 validsign方法。

不错，这是一个装饰器的定义。我们希望在接口访问的方法使用装饰器，就可以进行通用的接口校验。

`auth.py`

这一节的重点是实现登录注册和发短信接口，因此创建一个 auth.py的文件来写跟授权登录相关的接口，这样有利于我们组织代码。我们知道要实现接口的访问路径的定义与方法直接的对应，是使用 @route这个装饰器。这里我们在一个新文件中定义我们的接口，就需要用到 Blueprint

A blueprint is an object that allows defining application functions without requiring an application object ahead of time. It uses the same decorators as Flask, but defers the need for an application by recording them for later registration.

说白了，它的作用跟 @route差不多。

由于我们把登录注册当作一个接口来实现，即用户通过短信进行登录，后端会判断该用户是否为新用户，如果是新用户则自动注册。

0x02 短信接口

首先定义接口的访问路径为

{host:port}/api/auth/sendsms

请求方法：POST
参数：phone
请求成功
{
    "code": 0,
    "data": {
        "code": "97532",
        "phone": "18922986865"
    },
    "msg": "success"
}

根据接口定义我们会在 auth.py中定义一个 Blueprint对象用来映射我们的访问路径和方法。

bp = Blueprint("auth", __name__, url_prefix='/api/auth')

短信接口的实现这里会使用到 redis，将请求到的短信验证码保存在 redis中，并设置过期时间。然后登录时，再进行验证。

@bp.route("/sendsms", methods=['POST'], endpoint="sendsms")
@validsign
def send_sms():
    phone = request.form.get('phone')
    m = re.match(pattern_phone, phone)
    if not m:
        return make_response_error(300, 'phone number format error.')
    # 这里需要修改为对接短信服务
    code = '97532'
    key = f'{phone}-{code}'
    r.set(key, code, 60)
    return make_response_ok({'phone': phone, 'code': code})

注意这里的 endpoint="sendsms"是必需设置，因为 @validsign会修饰我们的方法，每个方法都是用一个通用的校验，方法名称会变成一样的，所以如果不设置 endpoint会导致 url映射失败。

0x03 登录注册接口

首先定义接口的访问路径为

{host:port}/api/auth/login

请求方法：POST
参数：phone
参数：code
请求成功
{
    "code": 0,
    "data": {
        "expire_time": "2019-08-10 07:34:20",
        "token": "5bea89727e7553284f162d35c9926414",
        "user_id": 100784
    },
    "msg": "success"
}

执行登录接口时，会先验证 redis中的验证码，然后查一下授权表 user_auth看看是否是新用户，最后返回用户的登录授权信息。

@bp.route("/login", methods=['POST'], endpoint='login')
@validsign
def login():
    phone = request.form.get('phone')
    code = request.form.get('code')
    key = f'{phone}-{code}'
    sms_code = r.get(key)
    if sms_code:
        sms_code = sms_code.decode()
    if code != sms_code:
        return make_response_error(503, 'sms code error')
    auth_info = UserAuth.query.filter_by(open_id=phone).first()
    if not auth_info:
        auth_info = register_by_phone(phone)
    else:
        auth_info = login_by_phone(auth_info)

    data = {'token': auth_info.token,
            'expired_time': auth_info.expired_time.strftime("%Y-%m-%d %H:%M:%S"),
            'user_id': auth_info.user_basic.id}

    r.set(f'auth_info_{auth_info.user_id}', str(data))
    return make_response_ok(data)

总体上逻辑还是比较清晰的，最后我们看一下 app.py

from flask import Flask

from api import auth, config
from models import db

app = Flask(__name__)
# 将blueprint注册到app中
app.register_blueprint(auth.bp)
# 配置app的config，将数据库信息配置好
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config["SQLALCHEMY_DATABASE_URI"] = config['DATABASE']['uri']
# 最好生成一个secret_key
app.secret_key = '8c2c0b555e6e6cb01a5fd36dd981bcee'

db.init_app(app)

@app.route('/')
def hello_world():
    return 'Hello World!'


if __name__ == '__main__':
    app.run()

配置文件 config.ini

# 配置数据库链接
[DATABASE]
uri = mysql+pymysql://user:password@127.0.0.1:3306/datingtoday

# 配置appkey和secret
[APP]
appkey = 432ABZ
appsecret = 1AE32B09224

0x04 单元测试

由于接口都需要动态计算校验码，所以单元测试是必需的。这里我使用最简单的方式，直接使用 unittest模块。

例如测试发短信的业务接口，首先生成一个随机数 nonce，然后计算校验码 sign参数，最后调用 flask中的 post方法模拟接口请求。

def test_sendsms(self):
    import math
    nonce = math.floor(random.uniform(100000, 1000000))
    params = {'phone': '18922986865', 'appkey': '432ABZ', 'timestamp': datetime.now().timestamp(),
              'nonce': nonce}
    sign = signature(params)
    params['sign'] = sign

    respdata = self.app.post("/api/auth/sendsms", data=params)
    resp = respdata.json
    self.assertEqual(resp['code'], 0, respdata.data)

如果请求成功，就认为通过测试。当然这里的逻辑还是比较简单，希望小伙伴们留言讨论。

0x05 项目地址

源码地址： https://github.com/hylinux1024/datingtoday

Flask官方地址： https://palletsprojects.com/p/flask/

注意本文会使用到 mysql和 redis数据库，需要自行安装。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2019-08-03，如有侵权请联系 cloudcommunity@tencent.com 删除

python

云数据库 Redis®

unix

本文分享自终身开发者微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

python

云数据库 Redis®

unix

登录后参与评论

0 条评论

热度