FastAPI
OAuth2.0授权模式
- 授权码授权模式
- 隐式授权模式
- 密码授权模式
- 客户端凭证授权模式
OAuth2 密码模式和 FastAPI 的 OAuth2PasswordBearer
import hashlib
from datetime import datetime, timedelta
from typing import Optional
from fastapi import APIRouter, status, HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from pydantic import BaseModel
from jose import JWTError, jwt
from mysql import MysqlHelper
from settings import mysql_config
mysql = MysqlHelper(mysql_config)
app06 = APIRouter()
"""OAuth2 密码模式和 FastAPI 的 OAuth2PasswordBearer"""
"""
OAuth2PasswordBearer是接收URL作为参数的一个类：客户端会向该URL发送username和password参数，然后得到一个Token值
OAuth2PasswordBearer并不会创建响应的URL路径操作，只是指明客户端用来请求Token的URL地址
当请求到来的时候，FastAPI会检查请求的Authorization头信息，如果没有找到Authorization头信息，或者头信息的内容不是Bearer token，它会返还401状态码(UNAUTHORIZED)
"""
oauth2_schema = OAuth2PasswordBearer(tokenUrl="/chapter06/token") # 请求Token的URL地址 http://127.0.0.1:8000/chapter06/token
@app06.get('/oauth2_password_bearer')
async def oauth2_password_bearer(token: str = Depends(oauth2_schema)):
return {"token": token}
"""基于 Password 和 Bearer token 的OAuth2 认证"""
fake_user_db = mysql.select_all("select username,password,position,company,phone,is_active from app01_user")
fake_user_db = {i['username']: i for i in fake_user_db}
# print(fake_user_db)
def fake_hash_password(password: str):
md5 = hashlib.md5()
md5.update(password.encode('utf-8'))
pwd = md5.hexdigest()
return pwd
class User(BaseModel):
username: str
position: Optional[str] = None
company: Optional[str] = None
phone: Optional[str] = None
is_active: Optional[bool] = None
class UserInDB(User):
password: str
@app06.post('/token')
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user_dic = fake_user_db.get(form_data.username)
print(user_dic)
if not user_dic:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="不正确的用户名或密码"
)
user = UserInDB(**user_dic)
hashed_password = fake_hash_password(form_data.password)
if not hashed_password == user.password:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="不正确的用户名或密码"
)
return {"access_token": user.username, "token_type": "bearer"}
def get_user(data, username: str):
if username in data:
user_dict = data[username]
print(user_dict)
return UserInDB(**user_dict)
def fake_decode_token(token: str):
user = get_user(fake_user_db, token)
return user
async def get_current_user(token: str = Depends(oauth2_schema)):
user = fake_decode_token(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="非法认证信息",
headers={"WWW-Authenticate": "Bearer"} # OAuth2 的规范，如果认证失败，请求头中返回"WWW-Authenticate"
)
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.is_active == 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="未激活用户"
)
return current_user
@app06.get('/users/me')
async def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
开发基于JSON Web Token的认证
"""OAuth2 with Password (and hashing), Bearer with JWT tokens 开发基于JSON Web Token的认证"""
fake_user_db.update({
"jhon snow": {
"username": "jhon snow",
"password": "123456",
"position": "xxx",
"company": '1',
"phone": '19912344321',
"is_active": 0,
}
})
# 生成密钥 openssl rand -hex 32 (linux 命令)
SECRET_KEY = "lem"
ALGORITHM = "HS256" # 加密算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30 # 访问令牌过期时间
class Token(BaseModel):
"""返回给用户的Token"""
access_token: str
token_type: str
pwd_context = CryptContext(schemes=['bcrypt'], deprecated="auto")
oauth2_schema = OAuth2PasswordBearer(tokenUrl="/chapter06/jwt/token")
def verity_password(plain_pwd: str, hash_pwd: str):
"""对密码进行校验"""
print(plain_pwd, hash_pwd)
return pwd_context.verify(plain_pwd, hash_pwd)
def jwt_get_user(db, username: str):
if username in db:
user_dic = db[username]
return UserInDB(**user_dic)
def jwt_authenticate_user(db, username: str, pwd: str):
user = jwt_get_user(db, username)
if not user:
return False
if not verity_password(plain_pwd=pwd, hash_pwd=user.password):
return False
return user
def create_access_token(data: dict, expire_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expire_delta:
expire = datetime.utcnow() + expire_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encode_jwt = jwt.encode(claims=to_encode, key=SECRET_KEY, algorithm=ALGORITHM)
return encode_jwt
@app06.post('/jwt/token', response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = jwt_authenticate_user(db=fake_user_db, username=form_data.username, pwd=form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名密码不正确",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username},
expire_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
async def jwt_get_current_user(token: str = Depends(oauth2_schema)):
"""解析jwt的字符串，将存在里面的user取出"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="证书校验失败",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token=token, key=SECRET_KEY, algorithms=ALGORITHM)
username = payload.get("sub")
if not username:
raise credentials_exception
except JWTError:
raise credentials_exception
user = jwt_get_user(db=fake_user_db, username=username)
if not user:
raise credentials_exception
return user
async def jwt_get_current_active_user(current_user: User = Depends(jwt_get_current_user)):
if current_user.is_active == 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="用户未激活"
)
return current_user
@app06.get('/jwt/users/me')
async def jwt_read_users_me(current_user: User = Depends(jwt_get_current_user)):
return current_user