60 秒系统安全认证实战 | python编程从入门到实践

点击python编程从入门到实践，置顶公众号重磅 python入门资料，第一时间送达

60 秒系统安全认证实战-LMLPHP

读完需要 18分钟

速读仅需 6 分钟

/ python 生产实战 60 秒系统安全认证实战 /

上节主要讲解了目前主流的认证规范/协议以及对 JWT 进行了深入的研究和分析并在最后给出了在生产环境中如何去生成一个有效的 Token，基于 Python 语言那在生产环境中是如何进行有效的安全认证的呢？上节我们也基于 JWT 的 Token 的认证过程进行了登陆认证、请求认证的理论分析以及用图示的方式给出了数据的流向，本节我们再带大家从代码层面走一次流程，一方面加深大家对上节理论部分的理解，另一方面也是给大家在做工程的过程中提供一套"模版"快速应用在项目中

准备工具

1.1

密码安全

为了数据安全，我们利用 PassLib 对入库的用户密码进行加密处理，推荐的加密算法是"Bcrypt"。我们需要安装依赖包：

pip install passlibpip install bcrypt

简单的介绍一下这两个库:
1.passlib 是 python2&3 的密码散列库，它提供超过 30 种密码散列算法的跨平台实现，以及作为管理现有密码哈希的框架。它被设计成有用的对于范围广泛的任务，从验证/etc/shadow 中找到的散列到为多用户应用程序提供全强度密码哈希。2.bcrypt 模块是一个用于在 Python 中生成强哈希值的库。
注意:若对于以上两个库的详细内容比较感兴趣的小伙伴可以自行到 python 官网查看相关资料，我们本节的核心是做整个流程的实现这个具体的库内容就先不做详细介绍了。

用户登陆流程

用户通过终端发送 username 和 password 到后端。后端收到数据后，进行一下操作:1.用户信息校验：查询当前系统是否存在该用户，以及密码是否正确
2.如果用户存在，则生成 JWT token 并返回，JWT payload 中可以携带自定义数据

# -*- encoding: utf-8 -*-from datetime import datetime, timedeltafrom typing import Optional
from fastapi import Depends, FastAPI, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormimport jwtfrom pydantic import BaseModelfrom passlib.context import CryptContext

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

# 模拟数据库数据fake_users_db = {    "hiashiniu": {        "username": "hiashiniu",        "full_name": "HaiShiNiu",        "email": "[email protected]",        "hashed_password": "$2b$12$uG4n5CK4.kCid/MfQ3ns9.uRslWuWr9EL5FR4HYNg6SNo6wEn8OPC",        "disabled": False,    }}

class Token(BaseModel):    access_token: str    token_type: str

class User(BaseModel):    username: str    email: Optional[str] = None    full_name: Optional[str] = None    disabled: Optional[bool] = None

class UserInDB(User):    hashed_password: str

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
app = FastAPI()

# 校验密码def verify_password(plain_password, hashed_password):    return pwd_context.verify(plain_password, hashed_password)

# 密码哈希def get_password_hash(password):    # pwd_context.hash('123456789')    # '$2b$12$uG4n5CK4.kCid/MfQ3ns9.uRslWuWr9EL5FR4HYNg6SNo6wEn8OPC'    return pwd_context.hash(password)

# 模拟从数据库读取用户信息def get_user(db, username: str):    user_dict = db.get(username,None)    if user_dict:        return UserInDB(**user_dict)

# 用户信息校验：username和password分别校验def authenticate_user(fake_db, username: str, password: str):    user = get_user(fake_db, username)    print(user)    if not user:        return False    if not verify_password(password, user.hashed_password):        return False    return user

# 生成token，带有过期时间def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):    to_encode = data.copy()    # 有一个实效性 默认的时间25分钟    if expires_delta:        expire = datetime.utcnow() + expires_delta    else:        expire = datetime.utcnow() + timedelta(minutes=15)    to_encode.update({"exp": expire})    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)    return encoded_jwt
@app.post("/token", response_model=Token)async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):    # 首先校验用户信息    print(form_data.password)    print(form_data.username)    user = authenticate_user(fake_users_db, form_data.username, form_data.password)    if not user:        raise HTTPException(            status_code=status.HTTP_401_UNAUTHORIZED,            detail="Incorrect username or password",            headers={"WWW-Authenticate": "Bearer"},        )
    # 生成并返回token信息    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)    access_token = create_access_token(        data={"sub": user.username}, expires_delta=access_token_expires    )    return {"access_token": access_token, "token_type": "bearer"}

请求数据流

终端获取到 token 信息后，必须在后续请求的 Authorization 头信息中带有 Bearer token，才能被允许访问。我们添加一个校验函数，对请求的合法性进行校验，读取 token 内容解析并进行验证。

# -*- encoding: utf-8 -*-
async def get_current_user(token: str = Depends(oauth2_scheme)):    credentials_exception = HTTPException(        status_code=status.HTTP_401_UNAUTHORIZED,        detail="Could not validate credentials",        headers={"WWW-Authenticate": "Bearer"},    )    try:        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])        username: str = payload.get("sub")        if username is None:            raise credentials_exception    except jwt.PyJWTError as ex:        print(ex)        raise credentials_exception    user = get_user(fake_users_db, username=username)    if user is None:        raise credentials_exception    return user

@app.get("/users/me/", response_model=User)async def read_users_me(current_user: User = Depends(get_current_user)):    return current_user

效果展示

本小节我们整合上述代码进行一下效果展示

# -*- encoding: utf-8 -*-from datetime import datetime, timedeltafrom typing import Optional
from fastapi import Depends, FastAPI, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormimport jwtfrom pydantic import BaseModelfrom passlib.context import CryptContext

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

# 模拟数据库数据fake_users_db = {    "hiashiniu": {        "username": "hiashiniu",        "full_name": "HaiShiNiu",        "email": "[email protected]",        "hashed_password": "$2b$12$uG4n5CK4.kCid/MfQ3ns9.uRslWuWr9EL5FR4HYNg6SNo6wEn8OPC",        "disabled": False,    }}

class Token(BaseModel):    access_token: str    token_type: str

class User(BaseModel):    username: str    email: Optional[str] = None    full_name: Optional[str] = None    disabled: Optional[bool] = None

class UserInDB(User):    hashed_password: str

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
app = FastAPI()

# 校验密码def verify_password(plain_password, hashed_password):    return pwd_context.verify(plain_password, hashed_password)

# 密码哈希def get_password_hash(password):    # pwd_context.hash('123456789')    # '$2b$12$uG4n5CK4.kCid/MfQ3ns9.uRslWuWr9EL5FR4HYNg6SNo6wEn8OPC'    return pwd_context.hash(password)

# 模拟从数据库读取用户信息def get_user(db, username: str):    user_dict = db.get(username,None)    if user_dict:        return UserInDB(**user_dict)

# 用户信息校验：username和password分别校验def authenticate_user(fake_db, username: str, password: str):    user = get_user(fake_db, username)    if not user:        return False    if not verify_password(password, user.hashed_password):        return False    return user

# 生成token，带有过期时间def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):    to_encode = data.copy()    # 有一个实效性 默认的时间25分钟    if expires_delta:        expire = datetime.utcnow() + expires_delta    else:        expire = datetime.utcnow() + timedelta(minutes=15)    to_encode.update({"exp": expire})    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)    return encoded_jwt

@app.post("/token", response_model=Token)async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):    # 首先校验用户信息    print(form_data.password)    print(form_data.username)    user = authenticate_user(fake_users_db, form_data.username, form_data.password)    if not user:        raise HTTPException(            status_code=status.HTTP_401_UNAUTHORIZED,            detail="Incorrect username or password",            headers={"WWW-Authenticate": "Bearer"},        )
    # 生成并返回token信息    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)    access_token = create_access_token(        data={"sub": user.username}, expires_delta=access_token_expires    )    return {"access_token": access_token, "token_type": "bearer"}

async def get_current_user(token: str = Depends(oauth2_scheme)):    credentials_exception = HTTPException(        status_code=status.HTTP_401_UNAUTHORIZED,        detail="Could not validate credentials",        headers={"WWW-Authenticate": "Bearer"},    )    try:        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])        username: str = payload.get("sub")        if username is None:            raise credentials_exception    except jwt.PyJWTError as ex:        print(ex)        raise credentials_exception    user = get_user(fake_users_db, username=username)    if user is None:        raise credentials_exception    return user

@app.get("/users/me/", response_model=User)async def read_users_me(current_user: User = Depends(get_current_user)):    return current_user

1.使用命令启动项目:

uvicorn main:app --reload

60 秒系统安全认证实战-LMLPHP

2.使用 Postman 模拟登陆获取验证 Token

60 秒系统安全认证实战-LMLPHP

3.使用 Postman 模拟携带 Token 获取正常逻辑信息

60 秒系统安全认证实战-LMLPHP

总结

本节核心:从代码层面实战了登陆认证、请求认证的数据流转

原创不易，只愿能帮助那些需要这些内容的同行或刚入行的小伙伴，你的每次点赞、分享都是我继续创作下去的动力，我希望能在推广 python 技术的道路上尽我一份力量，欢迎在评论区向我提问，我都会一一解答，记得一键三连支持一下哦！

加入python学习交流微信群，请后台回复「入群」

往期推荐

python生产实战 python 闭包之庖丁解牛篇

大型fastapi项目实战靠 python 中间件解决方案涨薪了

大型fastapi项目实战高并发请求神器之aiohttp(下)

大型fastapi项目实战高并发请求神器之aiohttp(上) [建议收藏]

本文分享自微信公众号 - python编程军火库（PythonCoder1024）。
如有侵权，请联系 [email protected] 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。