記錄一次完整的flask小型應用開發(2)
這一次,我們完成使用者認證的功能:
程式要進行使用者追蹤,程式知道使用者是誰之後,就能針對性的提供體驗。需要使用者提供使用者名稱和密碼。
要是想保證資料庫中存放密碼的安全性,那麼就不存放明文密碼,存放密碼的雜湊值,我們使用Werkzeug來實現密碼雜湊:
所以我們改變models.py中的User模型來支援密碼雜湊
#coding: utf-8 from . import db from werkzeug.security import generate_password_hash, check_password_hash class Role(db.Model): __tablename__ = 'roles' id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(64), unique=True) users = db.relationship('User', backref='role', lazy='dynamic') def __repr__(self): return '<Role %r>' % self.name class User(db.Model): __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(64), unique=True, index=True) role_id = db.Column(db.Integer, db.ForeignKey('roles.id')) password_hash = db.Column(db.String(128)) @property def password(self): raise AttributeError('password is not a readable attribute') @password.setter def password(self, password): self.password_hash = generate_password_hash(password) # 將原始密碼作為輸入,輸出密碼雜湊值 def verify_password(self, password): return check_password_hash(self.password_hash, password) # 對比資料庫中密碼雜湊值和使用者輸入的密碼,正確返回True def __repr__(self): return '<User %r>' % self.username
建立認證藍本:
與使用者認證相關的路由可以在auth藍本中定義,對於不同程式功能,我們儘量使用不同的藍本。所以建立app目錄下的auth資料夾,前面建立的main資料夾是主頁基礎功能。
這裡建立藍本的方式與前面差不多:
在auth/init.py中:
from flask import Blueprint
auth = Blueprint('auth', __name__)
from . import views
在auth/views.py中:
from flask import render_template from . import auth @auth.route('/login') def login(): return render_template('auth/login.html') # 這裡需要注意了,這個模板檔案需要儲存在auth這個資料夾中 # 但是這個資料夾又需要儲存在app/templates中 # flask認為模板的路徑是相對於程式模板資料夾而言的。
最後需要在create_app()中將藍本註冊到程式上:
from .auth import auth as auth_blueprint
app.register_blueprint(auth_blueprint, url_prefix='/auth')
# 這裡加上了prefix,註冊後藍本中定義的所有路由都會加上這個字首
# 所有views中定義的/login會變成/auth/login
使用Flask-Login來認證使用者:
使用者登入程式後,他們的認證狀態要被記錄下來,這樣瀏覽不同的頁面時才能記住這個狀態。
要想使用Flask-Login,程式User模型必須實現幾個方法,這個拓展提供了一個UserMixin類,包含了方法的預設實現。
修改User模型:
class User(UserMixin, db.Model): # 下面保持不變,新增一個email欄位
然後我們需要在工廠函式中初始化flask-login:
login_manager = LoginManager() # 建立一個登入例項
login_manager.session_protection = 'strong'
login_manager.login_view = 'auth.login' # 設定登入頁面的端點,路由在藍本中定義,所以要加上藍本的名字
最後在create_app()中初始化app:
login_manager.init_app(app)
最後flask-login要求程式使用指定的識別符號載入使用者:這是一個回撥函式
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
然後我們需要建立登入時用到的表單: 因為屬於auth功能,所以寫入auth/forms.py:
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Length, Email
class LoginForm(FlaskForm):
email = StringField('Email', validators=[DataRequired(), Length(1, 64),
Email()])
password = PasswordField('Password', validators=[DataRequired()])
remember_me = BooleanField('Keep me logged in')
submit = SubmitField('Log In')
這裡只是表單的功能,我們還需要一個html來展示表單的頁面,我們放在templates/auth/login.html中:
{% extends "/base.html" %}
{% import "bootstrap/wtf.html" as wtf %}
{% block title %}Flasky - Login{% endblock %}
{% block page_content %}
<div class="page-header">
<h1>Login</h1>
</div>
<div class="col-md-4">
{{ wtf.quick_form(form) }}
</div>
{% endblock %}
然後我們需要在auth的檢視函式中關聯上表單:
@auth.route('/login', methods=['GET', 'POST'])
def login():
form = LoginForm() # 建立一個物件
# get請求時,檢視函式直接渲染模板顯示錶單
# POST請求時,拓展的下面這個函式會驗證表單資料
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
if user is not None and user.verify_password(form.password.data):
login_user(user, form.remember_me.data)
next = request.args.get('next')
if next is None or not next.startswith('/'):
next = url_for('main.index')
return redirect(next)
flash('Invalid username or password.')
return render_template('auth/login.html', form=form)
然後我們執行,manager.py發現報錯:'A secret key is required to use CSRF.'
,所以我們需要在config檔案中加上SECRET_KEY = ‘’
結果執行成功!!!!!
然後我們需要一個登出的路由,重定向到首頁:
@auth.route('/logout')
@login_required
def logout():
logout_user()
flash('you logged out')
return redirect(url_for('main.index'))
下面我們實現註冊新使用者的功能:
新增使用者登錄檔單:
class RegistrationForm(FlaskForm):
email = StringField('Email', validators=[DataRequired(), Length(1, 64),
Email()])
username = StringField('Username', validators=[
DataRequired(), Length(1, 64),
Regexp('^[A-Za-z][A-Za-z0-9_.]*$', 0,
'Usernames must have only letters, numbers, dots or '
'underscores')])
password = PasswordField('Password', validators=[
DataRequired(), EqualTo('password2', message='Passwords must match.')])
password2 = PasswordField('Confirm password', validators=[DataRequired()])
submit = SubmitField('Register')
def validate_email(self, field):
if User.query.filter_by(email=field.data).first():
raise ValidationError('Email already registered.')
def validate_username(self, field):
if User.query.filter_by(username=field.data).first():
raise ValidationError('Username already in use.')
這還是隻實現了註冊新使用者的功能,我們需要一個註冊新功能的展示頁面:
{% extends "base.html" %}
{% import "bootstrap/wtf.html" as wtf %}
{% block title %}Flasky - Register{% endblock %}
{% block page_content %}
<div class="page-header">
<h1>Register</h1>
</div>
<div class="col-md-4">
{{ wtf.quick_form(form) }}
</div>
{% endblock %}
然後我們還需要登入頁面能有一個按鈕能讓使用者跳轉到註冊頁面:
<br>
<p>New user? <a href="{{ url_for('auth.register') }}">Click here to register</a>.</p>
最後別忘了,我們什麼都準備好了,只差完成註冊部分的路由:
@auth.route('/register', methods=['GET', 'POST'])
def register():
form = RegistrationForm()
if form.validate_on_submit():
user = User(email=form.email.data,
username=form.username.data,
password=form.password.data)
db.session.add(user)
db.session.commit()
flash('You can now login.')
return redirect(url_for('auth.login'))
return render_template('auth/register.html', form=form)
向用戶郵箱傳送確認註冊郵件:
使用者註冊後,新賬戶首先被標記為待確認狀態,需要使用者按照收到郵件中的說明操作後,才可以證明自己被聯絡上。往往使用者只需要點選一個包含確認令牌的特殊URL:
使用itsdangerous生成確認令牌:
這玩意兒有很多生成令牌的方法,其中TimedJSONWebSignatureSerializer類生成具有過期時間的JSON web簽名
所以我們將生成和檢驗這種令牌的功能新增到User模型:
from itsdangerous import TimedJSONWebSignatureSerializer
from flask import current_app
資料表新增一個欄位:
confirmed = db.Column(db.Boolean, default=False)
# 生成一個令牌,有效期預設一小時
def generate_confirmation_token(self, expiration=3600):
# 下面生成具有過期時間的JSON web簽名
s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'], expiration)
return s.dumps({'confirm': self.id}) # 為指定的資料生成一個加密簽名,然後生成令牌字串
# 檢驗令牌
def confirm(self, token):
s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
try:
data = s.loads(token) # 這個方法會檢驗簽名和過期時間,如果通過就返回原始資料
except:
return False
if data.get('confirm') != self.id:
return False
self.confirmed = True
db.session.add(self)
return True
現在令牌已經做出來了,我們需要向用戶傳送確認郵件,之前我們註冊後就直接重定向到index,現在我們需要在這之前,傳送一封確認郵件:
@auth.route('/register', methods=['GET', 'POST'])
def register():
form = RegistrationForm()
if form.validate_on_submit():
user = User(email=form.email.data,
username=form.username.data,
password=form.password.data)
db.session.add(user)
db.session.commit()
# flash('You can now login.')
# return redirect(url_for('auth.login'))
# 現在我們要生成令牌然後傳送郵件
token = user.generate_confirmation()
send_email(user.email, 'Confirmation of your new account', 'auth/email/confirm', user=user, token=token)
# 這裡給使用者傳送郵件:收件人,郵件標題,郵件模板,給模板傳的引數
# 一個電子郵件需要兩個模板,分別用於渲染純文字正文和富文字正文
flash('we have sent a confirmation email to you, please confirm it!!!')
return redirect(url_for('main.index'))
return render_template('auth/register.html', form=form)
郵件模板內容:
<p>Dear {{ user.username }},</p>
<p>Welcome to <b>Flasky</b>!</p>
<p>To confirm your account please <a href="{{ url_for('auth.confirm', token=token, _external=True) }}">click here</a>.</p>
<p>Alternatively, you can paste the following link in your browser's address bar:</p>
<p>{{ url_for('auth.confirm', token=token, _external=True) }}</p>
<p>Sincerely,</p>
<p>The Flasky Team</p>
<p><small>Note: replies to this email address are not monitored.</small></p>
這裡在郵件裡面使用者點選按鈕會開啟網頁,路由為auth.confirm。
所以接下來,我們還需要實現這個路由功能:
# 這個是傳送給使用者的郵件中的路由連結
from flask_login import current_user
@auth.route('/confirm/<token>')
@login_required # 這個修飾器會保護這個路由,只有使用者開啟連結登陸後,才可以執行下面的檢視函式
def confirm(token):
if current_user.confirmed:
# 首先檢查登入的使用者是否已經確認過,如果已經確認過,就不用再做什麼工作了,直接重定向到首頁
return redirect(url_for('main.index'))
if current_user.confirmed(token):
# 直接呼叫user模型中的驗證令牌方法,直接使用flash來顯示驗證結果。
flash('you have confirmed your acount!')
else:
flash('The confirmation link is invalid or it has expired')
return redirect(url_for('main.index'))
這裡有一個問題,如果使用者沒有在郵件裡面確認註冊就直接登入,我們可以讓使用者登入到一個頁面,在頁面裡面告訴使用者需要去郵箱裡面確認
這個步驟我們可以使用before_request來完成,但是這個只能應用到屬於藍本的請求上,如果要在藍本中使用針對全域性請求的鉤子,則需要使用before_app_request修飾器:
# 這個部分處理請求前驗證賬號是否被啟用
@auth.before_app_request
def before_request():
if current_user.is_authenticated \
and not current_user.confirmed \
and request.endpoint \
and request.blueprint != 'auth' \
and request.endpoint != 'static':
# 需要請求的端點不在認證的藍本當中
return redirect(url_for('auth.unconfirmed'))
# 如果請求驗證失敗,就跳轉到這個路由,顯示一個告訴你賬戶需要在郵件中確認的頁面
@auth.route('/unconfirmed')
def unconfirmed():
if current_user.is_anonymous or current_user.confirmed:
return redirect(url_for('main.index'))
return render_template('auth/unconfirmed.html')
這只是實現了功能,我們還是需要模板:
{% extends "base.html" %}
{% block title %}Flasky - Confirm your account{% endblock %}
{% block page_content %}
<div class="page-header">
<h1>
Hello, {{ current_user.username }}!
</h1>
<h3>You have not confirmed your account yet.</h3>
<p>
Before you can access this site you need to confirm your account.
Check your inbox, you should have received an email with a confirmation link.
</p>
<p>
Need another confirmation email?
<a href="{{ url_for('auth.resend_confirmation') }}">Click here</a>
</p>
</div>
{% endblock %}
可以看到這個模板有一個連結,功能是再次傳送郵件,我們來定義這個路由:
# 這個部分是告訴賬戶未啟用賬戶頁面的連結,用於再次傳送確認郵件
@auth.route('/confirm')
@login_required
def resend_confirmation():
token = current_user.generate_confirmation_token()
send_email(current_user.email, 'Confirm Your Account',
'auth/email/confirm', user=current_user, token=token)
flash('A new confirmation email has been sent to you by email.')
return redirect(url_for('main.index'))
這一部分算是徹底完成了,然後我們試著執行一下,發現config中差配置資訊,所以加上:
MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.googlemail.com')
MAIL_PORT = int(os.environ.get('MAIL_PORT', '587'))
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in \
['true', 'on', '1']
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]'
FLASKY_MAIL_SENDER = 'Flasky Admin <[email protected]>'
FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN')
SQLALCHEMY_TRACK_MODIFICATIONS = False
最後這裡還是有個坑,那就是在設定裡面配置的是gmail,但是國內會出現無網路連線的錯誤,所以可以使用qq郵箱。我們這裡先跳過這個錯誤,大家可在網上自行搜尋如何配置qq郵箱。
使用者角色
因為在web程式中,並不是所有的使用者都有同樣的地位,就比如管理員有著supreme的地位。
首先改進Role模型:
class Role(db.Model):
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(64), unique=True)
default = db.Column(db.Boolean, default=False, index=True)
permissions = db.Column(db.Integer)
users = db.relationship('User', backref='role', lazy='dynamic')
def __repr__(self):
return '<Role %r>' % self.name
然後列出使用者的許可權,許可權用數字表示
class Permission:
FOLLOW = 1
COMMENT = 2
WRITE = 4
MODERATE = 8
ADMIN = 16
現在我們可以建立角色了,但是將角色手動新增到資料庫很麻煩,所以我們在Role類中新增一個方法來完成這個功能。
@staticmethod
def insert_roles():
# 這個函式並不是直接建立新的角色,而是通過角色名來查詢現有的角色,再進行更新
roles = {
'User': [Permission.FOLLOW, Permission.COMMENT, Permission.WRITE],
'Moderator': [Permission.FOLLOW, Permission.COMMENT,
Permission.WRITE, Permission.MODERATE],
'Administrator': [Permission.FOLLOW, Permission.COMMENT,
Permission.WRITE, Permission.MODERATE,
Permission.ADMIN],
}
default_role = 'User'
for r in roles:
role = Role.query.filter_by(name=r).first()
if role is None:
role = Role(name=r)
role.reset_permissions()
for perm in roles[r]:
role.add_permission(perm)
role.default = (role.name == default_role)
db.session.add(role)
db.session.commit()
def add_permission(self, perm):
if not self.has_permission(perm):
self.permissions += perm
def remove_permission(self, perm):
if self.has_permission(perm):
self.permissions -= perm
def reset_permissions(self):
self.permissions = 0
def has_permission(self, perm):
return self.permissions & perm == perm
下面我們給使用者賦予角色,大多數使用者在註冊時候被賦予的角色就是普通使用者,但是當某一個郵箱出現用來註冊的時候,那就直接賦予管理員許可權,所以給User類新增一個初始化方法:
def __init__(self, **kwargs):
super(User, self).__init__(**kwargs)
if self.role is None:
if self.email == current_app.config['FLASKY_ADMIN']:
self.role = Role.query.filter_by(name='Administrator').first()
if self.role is None:
self.role = Role.query.filter_by(default=True).first()
下面我們實現角色驗證功能,即特定的路由只能讓擁有特定許可權的使用者才可以使用。
首先我們在User中增加一個方法用來檢查是否有指定的許可權:
def can(self, perm):
return self.role is not None and self.role.has_permission(perm)
def is_administrator(self):
return self.can(Permission.ADMIN)
class AnonymousUser(AnonymousUserMixin):
def can(self, permissions):
return False
def is_administrator(self):
return False
login_manager.anonymous_user = AnonymousUser
現在如果我們想讓特定的檢視函式只對特定的使用者開放,那麼我們就需要自定義一個裝飾器
在app目錄下新建一個decorators.py檔案:
from functools import wraps
from flask import abort
from flask_login import current_user
from .models import Permission
def permission_required(permission):
# 用來檢查常規許可權
def decorator(f):
@wraps(f)
# 使用裝飾器時,被裝飾後的函式已經是另外一個函數了,函式名等函式屬性會發生變化
# 這樣的改變會對測試有所影響,所以這個wraps裝飾器可以消除這樣的副作用。
def decorated_function(*args, **kwargs):
if not current_user.can(permission):
abort(403) # 使用者沒有許可權就返回403錯誤碼,即HTTP禁止錯誤
return f(*args, **kwargs)
return decorated_function
return decorator
def admin_required(f):
# 專門用來檢查管理員許可權
return permission_required(Permission.ADMIN)(f)
那麼如何使用我們剛剛定義的裝飾器呢?
from decorators import admin_requires, permission_required
from .models import Permission
@main.route('/admin')
@login_required
@admin_required
def for_admins_only():
return 'For administrators!'