Fastapi之OAuth2登錄認證

登錄認證概述

登錄是很多系統的基本功能, 有些頁面(如用戶信息頁面)需要登錄之后才能進行訪問. 實現這一功能的方案大體為:

1. 首先進行登錄, 登錄成功后, 給前端(瀏覽器)返回一個值"xxxx"(session或者token)
2. 前端(瀏覽器)去訪問需要登錄的頁面(如用戶信息頁面)時, 會帶上上面值"xxxx"
3. (后端)服務器根據傳入的值"xxxx"獲取到這個值對應的用戶是哪一個, 那么就返回這個用戶的信息

上面的方案再根據 返回值"xxx" 的方式不同, 可以細分為兩種:

• session:

1. 用戶登錄后, 在服務器這一方會生成一個隨機值session_id, 把session_id和用戶的唯一標識(如user_id)映射起來保存, 可以保存在數據庫中或者內存中, 然后給前端返回這個session_id
2. 前端下次請求時, 帶上這個session_id, 后端可以根據這個session_id找到對應的user_id, 即可知道請求的是哪個用戶的信息了

但是這種方式由於需要服務器端保存這個session_id, 因此會衍生出一些問題, 如:

a. 保存session_id需要耗費服務器資源

b. 當業務量比較大需要多台機器進行負載均衡, 統一提供服務時, 多台機器的session_id需要進行同步, 否則跨機器訪問時就獲取不到用戶了

於是就慢慢發展出了第二種方式: token

• token:

1. 用戶登錄后, 服務器端根據user_id和秘鑰(鹽值)進行簽名加密, 直接返回給前端(瀏覽器), 不進行保存操作
2. 前端下次請求時, 帶上這個token, 服務器端對這個token進行解密, 獲取到解密后的user_id, 即可知道請求的是哪個用戶的信息了

基於token的登錄認證解決了基於session方式帶來的問題, 且擴張性更強, 已經是現在的首選方案了

OAuth2

既然采用了上述的token方案, 那么可以考慮的再具體一點, 如何進行token的加密和解密? 后端返回token的格式是什么? 前端訪問時token是放在url參數中, 還是請求頭中或者請求體中? 前端攜帶token訪問時的格式是什么?

OAuth2就是對上述具體問題的一套規范的解決方案, 當然它不只是解決上面的問題, 也可以解決第三方應用的授權問題等.

在FastAPI中, 提供了多種認證解決方案工具, 其中也包括了OAuth2, 可以使用OAuth2PasswordBearer類來實現OAuth2的功能, 使用的是OAuth2中的一種認證方案, 通過bearer token來攜帶token, 具體做法就是:

在請求頭中添加參數Authorization, 其值為Bearer和token中間使用空格連接形成的字符串, 如Bearer your_token_string, 注意, Authorization和Bearer都是規范中固定的寫法, 不可修改

OAuth2PasswordBearer

簡單使用如下:

from fastapi import FastAPI, Depends
# 導入OAuth2PasswordBearer
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
# 實例化oauth2, tokenUrl暫時隨便給一個值, 后面會講解其用法
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='xxx')

# 定義一個API, 這個API需要登錄后才可以進行訪問
# 那么就需要設置一個參數, 這個參數依賴於上面的oauth2_scheme
@app.get('/')
async def test(s: str = Depends(oauth2_scheme)):
    return {'hello': s}

啟動項目, 來到FastAPI的文檔頁面http://127.0.0.1:8000/docs

可以看到多了這個testapi后面多了一把鎖, 直接執行訪問這個api, 返回401錯誤

在響應頭中, 也可以看到需要Bearer認證
點擊fastapi文檔網頁中的鎖, 會彈出一個賬戶密碼的輸入框, 這是fastapi文檔集成的一個類似於登錄頁面的界面

在這個界面可以看到Token URL為xxx, 即為上面實例化OAuth2PasswordBearer時傳入的tokenUrl參數

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='xxx')

這個參數的作用是當輸入完用戶密碼后, 點擊Authorize按鈕后, 會請求127.0.0.1:8000/xxx這個網址, 把用戶密碼通過Form表單的形式傳遞給這個請求, 即這個/xxx就是我們網站的后台登錄接口.

一般登錄接口我們可以通過json格式將用戶名密碼放在請求體中傳入, 但是OAuth2規范中要求用戶名密碼需要通過Form表單格式application/x-www-form-urlencoded傳入, 當然項目中可以結合實際情況來決定通過哪種形式來傳遞用戶名密碼.

我們隨便輸入用戶名密碼, 點擊Authorize按鈕, 發現它報錯說Auth Error Error: Not Found, 這是因為我們還沒有實現這個表單格式的登錄接口, 查看后台日志可以看到確實也請求了/xxx這個接口

INFO:     127.0.0.1:53600 - "POST /xxx HTTP/1.1" 404 Not Found

也就是說這個參數tokenUrl只是為了方便fastapi的文檔網頁的認證使用, 在前后端實際項目中並不會起到作用. 我們這里先跳過這個登錄接口的實現, 繼續回到接口認證的邏輯.我們這里先跳過這個登錄接口的實現, 繼續回到接口認證的邏輯.

前面試了直接訪問需要登錄才能訪問的接口時, 會報401錯誤, 是因為我們還沒有帶上認證的信息, 而OAuth2 Bearer的認證信息需要放到請求頭中,

由於在fastapi文檔網頁中不能修改請求頭, 我們可以來到postman或者其他api調試工具中, 例如, 在postman中, 提供了一個專門的認證選項(Authorization), 提供了一些常見的認證可供選擇

這里我們選擇Bearer Token認證, 在右邊會出現Token的輸入框, 我們隨便輸入一個值, 點擊發送, 可以看到返回的不再是401, 而是接口的正常返回值

我們也可以來到請求頭的標簽頁, 可以看到postman自動添加了一個參數Authorization, 其值為Bearer hahahadadw, 說明在前面的認證選項中設置了認證后, postman會自動幫我們轉換成對應實際格式

我們如果不使用postman的認證選項, 自己手動在請求頭中添加相同的Authorization參數, 也是可以正常訪問的

回顧一下我們的api代碼

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='xxx')

@app.get('/')
async def test(s: str = Depends(oauth2_scheme)):
    return {'hello': s}

結合輸出, 我們可以看到oauth2_scheme幫我們做的事情就是把請求頭中的Authorization中的token字符串提取出來, 並返回, api參數中的s接收這個返回值

函數中接下來的操作就是首先對這個s值進行驗證, 如果s是有效的, 即token是正確的, 那么就繼續進行后續的邏輯, 如果token不正確, 則返回錯誤信息.

那么這個token的生成規則和驗證規則, 不屬於OAuth2的范疇, 我們一般使用JWT

JWT

JWT全稱是Json Web Token, 具體原理可以自行百度或者查看我的博客:

python-JWT(Json Web Token)-pyjwt - Alex-GCX - 博客園 (cnblogs.com)

在python中我們一般使用pyjwt這個包操作JWT, 我上面的博客使用的就是這個包, Fastapi官網使用的是python-jose這個包, 因為它提供了 PyJWT 的所有功能，以及之后與其他工具進行集成時你可能需要的一些其他功能。

Python-jose需要一個額外的加密后端。這里我們使用的是推薦的后端:pyca/cryptography

pip install python-jose[cryptography]

登錄接口

token一般都是在登錄之后生成的, 因此我們需要做一個簡單的登錄接口, 在登錄接口中生成jwt token, 這里我們還是使用請求體json形式傳入用戶名和密碼

from datetime import datetime, timedelta

from fastapi import FastAPI, Depends, Body
from typing import Optional
from jose import JWTError, jwt

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "ed970259a19edfedf1010199c7002d183bd15bcaec612481b29bac1cb83d8137"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def get_user_id(user_name: str, password: str):
    return 123

def create_jwt_token(data: dict, expire_delta: Optional[timedelta] = None):
    # 如果傳入了過期時間, 那么就是用該時間, 否則使用默認的時間
    expire = datetime.utcnow() + expire_delta if expire_delta else datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    # 需要加密的數據data必須為一個字典類型, 在數據中添加過期時間鍵值對, 鍵exp的名稱是固定寫法
    data.update({'exp': expire})
    # 進行jwt加密
    token = jwt.encode(claims=data, key=SECRET_KEY, algorithm=ALGORITHM)
    return token

@app.post('/login/')
async def login(user_name: str = Body(...), password: str = Body(...)):
    # 校驗用戶密碼邏輯暫時省略, 這里我們不校驗, 認為都是用戶密碼都是對的, 返回一個固定user_id
    user_id = get_user_id(user_name, password)
    # 使用user_id生成jwt token
    data = {'user_id': user_id}
    token = create_jwt_token(data)
    return {'token': token}

登錄接口為/login/

1. 接受兩個參數user_name和password, 這兩個參數從請求體中傳入
2. 通過get_user_id函數獲取對應的user_id, 這里簡單返回一個固定值
3. 將user_id創建成一個字典, 調用create_jwt_token生成jwt token值
4. 返回token給前端

在create_jwt_token方法中介紹了jwt的簡單使用:

1. 加密的數據必須為字典類型, 如果需要設置token的有效時間, 那么需要在字典中添加一個鍵值對, 鍵名為固定的exp, 值為有效期的截止時間, 是一個日期類型.
2. 使用jwt.encode()方法進行加密, 該方法需要傳入三個參數:
   • claims: 需要加密的字典類型的數據
   • key: 加密需要使用的秘鑰, 也叫做鹽值
   • algorithm: 加密的算法, 默認為ALGORITHMS.HS256

在fastapi的文檔頁面進行測試, 在請求體中隨便輸入一個用戶名密碼, 能夠返回一個token字符串

驗證前端傳入的token

在上面OAuth2PasswordBearer這一節中, 定義了一個需要用戶登錄才能訪問的接口

@app.get('/')
async def test(s: str = Depends(oauth2_scheme)):
    return {'hello': s}

接下來需要對其進行改造, 添加驗證token的邏輯

from fastapi import FastAPI, Depends, Body, HTTPException, status
from jose import JWTError, jwt

@app.get('/')
async def test(token: str = Depends(oauth2_scheme)):
    # 定義一個驗證異常的返回
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="認證失敗",
        # 根據OAuth2規范, 認證失敗需要在響應頭中添加如下鍵值對
        headers={'WWW-Authenticate': "Bearer"}
    )
    # 驗證token
    try:
        # 解密token, 返回被加密的字典
        payload = jwt.decode(token=token, key=SECRET_KEY, algorithms=[ALGORITHM])
        print(f'payload: {payload}')
        # 從字典中獲取user_id數據
        user_id = payload.get('user_id')
        print(f'user_id: {user_id}')
        # 若沒有user_id, 則返回認證異常
        if not user_id:
            raise credentials_exception
    except JWTError as e:
        print(f'認證異常: {e}')
        # 如果解密過程出現異常, 則返回認證異常
        raise credentials_exception
    # 解密成功, 返回token中包含的user_id
    return {'hello': user_id}

使用jwt.decode()進行解密, 獲取加密前的數據payload, 在payload中獲取user_id

在postman中進行測試, 首先再隨便傳入一個token

可以看到返回認證失敗, 后台打印結果為

認證異常: Not enough segments

說明token的格式是錯誤的, 這次傳入上面登錄后生成的token

可以看到正常返回了結果, 后台打印為:

payload: {'user_id': 123, 'exp': 1621159181}
user_id: 123

至此就完成了最簡單的OAuth2登錄認證流程

小結

1. 登錄認證分為基於session和基於token, 目前最常用的是基於token的方式

2. OAuth2中有一個基於token認證的規范, 常用的是Bearer token方式, 即在請求頭中添加參數Authorization, 其值為Bearer和token中間使用空格連接形成的字符串, 如Bearer your_token_string, 注意, Authorization和Bearer都是規范中固定的寫法, 不可修改

3. Fastapi中集成了OAuth的Bearer方式, 通過OAuth2PasswordBearer類實現, 這個類主要做的就是把請求頭中的token信息提取出來, 具體對token的加密和解密需要我們自己實現

4. 一般使用JWT對token進行加密jwt.encode(claims=data, key=SECRET_KEY, algorithm=ALGORITHM)和解密jwt.decode(token=token, key=SECRET_KEY, algorithms=[ALGORITHM])

5. 在OAuth2中還有一些規范, 如

   • 登錄接口傳入用戶名和密碼時, 不能使用json格式, 而是使用表單形式application/x-www-form-urlencoded進行提交, 並且用戶名必須為username, 密碼必須為password, fastapi官網的案例也是使用表單形式提交, 需要安裝python-multipart這個包. 在登錄接口參數中, 使用了OAuth2PasswordRequestForm這個類接收用戶名和密碼
   • 登錄接口返回token時的json格式為: {"access_token": xxxxx, "token_type": "bearer"}
   • 如果token驗證失敗, 需要在請求頭中添加鍵值對: {'WWW-Authenticate': "Bearer"}
   • 這些規范我們可以看實際情況選擇是否遵守

   下面是官方案例的簡單登錄接口:

   from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
   
   @app.post("/token")
   async def login(form_data: OAuth2PasswordRequestForm = Depends()):
       # 從form_data中獲取用戶名和密碼
       user_name = form_data.username
       password = form_data.password
       # 校驗用戶名和密碼校驗是否正確
       .....
   	# 返回token
       return {"access_token": 'xxxxxx', "token_type": "bearer"}