Flask Login

使用Flask-Login注册登录

类似的登录页面,在前面也写过不少,也需要考虑很多的问题,比如session,账户的验证等等.

这是每一个项目,只要涉及到账户登录都会遇到的问题,因此有人为此做了一个轮子,就是Flask-Login.

1.安装

https://flask-login.readthedocs.io/en/latest/#flask-login

$ pip install flask-login

2.使用

1.初始化Flask-Login

和之前学习到的插件一样,都需要绑定到app上,才能真正的使用

from flask import Flask
from flask_login import LoginManager

app = Flask(__name__)
login_manager = LoginManager()
login_manager.init_app(app)

需要注意的是,默认情况下Flask-Login使用的是flask自带的session进行身份验证的,这就需要在config中设置SECRET_KEY.

# config.py
import os

DEBUG = True
TEMPLATES_AUTO_RELOAD = True
SECRET_KEY = os.urandom(24)

这样就建立其Flask app和Flask-Login插件之间的联系.然后让我们安装登录的常规思路一步一步如何实现.

• Flask-Login确定登录的URL
• 账户登录逻辑
• 验证账户密码
• 确认登录的账户是谁
• 登录可见

2.确认登录URL

为Flask-Login指定登录的URL.

from flask import Flask
from flask_login import LoginManager
import config

app = Flask(__name__)
app.config.from_object(config)
login_manager = LoginManager()

login_manager.init_app(app)
# 指定登录的URL
login_manager.login_view = 'login'
# 指定登录信息
login_manager.login_message = u'请先登录!'

Flask-Login是没有默认登录URL的,如果你使用的是蓝图(Blueprint),那么就应该指定其他的URL

login_manager.login_view = 'bpmodel.users.login'

指定好URL后,就可以为这个URL编写逻辑代码

@app.route('/login/', methods=['GET', 'POST'])
def login():
  pass

3.账户登陆逻辑

按照流程顺序来说,接下来就应该验证表单,在验证数据库,如果数据库中有数据,则对比后,选择是否添加session并运行登录.

# 类似如下
# 登录页
class LoginView(views.MethodView):
  def get(self):
      return render_template('login.html')
  def post(self):
      # 验证表单
      form = LoginForm()
      if form.validate_on_submit():
          # 表单验证成功,就跳转到数据库验证
          print('表单验证成功')
          email = form.email.data 
          password = form.password.data
          lifetime = form.lifetime.data
          # 数据库判断
          print(email, password, lifetime)
          exist_user = db.session.query(BankUsers).filter(BankUsers.email==email).filter(BankUsers.password==password).first()
          print(exist_user)
          if exist_user != None:
              if lifetime:
                  session.clear()
                  session['username'] = exist_user.username
                  session.permanent = True
                  return redirect(url_for('personal'))
              else:
                  session.clear()
                  session['username'] = exist_user.username
                  return redirect(url_for('personal'))
          else:
              return render_template('login.html', info=form.errors)
      else:
          return render_template('login.html', info="用户名密码不正确,请注册")

同样的,在Flask-Login中遵循同样的流程.

1.编写用户类(ORM)

要想使用Flask-Login扩展,程序的User模型(ORM)必须实现几个方法

方法	说明
is_authenticated	如果用户已经登录,返回True,否则返回False
is_active	如果允许用户登录,返回True,否则返回Falser,如果禁用用户,返回False
is_anonymous	匿名用户(未登录用户)返回False
get_id()	必须返回用户的唯一标识符作为ID,该ID必须为Unicode编码

这4个方法可以在模型类中作为方法直接实现,不过每次为User模型编写,会比较麻烦,Flask-Login提供了一个UserMixin类,其中包含这些方法的默认实现,并且可以实现大多数的需求,所以可以直接继承.

# sqlModel.py
# 类似如下

from flask_login import UserMixin
from exts import db 

class User(db.Model, UserMixin):
  	__tablename__ = 'users'
      pass

Flask_login还要求程序必须实现一个回调函数,这个回调函数用于通过session中存储的用户ID重新加载用户对象.它应该接受用户的Unicode ID,并返回相应的用户对象.

使用指定的标识符加载用户:

# sqlModel.py
from app import login_manager 

@login_manger.user_loader
def load_user(user_id):
  if query_user(username) is not None:
      curr_user = User()
      curr_user.id = username
  	return curr_user

如果ID无效,函数应该返回None(不引发异常),ID将从session中手动删除并且程序可以继续执行.

为什么需要这个回调函数?

is_authenticated就是用来判断用户是否登录是否有权来操作的,但是并不能知道当前登录的账户具体是谁,所以这里就需要一个回调函数用来判断是谁.

---

为了简便,这里不使用MySQL,简单的设计一个ORM

# 创建ORM映射
# 用户记录表
users = [
  {'username': 'Tom', 'password': '111111'},
  {'username': 'Michael', 'password': '123456'}
]


# 通过用户名,获取用户记录,如果不存在,返回None
def query_user(username, password):
  for user in users:
      if user['username'] == username:
          return user
      else:
          return None

class User(UserMixin):
  pass

# 如果用户名存在,就构造一个用户类对象,并使用用户名作为ID,如果不存在就返回None
# 回调函数
@login_manager.user_loader
def load_user(username):
  if query_user(username) is not None:
      curr_user = User()
      curr_user.id = username
      return curr_user
  else:
      return None

注意实现回调函数.

4.登录实现

实现如下

from flask import Flask, request, abort, redirect, url_for, render_template
from flask_login import LoginManager, login_user, UserMixin, login_required, logout_user
import config
from urllib.parse import urlparse, urljoin

app = Flask(__name__)
app.config.from_object(config)
login_manager = LoginManager()

login_manager.init_app(app)
# 指定登录的URL
login_manager.login_view = 'login'

# 创建ORM映射
# 用户记录表
users = [
  {'username': 'Tom', 'password': '111111'},
  {'username': 'Michael', 'password': '123456'}
]


# 通过用户名,获取用户记录,如果不存在,返回None
def query_user(username):
  for user in users:
      if user['username'] == username:
          return user
      else:
          return None

class User(UserMixin):
  pass

# 如果用户名存在,就构造一个用户类对象,并使用用户名作为ID,如果不存在就返回None
# 回调函数
@login_manager.user_loader
def load_user(username):
  if query_user(username) is not None:
      curr_user = User()
      curr_user.id = username
      return curr_user
  else:
      return None

def is_safe_url(target):
  ref_url = urlparse(request.host_url)
  test_url = urlparse(urljoin(request.host_url, target))
  return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc

@app.route('/login/', methods=['GET', 'POST'])
def login():
  # 假设通过表单验证
  # 假设通过数据库验证
  if request.method == 'POST':
      username = request.form.get('username')
      password = request.form.get('password')
      # 验证表单,数据库
      user = query_user(username)
      if user and password == user['password']:
          # curr_user 是 User类的一个实例
          curr_user = User()
          curr_user.id = username
          # 通过 Flask-login的login_user来登录账户
          login_user(curr_user)
          nextD = request.args.get('next')
          print(nextD)
          # is_safe_url 用来检查url是否可以安全的重定向
          # 避免重定向攻击
          # if not is_safe_url(nextD):
          #     return abort(404)
          # return redirect(next or url_for('index'))
          return redirect(url_for('index'))
  else:
      return  render_template('login.html')

@app.route('/')
def index():
  return 'Hello World!'

# 登录可见的页面
@app.route('/personal/', methods=['GET', 'POST'])
@login_required
def personal():
  pass

# 登出
@app.route('/logout/')
@login_required
def logout():
  logout_user()
  return 'logout'

if __name__ == '__main__':
  app.run()

这里使用了一个方法login_user,通过上面的对比,我们也知道在这一步,应该实现的是session的相关信息,为什么只使用这一个方法就可以实现,我们可以查看一下源码.

from flask import session

def login_user(user, remember=False, duration=None, force=False, fresh=True):
  if not force and not user.is_active:
      return False

  user_id = getattr(user, current_app.login_manager.id_attribute)()
  session['_user_id'] = user_id
  session['_fresh'] = fresh
  session['_id'] = current_app.login_manager._session_identifier_generator()

  if remember:
      session['_remember'] = 'set'
      if duration is not None:
          try:
              # equal to timedelta.total_seconds() but works with Python 2.6
              session['_remember_seconds'] = (duration.microseconds +
                                              (duration.seconds +
                                               duration.days * 24 * 3600) *
                                              10**6) / 10.0**6
          except AttributeError:
              raise Exception('duration must be a datetime.timedelta, '
                              'instead got: {0}'.format(duration))

  current_app.login_manager._update_request_context_with_user(user)
  user_logged_in.send(current_app._get_current_object(), user=_get_user())
  return True

以上包含的参数要内容是将用户信息放入

参数	说明
user	要登录的用户对象
remember(bool)	session过期后时候记住用户
duration	记住用户的过期时长
force(bool)	如果用户处于不活跃状态,设置这个参数(True)将强制登录用户
fresh(bool)	登录用户时,将用户session标记为fresh

可以看到login_user主要就是将用户信息加入到session中.确定是哪个user则需要使用回调函数,并且必须返回user本身

    def user_loader(self, callback):
    '''
    This sets the callback for reloading a user from the session. The
    function you set should take a user ID (a ``unicode``) and return a
    user object, or ``None`` if the user does not exist.
  
    :param callback: The callback for retrieving a user object.
    :type callback: callable
    '''
    self._user_callback = callback
    return callback
  
  
  
def _load_user(self):
    '''Loads user from session or remember_me cookie as applicable'''
  
    if self._user_callback is None and self._request_callback is None:
        raise Exception(
            "Missing user_loader or request_loader. Refer to "
            "http://flask-login.readthedocs.io/#how-it-works "
            "for more info.")
  
    user_accessed.send(current_app._get_current_object())
  
    # Check SESSION_PROTECTION
    if self._session_protection_failed():
        return self._update_request_context_with_user()
  
    user = None
  
    # Load user from Flask Session
    user_id = session.get('_user_id')
    if user_id is not None and self._user_callback is not None:
        user = self._user_callback(user_id)

可以看到一切都和self._user_callback有关,这就是为什么要设置回调函数,并且返回用户对象本身.

---

是否验证next,理论上说是需要验证next的,但是Flask-Login这里有所争议,具体参照

https://stackoverflow.com/questions/60532973/how-do-i-get-a-is-safe-url-function-to-use-with-flask-and-how-does-it-work

---

以上,通过login_user将用户登录到系统,通过user_loader判断登录的用户是谁.

登录可见

为了确保只有登录的用户才能访问页面,在Flask CSRF保护中,我们也自己实现了一个装饰器.

# limit.py
from functools import wraps
from flask import redirect,url_for,session

def limit_session(func):
  @wraps(func)
  def wrapper(*args,**kwargs):
      username = session.get('username',None)
      if username:
          return func(*args,**kwargs)
      else:
          session.clear()
          return redirect(url_for('login'))
  return wrapper

并把它应用到了视图中

class SaveMoneyView(views.MethodView):
  decorators = [limit_session]
  pass

同样的,Flask-Login也实现了相同的功能login_required

from  flask_login import login_required

# 登录可见的页面
@app.route('/personal/', methods=['GET', 'POST'])
@login_required
def personal():
  pass

同样可以查看源码,检查这一过程.

def login_required(func):
  @wraps(func)
  def decorated_view(*args, **kwargs):
      if request.method in EXEMPT_METHODS: # 通常该请求是获取服务器支持的HTTP请求方法
          return func(*args, **kwargs)
      elif current_app.config.get('LOGIN_DISABLED'): # 如果LOGIN_DISABLED 为True,则忽略验证
          return func(*args, **kwargs)
      elif not current_user.is_authenticated:  # 判断用户是否登录
          """
          没有登录的情况下:
          1.如果注册了 LoginManager.unauthorized_handler 则这个时候调用这个函数 
            2. 向用户提示 LoginManager.login_message信息 
            3.有 login_view的情况下,跳转到login_view,没有则返回abort(401)
            """
            return current_app.login_manager.unauthorized() 
        return func(*args, **kwargs)
    return decorated_view

以上涉及到了unauthorized()方法,这里可以简单看下

def unauthorized(self):
    # current_app._get_current_object() 指向了app本身
    # 查看 flask上下文相关
    user_unauthorized.send(current_app._get_current_object())
  
    if self.unauthorized_callback:
        return self.unauthorized_callback()
  
    if request.blueprint in self.blueprint_login_views:
        login_view = self.blueprint_login_views[request.blueprint]
    else:
        login_view = self.login_view
  
    if not login_view:
        abort(401)
  
    if self.login_message:
        if self.localize_callback is not None:
            flash(self.localize_callback(self.login_message),
                  category=self.login_message_category)
        else:
            flash(self.login_message, category=self.login_message_category)
  
    config = current_app.config
    if config.get('USE_SESSION_FOR_NEXT', USE_SESSION_FOR_NEXT):
        login_url = expand_login_view(login_view)
        session['_id'] = self._session_identifier_generator()
        session['next'] = make_next_param(login_url, request.url)
        redirect_url = make_login_url(login_view)
    else:
        redirect_url = make_login_url(login_view, next_url=request.url)
  
    return redirect(redirect_url)

退出登录

对于退出登录,只需要调用logout_user即可.

from flask_login import logout_user
# 登出
@app.route('/logout/')
@login_required
def logout():
  logout_user()
  return 'logout'

源码

def _get_user():
  if has_request_context() and not hasattr(_request_ctx_stack.top, 'user'):
      current_app.login_manager._load_user()

  return getattr(_request_ctx_stack.top, 'user', None)

def logout_user():
  '''
  Logs a user out. (You do not need to pass the actual user.) This will
  also clean up the remember me cookie if it exists.
  '''

  user = _get_user()

  if '_user_id' in session:
      session.pop('_user_id')

  if '_fresh' in session:
      session.pop('_fresh')

  if '_id' in session:
      session.pop('_id')

  cookie_name = current_app.config.get('REMEMBER_COOKIE_NAME', COOKIE_NAME)
  if cookie_name in request.cookies:
      session['_remember'] = 'clear'
      if '_remember_seconds' in session:
          session.pop('_remember_seconds')

  user_logged_out.send(current_app._get_current_object(), user=user)

  current_app.login_manager._update_request_context_with_user()
  return True

从堆栈结构中找到session中的user,实现登出操作.

自定义未授权访问的处理方法

@login_required装饰器对于未登录用户访问的默认处理是重定向到登录视图,如果我们不想它这么做的话,可以自定义处理方法:

@login_manager.unauthorized_handler
def unauthorized_handler():
  return 'Unauthorized'

这个@login_manager.unauthorized_handler装饰器所修饰的方法就会代替@login_required装饰器的默认处理方法.

有了上面的代码,当未登录用户访问index视图时,页面就会直接返回Unauthorized信息.

REmember Me

在登录视图中,调用login_user()方法,传入remeber=True,就可以实现记住我功能

login_user(curr_user, remember=True)

Flask-Login是通过在Cookie实现的,它会在Cookie中添加一个remember_token字段来记住之前登录的用户信息,所以禁用Cookie的话,该功能将无法工作.

fresh

当用户登录时,他们的会话会被标记为fresh(login_user()操作),即在session中设置_fresh=True.如果使用rember me的cookie从新登录,会话被标记成not fresh.

在某些情况下,会强制要求用户登录一次,比如修改登录密码,这就需要fresh_login_required.

fresh_login_required除了可以验证用户登录,也将确保他们的登录是fresh,如果不是fresh,它会重新将用户准到输入验证条件的页面,这主要是为了确保用户修改个人信息的敏感操作.

def fresh_login_required(func):

  @wraps(func)
  def decorated_view(*args, **kwargs):
      if request.method in EXEMPT_METHODS:
          return func(*args, **kwargs)
      elif current_app.config.get('LOGIN_DISABLED'):
          return func(*args, **kwargs)
      elif not current_user.is_authenticated:
          return current_app.login_manager.unauthorized()
      elif not login_fresh():
          return current_app.login_manager.needs_refresh()
      return func(*args, **kwargs)
  return decorated_view

# confirm_login 将会话重新标记为 fresh
def confirm_login():
  session['_fresh'] = True
  session['_id'] = current_app.login_manager._session_identifier_generator() 
  user_login_confirmed.send(current_app._get_current_object())

from flask_login import fresh_login_required

@app.route('/login/')
@fresh_login_required
def home():
  	return 'Logged in as: %s' % current_user.get_id()

Cookie设置

可以在config中设置cookie信息

REMEMBER_COOKIE_NAME	存储remember me信息的cookie名称,默认为remember_token
REMEMBER_COOKIE_DURATION	cookie过期时长,默认365天
REMEMBER_COOKIE_DOMAIN	默认为None,比如设置为.example.com将会允许所有子域名
REMEMBER_COOKIE_PATH	限制remember me到一个路径,默认为/
REMEMBER_COOKIE_SECURE	限制remember me到https
REMEMBER_COOKIE_HTTPONLY	限制remember me到客户端脚本访问
REMEMBER_COOKIE_REFRESH_EACH_REQUEST	Ture:cookie每次刷新会延长生命周期

session保护

Flask-Login自动启用会话保护功能.对于每个请求,它会验证用户标识,这个标识是由客户端IP地址和”User Agent”的值经SHA512编码而来.在用户登录成功时,Flask-Login就会将这个值保存起来以便后续检查.默认的会话保护模式是basic,为了加强安全性,你可以启用强会话保护模式.方法是配置LoginManager实例对象中的session_protection属性：

login_manager.session_protection = "strong"

在strong模式下，一旦用户标识检查失败，便会清空所用Session内容，并且”Remember Me”也失效。而basic模式下，只是将登录标为非Fresh登录。你还可以将login_manager.session_protection置为”None”来取消会话保护。

request_loader:禁用API的session cookie

在对API进行认证时,我们可能不会用到cookie.为此,可以使用一个自定义的session接口,该接口根据在请求中设计的标志跳过保存session

from flask import g
from flask.sessions import SecureCookieSessionInterface
from flask_login import user_loaded_from_header

class CustomSessionInterface(SecureCookieSessionInterface):
  """防止 API 请求创建 session。"""
  def save_session(self, *args, **kwargs):
      if g.get('login_via_header'):
          return
      return super(CustomSessionInterface, self).save_session(*args,
                                                              **kwargs)

app.session_interface = CustomSessionInterface()

@user_loaded_from_header.connect
def user_loaded_from_header(self, user=None):
  g.login_via_header = True

这可以放置在用户使用request_loader进行认证时设置Flask Session

current_user

Flask-Login中可以直接使用current_user访问已经登录的用户,current_user可以在每个模板中世界使用.

{% if current_user.is_authenticated %}
Hi {{ current_user.name }}!
{% endif %}

3.实例

1.保存cookie到memcache中,并设置超时时间

# app.py
from flask import Flask, request, abort, redirect, url_for, render_template, session
from flask_login import LoginManager, login_user, UserMixin, login_required, logout_user
import config
from urllib.parse import urlparse, urljoin
from flask_session import Session as Fsession
from datetime import timedelta

app = Flask(__name__)
app.config.from_object(config)
login_manager = LoginManager()
Fsession(app)

login_manager.init_app(app)
# 指定登录的URL
login_manager.login_view = 'login'

# 创建ORM映射
# 用户记录表
users = [
  {'username': 'Tom', 'password': '111111'},
  {'username': 'Michael', 'password': '123456'}
]


# 通过用户名,获取用户记录,如果不存在,返回None
def query_user(username):
  for user in users:
      if user['username'] == username:
          return user
      else:
          return None

class User(UserMixin):
  pass

# 如果用户名存在,就构造一个用户类对象,并使用用户名作为ID,如果不存在就返回None
# 回调函数
@login_manager.user_loader
def load_user(username):
  if query_user(username) is not None:
      curr_user = User()
      curr_user.id = username
      return curr_user
  else:
      return None

def is_safe_url(target):
  ref_url = urlparse(request.host_url)
  test_url = urlparse(urljoin(request.host_url, target))
  return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc

@app.route('/login/', methods=['GET', 'POST'])
def login():
  # 假设通过表单验证
  # 假设通过数据库验证
  if request.method == 'POST':
      username = request.form.get('username')
      password = request.form.get('password')
      # 验证表单,数据库
      user = query_user(username)
      if user and password == user['password']:
          # curr_user 是 User类的一个实例
          curr_user = User()
          curr_user.id = username
          # 通过 Flask-login的login_user来登录账户
          login_user(curr_user, remember=True, duration=timedelta(days=60))
          nextD = request.args.get('next')
          print(nextD)
          # is_safe_url 用来检查url是否可以安全的重定向
          # 避免重定向攻击
          # if not is_safe_url(nextD):
          #     return abort(404)
          # return redirect(next or url_for('index'))
          return redirect(url_for('index'))
  else:
      return  render_template('login.html')

@app.route('/')
def index():
  return 'Hello World!'

# 登录可见的页面
@app.route('/personal/', methods=['GET', 'POST'])
@login_required
def personal():
  pass

# 登出
@app.route('/logout/')
@login_required
def logout():
  print(session['_user_id'])
  logout_user()
  try:
      print(session['_user_id'])
  except Exception as e:
      print('session已经删除')
  return 'logout'

if __name__ == '__main__':
  app.run()

# config.py
#!/usr/bin/env python
# coding=utf-8
import os
from exts import client
from datetime import timedelta

DEBUG = True
TEMPLATES_AUTO_RELOAD = True
SECRET_KEY = os.urandom(24)

# 设置flask-session相关
SESSION_TYPE = 'memcached' # 指定类型
SESSION_MEMCACHED = client # 指定实例
SESSION_USE_SIGNER = True  # 设置加密
SESSION_KEY_PREFIX = 'session' # 指定值的前缀
PERMANENT_SESSION_LIFETIME = timedelta(days=0)

# exts.py
from pymemcache.client import Client
client = Client(('192.168.0.101', 11211), allow_unicode_keys=True, encoding='utf8')

<!-- login.html --->
<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <title>Title</title>
</head>
<body>
{% if current_user.is_authenticated %}
  Hi {{ current_user.name }}
{% endif %}
<form action="{{ url_for('login') }}" method="post">
  <label for="">username:</label><input type="text" name="username">
  <label for="">password:</label><input type="password" name="password">
  <input type="checkbox" value="remember me" name="remember">
  <input type="submit" name="submit">
</form>
</body>
</html>