60 秒系统安全认证实战-低调大师

60 秒系统安全认证实战

2021-04-18 678

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

读完需要 18分钟

速读仅需 6 分钟

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

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

1

准备工具

1.1

密码安全

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

pip install passlibpip install bcrypt

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

2

用户登陆流程

用户通过终端发送 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": "haishiniu@bbs.com",        "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"}

3

请求数据流

终端获取到 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

4

效果展示

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

# -*- 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": "haishiniu@bbs.com",        "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

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

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

5

总结

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

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

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

往期推荐

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

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

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

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

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

微信关注我们

原文链接：https://my.oschina.net/u/3699111/blog/5024126

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

QuickHttp 2.0.5 版本发布，一个 Java Http 客户端请求框架

本次更新内容如下: [新增]支持EventSource请求 [优化]解决http状态码错误时无法获取body输入流问题 [优化]修复日志功能请求头部无法显示完全问题。在debug级别下可显示http完整请求报文 [优化]依赖升级,fastjson升级至1.2.75版本 QuickHttp简介 QuickHttp是一个Java Http客户端框架,基于Java原生的HttpUrlConnection开发,设计了大量API方便开发者使用.QuickHttp的主要特点如下: 丰富的API设计 QuickHttp参考了Jsoup的API设计,同时添加了许多新的API.QuickHttp的Request接口定义了添加请求信息的方法,例如header,userAgent,requestBody等方法.Response接口定义获取返回结果的方法,例如body,bodyAsFile,bodyAsJSONObject等方法. Cookie管理,全局代理和日志记录 QuickHttp支持Cookie管理,自动提取Cookie信息和发送Cookie.支持设置全局代理.为了解决http请求调试困难问题,Qui...

2021-04-19

704

小诺cloud版本为小诺框架中微服务版本，创造之时秉承着能让所有用户在不装插件的情况下完全能跑起来使用微服务框架，之继承了eureka为服务的注册中心。项目采用SpringCloud Hoxton + SpringCloud Gateway + SpringBoot2+ MybatisPlus3为后端，AntDesignVue为前端，也是想的能玩微服务的用户应该会驾驭前后分离，同时在我们框架产品之中，微服务为技术含量最高的一款，后面的版本中会陆续接入Nacos、Sentinel、RocketMQ、Dubbo、Seata等多个套件。此次更新详细内容如下：【更新】代码生成转换字段加入enum、set类型，修改默认生成至本项目的路径【新增】新增人大金仓数据库备份文件及引用方式【更新】固定一个前端依赖版本【更新】更新角色授权菜单上下级关系【更新】去除无用的测试菜单【更新】banner替换为Snowy 【更新】数据库连接错误的信息打印详细化【更新】更新不同数据库配置案例，更新sql，去除druid监控密码，修复编辑空字段在不同数据库下的bug，pom默认放开其他jdbc的依赖 ...

2021-04-19

614

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。