一文把握异步web框架FastAPI（六）-- 安全（HTTP验证、Bearer Token、Sessi ...

目录

  
九、安全
1、HTTP 基自己份验证（Basic Authentication）
2、Bearer Token
3、Session 管理
1）简单实现
2）当需要session验证的接口会有多个时，使用中间件来做验证。
3）当有大量接口，此中一些需要登录而另一些不需要时，可以使用FastAPI的依赖注入系统来管理这些接口的访问控制。
4、OAuth2 和 OpenID Connect在fastapi中的应用
1）OAuth2
2）OpenID Connect
5、HTTPS 和 TLS 加密
1）获取 SSL/TLS 证书
自署名证书生成（仅用于开辟测试）
2）设置 FastAPI 使用 HTTPS
3）设置 TLS 加密
4）生产环境摆设发起
6、Rate Limiting (速率限定)

---

九、安全

1、HTTP 基自己份验证（Basic Authentication）

这是最简单的身份验证形式之一，它要求客户端发送带有用户名和暗码的 HTTP 头部。FastAPI 可以很容易地集成这种身份验证方式。使用 hashlib 对暗码进行哈希处理处罚以提高安全性。

1. from fastapi import FastAPI, Depends, HTTPException, status
2. from fastapi.security import HTTPBasic, HTTPBasicCredentials
3. from hashlib import sha256
5. app = FastAPI()
6. security = HTTPBasic()
7. # 假设我们存储的是密码的哈希值
8. STORED_USERNAME = "john"
9. STORED_PASSWORD_HASH = sha256("secret".encode()).hexdigest()
12. def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
13.     """获取当前用户名，并验证提供的凭据是否正确。"""
14.     correct_username = credentials.username == STORED_USERNAME
15.     password_hash = sha256(credentials.password.encode()).hexdigest()
16.     correct_password = password_hash == STORED_PASSWORD_HASH
17.     if not (correct_username and correct_password):
18.         raise HTTPException(
19.             status_code=status.HTTP_401_UNAUTHORIZED,
20.             detail="Incorrect user or password",
21.             headers={"WWW-Authenticate": "Basic"},
22.         )
23.     return credentials.username
26. @app.get("/users/me")
27. def read_user_me(current_username: str = Depends(get_current_username)):
28.     """获取当前用户的用户名。"""
29.     return {"username": current_username}

复制代码

请求：

1. import requests
2. from requests.auth import HTTPBasicAuth
4. # 定义服务器地址
5. BASE_URL = "http://127.0.0.1:8000"
8. def test_get_user_me():
9.     # 正确的凭据
10.     correct_credentials = HTTPBasicAuth('john', 'secret')
11.     response = requests.get(f"{BASE_URL}/users/me", auth=correct_credentials)
12.     print("Response with correct credentials:")
13.     print(response.status_code, response.json())
14.     # 错误的凭据
15.     incorrect_credentials = HTTPBasicAuth('john', 'wrongpassword')
16.     response = requests.get(f"{BASE_URL}/users/me", auth=incorrect_credentials)
17.     print("Response with incorrect credentials:")
18.     print(response.status_code, response.text)
21. if __name__ == "__main__":
22.     test_get_user_me()

复制代码

2、Bearer Token

Bearer token 身份验证是另一种常见的身份验证方法，通常与 OAuth 2.0 团结使用。FastAPI 提供了内置的支持来处理处罚 bearer tokens。

1. from fastapi import FastAPI, Depends, HTTPException, status
2. from fastapi.security import OAuth2PasswordBearer
4. app = FastAPI()
5. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
8. # 假设我们有一个函数来验证 token 是否有效
9. def fake_decode_token(token):
10.     """模拟解码令牌的函数，检查令牌是否有效并返回用户信息"""
11.     if token == "validtoken":
12.         return {"username": "john", "scope": "admin"}
13.     else:
14.         return None
17. async def get_current_user(token: str = Depends(oauth2_scheme)):
18.     """获取当前用户，如果令牌无效则抛出未授权异常"""
19.     user = fake_decode_token(token)
20.     if user is None:
21.         raise HTTPException(
22.             status_code=status.HTTP_401_UNAUTHORIZED,
23.             detail="Invalid token",
24.             headers={"WWW-Authenticate": "Bearer"},
25.         )
26.     return user
29. @app.get("/users/me")
30. def read_user_me(current_user=Depends(get_current_user)):
31.     """读取当前用户的信息"""
32.     return current_user

复制代码

请求：

1. import requests
3. # 定义服务器地址
4. BASE_URL = "http://127.0.0.1:8000"
7. def test_get_user_me():
8.     # 正确的令牌
9.     valid_token = "validtoken"
10.     headers = {"Authorization": f"Bearer {valid_token}"}
11.     response = requests.get(f"{BASE_URL}/users/me", headers=headers)
12.     print("Response with valid token:")
13.     print(response.status_code, response.json())
14.     # 错误的令牌
15.     invalid_token = "invalidtoken"
16.     headers = {"Authorization": f"Bearer {invalid_token}"}
17.     response = requests.get(f"{BASE_URL}/users/me", headers=headers)
18.     print("Response with invalid token:")
19.     print(response.status_code, response.text)
22. if __name__ == "__main__":
23.     test_get_user_me()

复制代码

3、Session 管理

1）简单实现

1. from fastapi import FastAPI, Request, HTTPException, status
2. from fastapi.responses import JSONResponse
3. from itsdangerous import URLSafeTimedSerializer
5. import hashlib
6. import os
9. def hash_password(password: str) -> str:
10.     # 生成随机盐
11.     salt = os.urandom(16)
12.     # 创建新的哈希对象
13.     hasher = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt, 100000)
14.     # 返回盐和哈希值的组合
15.     return salt.hex() + hasher.hex()
18. def verify_password(stored_password: str, provided_password: str) -> bool:
19.     # 从存储的密码中提取盐
20.     salt = bytes.fromhex(stored_password[:32])
21.     # 提取哈希值
22.     stored_hash = stored_password[32:]
23.     # 重新计算哈希值
24.     hasher = hashlib.pbkdf2_hmac('sha256', provided_password.encode('utf-8'), salt, 100000)
25.     # 比较哈希值
26.     return hasher.hex() == stored_hash
29. app = FastAPI()
32. serializer = URLSafeTimedSerializer("SECRET_KEY")
33. # 假设的用户数据库
34. fake_users_db = {
35.     "alice": {
36.         "username": "alice",
37.         "hashed_password": hash_password("secret_password")
38.     }
39. }
42. def authenticate_user(fake_db, username: str, password: str):
43.     user = fake_db.get(username)
44.     if not user or not verify_password(user["hashed_password"], password):
45.         return False
46.     return user
49. @app.post("/login")
50. async def login(request: Request):
51.     data = await request.json()
52.     username = data.get('username')
53.     password = data.get('password')
54.     user = authenticate_user(fake_users_db, username, password)
55.     if not user:
56.         raise HTTPException(
57.             status_code=status.HTTP_401_UNAUTHORIZED,
58.             detail="Incorrect username or password",
59.             headers={"WWW-Authenticate": "Bearer"},
60.         )
61.     session_token = serializer.dumps(username)
62.     response = JSONResponse({"message": "Login successful"})
63.     response.set_cookie(key="session_token", value=sessi

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。