FastAPI基础：7.安全认证-JWT

JWT(Json Web Token)

当下较多项目使用的一种认证方式，后端通过加密信息（用户名/角色）返回一串秘闻（实际可以解）；客户端通过请求头携带该秘闻，后端通过解密成功与否来确认是否是用户

密码加密

"""
密码加解密
pip install passlib bcrypt 
"""

from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """
    验证明文密码 vs hash密码
    :param plain_password: 明文密码
    :param hashed_password: hash密码
    :return:
    """
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """
    加密明文
    :param password: 明文密码
    :return:
    """
    return pwd_context.hash(password)

生成JWT

python-jose — python-jose 0.2.0 documentation

def generate_token(username: str, expires_delta: Optional[timedelta] = None):
    """
    生成token
    :param username: 用户名
    :param expires_delta: 有效时长
    :return: 
    """
    to_encode = {"sub": username}.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(
            minutes=1
        )
    to_encode.update(dict(exp=expire))
    encoded_jwt = jwt.encode(
        to_encode, "miyao", algorithm="HS256"
    )
    return encoded_jwt

解Token

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Depends, HTTPException

# HTTPBearer 实际从 Request header中取到了 Token
async def check_token(security: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
    """检查用户token"""
    token = security.credentials
    try:
        payload = jwt.decode(
            token, "miyao", algorithms=["HS256"]
        )
        username: str = payload.get("sub")
        if fake_db.get("username") == username:
            return fake_db
    except JWTError:
        raise HTTPException(status_code=401, detail="token g")

ps：前端请求头

ps：最终credentials 就是 token串，其实完全可以自己使用中间件或者依赖注入通过Request对象自己实现，因为想要接口文档上的验证按钮所以这么做

def check(request: Request):
    """
    自己检查token
    :param request:
    :return:
    """
    token = request.headers.get("x-token")
    try:
        payload = jwt.decode(
            token, "miyao", algorithms=["HS256"]
        )
        username: str = payload.get("sub")
        if fake_db.get("username") == username:
            return fake_db
    except JWTError:
        raise HTTPException(status_code=401, detail="token g")

登录认证

# 需要登录
@app.get("/info")
async def info(user=Depends(check_token)):
    return user

# 需要登录
@app.get("/user", dependencies=[Depends(check_token)])
async def infos():
    return 'G'

接口文档