feat: add security module (encryption + auth)
- Implement AES-256-GCM encryption for sensitive data - Implement decryption function - Implement Bearer token authentication verification - Add comprehensive tests for encryption/decryption roundtrip - Add tests for API token verification (success and failure cases) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
78
backend/app/security.py
Normal file
78
backend/app/security.py
Normal file
@@ -0,0 +1,78 @@
|
||||
import base64
|
||||
import os
|
||||
from typing import Optional
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
|
||||
|
||||
def encrypt_data(data: bytes, key_b64: str) -> bytes:
|
||||
"""
|
||||
使用 AES-256-GCM 加密数据.
|
||||
|
||||
Args:
|
||||
data: 原始数据
|
||||
key_b64: Base64 编码的加密密钥
|
||||
|
||||
Returns:
|
||||
加密后的数据 (nonce + ciphertext + tag)
|
||||
"""
|
||||
key = base64.b64decode(key_b64)
|
||||
nonce = os.urandom(12) # GCM 96-bit nonce
|
||||
|
||||
cipher = Cipher(
|
||||
algorithms.AES(key),
|
||||
modes.GCM(nonce),
|
||||
backend=default_backend()
|
||||
)
|
||||
encryptor = cipher.encryptor()
|
||||
ciphertext = encryptor.update(data) + encryptor.finalize()
|
||||
|
||||
# 返回 nonce + ciphertext + tag
|
||||
return nonce + ciphertext + encryptor.tag
|
||||
|
||||
|
||||
def decrypt_data(encrypted_data: bytes, key_b64: str) -> bytes:
|
||||
"""
|
||||
解密使用 encrypt_data 加密的数据.
|
||||
|
||||
Args:
|
||||
encrypted_data: 加密数据 (nonce + ciphertext + tag)
|
||||
key_b64: Base64 编码的加密密钥
|
||||
|
||||
Returns:
|
||||
解密后的原始数据
|
||||
"""
|
||||
key = base64.b64decode(key_b64)
|
||||
nonce = encrypted_data[:12]
|
||||
tag = encrypted_data[-16:]
|
||||
ciphertext = encrypted_data[12:-16]
|
||||
|
||||
cipher = Cipher(
|
||||
algorithms.AES(key),
|
||||
modes.GCM(nonce, tag),
|
||||
backend=default_backend()
|
||||
)
|
||||
decryptor = cipher.decryptor()
|
||||
return decryptor.update(ciphertext) + decryptor.finalize()
|
||||
|
||||
|
||||
def verify_api_token(authorization: Optional[str]) -> bool:
|
||||
"""
|
||||
验证 Bearer Token.
|
||||
|
||||
Args:
|
||||
authorization: Authorization 头部值
|
||||
|
||||
Returns:
|
||||
Token 是否有效
|
||||
"""
|
||||
if not authorization:
|
||||
return False
|
||||
|
||||
if not authorization.startswith('Bearer '):
|
||||
return False
|
||||
|
||||
from app.config import get_settings
|
||||
settings = get_settings()
|
||||
token = authorization[7:] # 移除 'Bearer ' 前缀
|
||||
return token == settings.api_token
|
||||
34
backend/tests/test_security.py
Normal file
34
backend/tests/test_security.py
Normal file
@@ -0,0 +1,34 @@
|
||||
import base64
|
||||
import pytest
|
||||
|
||||
|
||||
def test_encrypt_decrypt_roundtrip():
|
||||
"""测试加密解密往返."""
|
||||
from app.security import encrypt_data, decrypt_data
|
||||
|
||||
original = b"sensitive-secret-data"
|
||||
key = base64.b64encode(b'12345678901234567890123456789012').decode()
|
||||
|
||||
encrypted = encrypt_data(original, key)
|
||||
assert encrypted != original
|
||||
assert isinstance(encrypted, bytes)
|
||||
assert len(encrypted) > len(original) # nonce + ciphertext + tag
|
||||
|
||||
decrypted = decrypt_data(encrypted, key)
|
||||
assert decrypted == original
|
||||
|
||||
|
||||
def test_verify_api_token_success(test_env_vars):
|
||||
"""测试 API token 验证成功."""
|
||||
from app.security import verify_api_token
|
||||
|
||||
assert verify_api_token("Bearer test-token") is True
|
||||
|
||||
|
||||
def test_verify_api_token_failure():
|
||||
"""测试 API token 验证失败."""
|
||||
from app.security import verify_api_token
|
||||
|
||||
assert verify_api_token("Bearer wrong-token") is False
|
||||
assert verify_api_token(None) is False
|
||||
assert verify_api_token("Basic token") is False
|
||||
Reference in New Issue
Block a user