1. 程式人生 > >flask token認證

flask token認證

在前後端分離的專案中,我們現在多半會使用token認證機制實現登入許可權驗證。

token通常會給一個過期時間,這樣即使token洩露了,危害期也只是在有效時間內,超過這個有效時間,token過期了,就需要重新生成一個新的token。

如何生成token呢?

1、建立使用者資料庫,本文會使用flask-SQLAlchemy(ORM)去管理資料庫:

  首先建立一個使用者模型:包括了使用者暱稱,賬號(郵箱或者電話號碼等),密碼及擁有的許可權

 1 class User(Base):
 2     id = Column(Integer, primary_key=True)
3 nickname = Column(String(30), nullable=False) 4 account = Column(String(30), nullable=False) 5 _password = Column("password", String(100), nullable=False) 6 auth = Column(SmallInteger, default=1) 7 8 @property 9 def password(self): 10 return self._password 11 12
@password.setter 13 def password(self, row): 14 self._password = generate_password_hash(row) 15 16 @staticmethod 17 def register_by_email(nickname, account, password): 18 with db.auto_commit(): 19 user = User() 20 user.nickname = nickname 21 user.account = account
22 user.password = password 23 db.session.add(user) 24 25 @staticmethod 26 def checkUser(email, password): 27 # 驗證使用者名稱是否存在 28 user = User.query.filter_by(account=email).first_or_404() 29 res = user.checkPassword(password) 30 if not res: 31 raise AuthFailed() 32 scope = "adminScope" if user.auth=="2" else "scope" 33 return {"uid":user.id, "scope":scope} 34 35 def checkPassword(self, raw): 36 if not self._password: 37 return False 38 # check_password_hash將raw加密後和_password比較 39 p = generate_password_hash(raw) 40 print(p==self._password) 41 return check_password_hash(self._password, raw) 42 43 def delete(self): 44 self.status = "0"
View Code

  由於安全原因,資料庫的密碼是一定不能明文儲存的,所以此處將使用者名稱進行了加密

  本文使用的werkzeug.security下面的generate_password_hash()對密碼進行的加密,我們定義了password.setter方法,當在設定密碼時,會呼叫generate_password_hash(password)加密密碼,並將其賦值給_password

  當驗證密碼時,會呼叫werkzeug.security下面的check_password_hash(hashpwd, raw) 對使用者傳遞過來的密碼和加密後的密碼進行比對,如果正確返回True

2、註冊

  當前端傳遞過來使用者名稱,密碼時進行註冊時,我們需要對使用者名稱和密碼進行如下基本驗證

    1)非空性及長度等基本校驗

    2)使用者名稱是否已經存在

  郵箱註冊form:

class EmailRegisterForm(RegisterForm):
    nickname = StringField(validators=[DataRequired(), length(3,30)])
    account = StringField(validators=[DataRequired(message="account can not be blank"), length(
        min=3, max=32, message="account length wrong"), Email(message="format wrong")])
    password = StringField(validators=[DataRequired()])

    def validate_account(self, value):
        user = User.query.filter_by(account=value.data).first()
        if user:
            raise ParamsError(msg = "使用者已存在")

 

當驗證成功後,會呼叫我們在User模型下面定義的register_by_email() 方法進行註冊。

@api.router("/register", methods=["POST"])
def register():
    data = request.json
    form = RegisterForm(data=data).validate_for_api()
    promise = {
        ClientType.REGISTER_EMAIL:_register_by_email,
        ClientType.REGISTER_MOBILE:_register_by_mobile()
    }
    promise[form.type.data]()
    return Success()


def _register_by_email():
    form = EmailRegisterForm(data=request.json).validate_for_api()
    nickname = form.nickname.data
    account = form.account.data
    password = form.password.data
    User.register_by_email(nickname, account,password)

現在我們使用postman傳送一條註冊請求

 

 

如果使用者已經存在,會返回400

 

3、登入,生成token

  生成token的方式有很多種,如產生一個固定長度的隨機字串,和使用者名稱密碼及過期時間一起儲存在資料庫中,這樣token就是一個普通的字串,可以方便的和其他字串驗證比較並可以檢查是否過期

  比較複雜一點的做法就是,不要將token儲存在資料庫,而是使用數字簽名作為token,這樣做的好處是經過使用者數字簽名的token是可以防止篡改的。

  flask使用與數字簽名類似的方法去實現加密的token,我們可以直接使用itsdangerous庫去實現。

  生成token,需要用到itsdangerous下面的TimedJSONWebSignatureSerializer

  首先我們例項化一個Serializer,並將我們的祕鑰SECRET_KEY和過期時間作為引數,返回一個TimedJSONWebSignatureSerializer型別物件

  然後呼叫TimedJSONWebSignatureSerializer物件的dumps方法,將我們想要寫入到token中的資訊以字典形式傳遞進去即可。

from itsdangerous import TimedJSONWebSignatureSerializer as Serializer

def generate_auth_token(uid, type, scope, expiration=7200):
    serializer = Serializer(current_app.config["SECRET_KEY"], expires_in=expiration)
    token = serializer.dumps({"uid":uid, "type":type.value, "scope":scope})
    return token

  當前端傳遞使用者名稱,密碼到服務端時,服務端校驗使用者存在並且密碼正確時候,就會呼叫generate_auth_token函式,生成token

  值得注意的一點是,這裡生成的token是二進位制的,所以我們在返回給前端時,需要將二進位制解碼token.decode("ascii")

  後面使用者在訪問需要登入才能訪問的的介面時,就不需要再登入,只需要將token傳遞過來即可。

  1)驗證使用者名稱是否存在,此方法作為靜態方法放在User模型下

    @staticmethod
    def checkUser(email, password):
        user = User.query.filter_by(account=email).first_or_404()
        res = user.checkPassword(password)
        if not res:
            raise AuthFailed()
        scope = "adminScope" if user.auth=="2" else "scope"
        return {"uid":user.id, "scope":scope}

  2)校驗密碼是否匹配  

    def checkPassword(self, raw):
        if not self._password:
            return False
        # check_password_hash將raw加密後和_password比較
        return check_password_hash(self._password, raw)

  3)校驗通過後,呼叫generate_auth_token方法生成token

@api.router("/", methods=["POST"])
def get_token():
    data = request.json
    form = EmailLoginForm(data=data).validate_for_api()
    type = form.type.data
    promise = {
        ClientType.REGISTER_EMAIL:User.checkUser
    }
    identify = promise[ClientType(type)](form.account.data, form.password.data)
    expiration = current_app.config["EXPIRATION"]
    token = generate_auth_token(identify["uid"], type,identify["scope"], expiration)
    r = {
        "token":token.decode("ascii")
    }
    return jsonify(r)

 

4、token認證

  如使用者想要獲取使用者資訊,這個是要登入後才能訪問的介面,我們可以使用一個裝飾器 @auth.login_required 保護,即表示只有正常登入的使用者才可以訪問

  這個裝飾器用到了flask_httpauth庫下面的HTTPBasicAuth

  auth = HTTPBasicAuth

  HTTP Basic Authentication 協議沒有具體要求必須使用使用者名稱密碼進行驗證,HTTP頭可以使用兩個欄位去傳輸認證資訊,對於token,我們只需要將token作為使用者名稱傳遞過去即可,密碼欄位可以不填

  @auth.verify_password將作為@auth.login_required的中校驗密碼的回撥函式被呼叫。

  我們前面生成token的時候,用到了我們自定義了SECRET_KEY加密,同樣解密也需要使用我們的祕鑰SECRET_KEY,加密呼叫的是serializer.dumps(),解密對應的需要使用serializer.loads()

  呼叫serializer.loads(token)時,如果捕捉到下面兩個錯誤:

    BadSignature:簽名錯誤,簽名可能被篡改

    SignatureExpired:簽名已過期

  表示驗證token失敗,直接丟擲自定義異常,如果沒有捕捉到錯誤,表示,驗證通過。可以從中取得前面加密的使用者資訊,並將資訊儲存在g變數中,留做他用。

  這裡的g變數和request一樣,都是代理模式的實現,而且是執行緒隔離的,所以也不用擔心多個請求執行緒導致資料錯亂。

@auth.verify_password
def check_authorization(token, pwd):
    user_info = check_auth_token(token)
    if not user_info:
        return False
    else:
        g.user = user_info
        return True

def check_auth_token(token):
    serialzer = Serializer(current_app.config["SECRET_KEY"])
    try:
        s = serialzer.loads(token)
    except BadSignature:
        raise AuthFailed(msg="token is invalid", error_code=1004)
    except SignatureExpired:
        raise AuthFailed(msg="token is expired", error_code=1004)
    uid = s["uid"]
    type = s["type"]
    scope = s["scope"]
    return user(uid, type, scope)