buff3.0/backend-admin/app/api/admin/keys.py

"""
密钥管理接口
"""
from fastapi import APIRouter, Depends, Request
from sqlalchemy.orm import Session
from fastapi import Body
from app.database import get_db
from app.api.deps import get_current_admin
from app.models.api_key import APIKey
from app.models.key_rotation_config import KeyRotationConfig
from app.models.system_config import SystemConfig
from app.core.encrypt import encryptor
from datetime import datetime, timedelta
from app.utils.timezone import beijing_now
from app.config import settings
from app.core.logger import logger

router = APIRouter()

@router.get("/current")
async def get_current_key(
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """获取当前密钥"""
    key = db.query(APIKey).filter(APIKey.is_active == 1).order_by(APIKey.key_version.desc()).first()
    
    if not key:
        return {"code": 404, "msg": "未找到激活的密钥"}
    
    return {
        "code": 200,
        "msg": "成功",
        "data": {
            "public_key": key.public_key,
            "private_key": key.private_key,
            "key_version": key.key_version,
            "expire_at": key.expire_at.isoformat() if key.expire_at else None
        }
    }

@router.get("/history")
async def get_key_history(
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """获取历史密钥"""
    keys = db.query(APIKey).order_by(APIKey.key_version.desc()).all()

    return {
        "code": 200,
        "msg": "成功",
        "data": [
            {
                "id": k.id,
                "key_version": k.key_version,
                "is_active": bool(k.is_active),
                "public_key": k.public_key,
                "private_key": k.private_key,
                "created_at": k.created_at.isoformat() if k.created_at else None,
                "expire_at": k.expire_at.isoformat() if k.expire_at else None
            }
            for k in keys
        ]
    }

@router.post("/rotate")
async def rotate_key(
    request: Request,
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """立即更新密钥"""
    # 获取当前激活的密钥
    current_key = db.query(APIKey).filter(APIKey.is_active == 1).order_by(APIKey.key_version.desc()).first()

    # 生成新密钥
    new_key_pair = encryptor.generate_key_pair()
    new_version = (current_key.key_version + 1) if current_key else 1

    # 从配置中获取密钥过期时间（默认60分钟）
    expire_minutes = SystemConfig.get_config_value('key_expire_minutes', 60)

    logger.info(f"执行手动密钥轮换：新版本={new_version}，过期时间={expire_minutes}分钟")

    expire_at = beijing_now() + timedelta(minutes=expire_minutes)

    # 创建新密钥
    new_key = APIKey(
        public_key=new_key_pair[1],  # 公钥 (public_pem)
        private_key=new_key_pair[0], # 私钥 (private_pem)
        key_version=new_version,
        is_active=1,
        expire_at=expire_at,
        created_at=beijing_now(),
        updated_at=beijing_now()
    )
    db.add(new_key)

    # 刷新以获取ID
    db.flush()

    # 创建历史记录
    from app.utils.history import create_api_key_history, create_operation_log
    create_api_key_history(db, new_key, "INSERT")

    # 记录操作日志
    client_ip = request.client.host if request.client else None
    old_version = current_key.key_version if current_key else None
    create_operation_log(
        db=db,
        admin_id=current_admin.id,
        module="密钥管理",
        action="手动密钥轮换",
        description=f"执行手动密钥轮换：版本 {old_version} -> {new_version}",
        old_data={"old_version": old_version},
        new_data={"new_version": new_version, "expire_minutes": expire_minutes},
        ip_address=client_ip
    )

    # 一次性提交所有更改
    db.commit()

    return {
        "code": 200,
        "msg": "密钥更新成功",
        "data": {
            "public_key": new_key.public_key,
            "private_key": new_key.private_key,
            "key_version": new_key.key_version,
            "expire_at": new_key.expire_at.isoformat() if new_key.expire_at else None
        }
    }

@router.put("/cycle")
async def update_key_cycle(
    req: dict = Body(...),
    request: Request = None,
    db: Session = Depends(get_db),
    current_admin = Depends(get_current_admin)
):
    """更新密钥轮换周期"""
    rotation_days = req.get("rotation_days")
    if not rotation_days or rotation_days < 1:
        return {"code": 400, "msg": "轮换周期必须大于0"}

    # 获取或创建配置
    config = db.query(KeyRotationConfig).first()
    old_rotation_days = config.rotation_days if config else None

    if not config:
        config = KeyRotationConfig(
            rotation_days=rotation_days,
            is_auto_rotate=1,
            updated_at=beijing_now()
        )
        db.add(config)
    else:
        config.rotation_days = rotation_days
        config.updated_at = beijing_now()

    # 创建历史记录
    from app.utils.history import create_key_rotation_config_history, create_operation_log
    create_key_rotation_config_history(db, config, "INSERT" if not old_rotation_days else "UPDATE")

    # 记录操作日志
    client_ip = request.client.host if request and request.client else None
    create_operation_log(
        db=db,
        admin_id=current_admin.id,
        module="密钥管理",
        action="更新轮换周期",
        description=f"更新密钥轮换周期：{old_rotation_days}天 -> {rotation_days}天",
        old_data={"rotation_days": old_rotation_days},
        new_data={"rotation_days": rotation_days},
        ip_address=client_ip
    )

    # 一次性提交所有更改
    db.commit()

    return {"code": 200, "msg": "更新成功"}