buff/backend/app/api/client/announcement.py

"""
公告接口（客户端使用）- 加密版本
"""
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 get_client_logger
logger = get_client_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. 记录明文响应数据日志
        logger.debug(f"[公告接口响应] 明文响应数据: {json.dumps(response_data, ensure_ascii=False, indent=2)}")

        # 9. 使用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()
        }