""" 公告接口（客户端使用）- 加密版本 """ import hmac import hashlib import base64 import json import secrets from typing import Optional, Dict, Any from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from sqlalchemy import or_ from pydantic import BaseModel from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding from app.database import get_db from app.models.announcement import Announcement from app.core.encrypt import encryptor from app.utils.timestamp_validator import TimestampValidator from app.config import settings from app.core.logger import logger router = APIRouter() class AnnouncementRequest(BaseModel): """公告获取请求""" encrypted_params: str # Base64编码的AES加密参数 aes_key: str # Base64编码的AES密钥（RSA加密后） key_version: int # 公钥版本 timestamp: int # 请求时间戳 nonce: str # 随机字符串，防止重放攻击 signature: str # HMAC签名 def _generate_signature(data: str, secret_key: str) -> str: """生成HMAC签名""" signature = hmac.new( secret_key.encode(), data.encode(), hashlib.sha256 ).digest() return base64.b64encode(signature).decode() def _verify_signature(data: str, signature: str, secret_key: str) -> bool: """验证HMAC签名""" try: expected_signature = _generate_signature(data, secret_key) return signature == expected_signature except Exception as e: logger.exception("签名验证异常") return False @router.post("") async def get_announcement( request: AnnouncementRequest, db: Session = Depends(get_db), ): """ 获取公告列表（支持检索与分页）- 加密版本 安全特性: - AES加密传输请求参数 - RSA加密AES密钥传输 - 时间戳验证防止重放攻击 - HMAC签名确保数据完整性 """ try: # 1. 验证时间戳 is_valid, error_msg = TimestampValidator.validate_timestamp(request.timestamp, db) if not is_valid: logger.warning(f"公告获取请求时间戳验证失败: {error_msg}") return { "code": 400, "msg": f"时间戳验证失败: {error_msg}", "timestamp": TimestampValidator.generate_timestamp() } # 2. 验证签名 signature_data = f"{request.timestamp}:{request.nonce}:{request.encrypted_params}:{request.aes_key}:{request.key_version}" if not _verify_signature(signature_data, request.signature, settings.SECRET_KEY): logger.warning("公告获取请求签名验证失败") return { "code": 400, "msg": "签名验证失败", "timestamp": TimestampValidator.generate_timestamp() } # 3. 使用RSA私钥解密AES密钥 try: # 确保使用正确的密钥版本 if not encryptor.load_key_by_version(db, request.key_version): logger.error(f"无法加载密钥版本 {request.key_version}") return { "code": 400, "msg": "无效的密钥版本", "timestamp": TimestampValidator.generate_timestamp() } # 解密AES密钥 encrypted_aes_key_bytes = base64.b64decode(request.aes_key) aes_key_bytes = encryptor.private_key.decrypt( encrypted_aes_key_bytes, padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None ) ) except Exception as e: logger.exception("AES密钥解密失败") return { "code": 400, "msg": "密钥解密失败", "timestamp": TimestampValidator.generate_timestamp() } # 4. 使用AES解密请求参数 (CTR模式，与前端保持一致) try: encrypted_params_bytes = base64.b64decode(request.encrypted_params) # 解析AES加密的数据：IV(16) + 加密数据 if len(encrypted_params_bytes) < 16: raise ValueError("加密数据长度无效") iv = encrypted_params_bytes[:16] encrypted_payload = encrypted_params_bytes[16:] # AES CTR模式解密 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend cipher = Cipher(algorithms.AES(aes_key_bytes), modes.CTR(iv), backend=default_backend()) decryptor = cipher.decryptor() decrypted_bytes = decryptor.update(encrypted_payload) + decryptor.finalize() # JSON解析 params: Dict[str, Any] = json.loads(decrypted_bytes.decode('utf-8')) # 记录解密后的参数（用于调试） logger.debug(f"[公告接口解密] 解密后的请求参数: {json.dumps(params, ensure_ascii=False, indent=2)}") except Exception as e: logger.exception("请求参数解密失败") return { "code": 400, "msg": "参数解密失败", "timestamp": TimestampValidator.generate_timestamp() } # 5. 提取业务参数 page = params.get('page', 1) page_size = params.get('page_size', 10) keyword = params.get('keyword') # 参数验证 if not isinstance(page, int) or page < 1: page = 1 if not isinstance(page_size, int) or page_size < 1 or page_size > 100: page_size = 10 if keyword and not isinstance(keyword, str): keyword = None # 6. 执行业务逻辑 offset = (page - 1) * page_size query = db.query(Announcement).filter(Announcement.is_active == 1, Announcement.is_deleted == 0) if keyword: like_pattern = f"%{keyword}%" query = query.filter( or_( Announcement.title.like(like_pattern), Announcement.content.like(like_pattern), ) ) # 按优先级降序（紧急->重要->普通），然后按创建时间降序 query = query.order_by(Announcement.priority.desc(), Announcement.created_at.desc(), Announcement.id.desc()) total = query.count() announcements = query.offset(offset).limit(page_size).all() # 7. 准备响应数据 response_data = { "data": [ { "id": a.id, "title": a.title, "content": a.content, "publish_time": a.publish_time.isoformat() if a.publish_time else None, "created_at": a.created_at.isoformat() if a.created_at else None, } for a in announcements ], "total": total, "page": page, "page_size": page_size, } # 8. 使用AES加密响应数据 (CTR模式，与前端保持一致) # 🔐 安全设计：使用请求中解密出的同一个AES密钥加密响应 # 这样前端可以用本地保存的AES密钥直接解密，无需额外的RSA密钥交换 response_json = json.dumps(response_data, ensure_ascii=False, separators=(',', ':')) # 使用解密出的AES密钥加密响应 response_iv = secrets.token_bytes(16) from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend cipher = Cipher(algorithms.AES(aes_key_bytes), modes.CTR(response_iv), backend=default_backend()) encryptor_obj = cipher.encryptor() data_bytes = response_json.encode('utf-8') # CTR模式不需要填充 encrypted_response = encryptor_obj.update(data_bytes) + encryptor_obj.finalize() # 组合：IV(16) + 加密数据 encrypted_response_with_iv = response_iv + encrypted_response # 9. 生成响应签名 timestamp = TimestampValidator.generate_timestamp() response_signature_data = f"{timestamp}:{request.nonce}:{base64.b64encode(encrypted_response_with_iv).decode()}" response_signature = _generate_signature(response_signature_data, settings.SECRET_KEY) # 10. 返回加密响应 # 注意：这里不返回encrypted_key，因为前端已经知道AES密钥 return { "code": 200, "msg": "成功", "encrypted_data": base64.b64encode(encrypted_response_with_iv).decode(), "signature": response_signature, "timestamp": timestamp, "nonce": request.nonce } except Exception as e: logger.exception("公告获取异常") return { "code": 500, "msg": "系统异常，请稍后重试；若持续发生请联系客服", "timestamp": TimestampValidator.generate_timestamp() }