240 lines
8.4 KiB
Python
240 lines
8.4 KiB
Python
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)
|
|
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)
|
|
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] = []
|
|
|
|
class Meta:
|
|
db_table = "user_access"
|
|
ordering = ["user_name"]
|
|
|
|
def __str__(self) -> str:
|
|
return self.user_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"])
|