# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
import datetime
from dataclasses import dataclass
from datetime import timezone, timedelta
from uuid import UUID
import jwt
from django.conf import settings
from django.contrib.auth.models import (
AbstractUser,
UserManager as DjangoUserManager,
Group,
)
from django.db import models
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.emiflow.metrics import TonnageLimit
from ferrosoft.apps.ferrobase.auth.roles import Role, Registry
from ferrosoft.apps.ferrobase.models.shared import (
ColorTheme,
Language,
LocalUser,
TimeZone,
)
from ferrosoft.apps.ferrobase.models.base import (
AccessKeyMixin,
CommonLimit,
UUIDModel,
)
from ferrosoft.apps.ferrobase.models.subscription import SubscriptionPlan
from ferrosoft.config import require_settings
[docs]
class OrganizationManager(models.Manager):
"""Manager for ``Organization`` that eagerly loads subscription and tenant relations."""
[docs]
def get_queryset(self):
return (
super()
.get_queryset()
.select_related("subscription")
.prefetch_related("tenants")
)
[docs]
class Organization(UUIDModel):
"""Top-level entity representing a customer organisation.
An organisation owns one ``SubscriptionPlan`` and contains one or more
``Tenant`` databases and ``User`` accounts. On save, the subscription
tonnage limit Prometheus metric is updated so dashboards stay in sync.
"""
class Meta:
verbose_name = _("Organization")
verbose_name_plural = _("Organizations")
constraints = (
models.UniqueConstraint(
name="uniq_organization",
fields=["name"],
),
models.UniqueConstraint(
name="uniq_organization_iam_reference",
fields=["iam_reference"],
),
)
indexes = [
models.Index(fields=["contact_email"]),
models.Index(fields=["contact_first_name"]),
models.Index(fields=["contact_last_name"]),
]
objects = OrganizationManager()
name = models.CharField(
verbose_name=_("Name"),
)
iam_reference = models.CharField(
verbose_name=_("IAM reference"),
null=True,
)
contact_email = models.EmailField(
default="",
verbose_name=_("Contact Email"),
)
contact_first_name = models.CharField(
default="",
verbose_name=_("Contact First Name"),
)
contact_last_name = models.CharField(
default="",
verbose_name=_("Contact Last Name"),
)
subscription = models.ForeignKey(
SubscriptionPlan,
on_delete=models.RESTRICT,
null=True,
related_name="+",
verbose_name=_("Subscription Plan"),
)
def __str__(self):
return self.name
[docs]
def save(
self, *, force_insert=False, force_update=False, using=None, update_fields=None
):
super().save(
force_insert=force_insert,
force_update=force_update,
using=using,
update_fields=update_fields,
)
if self.subscription is not None:
TonnageLimit.labels(organization=self.name).set(
float(self.subscription.max_tonnage or 0)
)
[docs]
def get_absolute_url(self):
return reverse("ferrobase:organization_update", args=[self.pk])
[docs]
class TenantManager(models.Manager):
"""Manager for ``Tenant`` that eagerly loads the related organisation."""
[docs]
def get_queryset(self):
return super().get_queryset().select_related("organization")
[docs]
class Tenant(UUIDModel):
"""
Tenant segregates entities from each other.
This model is global to all tenants.
"""
class Meta:
verbose_name = _("Tenant")
verbose_name_plural = _("Tenants")
indexes = [
models.Index(fields=["expiry_date", "active"]),
]
objects = TenantManager()
organization = models.ForeignKey(
Organization,
on_delete=models.CASCADE,
related_name="tenants",
verbose_name=_("Organization"),
)
name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Name"),
unique=True,
)
active = models.BooleanField(
verbose_name=_("Active"),
help_text=_(
"Active tenants can be selected in the tenant menu and are usable by the user."
),
default=True,
)
expiry_date = models.DateTimeField(
verbose_name=_("Expiry date"),
help_text=_(
"Set expiry date to automatically delete the tenant after the specified time."
),
null=True,
default=None,
blank=True,
)
database_hostname = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Database connection hostname"),
default="localhost",
)
database_port = models.IntegerField(
verbose_name=_("Database connection port number"),
default=5432,
)
database_username = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Database username"),
default="",
)
database_password = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Database connection password"),
default="",
)
database_name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Database name"),
default="",
)
def __str__(self):
return self.name
[docs]
@classmethod
def find(cls, tenant_id, active=True) -> "Tenant":
"""
Find tenant with given ID.
Args:
tenant_id:
active: Find only tenants which are active or not active (active by default).
"""
return cls.objects.filter(pk=tenant_id, active=active).first()
[docs]
@classmethod
async def afind(cls, tenant_id, active=True):
return await cls.objects.filter(pk=tenant_id, active=active).afirst()
[docs]
@classmethod
def lookup(cls, tenant: "str | UUID | Tenant") -> "Tenant | None":
if isinstance(tenant, Tenant):
return tenant
elif isinstance(tenant, str):
return cls.objects.filter(name=tenant).first()
elif isinstance(tenant, UUID):
return cls.objects.filter(id=tenant).first()
else:
raise ValueError("Cannot look up tenant by %s" % type(tenant))
[docs]
@classmethod
def all_ordered(cls):
return cls.objects.filter(active=True).order_by("name")
@property
def django_connection(self):
return {
"ATOMIC_REQUESTS": False,
"AUTOCOMMIT": True,
"CONN_HEALTH_CHECKS": False,
"CONN_MAX_AGE": 0,
"ENGINE": "django_prometheus.db.backends.postgresql",
"HOST": self.database_hostname,
"NAME": self.database_name,
"OPTIONS": {},
"PASSWORD": self.database_password,
"PORT": self.database_port,
"TEST": {
"CHARSET": None,
"COLLATION": None,
"MIGRATE": None,
"MIRROR": None,
"NAME": None,
},
"TIME_ZONE": None,
"USER": self.database_username,
}
[docs]
def activate_connection(self, default: bool = None):
"""
Fudge tenant database connection details into runtime configuration.
It is advisable to use activate_tenant_database instead of using this
method directly.
"""
settings.DATABASES[self.database_name] = self.django_connection
if default:
settings.DATABASES["default"] = self.django_connection
[docs]
def get_absolute_url(self):
return reverse("ferrobase:tenant_update", args=[self.pk])
[docs]
class UserManager(DjangoUserManager):
"""Manager for ``User`` that eagerly loads organisation, language, and timezone."""
[docs]
def get_queryset(self):
return (
super()
.get_queryset()
.select_related("organization", "language", "timezone")
)
[docs]
class User(AbstractUser):
"""
Extends Django user model with fields required for Ferrosoft Platform.
This model is global to all tenants.
"""
[docs]
class PasswordsDoNotMatch(Exception):
def __str__(self):
return "passwords do not match"
[docs]
class PasswordInsufficientSize(Exception):
def __str__(self):
return "password is of insufficient size"
objects = UserManager()
MINIMUM_PASSWORD_LENGTH = 15
REQUIRED_FIELDS = ["email", "organization_id"]
organization = models.ForeignKey(
Organization,
related_name="users",
on_delete=models.CASCADE,
verbose_name=_("Organization"),
)
language = models.ForeignKey(
Language,
on_delete=models.RESTRICT,
verbose_name=_("Preferred language"),
null=True,
blank=True,
)
timezone = models.ForeignKey(
TimeZone,
on_delete=models.RESTRICT,
verbose_name=_("Time zone"),
null=True,
blank=True,
)
color_theme = models.CharField(
max_length=1,
choices=ColorTheme.choices,
default=ColorTheme.AUTO,
verbose_name=_("Color theme"),
)
[docs]
@classmethod
def all_ordered(cls):
return cls.objects.order_by("username")
[docs]
@classmethod
def get_by_username(cls, name: str) -> "User":
return cls.objects.get(username=name)
[docs]
def get_absolute_url(self):
return reverse("ferrobase:user_update", kwargs={"pk": self.pk})
[docs]
def get_assigned_feature(self, feature_code: str):
subscription: SubscriptionPlan | None = getattr(
self.organization, "subscription", None
)
if subscription is None:
return None, None
return subscription.get_assigned_feature(feature_code), subscription
[docs]
def assign_roles(self, *roles: Role):
groups = list(
Group.objects.filter(
name__in=[
perm_set.group
for required_role, perm_set in Registry.get_permission_sets().items()
if required_role.value in roles
]
)
)
self.groups.add(*groups)
[docs]
def set_password_validated(self, password: str, password_again: str):
if password != password_again:
raise self.PasswordsDoNotMatch()
if len(password) < self.MINIMUM_PASSWORD_LENGTH:
raise self.PasswordInsufficientSize()
self.set_password(password)
[docs]
def set_tenants(self, tenant_ids: list):
raise NotImplementedError
[docs]
def tenant_allowed(self, tenant: Tenant):
return self.is_superuser or self.organization_id == tenant.organization_id
[docs]
def tenant_options(self):
return [
(
str(tenant.pk),
tenant.name,
True,
)
for tenant in self.organization.tenants.all()
]
[docs]
def local_user(self):
"""
Find LocalUser corresponding to this user.
If the LocalUser does not exist, it is created implicitly.
Returns: LocalUser
"""
return LocalUser.load(self)
[docs]
class PasswordResetRequest(AccessKeyMixin, UUIDModel):
"""
A password reset request is created upon the users request to reset a password.
It is deleted once the user changes the password.
This model is global to all tenants.
"""
user = models.ForeignKey(
User,
on_delete=models.CASCADE,
verbose_name=_("User"),
)
[docs]
@classmethod
def create(cls, user: User):
"""
Create new request with random access key.
Args:
user: For which user the request is created.
"""
return cls.objects.create(user=user, **cls.new_access_key_params())
def __str__(self):
return "password reset request %s for user %s valid until %s" % (
str(self.pk),
self.user.username,
self.access_key_expiry.isoformat(),
)
[docs]
class AccountActivationRequest(AccessKeyMixin, UUIDModel):
"""Time-limited token that activates a freshly provisioned tenant account.
Created when a new tenant is set up and sent to the contact email. The
recipient follows the link containing ``access_key`` to confirm the account.
Expired after ``AccessKeyMixin._DEFAULT_VALIDITY``.
"""
tenant = models.ForeignKey(
Tenant,
on_delete=models.CASCADE,
related_name="+",
)
[docs]
@classmethod
def create(cls, tenant: Tenant, expiry: datetime.datetime | None = None):
return cls.objects.create(tenant=tenant, **cls.new_access_key_params(expiry))
def __str__(self):
return "account activation request %s for tenant %s valid until %s" % (
str(self.pk),
self.tenant.name,
self.access_key_expiry.isoformat(),
)
[docs]
def get_absolute_url(self):
return reverse(
"ferrobase:activate_account", kwargs={"access_key": self.access_key}
)
[docs]
class APITokenStatus(models.IntegerChoices):
"""Lifecycle state of an ``APIToken``."""
ACTIVE = 1, _("active")
INACTIVE = 2, _("inactive")
BLOCKED = 3, _("blocked")
[docs]
class APITokenPrefetchedManager(models.Manager):
"""Manager for ``APIToken`` that deeply pre-fetches user, organisation, and tenant."""
[docs]
def get_queryset(self):
return (
super()
.get_queryset()
.select_related(
"user__organization",
"user__organization__subscription",
"user__language",
"tenant__organization",
)
)
[docs]
class APIToken(UUIDModel):
"""
REST interface is authenticated with JWT's. Token claims include a reference
to this model, which informs about the user and which tenant to use for the
request.
API middleware also inspects status and expiration and grants access only if
the status is ACTIVE and the current time does not exceed expiration time.
In order to not rely too much on JWT, these fields are not encoded directly
in the JWT, allowing for other authentication methods to be added.
Token name is purely informational for display in the UI.
"""
class Meta:
verbose_name = _("API Token")
verbose_name_plural = _("API Tokens")
indexes = [
models.Index(fields=["name"]),
models.Index(fields=["status"]),
]
[docs]
@dataclass
class Settings:
jwt_secret_key: str
expiration_days: int = 30
jwt_algorithm: str = "HS256"
[docs]
@classmethod
def from_django_settings(cls):
settings = require_settings("FERROBASE_REST_API")
options = settings["FERROBASE_REST_API"]
return cls(
options["jwt_secret_key"],
options.get("expiration_days", cls.expiration_days),
options.get("jwt_algorithm", cls.jwt_algorithm),
)
TOKEN_ID_CLAIM = "token_id"
objects = APITokenPrefetchedManager()
user = models.ForeignKey(
User,
on_delete=models.CASCADE,
related_name="api_tokens",
verbose_name=_("User"),
)
tenant = models.ForeignKey(
Tenant,
on_delete=models.CASCADE,
related_name="+",
verbose_name=_("Tenant"),
)
name = models.CharField(
verbose_name=_("Name"),
blank=True,
default="",
)
status = models.IntegerField(
choices=APITokenStatus.choices,
default=APITokenStatus.ACTIVE,
verbose_name=_("Status"),
)
expiration = models.DateTimeField(
verbose_name=_("Expiration"),
)
[docs]
@classmethod
def create(cls, user: User, tenant: Tenant, name: str = ""):
options = cls.Settings.from_django_settings()
expiration = cls._get_expiration(options)
token_model = cls.objects.create(
user=user,
tenant=tenant,
expiration=expiration,
name=name,
)
token = cls._encode_token(token_model, options)
return token_model, token
[docs]
@classmethod
async def acreate(cls, user: User, tenant: Tenant):
options = cls.Settings.from_django_settings()
expiration = cls._get_expiration(options)
token_model = await cls.objects.acreate(
user=user, tenant=tenant, expiration=expiration
)
token = cls._encode_token(token_model, options)
return token_model, token
@classmethod
def _get_expiration(cls, options: Settings):
return now_utc() + timedelta(days=options.expiration_days)
@classmethod
def _encode_token(cls, token_model, options: Settings):
return jwt.encode(
{cls.TOKEN_ID_CLAIM: str(token_model.pk)},
options.jwt_secret_key,
algorithm=options.jwt_algorithm,
)
[docs]
@classmethod
async def averify(cls, token: str | bytes) -> APIToken | None:
options = cls.Settings.from_django_settings()
claims = jwt.decode(
token,
options.jwt_secret_key,
algorithms=[options.jwt_algorithm],
options={
"require": [cls.TOKEN_ID_CLAIM],
},
)
token = await cls.objects.aget(pk=claims[cls.TOKEN_ID_CLAIM])
if not token.is_valid():
return None
return token
[docs]
def is_valid(self) -> bool:
return (
self.status == APITokenStatus.ACTIVE.value
and now_utc() < self.expiration.astimezone(tz=timezone.utc)
)