赞
踩
在Flask框架中,实现Token认证机制并不是一件复杂的事情。除了使用官方提供的flask_httpauth
模块或者第三方模块flask-jwt
,我们还可以考虑自己实现一个简易版的Token认证工具。自定义Token认证机制的本质是生成一个令牌(Token),并在用户每次请求时验证这个令牌的有效性。
整个过程可以分为以下几个步骤:
这种自定义的Token认证机制相对简单,适用于一些小型应用或者对于Token认证机制有特殊需求的场景。搭建这样一个简易的认证系统有助于理解Token认证的基本原理,并可以根据实际需求进行灵活的定制。
通过表结构的创建,建立用户认证和会话管理表。UserAuthDB
表存储了用户的账号密码信息,而SessionAuthDB
表则存储了用户登录后生成的Token
信息,包括用户名、Token本身以及Token的过期时间。这为后续实现用户注册、登录以及Token认证等功能提供了数据库支持。
UserAuthDB表:
id
: 主键,自增,唯一标识每个用户。username
: 用户名,非空,唯一,用于登录时识别用户。password
: 密码,非空,用于验证用户身份。SessionAuthDB表:
id
: 主键,自增,唯一标识每个登录会话。username
: 用户名,非空,唯一,关联到UserAuthDB
表的用户名。token
: 用户登录后生成的Token,非空,唯一,用于身份验证。invalid_date
: Token的过期时间,用于判断Token是否过期。代码通过Flask路由/create
实现了数据库表结构的创建,主要包括两张表,分别是UserAuthDB
和SessionAuthDB
。
@app.route("/create",methods=["GET"]) def create(): conn = sqlite3.connect("./database.db") cursor = conn.cursor() create_auth = "create table UserAuthDB(" \ "id INTEGER primary key AUTOINCREMENT not null unique," \ "username varchar(64) not null unique," \ "password varchar(64) not null" \ ")" cursor.execute(create_auth) create_session = "create table SessionAuthDB(" \ "id INTEGER primary key AUTOINCREMENT not null unique," \ "username varchar(64) not null unique," \ "token varchar(128) not null unique," \ "invalid_date int not null" \ ")" cursor.execute(create_session) conn.commit() cursor.close() conn.close() return "create success"
该验证函数用于保证传入的用户名和密码满足一定的安全性和格式要求。通过对长度和字符内容的检查,确保了传入的参数不会导致潜在的安全问题。这样的验证机制在用户注册、登录等场景中可以有效地防止一些常见的安全漏洞。
参数验证:
*kwargs
,可传入多个参数。字符串处理:
字符内容验证:
返回结果:
True
,表示参数合法。False
,表示参数存在非法字符或超出长度限制。代码定义了一个名为CheckParameters
的验证函数,该函数用于验证传入的参数是否合法。主要验证的对象是用户名和密码,具体概述如下:
def CheckParameters(*kwargs):
for item in range(len(kwargs)):
# 先验证长度
if len(kwargs[item]) >= 128 or len(kwargs[item]) == 0:
return False
# 先小写,然后去掉两侧空格,去掉所有空格
local_string = kwargs[item].lower().strip().replace(" ","")
# 判断是否只包含 大写 小写 数字
for kw in local_string:
if kw.isupper() != True and kw.islower() != True and kw.isdigit() != True:
return False
return True
该函数实现了用户登录认证的核心逻辑。首先对输入的用户名和密码进行验证,然后检查用户是否存在以及是否已经有生成的Token。如果用户存在但Token不存在,生成一个新的Token并存入数据库,最终返回生成的Token。
路由定义:
@app.route("/login", methods=["POST"])
定义了一个POST请求的路由,用于处理用户登录请求。参数获取:
request.form.to_dict()
获取POST请求中的参数,包括用户名(username)和密码(password)。参数验证:
CheckParameters
函数对获取的用户名和密码进行合法性验证,确保其符合安全性和格式要求。用户存在性验证:
RunSqlite
函数查询UserAuthDB
表,验证用户名和密码是否匹配。如果存在匹配的用户,则继续执行下一步。生成Token:
SessionAuthDB
表,检查是否存在该用户的Token记录。如果存在,则直接返回该Token。Token写入数据库:
SessionAuthDB
表。返回结果:
@app.route("/login",methods=["POST"]) def login(): if request.method == "POST": # 获取参数信息 obtain_dict = request.form.to_dict() if len(obtain_dict) != 0 and len(obtain_dict) == 2: username = obtain_dict["username"] password = obtain_dict["password"] # 验证是否合法 is_true = CheckParameters(username,password) if is_true == True: # 查询是否存在该用户 select = RunSqlite("./database.db", "UserAuthDB", "select", "username,password", f"username='{username}'") if select[0][0] == username and select[0][1] == password: # 查询Session列表是否存在 select_session = RunSqlite("./database.db","SessionAuthDB","select","token",f"username='{username}'") if select_session != []: ref = {"message": ""} ref["message"] = select_session[0][0] return json.dumps(ref, ensure_ascii=False) # Session不存在则需要重新生成 else: # 生成并写入token和过期时间戳 token = ''.join(random.sample(string.ascii_letters + string.digits, 32)) # 设置360秒周期,过期时间 time_stamp = int(time.time()) + 360 insert = RunSqlite("./database.db", "SessionAuthDB", "insert", "username,token,invalid_date", f"'{username}','{token}',{time_stamp}") if insert == True: ref = {"message": ""} ref["message"] = token return json.dumps(ref, ensure_ascii=False) else: return json.dumps("{'message': '用户名或密码错误'}", ensure_ascii=False) else: return json.dumps("{'message': '输入参数不可用'}", ensure_ascii=False) return json.dumps("{'message': '未知错误'}", ensure_ascii=False)
检查用户登录状态Token是否过期的装饰器,装饰器用于装饰某一些函数,当主调函数被调用时,会优先执行装饰器内的代码,执行后根据装饰器执行结果返回或退出,装饰器分为两种模式,一种是FBV模式,另一种是CBV模式。
FBV(Function-Based Views)和CBV(Class-Based Views)是两种不同的视图设计模式,用于处理Web框架中的请求和生成响应。这两种模式在Django框架中被广泛使用。
示例:
def my_view(request):
# 处理逻辑
return HttpResponse("Hello, World!")
示例:
class MyView(View):
def get(self, request):
# 处理 GET 请求的逻辑
return HttpResponse("Hello, World!")
def post(self, request):
# 处理 POST 请求的逻辑
return HttpResponse("Received a POST request")
在Flask中,两种设计模式都可以使用,开发者可以根据项目的需求和个人喜好选择使用FBV或CBV。
基于FBV的装饰器设置使用时,需要注意装饰器嵌入的位置,装饰器需要在请求进入路由之前,即在请求未走原逻辑代码的时候介入,对原业务逻辑进行业务拓展。
from flask import Flask, request,render_template from functools import wraps app = Flask(__name__) def login(func): @wraps(func) def wrapper(*args, **kwargs): print("登录请求: {}".format(request.url)) value = request.form.get("value") if value == "lyshark": # 调用原函数,并返回 function_ptr = func(*args, **kwargs) return function_ptr else: return "登录失败" return wrapper @app.route('/', methods=['GET', 'POST']) @login def index(): if request.method == "POST": value = request.form.get("value") return "index" if __name__ == '__main__': app.run()
而基于CBV的装饰器设置,使用就显得更加细分化,可以定制管理专属功能,在外部定义装饰器可以全局使用,内部定义可以针对特定路由函数特殊处理。
from flask import Flask, request,render_template,views from functools import wraps app = Flask(__name__) # 装饰器 def login(func): @wraps(func) def wrapper(*args, **kwargs): print("登录请求: {}".format(request.url)) value = request.form.get("value") if value == "lyshark": # 调用原函数,并返回 function_ptr = func(*args, **kwargs) return function_ptr else: return "登录失败" return wrapper # 类视图 class index(views.MethodView): @login def get(self): return request.args @login def post(self): return "success" # 增加路由 app.add_url_rule(rule='/', view_func=index.as_view('index')) if __name__ == '__main__': app.run()
此处为了实现起来更简单一些此处直接使用FBV模式,我们实现的login_check
装饰器通过FVB模式构建,代码中取得用户的Token以及用户名对用户身份进行验证。
def login_check(func): @wraps(func) def wrapper(*args, **kwargs): print("处理登录逻辑部分: {}".format(request.url)) # 得到token 验证是否登陆了,且token没有过期 local_timestamp = int(time.time()) get_token = request.headers.get("token") # 验证传入参数是否合法 if CheckParameters(get_token) == True: select = RunSqlite("database.db","SessionAuthDB","select","token,invalid_date",f"token='{get_token}'") print(select) # 判断是否存在记录,如果存在,在判断时间戳是否合理 if select != []: # 如果当前时间与数据库比对,大于说明过期了需要删除原来的,让用户重新登录 if local_timestamp >= int(select[0][1]): print("时间戳过期了") # 删除原来的Token delete = RunSqlite("database.db","SessionAuthDB","delete",f"token='{get_token}'","none") if delete == True: return json.dumps("{'token': 'Token 已过期,请重新登录获取'}", ensure_ascii=False) else: return json.dumps("{'token': '数据库删除异常,请联系开发者'}", ensure_ascii=False) else: # 验证Token是否一致 if select[0][0] == get_token: print("Token验证正常,继续执行function_ptr指向代码.") # 返回到原函数 return func(*args, **kwargs) else: print("Token验证错误 {}".format(select)) return json.dumps("{'token': 'Token 传入错误'}", ensure_ascii=False) # 装饰器调用原函数 # function_ptr = func(*args, **kwargs) return json.dumps("{'token': 'Token 验证失败'}", ensure_ascii=False) return wrapper
主调用函数则是具体的功能实现可以自定义扩展,当用户访问该路由时会优先调用login_check
装饰器来验证用户携带Token的合法性,如果合法则会通过return func(*args, **kwargs)
返回执行主调函数,否则直接返回验证失败的消息。
# 获取参数函数 @app.route("/GetPage", methods=["POST"]) @login_check def GetPage(): if request.method == "POST": # 获取参数信息 obtain_dict = request.form.to_dict() if len(obtain_dict) != 0 and len(obtain_dict) == 1: pagename = obtain_dict["pagename"] print("查询名称: {}".format(obtain_dict["pagename"])) # 相应头的完整写法 req = Response(response="ok", status=200, mimetype="application/json") req.headers["Content-Type"] = "text/json; charset=utf-8" req.headers["Server"] = "LyShark Server 1.0" req.data = json.dumps("{'message': 'hello world'}") return req else: return json.dumps("{'message': '传入参数错误,请携带正确参数请求'}", ensure_ascii=False) return json.dumps("{'token': '未知错误'}", ensure_ascii=False) # 用户注册函数 @app.route("/register", methods=["POST"]) def Register(): if request.method == "POST": obtain_dict = request.form.to_dict() if len(obtain_dict) != 0 and len(obtain_dict) == 2: print("用户名: {} 密码: {}".format(obtain_dict["username"], obtain_dict["password"])) reg_username = obtain_dict["username"] reg_password = obtain_dict["password"] # 验证是否合法 if CheckParameters(reg_username, reg_password) == False: return json.dumps("{'message': '传入用户名密码不合法'}", ensure_ascii=False) # 查询用户是否存在 select = RunSqlite("database.db","UserAuthDB","select","id",f"username='{reg_username}'") if select != []: return json.dumps("{'message': '用户名已被注册'}", ensure_ascii=False) else: insert = RunSqlite("database.db","UserAuthDB","insert","username,password",f"'{reg_username}','{reg_password}'") if insert == True: return json.dumps("{'message': '注册成功'}", ensure_ascii=False) else: return json.dumps("{'message': '注册失败'}", ensure_ascii=False) else: return json.dumps("{'message': '传入参数个数不正确'}", ensure_ascii=False) return json.dumps("{'message': '未知错误'}", ensure_ascii=False) # 密码修改函数 @app.route("/modify", methods=["POST"]) @login_check def modify(): if request.method == "POST": obtain_dict = request.form.to_dict() if len(obtain_dict) != 0 and len(obtain_dict) == 1: mdf_password = obtain_dict["password"] get_token = request.headers.get("token") print("获取token: {} 修改后密码: {}".format(get_token,mdf_password)) # 验证是否合法 if CheckParameters(get_token, mdf_password) == False: return json.dumps("{'message': '传入密码不合法'}", ensure_ascii=False) # 先得到token对应用户名 select = RunSqlite("database.db","SessionAuthDB","select","username",f"token='{get_token}'") if select != []: # 接着直接修改密码即可 modify_username = str(select[0][0]) print("得到的用户名: {}".format(modify_username)) update = RunSqlite("database.db","UserAuthDB","update",f"username='{modify_username}'",f"password='{mdf_password}'") if update == True: # 删除原来的token,让用户重新获取 delete = RunSqlite("database.db","SessionAuthDB","delete",f"username='{modify_username}'","none") print("删除token状态: {}".format(delete)) return json.dumps("{'message': '修改成功,请重新登录获取Token'}", ensure_ascii=False) else: return json.dumps("{'message': '修改失败'}", ensure_ascii=False) else: return json.dumps("{'message': '不存在该Token,无法修改密码'}", ensure_ascii=False) else: return json.dumps("{'message': '传入参数个数不正确'}", ensure_ascii=False) return json.dumps("{'message': '未知错误'}", ensure_ascii=False)
FBV模式下的完整代码,以下是对代码的概述:
RunSqlite
函数)。login_check
来验证 Token 的有效性。/create
。/GetPage
接口,使用了 login_check
装饰器来验证用户登录状态,仅对已登录用户提供页面信息。/create
:创建数据库表结构。/login
:用户登录接口,返回用户的 Token。/GetPage
:获取页面信息,需要用户登录并携带有效 Token。/register
:用户注册接口。/modify
:修改用户密码接口,需要用户登录并携带有效 Token。login_check
对需要登录的路由进行认证。from flask import Flask,render_template,request,Response,redirect,jsonify from functools import wraps import json,sqlite3,random,string,time app = Flask(__name__) # 增删改查简单封装 def RunSqlite(db,table,action,field,value): connect = sqlite3.connect(db) cursor = connect.cursor() # 执行插入动作 if action == "insert": insert = f"insert into {table}({field}) values({value});" if insert == None or len(insert) == 0: return False try: cursor.execute(insert) except Exception: return False # 执行更新操作 elif action == "update": update = f"update {table} set {value} where {field};" if update == None or len(update) == 0: return False try: cursor.execute(update) except Exception: return False # 执行查询操作 elif action == "select": # 查询条件是否为空 if value == "none": select = f"select {field} from {table};" else: select = f"select {field} from {table} where {value};" try: ref = cursor.execute(select) ref_data = ref.fetchall() connect.commit() connect.close() return ref_data except Exception: return False # 执行删除操作 elif action == "delete": delete = f"delete from {table} where {field};" if delete == None or len(delete) == 0: return False try: cursor.execute(delete) except Exception: return False try: connect.commit() connect.close() return True except Exception: return False @app.route("/create",methods=["GET"]) def create(): conn = sqlite3.connect("./database.db") cursor = conn.cursor() create_auth = "create table UserAuthDB(" \ "id INTEGER primary key AUTOINCREMENT not null unique," \ "username varchar(64) not null unique," \ "password varchar(64) not null" \ ")" cursor.execute(create_auth) create_session = "create table SessionAuthDB(" \ "id INTEGER primary key AUTOINCREMENT not null unique," \ "username varchar(64) not null unique," \ "token varchar(128) not null unique," \ "invalid_date int not null" \ ")" cursor.execute(create_session) conn.commit() cursor.close() conn.close() return "create success" # 验证用户名密码是否合法 def CheckParameters(*kwargs): for item in range(len(kwargs)): # 先验证长度 if len(kwargs[item]) >= 256 or len(kwargs[item]) == 0: return False # 先小写,然后去掉两侧空格,去掉所有空格 local_string = kwargs[item].lower().strip().replace(" ","") # 判断是否只包含 大写 小写 数字 for kw in local_string: if kw.isupper() != True and kw.islower() != True and kw.isdigit() != True: return False return True # 登录认证模块 @app.route("/login",methods=["POST"]) def login(): if request.method == "POST": # 获取参数信息 obtain_dict = request.form.to_dict() if len(obtain_dict) != 0 and len(obtain_dict) == 2: username = obtain_dict["username"] password = obtain_dict["password"] # 验证是否合法 is_true = CheckParameters(username,password) if is_true == True: # 查询是否存在该用户 select = RunSqlite("./database.db", "UserAuthDB", "select", "username,password", f"username='{username}'") if select[0][0] == username and select[0][1] == password: # 查询Session列表是否存在 select_session = RunSqlite("./database.db","SessionAuthDB","select","token",f"username='{username}'") if select_session != []: ref = {"message": ""} ref["message"] = select_session[0][0] return json.dumps(ref, ensure_ascii=False) # Session不存在则需要重新生成 else: # 生成并写入token和过期时间戳 token = ''.join(random.sample(string.ascii_letters + string.digits, 32)) # 设置360秒周期,过期时间 time_stamp = int(time.time()) + 360 insert = RunSqlite("./database.db", "SessionAuthDB", "insert", "username,token,invalid_date", f"'{username}','{token}',{time_stamp}") if insert == True: ref = {"message": ""} ref["message"] = token return json.dumps(ref, ensure_ascii=False) else: return json.dumps("{'message': '用户名或密码错误'}", ensure_ascii=False) else: return json.dumps("{'message': '输入参数不可用'}", ensure_ascii=False) return json.dumps("{'message': '未知错误'}", ensure_ascii=False) # 检查登录状态 token是否过期的装饰器 def login_check(func): @wraps(func) def wrapper(*args, **kwargs): print("处理登录逻辑部分: {}".format(request.url)) # 得到token 验证是否登陆了,且token没有过期 local_timestamp = int(time.time()) get_token = request.headers.get("token") # 验证传入参数是否合法 if CheckParameters(get_token) == True: select = RunSqlite("./database.db","SessionAuthDB","select","token,invalid_date",f"token='{get_token}'") print(select) # 判断是否存在记录,如果存在,在判断时间戳是否合理 if select != []: # 如果当前时间与数据库比对,大于说明过期了需要删除原来的,让用户重新登录 if local_timestamp >= int(select[0][1]): print("时间戳过期了") # 删除原来的Token delete = RunSqlite("./database.db","SessionAuthDB","delete",f"token='{get_token}'","none") if delete == True: return json.dumps("{'token': 'Token 已过期,请重新登录获取'}", ensure_ascii=False) else: return json.dumps("{'token': '数据库删除异常,请联系开发者'}", ensure_ascii=False) else: # 验证Token是否一致 if select[0][0] == get_token: print("Token验证正常,继续执行function_ptr指向代码.") # 返回到原函数 return func(*args, **kwargs) else: print("Token验证错误 {}".format(select)) return json.dumps("{'token': 'Token 传入错误'}", ensure_ascii=False) # 装饰器调用原函数 # function_ptr = func(*args, **kwargs) return json.dumps("{'token': 'Token 验证失败'}", ensure_ascii=False) return wrapper # 获取参数函数 @app.route("/GetPage", methods=["POST"]) @login_check def GetPage(): if request.method == "POST": # 获取参数信息 obtain_dict = request.form.to_dict() if len(obtain_dict) != 0 and len(obtain_dict) == 1: pagename = obtain_dict["pagename"] print("查询名称: {}".format(obtain_dict["pagename"])) # 相应头的完整写法 req = Response(response="ok", status=200, mimetype="application/json") req.headers["Content-Type"] = "text/json; charset=utf-8" req.headers["Server"] = "LyShark Server 1.0" req.data = json.dumps("{'message': 'hello world'}") return req else: return json.dumps("{'message': '传入参数错误,请携带正确参数请求'}", ensure_ascii=False) return json.dumps("{'token': '未知错误'}", ensure_ascii=False) # 用户注册函数 @app.route("/register", methods=["POST"]) def Register(): if request.method == "POST": obtain_dict = request.form.to_dict() if len(obtain_dict) != 0 and len(obtain_dict) == 2: print("用户名: {} 密码: {}".format(obtain_dict["username"], obtain_dict["password"])) reg_username = obtain_dict["username"] reg_password = obtain_dict["password"] # 验证是否合法 if CheckParameters(reg_username, reg_password) == False: return json.dumps("{'message': '传入用户名密码不合法'}", ensure_ascii=False) # 查询用户是否存在 select = RunSqlite("database.db","UserAuthDB","select","id",f"username='{reg_username}'") if select != []: return json.dumps("{'message': '用户名已被注册'}", ensure_ascii=False) else: insert = RunSqlite("database.db","UserAuthDB","insert","username,password",f"'{reg_username}','{reg_password}'") if insert == True: return json.dumps("{'message': '注册成功'}", ensure_ascii=False) else: return json.dumps("{'message': '注册失败'}", ensure_ascii=False) else: return json.dumps("{'message': '传入参数个数不正确'}", ensure_ascii=False) return json.dumps("{'message': '未知错误'}", ensure_ascii=False) # 密码修改函数 @app.route("/modify", methods=["POST"]) @login_check def modify(): if request.method == "POST": obtain_dict = request.form.to_dict() if len(obtain_dict) != 0 and len(obtain_dict) == 1: mdf_password = obtain_dict["password"] get_token = request.headers.get("token") print("获取token: {} 修改后密码: {}".format(get_token,mdf_password)) # 验证是否合法 if CheckParameters(get_token, mdf_password) == False: return json.dumps("{'message': '传入密码不合法'}", ensure_ascii=False) # 先得到token对应用户名 select = RunSqlite("./database.db","SessionAuthDB","select","username",f"token='{get_token}'") if select != []: # 接着直接修改密码即可 modify_username = str(select[0][0]) print("得到的用户名: {}".format(modify_username)) update = RunSqlite("database.db","UserAuthDB","update",f"username='{modify_username}'",f"password='{mdf_password}'") if update == True: # 删除原来的token,让用户重新获取 delete = RunSqlite("./database.db","SessionAuthDB","delete",f"username='{modify_username}'","none") print("删除token状态: {}".format(delete)) return json.dumps("{'message': '修改成功,请重新登录获取Token'}", ensure_ascii=False) else: return json.dumps("{'message': '修改失败'}", ensure_ascii=False) else: return json.dumps("{'message': '不存在该Token,无法修改密码'}", ensure_ascii=False) else: return json.dumps("{'message': '传入参数个数不正确'}", ensure_ascii=False) return json.dumps("{'message': '未知错误'}", ensure_ascii=False) if __name__ == '__main__': app.run(debug=True)
首先需要在Web页面访问http://127.0.0.1/create
路径实现对数据库的初始化,并打开Postman
工具,通过传入参数来使用这个案例。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。