63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
|
"""
|
|||
|
|
API Key 模型 — 用于第三方应用/脚本通过 API 调用平台服务
|
|||
|
|
"""
|
|||
|
|
import hashlib
|
|||
|
|
import logging
|
|||
|
|
from datetime import datetime
|
|||
|
|
from typing import Optional
|
|||
|
|
|
|||
|
|
from sqlalchemy import Column, String, DateTime, JSON, func, ForeignKey
|
|||
|
|
from sqlalchemy.dialects.mysql import CHAR
|
|||
|
|
from sqlalchemy.orm import relationship, Session
|
|||
|
|
from app.core.database import Base
|
|||
|
|
import uuid
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
KEY_PREFIX = "sk-tg-"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class ApiKey(Base):
|
|||
|
|
"""API Key 表"""
|
|||
|
|
__tablename__ = "api_keys"
|
|||
|
|
|
|||
|
|
id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()), comment="密钥ID")
|
|||
|
|
user_id = Column(CHAR(36), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, comment="所属用户ID")
|
|||
|
|
workspace_id = Column(CHAR(36), ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False, comment="所属工作区ID")
|
|||
|
|
name = Column(String(100), nullable=False, comment="密钥名称（用于标识）")
|
|||
|
|
key_prefix = Column(String(12), nullable=False, comment="密钥前缀（如 sk-tg-）")
|
|||
|
|
key_hash = Column(String(255), nullable=False, unique=True, comment="SHA256 哈希存储")
|
|||
|
|
permissions = Column(JSON, nullable=True, comment="权限范围（null=全部权限）")
|
|||
|
|
last_used_at = Column(DateTime, nullable=True, comment="最后使用时间")
|
|||
|
|
expires_at = Column(DateTime, nullable=True, comment="过期时间（null=永不过期）")
|
|||
|
|
status = Column(String(20), default="active", comment="状态: active/revoked")
|
|||
|
|
created_at = Column(DateTime, default=func.now(), comment="创建时间")
|
|||
|
|
|
|||
|
|
# 关系
|
|||
|
|
user = relationship("User", backref="api_keys")
|
|||
|
|
|
|||
|
|
def __repr__(self):
|
|||
|
|
return f"<ApiKey(id={self.id}, name={self.name}, status={self.status})>"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def verify_api_key(api_key_str: str, db: Session) -> Optional["ApiKey"]:
|
|||
|
|
"""验证 API Key 字符串，返回 ApiKey 对象或 None。"""
|
|||
|
|
if not api_key_str or not api_key_str.startswith(KEY_PREFIX):
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
key_hash = hashlib.sha256(api_key_str.encode()).hexdigest()
|
|||
|
|
|
|||
|
|
api_key = db.query(ApiKey).filter(
|
|||
|
|
ApiKey.key_hash == key_hash,
|
|||
|
|
ApiKey.status == "active",
|
|||
|
|
).first()
|
|||
|
|
|
|||
|
|
if not api_key:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
if api_key.expires_at and api_key.expires_at < datetime.utcnow():
|
|||
|
|
logger.warning("API Key %s 已过期", api_key.id)
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
return api_key
|