from __future__ import annotations import hashlib import hmac import secrets from django.conf import settings from django.db import models 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 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}"