I、一个综上小结的实例--转账demo
需求:用户通过username = admin,password = 1234登录转账界面,点击转账按钮后提示转账xxx元到xxx账户成功
分析:此处我们应用wtf和cookie来进行验证用户登录,然后用两个视图函数模拟转账
首先,根据创建一个app-transferdemo,根据以往的经验我们先导入几个包
from flask import Flask, request, render_template, redirect, url_for, flash, make_response
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, EqualTo
app = Flask(__name__)
if __name__ == '__main__':
app.run(port=5003, debug=True)
其中make_response
相当于Django内的Httpresponse
继而,根据wtf的验证方式,我们需要定义一个登陆验证表单
class RegisterForm(FlaskForm):
username = StringField("用户名:", validators=[DataRequired("请输入用户名")], render_kw={"placeholder": "请输入用户名"})
password = PasswordField("密码", validators=[DataRequired("请输入密码"),render_kw={"placeholder": "请输入密码"}])
submit = SubmitField("登陆")
编写相应的登录界面logintrans.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="" method="post">
{{ form.username.label }}{{ form.username }}<br>
{{ form.password.label }}{{ form.password }}<br>
{{ form.submit }}<br>
{{ form.csrf_token }}
</form>
{% for message in get_flashed_messages() %}
{{ message }}
{% endfor %}
</body>
</html>
编写登录验证的视图函数index
@app.route('/', methods=["get", "post"])
def index():
register_form = RegisterForm()
if register_form.validate_on_submit():
# 如果代码走到if里面证明表单验证有效
username = request.form.get("username")
password = request.form.get("password", "")
if username == 'admin' and password == '1234':
return redirect(url_for('transfer'))
else:
if request.method == 'POST':
flash("表单参数有误或者不完整")
return render_template("logintrans.html", form=register_form)
在登录后,跳转到transfer页面
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>转账页面</title>
</head>
<body>
<form action="" method="post">
<label>账户</label><input type="text" name="account">
<label>金额</label><input type="number" name="money">
<input type="submit" value="转账">
</form>
</body>
</html>
编写相应的transfer视图函数
@app.route('/transfer',methods=["get", "post"])
def transfer():
# 取出cookie确保登录
username = request.cookies.get('username', '')
if not username:
# 无cookie说明未登录,需要登录
return redirect(url_for('index'))
if request.method == 'POST':
account = request.form.get('account')
money = request.form.get('money')
return '转账{}元到{}账户成功'.format(account, money)
# makeresponse 相当于Django内的Httpresponse
response = make_response(render_template('transfer.html'))
return response
启动服务,观察效果
II、CSRF攻击:跨站请求伪造
首先我们创建两个应用weba、webb,基于转账功能,此处我们不应用wtf验证
weba.py
from flask import Flask, request, render_template, redirect, url_for, flash, make_response
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, EqualTo
app = Flask(__name__)
# 自定义一个表单类
class RegisterForm(FlaskForm):
username = StringField("用户名:", validators=[DataRequired("请输入用户名")], render_kw={"placeholder": "请输入用户名"})
password = PasswordField("密码", validators=[DataRequired("请输入密码")])
password2 = PasswordField("确认密码", validators=[DataRequired("请输入确认密码"), EqualTo("password", "两次密码不一致")])
submit = SubmitField("注册")
app.secret_key = 'dasdasdsadas'
@app.route('/', methods=["get", "post"])
def index():
if request.method == 'POST':
# 取出表单数据
username = request.form.get("username")
password = request.form.get("password", "")
password2 = request.form.get("password2", "")
if not all([username, password]):
print('参数error')
else:
print(password, username)
if username == 'admin' and password == '1234':
# 登录成功跳转转账页面
response = redirect(url_for('transfer'))
response.set_cookie('username', username)
return response
else:
print('password error')
return render_template("login2.html")
@app.route('/transfer',methods=["get", "post"])
def transfer():
# 取出cookie确保登录
username = request.cookies.get('username', '')
if not username:
# 无cookie说明未登录,需要登录
return redirect(url_for('index'))
if request.method == 'POST':
account = request.form.get('account')
money = request.form.get('money')
return '转账{}元到{}账户成功'.format(account, money)
# makeresponse 相当于Django内的Httpresponse
response = make_response(render_template('transfer.html')
)
return response
if __name__ == '__main__':
app.run(port=5003, debug=True)
login2.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="" method="post">
<label>用户名</label><input type="text" name="username">
<label>密码</label><input type="password" name="password">
<input type="submit" value="登录">
</form>
</body>
</html>
webb.py
from flask import Flask, render_template
app = Flask(__name__)
@app.route('/', methods=["get", "post"])
def index():
return render_template('csrf-index.html')
if __name__ == '__main__':
app.run(port=5004, debug=True)
csrf-index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>袭击网站A</title>
</head>
<body>
<form action="" method="post">
<input type="hidden" name="account" value="999">
<input type="hidden" name="money" value="111">
<input type="submit" value='点击领取优惠券'>
</form>
</body>
</html>
在建立好如上应用后,我们尝试模拟csrf攻击,由webb攻击weba
由于weba是一个登录后的转账页面,为了攻击该转账过程,尝试修改csrf-index.html内form
的action
<form action="http://127.0.0.1:5003/transfer" method="post">
启动两个app,清除网站的cookie,然后点击点击领取优惠券
该按钮查看过程。
我们可以发现,当admin未登录时,点击按钮返回登录界面。当用户admin登录后,我们点击按钮时,会提示转账信息,而该转账信息并未来自于admin用户,而如何防治这种攻击呢?
我们可以利用csrf_token的方式进行校验
首先我们编辑生成csrf_token的方法csrf_g.py
import base64
import os
def generate_csrf():
return bytes.decode(base64.b64encode(os.urandom(48)))
print(generate_csrf())
在每次转账前都生成一个csrf_token
使得每次生成的csrf_token
值都是不一样的,利用浏览器的无法跨站请求cookie的协议,webb无法请求weba的cookie
修改weba内的transfer视图,添加验证csrf_token的内容:
import csrf_g
def transfer():
# 取出cookie确保登录
username = request.cookies.get('username', '')
if not username:
# 无cookie说明未登录,需要登录
return redirect(url_for('index'))
if request.method == 'POST':
account = request.form.get('account')
money = request.form.get('money')
# 取表单中的csrf_token
form_csrf_token = request.form.get('csrf_token')
# 取cookie内csrf_token
cookie_csrf_token = request.cookies.get('csrf_token')
if form_csrf_token != cookie_csrf_token:
return "非法操作"
else:
return '转账{}元到{}账户成功'.format(account, money)
csrf_token = csrf_g.generate_csrf()
# makeresponse 相当于Django内的Httpresponse
response = make_response(render_template('transfer.html', csrf_token=csrf_token))
response.set_cookie('csrf_token', csrf_token)
return response
而在多数实际操作的过程中,我们不必过于麻烦,可以使用flask已封装的方法,即CsrfProtect
模块,此处我们应用的是Flask 1.1.2。
我们找到之前我们用wtf表单进行的登录验证app,利用CsrfProtect
模块添加验证内容
from flask_wtf.csrf import CSRFProtect, CSRFError
CSRFProtect(app)
# 添加错误装饰器
@app.errorhandler(CSRFError)
def csrf_error(reason):
return render_template('csrf_error.html', reason=reason), 400
新建csrf_error.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
ERROR
</body>
</html>
修改wtf.html
<form action="" method="post">
{{ form.username.label }}{{ form.username }}<br>
{{ form.password.label }}{{ form.password }}<br>
{{ form.password2.label }}{{ form.password2 }}<br>
{{ form.submit }}<br>
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}" />
</form>
实现csrf保护
而对于添加csrf保护的方法不仅仅于此,具体内容可通过官方文档 CsrfProtect深入了解
III、关联数据库方式SQLAlchemy
在学习完之前的预备内容后,我们大致了解了Flask相关的一些核心内容,而我们接下来要继续深入了解Flask,而这就不得不了解Flask的ORM框架结构,因此,我们需要先体会一下Flask内管理数据库的方式。在此处我们仅仅做一个引例,在下一章中将继续深入了解该内容。
在引例子中,我们将涉及一些关联数据库的基础操作,请自行体会。
预备过程需要安装 flask_sqlalchemy、mysqlclient并创建一个pure python 的项目test4,创建数据库flask_test
编写一个db_demo.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 设置数据库连接
# 设置格式app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://用户名:密码@数据库端口地址/数据库名称'
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:123456@127.0.0.1:3306/test_flask'
# 动态追踪设置,可用于关闭某些Warning
app.config['SQLALCHEMY_TRACK_MODUFICATIONS'] = True
# 显示原始sql
app.config['SQLALCHEMY_ECHO'] = True
# 设置每次请求后自动提交数据库的改动
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
# 数据库要和 app关联,db为SQLAlchemy的一个实例
db = SQLAlchemy(app)
# 创建一个角色类作为例子,继承于db
class Role(db.Model):
# 定义表名
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True)
@app.route('/')
def index():
return 'index'
if __name__ == '__main__':
# 删除表
db.drop_all()
# 创建表
db.create_all()
ro1 = Role(name='admin')
ro2 = Role(name='user')
db.session.add_all([ro1, ro2])
db.session.commit()
app.run(port=5005)
注:Flask在创建表时并不会自动创建id主键值和表名,需要自行设计,这与Django是有区别的
运行服务后,打开数据库,观察结果。