from __future__ import annotations import hashlib import hmac import secrets from django.conf import settings from django.core.exceptions import ValidationError from django.db import models from django.db.models import Q from django.utils import timezone class ClientDevice(models.Model): class ProductCode(models.TextChoices): CMSHOPEE = "cmshopee", "虾皮圈优化助手" class Platform(models.TextChoices): WINDOWS = "windows", "Windows" MACOS = "macos", "macOS" LINUX = "linux", "Linux" class Status(models.TextChoices): ACTIVE = "active", "有效" REVOKED = "revoked", "已吊销" user = models.ForeignKey( settings.AUTH_USER_MODEL, verbose_name="用户", on_delete=models.PROTECT, related_name="client_devices", ) product_code = models.CharField( "产品代码", max_length=32, choices=ProductCode.choices, ) device_id_version = models.CharField("设备标识版本", max_length=32) device_fingerprint = models.CharField("设备摘要", max_length=64, editable=False) public_key_fingerprint = models.CharField("安装公钥摘要", max_length=64, editable=False) platform = models.CharField("平台", max_length=16, choices=Platform.choices) client_version = models.CharField("客户端版本", max_length=64) status = models.CharField( "状态", max_length=20, choices=Status.choices, default=Status.ACTIVE, ) first_seen_at = models.DateTimeField("首次登记时间", auto_now_add=True) last_seen_at = models.DateTimeField("最后活跃时间", auto_now_add=True) created_at = models.DateTimeField("创建时间", auto_now_add=True) updated_at = models.DateTimeField("更新时间", auto_now=True) class Meta: db_table = "client_device" verbose_name = "客户端设备" verbose_name_plural = "客户端设备" ordering = ("-last_seen_at", "-id") constraints = [ models.UniqueConstraint( fields=("user", "product_code", "device_fingerprint"), name="client_device_user_product_fingerprint_unique", ), ] indexes = [ models.Index(fields=("user", "product_code", "status")), models.Index(fields=("product_code", "last_seen_at")), ] def __str__(self) -> str: return f"{self.user} {self.product_code} {self.platform}" @staticmethod def fingerprint_device_id(device_id_version: str, device_id: str) -> str: payload = f"{device_id_version}:{device_id}".encode("utf-8") pepper = settings.DEVICE_IDENTIFIER_PEPPER.encode("utf-8") return hmac.new(pepper, payload, hashlib.sha256).hexdigest() @staticmethod def fingerprint_public_key(public_key: str) -> str: return hashlib.sha256(public_key.encode("utf-8")).hexdigest() class SoftwarePlan(models.Model): class Status(models.TextChoices): ACTIVE = "active", "启用" INACTIVE = "inactive", "停用" product_code = models.CharField( "产品代码", max_length=32, choices=ClientDevice.ProductCode.choices, ) name = models.CharField("套餐名称", max_length=120) duration_days = models.PositiveIntegerField("有效天数") price = models.DecimalField("套餐价格", max_digits=12, decimal_places=2) device_limit = models.PositiveSmallIntegerField("设备数量") grace_days = models.PositiveSmallIntegerField("宽限天数", default=0) status = models.CharField( "状态", max_length=20, choices=Status.choices, default=Status.ACTIVE, ) created_at = models.DateTimeField("创建时间", auto_now_add=True) updated_at = models.DateTimeField("更新时间", auto_now=True) class Meta: db_table = "software_plan" verbose_name = "软件套餐" verbose_name_plural = "软件套餐" ordering = ("product_code", "name", "id") constraints = [ models.CheckConstraint( condition=Q(duration_days__gt=0), name="software_plan_duration_days_positive", ), models.CheckConstraint( condition=Q(price__gt=0), name="software_plan_price_positive", ), models.CheckConstraint( condition=Q(device_limit__gt=0), name="software_plan_device_limit_positive", ), ] indexes = [ models.Index(fields=("product_code", "status")), ] def __str__(self) -> str: return f"{self.product_code} {self.name}" class SoftwareEntitlement(models.Model): class Status(models.TextChoices): ACTIVE = "active", "有效" REVOKED = "revoked", "已撤销" EXPIRED = "expired", "已过期" user = models.ForeignKey( settings.AUTH_USER_MODEL, verbose_name="用户", on_delete=models.PROTECT, related_name="software_entitlements", ) product_code = models.CharField( "产品代码", max_length=32, choices=ClientDevice.ProductCode.choices, ) source_plan = models.ForeignKey( SoftwarePlan, verbose_name="来源套餐", null=True, blank=True, on_delete=models.SET_NULL, related_name="entitlements", ) plan_name = models.CharField("套餐名称快照", max_length=120) plan_duration_days = models.PositiveIntegerField("套餐有效天数快照") plan_price = models.DecimalField("套餐价格快照", max_digits=12, decimal_places=2) plan_device_limit = models.PositiveSmallIntegerField("设备数量快照") plan_grace_days = models.PositiveSmallIntegerField("宽限天数快照", default=0) status = models.CharField( "状态", max_length=20, choices=Status.choices, default=Status.ACTIVE, ) starts_at = models.DateTimeField("开始时间") expires_at = models.DateTimeField("到期时间") grace_expires_at = models.DateTimeField("宽限截止时间") revoked_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 = "software_entitlement" verbose_name = "软件权益" verbose_name_plural = "软件权益" ordering = ("-expires_at", "-id") constraints = [ models.CheckConstraint( condition=Q(plan_duration_days__gt=0), name="software_entitlement_duration_positive", ), models.CheckConstraint( condition=Q(plan_price__gt=0), name="software_entitlement_price_positive", ), models.CheckConstraint( condition=Q(plan_device_limit__gt=0), name="software_entitlement_device_limit_positive", ), ] indexes = [ models.Index(fields=("user", "product_code", "status")), models.Index(fields=("product_code", "expires_at")), ] def __str__(self) -> str: return f"{self.user} {self.product_code} {self.plan_name}" def clean(self) -> None: super().clean() if self.expires_at and self.starts_at and self.expires_at <= self.starts_at: raise ValidationError({"expires_at": "到期时间必须晚于开始时间。"}) if ( self.grace_expires_at and self.expires_at and self.grace_expires_at < self.expires_at ): raise ValidationError({"grace_expires_at": "宽限截止时间不能早于到期时间。"}) def is_usable_at(self, now=None) -> bool: now = now or timezone.now() return self.status == self.Status.ACTIVE and self.grace_expires_at > now class LicenseSeat(models.Model): entitlement = models.ForeignKey( SoftwareEntitlement, verbose_name="软件权益", on_delete=models.PROTECT, related_name="seats", ) seat_number = models.PositiveSmallIntegerField("席位序号") device = models.ForeignKey( ClientDevice, verbose_name="绑定设备", null=True, blank=True, on_delete=models.SET_NULL, related_name="license_seats", ) bound_at = models.DateTimeField("绑定时间", null=True, blank=True) released_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 = "license_seat" verbose_name = "授权席位" verbose_name_plural = "授权席位" ordering = ("entitlement_id", "seat_number") constraints = [ models.UniqueConstraint( fields=("entitlement", "seat_number"), name="license_seat_entitlement_number_unique", ), models.CheckConstraint( condition=Q(seat_number__gt=0), name="license_seat_number_positive", ), ] indexes = [ models.Index(fields=("device", "updated_at")), ] def __str__(self) -> str: return f"{self.entitlement} #{self.seat_number}" class LicenseEvent(models.Model): class Action(models.TextChoices): GRANTED = "granted", "人工授予" RENEWED = "renewed", "续期" REVOKED = "revoked", "撤销" SEAT_ASSIGNED = "seat_assigned", "绑定席位" SEAT_RELEASED = "seat_released", "解绑席位" entitlement = models.ForeignKey( SoftwareEntitlement, verbose_name="软件权益", on_delete=models.PROTECT, related_name="events", ) seat = models.ForeignKey( LicenseSeat, verbose_name="授权席位", null=True, blank=True, on_delete=models.SET_NULL, related_name="events", ) device = models.ForeignKey( ClientDevice, verbose_name="关联设备", null=True, blank=True, on_delete=models.SET_NULL, related_name="license_events", ) actor = models.ForeignKey( settings.AUTH_USER_MODEL, verbose_name="操作人", null=True, blank=True, on_delete=models.SET_NULL, related_name="license_events_performed", ) action = models.CharField("动作", max_length=32, choices=Action.choices) reason = models.CharField("原因", max_length=255) metadata = models.JSONField("附加信息", default=dict, blank=True) created_at = models.DateTimeField("创建时间", auto_now_add=True) class Meta: db_table = "license_event" verbose_name = "授权事件" verbose_name_plural = "授权事件" ordering = ("-created_at", "-id") indexes = [ models.Index(fields=("entitlement", "created_at")), models.Index(fields=("device", "created_at")), ] def __str__(self) -> str: return f"{self.entitlement} {self.action}" class DeviceSession(models.Model): TOKEN_PREFIX_LENGTH = 12 device = models.ForeignKey( ClientDevice, verbose_name="客户端设备", on_delete=models.PROTECT, related_name="sessions", ) token_hash = models.CharField("会话令牌哈希", max_length=64, unique=True, editable=False) expires_at = models.DateTimeField("过期时间") last_used_at = models.DateTimeField("最后使用时间", null=True, blank=True) revoked_at = models.DateTimeField("吊销时间", null=True, blank=True) created_at = models.DateTimeField("创建时间", auto_now_add=True) class Meta: db_table = "device_session" verbose_name = "设备会话" verbose_name_plural = "设备会话" ordering = ("-created_at", "-id") indexes = [ models.Index(fields=("device", "expires_at")), models.Index(fields=("expires_at", "revoked_at")), ] def __str__(self) -> str: return f"{self.device} session" @classmethod def generate_plaintext_token(cls) -> str: return f"dvs_cmhub_{secrets.token_urlsafe(32)}" @staticmethod def hash_token(raw_token: str) -> str: return hashlib.sha256(raw_token.encode("utf-8")).hexdigest() def matches_token(self, raw_token: str) -> bool: return hmac.compare_digest(self.token_hash, self.hash_token(raw_token)) def is_active_at(self, now=None) -> bool: now = now or timezone.now() return self.revoked_at is None and self.expires_at > now class DeviceBindingAudit(models.Model): class Action(models.TextChoices): REGISTERED = "registered", "已登记" SESSION_ISSUED = "session_issued", "会话已签发" HEARTBEAT = "heartbeat", "心跳" REJECTED = "rejected", "请求被拒绝" REVOKED = "revoked", "设备已吊销" user = models.ForeignKey( settings.AUTH_USER_MODEL, verbose_name="用户", on_delete=models.PROTECT, related_name="device_binding_audits", ) device = models.ForeignKey( ClientDevice, verbose_name="客户端设备", null=True, blank=True, on_delete=models.SET_NULL, related_name="audits", ) api_key = models.ForeignKey( "users.ApiKey", verbose_name="来源 API 密钥", null=True, blank=True, on_delete=models.SET_NULL, related_name="device_binding_audits", ) action = models.CharField("动作", max_length=32, choices=Action.choices) reason = models.CharField("原因", max_length=120, blank=True) client_version = models.CharField("客户端版本", max_length=64, blank=True) created_at = models.DateTimeField("创建时间", auto_now_add=True) class Meta: db_table = "device_binding_audit" verbose_name = "设备绑定审计" verbose_name_plural = "设备绑定审计" ordering = ("-created_at", "-id") indexes = [ models.Index(fields=("user", "created_at")), models.Index(fields=("device", "created_at")), ] def __str__(self) -> str: return f"{self.user} {self.action}"