from __future__ import annotations import hashlib import secrets from django.contrib.auth.models import AbstractBaseUser, BaseUserManager, PermissionsMixin from django.db import models from django.utils import timezone class UserManager(BaseUserManager): def create_user(self, user_name: str, password: str | None = None, **extra_fields): if not user_name: raise ValueError("user_name is required") extra_fields.setdefault("status", User.STATUS_GM) extra_fields.setdefault("is_superuser", False) extra_fields.setdefault("display_name", user_name) user = self.model(user_name=user_name, **extra_fields) user.set_password(password) user.save(using=self._db) return user def create_superuser(self, user_name: str, password: str | None = None, **extra_fields): """Break-glass: status=superuser (ops + user management). Hidden from user lists.""" extra_fields["status"] = User.STATUS_SUPERUSER extra_fields["is_superuser"] = True return self.create_user(user_name, password, **extra_fields) class User(AbstractBaseUser, PermissionsMixin): """HQ executive user. Roles are distinct from site dashboard active/superadmin.""" STATUS_INACTIVE = "inactive" STATUS_GM = "gm" STATUS_GM_ADMIN = "gm_admin" STATUS_BUH = "buh" STATUS_BU_ADMIN = "bu_admin" STATUS_DIRECTOR = "director" STATUS_DIRECTOR_ADMIN = "director_admin" STATUS_ACCOUNT_ADMIN = "account_admin" STATUS_SUPERUSER = "superuser" STATUS_CHOICES = [ (STATUS_INACTIVE, "Inactive"), (STATUS_GM, "GM"), (STATUS_GM_ADMIN, "GM Admin"), (STATUS_BUH, "BUH"), (STATUS_BU_ADMIN, "BU Admin"), (STATUS_DIRECTOR, "Director"), (STATUS_DIRECTOR_ADMIN, "Director Admin"), (STATUS_ACCOUNT_ADMIN, "Account Admin"), (STATUS_SUPERUSER, "Superuser"), ] # Roles that may sign in and use the executive dashboard. EXECUTIVE_STATUSES = frozenset( { STATUS_GM, STATUS_GM_ADMIN, STATUS_BUH, STATUS_BU_ADMIN, STATUS_DIRECTOR, STATUS_DIRECTOR_ADMIN, STATUS_ACCOUNT_ADMIN, STATUS_SUPERUSER, } ) # BU / Director "*_admin" are view-only (same as buh / director). # Kept for labeling only — no extra write powers. ADMIN_STATUSES = frozenset( { STATUS_BU_ADMIN, STATUS_DIRECTOR_ADMIN, } ) user_id = models.BigAutoField(primary_key=True) user_name = models.CharField(max_length=30, unique=True) display_name = models.CharField(max_length=30, unique=True) status = models.CharField(max_length=30, choices=STATUS_CHOICES, default=STATUS_GM) # BUH who manages this GM / GM Admin (unused for director / buh themselves). managed_by = models.ForeignKey( "self", null=True, blank=True, on_delete=models.SET_NULL, related_name="managed_gms", limit_choices_to={"status__in": ("buh", "bu_admin")}, ) # GM-requested BUH waiting for confirmation (BUH only — not BU Admin). managed_by_pending = models.ForeignKey( "self", null=True, blank=True, on_delete=models.SET_NULL, related_name="pending_managed_gms", limit_choices_to={"status": "buh"}, ) managed_by_requested_at = models.DateTimeField(null=True, blank=True) is_active = models.BooleanField(default=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) objects = UserManager() USERNAME_FIELD = "user_name" REQUIRED_FIELDS: list[str] = ["display_name"] class Meta: db_table = "user_access" ordering = ["display_name"] def __str__(self) -> str: return self.display_name @property def is_executive(self) -> bool: return self.status in self.EXECUTIVE_STATUSES @property def is_executive_admin(self) -> bool: return self.status in self.ADMIN_STATUSES def request_managed_by(self, buh: "User") -> None: """GM asks to report under a BUH (pending until that BUH confirms).""" from django.core.exceptions import ValidationError if self.status != self.STATUS_GM: raise ValidationError("Only GM can request a BUH.") if buh.status != self.STATUS_BUH: raise ValidationError("Target must be a BUH (not BU Admin).") if self.managed_by_id == buh.pk and self.managed_by_pending_id is None: raise ValidationError("Already managed by this BUH.") self.managed_by_pending = buh self.managed_by_requested_at = timezone.now() self.save( update_fields=["managed_by_pending", "managed_by_requested_at", "updated_at"] ) def cancel_managed_by_request(self) -> None: if self.managed_by_pending_id is None: return self.managed_by_pending = None self.managed_by_requested_at = None self.save( update_fields=["managed_by_pending", "managed_by_requested_at", "updated_at"] ) def approve_managed_by(self, approver: "User") -> None: from django.core.exceptions import ValidationError if self.managed_by_pending_id is None: raise ValidationError("No pending BUH request.") if approver.pk != self.managed_by_pending_id: raise ValidationError("Only the requested BUH can approve this request.") if approver.status != self.STATUS_BUH: raise ValidationError("Only a BUH (not BU Admin) can approve this request.") self.managed_by = self.managed_by_pending self.managed_by_pending = None self.managed_by_requested_at = None self.save( update_fields=[ "managed_by", "managed_by_pending", "managed_by_requested_at", "updated_at", ] ) def reject_managed_by(self, approver: "User") -> None: from django.core.exceptions import ValidationError if self.managed_by_pending_id is None: raise ValidationError("No pending BUH request.") if approver.pk != self.managed_by_pending_id: raise ValidationError("Only the requested BUH can reject this request.") if approver.status != self.STATUS_BUH: raise ValidationError("Only a BUH (not BU Admin) can reject this request.") self.managed_by_pending = None self.managed_by_requested_at = None self.save( update_fields=["managed_by_pending", "managed_by_requested_at", "updated_at"] ) def save(self, *args, **kwargs): # Product roles never use Django is_staff. status=superuser is break-glass # (ops + user management) and always carries is_superuser. if self.status == self.STATUS_INACTIVE: self.is_active = False self.is_superuser = False elif self.status == self.STATUS_SUPERUSER: self.is_active = True self.is_superuser = True else: self.is_active = True self.is_superuser = False super().save(*args, **kwargs) class ApiKey(models.Model): """Hashed API keys for X-API-Key authentication (inbound to HQ API).""" user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="api_keys") name = models.CharField(max_length=100) prefix = models.CharField(max_length=12, db_index=True) key_hash = models.CharField(max_length=64) is_active = models.BooleanField(default=True) 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_keys" ordering = ["-created_at"] def __str__(self) -> str: return f"{self.name} ({self.prefix}…)" @staticmethod def hash_key(raw_key: str) -> str: return hashlib.sha256(raw_key.encode("utf-8")).hexdigest() @classmethod def generate(cls, user: User, name: str) -> tuple[ApiKey, str]: raw = secrets.token_urlsafe(32) prefix = raw[:8] instance = cls.objects.create( user=user, name=name, prefix=prefix, key_hash=cls.hash_key(raw), ) return instance, raw def verify(self, raw_key: str) -> bool: return secrets.compare_digest(self.key_hash, self.hash_key(raw_key)) def touch(self) -> None: self.last_used_at = timezone.now() self.save(update_fields=["last_used_at", "updated_at"])