from __future__ import annotations import hashlib import hmac import secrets from django.conf import settings from django.contrib.auth.models import AbstractUser from django.db import models from django.db.models import Q class User(AbstractUser): class Status(models.TextChoices): ACTIVE = "active", "正常" DISABLED = "disabled", "禁用" email = models.EmailField("邮箱", unique=True) payment_user_id = models.CharField( "支付系统用户 ID", max_length=128, blank=True, help_text="外部支付系统用户标识,可为空。", ) status = models.CharField( "状态", max_length=20, choices=Status.choices, default=Status.ACTIVE, ) created_at = models.DateTimeField("创建时间", auto_now_add=True) @property def is_business_active(self) -> bool: return self.status == self.Status.ACTIVE and self.is_active class Meta: db_table = "user" verbose_name = "用户" verbose_name_plural = "用户" class UserWallet(models.Model): user = models.OneToOneField( settings.AUTH_USER_MODEL, verbose_name="用户", on_delete=models.PROTECT, related_name="wallet", ) points_balance = models.BigIntegerField("点数余额", default=0) created_at = models.DateTimeField("创建时间", auto_now_add=True) updated_at = models.DateTimeField("更新时间", auto_now=True) class Meta: db_table = "user_wallet" verbose_name = "用户钱包" verbose_name_plural = "用户钱包" ordering = ("user_id",) constraints = [ models.CheckConstraint( condition=Q(points_balance__gte=0), name="user_wallet_points_balance_non_negative", ), ] def __str__(self) -> str: return f"{self.user} wallet: {self.points_balance}" class ApiKey(models.Model): KEY_PREFIX_LENGTH = 16 class Status(models.TextChoices): ACTIVE = "active", "有效" REVOKED = "revoked", "已吊销" user = models.ForeignKey( settings.AUTH_USER_MODEL, verbose_name="用户", on_delete=models.PROTECT, related_name="api_keys", ) name = models.CharField("名称", max_length=80, blank=True) key_hash = models.CharField("API Key 哈希", max_length=64, unique=True, editable=False) key_prefix = models.CharField("Key 前缀", max_length=32, editable=False) status = models.CharField( "状态", max_length=20, choices=Status.choices, default=Status.ACTIVE, ) last_used_at = models.DateTimeField("最后使用时间", null=True, blank=True) created_at = models.DateTimeField("创建时间", auto_now_add=True) updated_at = models.DateTimeField("更新时间", auto_now=True) class Meta: db_table = "api_key" verbose_name = "API 密钥" verbose_name_plural = "API 密钥" ordering = ("-created_at", "-id") indexes = [ models.Index(fields=("user", "status")), models.Index(fields=("key_prefix",)), ] def __str__(self) -> str: return f"{self.user} {self.key_prefix}" @classmethod def generate_plaintext_key(cls) -> str: return f"sk_cmhub_{secrets.token_urlsafe(32)}" @classmethod def hash_key(cls, raw_key: str) -> str: return hashlib.sha256(raw_key.encode("utf-8")).hexdigest() @classmethod def build_prefix(cls, raw_key: str) -> str: return raw_key[: cls.KEY_PREFIX_LENGTH] @classmethod def create_for_user(cls, user, *, name: str = "") -> tuple["ApiKey", str]: raw_key = cls.generate_plaintext_key() api_key = cls(user=user, name=name) api_key.set_key(raw_key) api_key.save() return api_key, raw_key def set_key(self, raw_key: str) -> None: self.key_hash = self.hash_key(raw_key) self.key_prefix = self.build_prefix(raw_key) def matches_key(self, raw_key: str) -> bool: return hmac.compare_digest(self.key_hash, self.hash_key(raw_key)) @property def is_active_key(self) -> bool: return self.status == self.Status.ACTIVE