从头搭建一个flask鉴权系统之注册-阿里云开发者社区

知识树

1.Python发送email

2.token处理

3.用户账号管理

一个微型的Email工具

因为当用户注册时，可能会涉及到给用户发送邮件，所以这里先写一个发送Email的小工具。

直接使用Python自带的smtplib库和email库，我这里使用的是Google的smtp服务，并且把登陆用户名和密码设置在环境变量中，毕竟属于敏感信息嘛。然后就是定义消息体、邮件主题等信息。另外为了提高发送邮件的效率，采用了多线程的方式。

def sendmail(to, subject, text):
    # 第三方 SMTP 服务
    mail_host = "smtp.gmail.com"  # 设置服务器
    mail_user = os.environ.get('MAIL_USERNAME')  # 用户名， 从环境变量中获取
    mail_pass = os.environ.get('MAIL_PASSWORD')  # 口令
    sender = os.environ.get('MAIL_USERNAME')
    receivers = [to]
    message = MIMEText(text, 'plain', 'utf-8')
    message['From'] = Header('萝卜大杂烩', 'utf-8')
    message['to'] = Header(to, 'utf-8')
    subject = subject
    message['Subject'] = Header(subject, 'utf-8')
    smtpobj = smtplib.SMTP(mail_host, 25)
    smtpobj.ehlo()
    smtpobj.starttls()
    smtpobj.login(mail_user, mail_pass)
    thr = Thread(target=smtpobj.sendmail, args=[sender, receivers, message.as_string()])
    # smtpobj.sendmail(sender, receivers, message.as_string())
    thr.start()
    return thr

更新表结构

生成token的工具

这里使用itsdangerous来生成命令牌，它的TimedJSONWebSignatureSerializer 类可以生成具有过期时间的 JSON Web 签名（JSON Web Signatures，JWS）。这个类的构造函数接收的参数是一个密钥，在 Flask 程序中可使用 SECRET_KEY 设置。

dumps() 方法为指定的数据生成一个加密签名，然后再对数据和签名进行序列化，生成令牌字符串。expires_in 参数设置令牌的过期时间，单位为秒。

为了解码令牌，序列化对象提供了 loads() 方法，其唯一的参数是令牌字符串。这个方法会检验签名和过期时间，如果通过，返回原始数据。如果提供给 loads() 方法的令牌不正确或过期了，则抛出异常。

新的表结构

把生成命令牌的动作定义在WebUser里面，同时再为该类增加一个confirmed字段，用来标识用户是否确认，所以现在完整的WebUser类定义为

class WebUser(UserMixin, db.Model):
    __tablename__ = 'webuser'
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.String(64), unique=True, index=True)
    email = db.Column(db.String(64), unique=True, index=True)
    username = db.Column(db.String(64), unique=True, index=True)
    password_hash = db.Column(db.String(128))
    confirmed = db.Column(db.Boolean, default=False)
    @staticmethod
    def init_user():
        users = WebUser.query.filter_by(username='admin').first()
        if users is None:
            users = WebUser(user_id=time.time(), email='admin@123.com', username='admin', confirmed=True)
        users.password = '123456'
        db.session.add(users)
        db.session.commit()
    @property
    def password(self):
        raise AttributeError('password is not readable attribute')
    @password.setter
    def password(self, password):
        self.password_hash = generate_password_hash(password)
    def verify_password(self, password):
        if self.password_hash is not None:
            return check_password_hash(self.password_hash, password)
    def generate_confirmation_token(self, expiration=3600):
        s = Serializer(current_app.config['SECRET_KEY'], expiration)
        return s.dumps({'confirm': self.id})
    def confirm(self, token):
        s = Serializer(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)
        db.session.commit()
        return True

也可以看到，初始化用户的代码里，我设置了confirmed字段为True，即他默认为确认状态。

注册路由函数

这里定义了两个注册的路由函数，一个是需要用户使用邮件确认的，一个是不需要的。

其实也很好实现，不需要确认的函数，只要设置该用户的confirmed字段为True即可

@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegisterForm()
    if form.validate_on_submit():
        user = WebUser(email=form.email.data,
                       username=form.username.data, password=form.password.data,
                       user_id=time.time(), confirmed=True)
        db.session.add(user)
        db.session.commit()
        flash('You can login now.')
        return redirect(url_for('auth.login'))
    return render_template('auth/register.html', form=form)

对于需要确认的函数，需要发送邮件，这里就使用到我们刚刚定义的sendemail函数了

@auth.route('/registerconfirm', methods=['GET', 'POST'])
def register_confirm():
    form = RegisterForm()
    if form.validate_on_submit():
        user = WebUser(email=form.email.data,
                       username=form.username.data, password=form.password.data,
                       user_id=time.time())
        db.session.add(user)
        db.session.commit()
        token = user.generate_confirmation_token()
        text = render_template('auth/email/confirm.txt', user=user, token=token)
        sendmail(user.email, 'Confirm Your Account', text)
        flash('A Confirmation email has been sent to you by your registered email.')
        return redirect(url_for('auth.login'))
    return render_template('auth/registerconfirm.html', form=form)

而当用户点击确认链接后，跳转的确认函数逻辑为

@auth.route('/confirm/<token>')
@login_required
def confirm(token):
    if current_user.confirmed:
        return redirect(url_for('main.index'))
    if current_user.confirm(token):
        flash('You have confirmed your account')
    else:
        flash('The confirmation link is invalid or has expired.')
    return redirect(url_for('main.index'))

拿着用户确认链接里的token，调用confirm方法来更新confirmed字段。

重新发送confirm邮件

定义确认函数

如果用户没有在token超时时间内完成确认，而某些页面又是必须要确认才能访问的，所以就需要重新发送一个确认链接给用户，所以定义一个resend confirm函数，重新生成一个确认链接。因为这个是用户的主动行为，所以用login_required函数限制只有在用户登陆的情况下才可以使用。

@auth.route('/resendconfirm')
@login_required
def resend_confirmation():
    token = current_user.generate_confirmation_token()
    text = render_template('auth/email/confirm.txt', user=current_user, token=token)
    sendmail(current_user.email, 'Confirm Your Account', text)
    flash('A new confirmation email has been sent to you by your registered email')
    return redirect(url_for('main.index'))

定义哪些页面非确认用户不可访问

Flask有两个钩子函数，分别是 before_request和before_app_request 修饰器。

before_request：注册一个函数，在每次请求之前运行。

before_app_request：针对程序全局，每次请求之前运行。

这里使用before_app_request注册一个函数，用来区别对待不同的页面

@auth.before_app_request
def before_request():
    if current_user.is_authenticated and not current_user.confirmed and request.endpoint == 'main.needconfirm':
        return redirect(url_for('auth.unconfirmed'))

可以看到，只有当用户为登陆状态，且没有确认，并且endpoint为main.needconfirm时，before_app_request才会拦截请求，并跳转到auth.unconfirmed页面。

其中needconfirm函数定义如下

@main.route('/needconfirm')
@login_required
def needconfirm():
    print(request.endpoint)
    return 'Only confirmed users are allowed!'

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')

模板定义

现在新增了几个模板，分别为register.html、confirm.txt和unconfirmed.html等

对于register.html模板，是从login页面跳转过来的，在login页面添加不同的提示，指向不同的逻辑函数

<div class="col-md-12">
    <a href="{{ url_for('auth.register') }}">
        点击此处注册,不需要确认邮箱
    </a>
</div>
<div class="col-md-12">
    <a href="{{ url_for('auth.register_confirm') }}">
        点击此处注册,需要确认邮箱
    </a>
</div>

confirm.txt是邮件模板，里面需要展示用户的确认url，而且这个url需要是绝对url，故使用_external=True来生成绝对url

{{ url_for('auth.confirm', token=token, _external=True) }}

而unconfirmed.html模板只需要调用resend_confirmation函数即可

<a href="{{ url_for('auth.resend_confirmation') }}">Click here</a>

用户账号管理

修改密码

这个功能相对简单一些，只要用户在登陆状态下，就可以展示一个表单，供用户修改密码

class ChangePwdForm(FlaskForm):
    oldpwd = PasswordField('Old Password', validators=[DataRequired()])
    newpwd = PasswordField('New Password', validators=[DataRequired(),
                                                       EqualTo('newpwd2', message='两次输入的密码需要一致。')])
    newpwd2 = PasswordField('Confirm New Password', validators=[DataRequired()])
    submit = SubmitField('Change Password')

然后使用WebUser模型的verity_password方法来检测老的密码是否正确，如果正确，则更新新密码

@auth.route('/changepwd', methods=['GET', 'POST'])
@login_required
def changepwd():
    form = ChangePwdForm()
    if form.validate_on_submit():
        if current_user.verify_password(form.oldpwd.data):
            current_user.password = form.newpwd.data
            db.session.add(current_user)
            db.session.commit()
            flash('You have changed your password!')
            return redirect(url_for('main.index'))
        else:
            flash('Invalid password')
    return render_template('auth/changepwd.html', form=form)

重设密码

重设密码相对复杂些，首先要在WebUser模型中增加两个静态方法，分别用来产生新的token和重置密码

def generate_reset_token(self, expiration=3600):
        s = Serializer(current_app.config['SECRET_KEY'], expiration)
        return s.dumps({'reset': self.id})
    def reset_password(self, token, newpwd):
        s = Serializer(current_app.config['SECRET_KEY'])
        try:
            data = s.loads(token)
        except:
            return False
        if data.get('reset') != self.id:
            return False
        self.password = newpwd
        db.session.add(self)
        db.session.commit()
        return True

然后定义两个表单，一个为重置密码请求表单，包含一个Email输入框，用来输入用户注册的邮箱；另一个为重置密码表单，如果在重设密码时输入的Email是错误的邮箱，则直接报错

class ResetPwdEmailForm(FlaskForm):
    email = StringField('Your Register Email', validators=[DataRequired(), Length(1, 64), Email()])
    submit = SubmitField('Reset Password')
class ResetPwdForm(FlaskForm):
    email = StringField('Email', validators=[DataRequired(), Length(1, 64), Email()])
    newpwd = PasswordField('New Password', validators=[DataRequired(),
                                                       EqualTo('newpwd2', message='两次输入的密码需要一致。')])
    newpwd2 = PasswordField('Confirm New Password', validators=[DataRequired()])
    submit = SubmitField('Reset Password')
    def validate_email(self, field):
        if WebUser.query.filter_by(email=field.data).first() is None:
            raise ValidationError('This Email is Invalid')

页面演示

我将完整的代码上传到GitHub上了，有兴趣的同学可以戳这里：

https://github.com/zhouwei713/flask-webauth

另外还配置了一个演示web，地址为：

https://luobodazahui.top/

感兴趣的同学可以来玩玩

从头搭建一个flask鉴权系统之注册

知识树

一个微型的Email工具

更新表结构

生成token的工具

新的表结构

注册路由函数

重新发送confirm邮件

定义确认函数

定义哪些页面非确认用户不可访问

模板定义

用户账号管理

修改密码

重设密码

页面演示

热门文章

最新文章

相关课程

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

从头搭建一个flask鉴权系统之注册

知识树

一个微型的Email工具

更新表结构

生成token的工具

新的表结构

注册路由函数

重新发送confirm邮件

定义确认函数

定义哪些页面非确认用户不可访问

模板定义

用户账号管理

修改密码

重设密码

页面演示

热门文章

最新文章

相关课程

相关电子书