init_keys.py 1.7 KB

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