59 lines
2.4 KiB
Python
59 lines
2.4 KiB
Python
"""开放平台模型：
|
|
APIKey 外部系统访问令牌（哈希安全存储、Scope 细粒度权限控制、调用频率与最后使用审计）
|
|
"""
|
|
|
|
import hashlib
|
|
import secrets
|
|
from django.db import models
|
|
from apps.core.base_models import TenantScopedModel
|
|
|
|
|
|
class APIKey(TenantScopedModel):
|
|
"""外部第三方对接 API Key。"""
|
|
|
|
name = models.CharField(max_length=128, help_text="接入方系统名称，如'下游分销A系统'")
|
|
prefix = models.CharField(max_length=16, db_index=True, help_text="Key前缀用于检索")
|
|
hashed_key = models.CharField(max_length=128, help_text="SHA256 哈希值")
|
|
|
|
# 权限范围：["products:read", "stocks:read", "orders:write"]
|
|
scopes = models.JSONField(default=list, blank=True, help_text="授权的作用域列表")
|
|
is_active = models.BooleanField(default=True)
|
|
rate_limit = models.IntegerField(default=120, help_text="每分钟最大请求次数")
|
|
|
|
expires_at = models.DateTimeField(null=True, blank=True, help_text="过期时间")
|
|
last_used_at = models.DateTimeField(null=True, blank=True, help_text="最后一次调用时间")
|
|
|
|
class Meta:
|
|
db_table = "openapi_api_key"
|
|
verbose_name_plural = "开放平台APIKey"
|
|
ordering = ["-created_at"]
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.prefix}****)"
|
|
|
|
@classmethod
|
|
def generate(cls, tenant, name: str, scopes: list = None, expires_at=None, created_by=None):
|
|
"""生成并安全存储一个新的 API Key，返回 (instance, raw_key_string)。
|
|
注意：raw_key_string 仅在创建时返回一次！
|
|
"""
|
|
raw_secret = secrets.token_urlsafe(32)
|
|
prefix = f"dh_{secrets.token_hex(4)}"
|
|
raw_key = f"{prefix}.{raw_secret}"
|
|
hashed_key = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()
|
|
|
|
instance = cls.objects.create(
|
|
tenant=tenant,
|
|
name=name,
|
|
prefix=prefix,
|
|
hashed_key=hashed_key,
|
|
scopes=scopes or ["products:read", "stocks:read", "orders:write"],
|
|
expires_at=expires_at,
|
|
created_by=created_by,
|
|
)
|
|
return instance, raw_key
|
|
|
|
def verify_key(self, raw_key: str) -> bool:
|
|
"""校验原始 Key 与哈希是否匹配。"""
|
|
computed = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()
|
|
return secrets.compare_digest(self.hashed_key, computed)
|