init_keys.py 1.9 KB

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