【Sanic框架】第三章：注册登录token验证

妖狐艹你老母 2023-02-12 15:23 15阅读 0赞

本来这一张是想写中间件的使用的，但是又一想，单纯的解释中间件的使用又有点调阅，还是先写个前奏就当是举个例子了。

### 注册 ###

在写接口的时候需要注意的是在传参的过程中我们程序并不知道他有没有传参，需不需要传参，所以就得判断参数的必选项。这个要和前段商量，是否是必选项就不说了，这里说一下判断是否是必填项。一般普通代码都会在接口中判断是否有这个参数，原理是这么回事，但是最好不要在接口中写，要单独写一个方法，以便适用于所有接口。首先在建文件/until/public/pub\_func.py文件。代码：

#/until/public/pub_func.py
    # 0选填， 1必填
    class Param:
        def __init__(self, request):
            content_type = request.content_type if request.content_type else ""
            # 这里需要自己判断自己需要什么方法传参，get传参都一样，主要post，我这里用raw中的json方法
            self.__param = request.args if request.method == "GET" or "multipart/form-data" in content_type else request.json
    
        def checkParam(self, key, code):
            if self.__param is None:
                raise RuntimeError({'runCode': "1000", "runMsg": key + "参数不存在"})
            else:
                param = self.__param.get(key)
                if param is None:
                    if code == 1:
                        raise RuntimeError({'runCode': "1000", "runMsg": key + "参数不存在"})
                    else:
                        return ''
                else:
                    if code == 1:
                        if len(param) == 0:
                            raise RuntimeError({'runCode': "1000", "runMsg": key + "参数不存在"})
                    return param
    
        def HandleParam(self, ParamStr):
            RealParam = {k: self.checkParam(k, v) for (k, v) in ParamStr.items()}
            return RealParam

这里有个验证码，就不说了，想了解的可以去百度一下，我这里就不写这个功能了，随便什么验证码都可以。代码：

#/routes/api_v1/login.py
    from routes import api
    from dal import login_dal
    from sanic.response import json
    from until.config import rsa_public
    from until.public import pub_func
    import uuid
    
    
    @api.route('/app/register', methods=['POST'])
    def register(request):
        param = pub_func.Param(request)
        try:
            ParamStr = {'iPhone': 1, "code": 1, "user_pw": 1}
            RealParam = param.HandleParam(ParamStr)
            iPhone = RealParam.get('iPhone')
            # 验证是否是手机号
            is_iphone = pub_func.is_phone(iPhone)
            if not is_iphone:
                return json({"error": "请输入正确的手机号"})
            code = RealParam.get('code')
            # 验证验证码省略。。
            # 判断手机号是否已注册过？
            result = login_dal.get_user_iphone(iPhone)
            if result:
                return json({"error": "改手机号已注册"})
            user_pw = RealParam.get('user_pw')
            md5_pw = rsa_public.MD5(user_pw)
            user_id = str(uuid.uuid4()).replace('-', '')
            result = login_dal.set_user(user_id, md5_pw, iPhone)
            if result:
                return json({"success": "注册成功"})
            else:
                return json({"error": "注册失败"})
        except Exception as e:
            return json({"error": e.args})

sql语句代码：

#/dal/login_dal.py
    from until.base.mysql import mysqlHelper
    
    mySql = mysqlHelper()
    
    def set_user(user_id, user_pw, iPhone):
        return mySql.INSERT(f"insert into `user`(user_id, user_pw, iPhone, registerTime) values ('{user_id}', '{user_pw}', '{iPhone}', NOW())")
    
    
    def get_user_iphone(iPhone):
        return mySql.selectOne(f"select * from `user` where iPhone = {iPhone}")

验证下是否可以正常注册。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J0YV9iaA_size_16_color_FFFFFF_t_70][]

### 登录 ###

首先在数据库中新建一个用户表格user,添加字段账号和密码，添加一个账号和密码，在router/api\_v1下建一个路由文件模块login.py文件，然后再在dal文件夹下建一个login\_dal.py文件用来写sql语句，代码：

#/routes/api_v1/login.py
    
    from routes import api
    from dal import login_dal
    from sanic.response import json
    from until.config import rsa_public
    from until.public import pub_func
    import uuid
    
    
    @api.route('/app/login', methods=['POST'])
    def login(request):
        param = pub_func.Param(request)
        try:
            ParamStr = {'iPhone': 1, "user_pw": 1}
            RealParam = param.HandleParam(ParamStr)
            iPhone = RealParam.get('iPhone')
            login_data = login_dal.get_user_login(iPhone)
            if login_data:
                user_pw = RealParam.get('user_pw')
                md5_pw = rsa_public.MD5(user_pw)
                get_pw = login_data.get('user_pw')
                if get_pw == md5_pw:
                    return json({'success': '登录成功'})
                else:
                    return json({'error': '密码错误'})
            else:
                return json({'error': '账户不存在'})
        except Exception as e:
            return json({"error": e.args})

#/dal/login_dal.py
    
    from until.base.mysql import mysqlHelper
    
    mySql = mysqlHelper()
    
    
    def get_user_login(iPhone):
        return mySql.selectOne(f"select * from `user` where iPhone = {iPhone}")

然后请求测试，结果出来了：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J0YV9iaA_size_16_color_FFFFFF_t_70 1][]

但是感觉是不是太简单了，一般登录肯定是要获取一个唯一值的，方便后面所有接口的验证，不然要登录干嘛呢。那这个唯一值应该怎么获得呢，肯定是公钥加密和私钥解密的token啊，但是本片文章只简单的生成token，验证放到下章去讲，因为验证就要和中间件相关了。我这里上面的密码也是加密过的，当然这个取决于自己的决定。

### 生成token ###

在生成token之前先生成公钥和私钥，因为token是用公钥和私钥加密解密的，这里用到了Crypto模块直接下载是不行的，这里得下载pycryptodome模块，下载完之后就可以继续写代码了（注意要下载的东西要放到版本库中）。

生成公钥和私钥，先在until/config文件夹下创建rsa\_public.py文件用来写和加密解密相关的代码。当然，上面的密码加密已经用到这个文件了，但是他是密码加密，和这个没有关系。

#/until/config/rsa_public.py
    import hashlib, base64
    from Crypto import Random
    from Crypto.PublicKey import RSA
    from Crypto.Cipher import PKCS1_v1_5 as Cipher_pkcs1_v1_5
    
    
    # 获取一个伪随机数生成器
    random_generator = Random.new().read
    # 获取一个rsa算法对应的密钥对生成器实例
    rsa = RSA.generate(2048, random_generator)
    
    # 生成公钥并保存
    public_pem = rsa.publickey().exportKey()
    print(public_pem)
    
    # 生成私钥并保存
    private_pem = rsa.exportKey()
    print(private_pem)
    
    #用公钥加密
    def encrypt(value): #value是需要加密的字符串，比如user_id可以放到里面，之后的传参就不需要传了
        pubkey = RSA.import_key(public_pem)
        original_text = value.encode('utf8')
        cipher = Cipher_pkcs1_v1_5.new(pubkey)
        cipher_text = base64.b64encode(cipher.encrypt(original_text)).decode()
        return cipher_text
    
    #用私钥解密
    def decrypt(app_token):
        try:
            privateKey = RSA.importKey(private_pem)
            cipher = Cipher_pkcs1_v1_5.new(privateKey)
            text = cipher.decrypt(base64.b64decode(app_token), None).decode()
            return text
        except Exception as e:
            raise Exception("token 无效")

下面我们对login代码升级一下

@api.route('/app/login', methods=['POST'])
    def login(request):
        """
        @api {POST} /blog/app/login  登录
        """
        param = pub_func.Param(request)
        try:
            ParamStr = {'iPhone': 1, "user_pw": 1}
            RealParam = param.HandleParam(ParamStr)
            iPhone = RealParam.get('iPhone')
            login_data = login_dal.get_user_login(iPhone)
            if login_data:
                user_pw = RealParam.get('user_pw')
                md5_pw = rsa_public.MD5(user_pw)
                get_pw = login_data.get('user_pw')
                if get_pw == md5_pw:
                    user_id = login_data.get('user_id')
                    app_token = rsa_public.encrypt(user_id)
                    return json({'app_token': app_token})
                else:
                    return json({'error': '密码错误'})
            else:
                return json({'error': '账户不存在'})
        except Exception as e:
            return json({"error": e.args})

然后就可以获得token啦。之后再每次请求接口的时候都要带着token，后端就可以验证了

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J0YV9iaA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200527142854634.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J0YV9iaA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J0YV9iaA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200527144441741.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J0YV9iaA==,size_16,color_FFFFFF,t_70