init_keys.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. """
  2. 初始化密钥对
  3. """
  4. from sqlalchemy.orm import Session
  5. from app.database import SessionLocal
  6. from app.models.api_key import APIKey
  7. from app.core.encrypt import encryptor
  8. from app.core.logger import logger
  9. from datetime import datetime, timedelta
  10. from app.utils.timezone import beijing_now
  11. from app.config import settings
  12. def init_keys():
  13. """初始化密钥对"""
  14. db = SessionLocal()
  15. try:
  16. # 检查是否已有激活的密钥
  17. existing_key = db.query(APIKey).filter(APIKey.is_active == 1).first()
  18. if existing_key:
  19. logger.info("已存在激活的密钥,加载私钥")
  20. encryptor.load_private_key(existing_key.private_key)
  21. return
  22. # 生成新密钥对
  23. logger.info("生成新的RSA密钥对")
  24. private_pem, public_pem = encryptor.generate_key_pair()
  25. # 获取当前最大版本号
  26. max_version = db.query(APIKey).order_by(APIKey.key_version.desc()).first()
  27. next_version = (max_version.key_version + 1) if max_version else 1
  28. # 计算过期时间
  29. expire_at = beijing_now() + timedelta(days=settings.KEY_ROTATION_DAYS)
  30. # 创建密钥记录
  31. api_key = APIKey(
  32. public_key=public_pem,
  33. private_key=private_pem,
  34. key_version=next_version,
  35. is_active=1,
  36. created_at=beijing_now(),
  37. expire_at=expire_at
  38. )
  39. db.add(api_key)
  40. db.commit()
  41. logger.info(f"密钥对生成成功,版本号: {next_version}")
  42. except Exception as e:
  43. db.rollback()
  44. logger.exception("初始化密钥失败")
  45. raise
  46. finally:
  47. db.close()