mirror of
https://git.oceanpay.cc/danial/kami_itunes_third_api.git
synced 2025-12-18 21:19:19 +00:00
584a258c50
- 实现RSA密钥对的生成、导出、加载功能 - 新增RSA公钥加密与私钥解密接口，支持Base64和Hex编码 - 实现RSA私钥数字签名及公钥签名验证功能 - 添加RSA密钥对文件读写辅助函数 - 新增非对称加密使用示例，涵盖密钥生成、加解密、签名验证等场景 - 提供项目中集成非对称加密的应用示例，包括安全API请求、数据加密、证书签发与认证令牌 - 添加完善的单元测试覆盖密钥管理、加密解密、签名验证及边界情况 - 调整依赖版本，替换pycryptodome为cryptography包 - 优化部分已有加密代码，改用cryptography库实现AES加密，增强兼容性和安全性
349 lines
9.9 KiB
Python
349 lines
9.9 KiB
Python
"""
|
||
非对称加密辅助函数在项目中的集成示例
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime
|
||
from typing import Optional, Dict, Any
|
||
|
||
from src.utils.crypto import (
|
||
RSAKeyPair,
|
||
encrypt_with_rsa,
|
||
decrypt_with_rsa,
|
||
sign_with_rsa,
|
||
verify_rsa_signature,
|
||
)
|
||
|
||
|
||
class SecureAPIRequest:
|
||
"""
|
||
安全的 API 请求签名示例
|
||
|
||
用于与第三方 API 进行安全交互，确保请求的真实性和完整性
|
||
"""
|
||
|
||
def __init__(self, private_key_pem: str, app_id: str):
|
||
"""
|
||
初始化安全 API 请求
|
||
|
||
Args:
|
||
private_key_pem: 应用的 RSA 私钥（PEM 格式）
|
||
app_id: 应用 ID
|
||
"""
|
||
self.private_key_pem = private_key_pem
|
||
self.app_id = app_id
|
||
|
||
def create_signed_request(
|
||
self, endpoint: str, data: Dict[str, Any]
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
创建带签名的 API 请求
|
||
|
||
Args:
|
||
endpoint: API 端点路径
|
||
data: 请求数据
|
||
|
||
Returns:
|
||
Dict: 包含签名的完整请求体
|
||
"""
|
||
timestamp = str(int(datetime.now().timestamp()))
|
||
|
||
# 构建请求内容
|
||
request_content = json.dumps(
|
||
{
|
||
"app_id": self.app_id,
|
||
"endpoint": endpoint,
|
||
"timestamp": timestamp,
|
||
"data": data,
|
||
},
|
||
separators=(",", ":"),
|
||
)
|
||
|
||
# 对请求内容签名
|
||
signature = sign_with_rsa(request_content, self.private_key_pem)
|
||
|
||
return {
|
||
"request": request_content,
|
||
"signature": signature,
|
||
"app_id": self.app_id,
|
||
}
|
||
|
||
@staticmethod
|
||
def verify_request(request_json: str, signature: str, public_key_pem: str) -> bool:
|
||
"""
|
||
验证 API 请求签名
|
||
|
||
Args:
|
||
request_json: 请求内容（JSON 字符串）
|
||
signature: 请求签名
|
||
public_key_pem: 应用的 RSA 公钥（PEM 格式）
|
||
|
||
Returns:
|
||
bool: 签名是否有效
|
||
"""
|
||
return verify_rsa_signature(request_json, signature, public_key_pem)
|
||
|
||
|
||
class SecureDataEncryption:
|
||
"""
|
||
敏感数据加密示例
|
||
|
||
用于在存储或传输敏感数据时进行加密
|
||
"""
|
||
|
||
def __init__(self, public_key_pem: str, private_key_pem: str):
|
||
"""
|
||
初始化数据加密工具
|
||
|
||
Args:
|
||
public_key_pem: 公钥（用于加密）
|
||
private_key_pem: 私钥（用于解密）
|
||
"""
|
||
self.public_key_pem = public_key_pem
|
||
self.private_key_pem = private_key_pem
|
||
|
||
def encrypt_user_data(self, user_id: str, sensitive_data: str) -> Dict[str, str]:
|
||
"""
|
||
加密用户敏感数据
|
||
|
||
Args:
|
||
user_id: 用户 ID
|
||
sensitive_data: 敏感数据（如密码、令牌等）
|
||
|
||
Returns:
|
||
Dict: 包含加密数据和元数据的字典
|
||
"""
|
||
# 为了避免超过 RSA 加密限制，先对数据进行摘要处理
|
||
# 如果数据较大，应使用混合加密方式
|
||
encrypted = encrypt_with_rsa(sensitive_data, self.public_key_pem)
|
||
|
||
return {
|
||
"user_id": user_id,
|
||
"encrypted_data": encrypted,
|
||
"timestamp": datetime.now().isoformat(),
|
||
"algorithm": "RSA-OAEP",
|
||
}
|
||
|
||
def decrypt_user_data(self, encrypted_data: str) -> str:
|
||
"""
|
||
解密用户敏感数据
|
||
|
||
Args:
|
||
encrypted_data: 加密的数据
|
||
|
||
Returns:
|
||
str: 解密后的原始数据
|
||
"""
|
||
return decrypt_with_rsa(encrypted_data, self.private_key_pem)
|
||
|
||
|
||
class CertificateAuthority:
|
||
"""
|
||
简单的证书颁发机构实现
|
||
|
||
用于证书的签发和验证
|
||
"""
|
||
|
||
def __init__(self, ca_private_key_pem: str, ca_public_key_pem: str):
|
||
"""
|
||
初始化 CA
|
||
|
||
Args:
|
||
ca_private_key_pem: CA 的私钥
|
||
ca_public_key_pem: CA 的公钥
|
||
"""
|
||
self.ca_private_key_pem = ca_private_key_pem
|
||
self.ca_public_key_pem = ca_public_key_pem
|
||
|
||
def issue_certificate(
|
||
self, subject: str, subject_public_key_pem: str, validity_days: int = 365
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
颁发证书
|
||
|
||
Args:
|
||
subject: 证书主体（如用户 ID、服务器名称等）
|
||
subject_public_key_pem: 主体的公钥
|
||
validity_days: 有效期（天数）
|
||
|
||
Returns:
|
||
Dict: 颁发的证书
|
||
"""
|
||
now = datetime.now().isoformat()
|
||
valid_until = datetime.fromtimestamp(
|
||
datetime.now().timestamp() + validity_days * 86400
|
||
).isoformat()
|
||
|
||
cert_data = json.dumps(
|
||
{
|
||
"subject": subject,
|
||
"public_key": subject_public_key_pem,
|
||
"issued_at": now,
|
||
"valid_until": valid_until,
|
||
"issuer": "CA",
|
||
}
|
||
)
|
||
|
||
# CA 对证书进行签名
|
||
signature = sign_with_rsa(cert_data, self.ca_private_key_pem)
|
||
|
||
return {
|
||
"certificate": cert_data,
|
||
"signature": signature,
|
||
"issued_at": now,
|
||
}
|
||
|
||
def verify_certificate(self, cert_data: str, signature: str) -> bool:
|
||
"""
|
||
验证证书签名
|
||
|
||
Args:
|
||
cert_data: 证书数据
|
||
signature: 证书签名
|
||
|
||
Returns:
|
||
bool: 证书是否有效
|
||
"""
|
||
return verify_rsa_signature(cert_data, signature, self.ca_public_key_pem)
|
||
|
||
|
||
class AuthenticationToken:
|
||
"""
|
||
认证令牌签名示例
|
||
|
||
用于生成和验证认证令牌
|
||
"""
|
||
|
||
def __init__(self, private_key_pem: str, public_key_pem: str, issuer: str):
|
||
"""
|
||
初始化令牌生成器
|
||
|
||
Args:
|
||
private_key_pem: 私钥（用于签名）
|
||
public_key_pem: 公钥（用于验证）
|
||
issuer: 令牌颁发者
|
||
"""
|
||
self.private_key_pem = private_key_pem
|
||
self.public_key_pem = public_key_pem
|
||
self.issuer = issuer
|
||
|
||
def generate_token(
|
||
self, user_id: str, permissions: list, expires_in_hours: int = 24
|
||
) -> str:
|
||
"""
|
||
生成签名的认证令牌
|
||
|
||
Args:
|
||
user_id: 用户 ID
|
||
permissions: 权限列表
|
||
expires_in_hours: 过期时间（小时）
|
||
|
||
Returns:
|
||
str: 签名的令牌
|
||
"""
|
||
issued_at = datetime.now().isoformat()
|
||
expires_at = datetime.fromtimestamp(
|
||
datetime.now().timestamp() + expires_in_hours * 3600
|
||
).isoformat()
|
||
|
||
token_data = json.dumps(
|
||
{
|
||
"user_id": user_id,
|
||
"permissions": permissions,
|
||
"issuer": self.issuer,
|
||
"issued_at": issued_at,
|
||
"expires_at": expires_at,
|
||
}
|
||
)
|
||
|
||
# 对令牌签名
|
||
signature = sign_with_rsa(token_data, self.private_key_pem)
|
||
|
||
# 将令牌和签名组合（使用分隔符）
|
||
return f"{token_data}.{signature}"
|
||
|
||
def verify_token(self, token: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
验证令牌并返回令牌数据
|
||
|
||
Args:
|
||
token: 签名的令牌字符串
|
||
|
||
Returns:
|
||
Optional[Dict]: 令牌数据（如果有效）或 None（如果无效）
|
||
"""
|
||
try:
|
||
token_data, signature = token.rsplit(".", 1)
|
||
|
||
# 验证签名
|
||
if not verify_rsa_signature(token_data, signature, self.public_key_pem):
|
||
return None
|
||
|
||
# 解析令牌数据
|
||
token_dict = json.loads(token_data)
|
||
|
||
# 验证过期时间
|
||
expires_at = datetime.fromisoformat(token_dict["expires_at"])
|
||
if datetime.now() > expires_at:
|
||
return None
|
||
|
||
return token_dict
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
# 使用示例
|
||
if __name__ == "__main__":
|
||
# 生成密钥对
|
||
print("=== 生成密钥对 ===")
|
||
key_pair = RSAKeyPair.generate(key_size=2048)
|
||
private_pem = key_pair.get_private_key_pem()
|
||
public_pem = key_pair.get_public_key_pem()
|
||
|
||
# 示例 1: 安全 API 请求
|
||
print("\n=== 安全 API 请求 ===")
|
||
api_request = SecureAPIRequest(private_pem, app_id="app_001")
|
||
signed_request = api_request.create_signed_request(
|
||
endpoint="/api/users/login",
|
||
data={"username": "user@example.com", "password": "secret"},
|
||
)
|
||
print(f"签名请求: {signed_request['signature'][:50]}...")
|
||
is_valid = SecureAPIRequest.verify_request(
|
||
signed_request["request"], signed_request["signature"], public_pem
|
||
)
|
||
print(f"请求签名有效: {is_valid}")
|
||
|
||
# 示例 2: 数据加密
|
||
print("\n=== 数据加密 ===")
|
||
data_encryption = SecureDataEncryption(public_pem, private_pem)
|
||
encrypted = data_encryption.encrypt_user_data(
|
||
user_id="user_123", sensitive_data="my_secret_token_12345"
|
||
)
|
||
print(f"加密数据: {encrypted['encrypted_data'][:50]}...")
|
||
decrypted = data_encryption.decrypt_user_data(encrypted["encrypted_data"])
|
||
print(f"解密后: {decrypted}")
|
||
|
||
# 示例 3: 证书签发
|
||
print("\n=== 证书颁发 ===")
|
||
ca = CertificateAuthority(private_pem, public_pem)
|
||
certificate = ca.issue_certificate(
|
||
subject="server.example.com", subject_public_key_pem=public_pem
|
||
)
|
||
print(f"证书签名: {certificate['signature'][:50]}...")
|
||
is_valid = ca.verify_certificate(
|
||
certificate["certificate"], certificate["signature"]
|
||
)
|
||
print(f"证书有效: {is_valid}")
|
||
|
||
# 示例 4: 认证令牌
|
||
print("\n=== 认证令牌 ===")
|
||
auth_token = AuthenticationToken(private_pem, public_pem, issuer="auth_server")
|
||
token = auth_token.generate_token(
|
||
user_id="user_456", permissions=["read", "write"], expires_in_hours=24
|
||
)
|
||
print(f"生成的令牌: {token[:50]}...")
|
||
token_data = auth_token.verify_token(token)
|
||
print(f"令牌数据: {token_data}")
|
||
|
||
print("\n✓ 所有集成示例执行完成！")
|