authentik.tenants.models

Tenant models

  1"""Tenant models"""
  2
  3import re
  4from uuid import uuid4
  5
  6from django.apps import apps
  7from django.conf import settings
  8from django.core.exceptions import ValidationError
  9from django.core.validators import MaxValueValidator, MinValueValidator
 10from django.db import models
 11from django.db.utils import IntegrityError
 12from django.dispatch import receiver
 13from django.utils.translation import gettext_lazy as _
 14from django_tenants.models import DomainMixin, TenantMixin, post_schema_sync
 15from django_tenants.utils import get_tenant_base_schema
 16from rest_framework.serializers import Serializer
 17from structlog.stdlib import get_logger
 18
 19from authentik.blueprints.apps import ManagedAppConfig
 20from authentik.lib.models import InternallyManagedMixin, SerializerModel
 21from authentik.lib.utils.time import timedelta_string_validator
 22
 23LOGGER = get_logger()
 24
 25
 26VALID_SCHEMA_NAME = re.compile(r"^t_[a-z0-9]{1,61}$")
 27
 28DEFAULT_TOKEN_DURATION = "days=1"  # nosec
 29DEFAULT_TOKEN_LENGTH = 60
 30DEFAULT_REPUTATION_LOWER_LIMIT = -5
 31DEFAULT_REPUTATION_UPPER_LIMIT = 5
 32
 33
 34def _validate_schema_name(name):
 35    if not VALID_SCHEMA_NAME.match(name):
 36        raise ValidationError(
 37            _(
 38                "Schema name must start with t_, only contain lowercase letters and numbers and "
 39                "be less than 63 characters."
 40            )
 41        )
 42
 43
 44class Tenant(InternallyManagedMixin, TenantMixin, SerializerModel):
 45    """Tenant"""
 46
 47    tenant_uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
 48    schema_name = models.CharField(
 49        max_length=63, unique=True, db_index=True, validators=[_validate_schema_name]
 50    )
 51    name = models.TextField()
 52
 53    auto_create_schema = True
 54    auto_drop_schema = True
 55    ready = models.BooleanField(default=False)
 56
 57    avatars = models.TextField(
 58        help_text=_("Configure how authentik should show avatars for users."),
 59        default="gravatar,initials",
 60    )
 61    default_user_change_name = models.BooleanField(
 62        help_text=_("Enable the ability for users to change their name."), default=True
 63    )
 64    default_user_change_email = models.BooleanField(
 65        help_text=_("Enable the ability for users to change their email address."), default=False
 66    )
 67    default_user_change_username = models.BooleanField(
 68        help_text=_("Enable the ability for users to change their username."), default=False
 69    )
 70    event_retention = models.TextField(
 71        default="days=365",
 72        validators=[timedelta_string_validator],
 73        help_text=_(
 74            "Events will be deleted after this duration.(Format: weeks=3;days=2;hours=3,seconds=2)."
 75        ),
 76    )
 77    reputation_lower_limit = models.IntegerField(
 78        help_text=_("Reputation cannot decrease lower than this value. Zero or negative."),
 79        default=DEFAULT_REPUTATION_LOWER_LIMIT,
 80        validators=[MaxValueValidator(0)],
 81    )
 82    reputation_upper_limit = models.IntegerField(
 83        help_text=_("Reputation cannot increase higher than this value. Zero or positive."),
 84        default=DEFAULT_REPUTATION_UPPER_LIMIT,
 85        validators=[MinValueValidator(0)],
 86    )
 87    footer_links = models.JSONField(
 88        help_text=_("The option configures the footer links on the flow executor pages."),
 89        default=list,
 90        blank=True,
 91    )
 92    gdpr_compliance = models.BooleanField(
 93        help_text=_(
 94            "When enabled, all the events caused by a user "
 95            "will be deleted upon the user's deletion."
 96        ),
 97        default=True,
 98    )
 99    impersonation = models.BooleanField(
100        help_text=_("Globally enable/disable impersonation."), default=True
101    )
102    impersonation_require_reason = models.BooleanField(
103        help_text=_("Require administrators to provide a reason for impersonating a user."),
104        default=True,
105    )
106    default_token_duration = models.TextField(
107        help_text=_("Default token duration"),
108        default=DEFAULT_TOKEN_DURATION,
109        validators=[timedelta_string_validator],
110    )
111    default_token_length = models.PositiveIntegerField(
112        help_text=_("Default token length"),
113        default=DEFAULT_TOKEN_LENGTH,
114        validators=[MinValueValidator(1)],
115    )
116
117    pagination_default_page_size = models.PositiveIntegerField(
118        help_text=_("Default page size for API responses, if no size was requested."),
119        default=20,
120    )
121    pagination_max_page_size = models.PositiveIntegerField(
122        help_text=_("Maximum page size"),
123        default=100,
124    )
125
126    flags = models.JSONField(default=dict)
127
128    def save(self, *args, **kwargs):
129        if self.schema_name == get_tenant_base_schema() and not settings.TEST:
130            raise IntegrityError(f"Cannot create schema named {self.schema_name}")
131        super().save(*args, **kwargs)
132
133    @property
134    def serializer(self) -> Serializer:
135        from authentik.tenants.api.tenants import TenantSerializer
136
137        return TenantSerializer
138
139    def __str__(self) -> str:
140        return f"Tenant {self.name}"
141
142    class Meta:
143        verbose_name = _("Tenant")
144        verbose_name_plural = _("Tenants")
145
146
147class Domain(DomainMixin, SerializerModel):
148    """Tenant domain"""
149
150    tenant = models.ForeignKey(
151        Tenant, db_index=True, related_name="domains", on_delete=models.CASCADE
152    )
153
154    def __str__(self) -> str:
155        return f"Domain {self.domain}"
156
157    @property
158    def serializer(self) -> Serializer:
159        from authentik.tenants.api.domains import DomainSerializer
160
161        return DomainSerializer
162
163    class Meta:
164        verbose_name = _("Domain")
165        verbose_name_plural = _("Domains")
166
167
168@receiver(post_schema_sync, sender=TenantMixin)
169def tenant_needs_sync(sender, tenant, **kwargs):
170    """Reconcile apps for a specific tenant on creation"""
171    if tenant.ready:
172        return
173
174    with tenant:
175        for app in apps.get_app_configs():
176            if isinstance(app, ManagedAppConfig):
177                app._reconcile(ManagedAppConfig.RECONCILE_TENANT_CATEGORY)
178
179    tenant.ready = True
180    tenant.save()
LOGGER = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
VALID_SCHEMA_NAME = re.compile('^t_[a-z0-9]{1,61}$')
DEFAULT_TOKEN_DURATION = 'days=1'
DEFAULT_TOKEN_LENGTH = 60
DEFAULT_REPUTATION_LOWER_LIMIT = -5
DEFAULT_REPUTATION_UPPER_LIMIT = 5
class Tenant(authentik.lib.models.InternallyManagedMixin, django_tenants.models.TenantMixin, authentik.lib.models.SerializerModel): 
 45class Tenant(InternallyManagedMixin, TenantMixin, SerializerModel):
 46    """Tenant"""
 47
 48    tenant_uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
 49    schema_name = models.CharField(
 50        max_length=63, unique=True, db_index=True, validators=[_validate_schema_name]
 51    )
 52    name = models.TextField()
 53
 54    auto_create_schema = True
 55    auto_drop_schema = True
 56    ready = models.BooleanField(default=False)
 57
 58    avatars = models.TextField(
 59        help_text=_("Configure how authentik should show avatars for users."),
 60        default="gravatar,initials",
 61    )
 62    default_user_change_name = models.BooleanField(
 63        help_text=_("Enable the ability for users to change their name."), default=True
 64    )
 65    default_user_change_email = models.BooleanField(
 66        help_text=_("Enable the ability for users to change their email address."), default=False
 67    )
 68    default_user_change_username = models.BooleanField(
 69        help_text=_("Enable the ability for users to change their username."), default=False
 70    )
 71    event_retention = models.TextField(
 72        default="days=365",
 73        validators=[timedelta_string_validator],
 74        help_text=_(
 75            "Events will be deleted after this duration.(Format: weeks=3;days=2;hours=3,seconds=2)."
 76        ),
 77    )
 78    reputation_lower_limit = models.IntegerField(
 79        help_text=_("Reputation cannot decrease lower than this value. Zero or negative."),
 80        default=DEFAULT_REPUTATION_LOWER_LIMIT,
 81        validators=[MaxValueValidator(0)],
 82    )
 83    reputation_upper_limit = models.IntegerField(
 84        help_text=_("Reputation cannot increase higher than this value. Zero or positive."),
 85        default=DEFAULT_REPUTATION_UPPER_LIMIT,
 86        validators=[MinValueValidator(0)],
 87    )
 88    footer_links = models.JSONField(
 89        help_text=_("The option configures the footer links on the flow executor pages."),
 90        default=list,
 91        blank=True,
 92    )
 93    gdpr_compliance = models.BooleanField(
 94        help_text=_(
 95            "When enabled, all the events caused by a user "
 96            "will be deleted upon the user's deletion."
 97        ),
 98        default=True,
 99    )
100    impersonation = models.BooleanField(
101        help_text=_("Globally enable/disable impersonation."), default=True
102    )
103    impersonation_require_reason = models.BooleanField(
104        help_text=_("Require administrators to provide a reason for impersonating a user."),
105        default=True,
106    )
107    default_token_duration = models.TextField(
108        help_text=_("Default token duration"),
109        default=DEFAULT_TOKEN_DURATION,
110        validators=[timedelta_string_validator],
111    )
112    default_token_length = models.PositiveIntegerField(
113        help_text=_("Default token length"),
114        default=DEFAULT_TOKEN_LENGTH,
115        validators=[MinValueValidator(1)],
116    )
117
118    pagination_default_page_size = models.PositiveIntegerField(
119        help_text=_("Default page size for API responses, if no size was requested."),
120        default=20,
121    )
122    pagination_max_page_size = models.PositiveIntegerField(
123        help_text=_("Maximum page size"),
124        default=100,
125    )
126
127    flags = models.JSONField(default=dict)
128
129    def save(self, *args, **kwargs):
130        if self.schema_name == get_tenant_base_schema() and not settings.TEST:
131            raise IntegrityError(f"Cannot create schema named {self.schema_name}")
132        super().save(*args, **kwargs)
133
134    @property
135    def serializer(self) -> Serializer:
136        from authentik.tenants.api.tenants import TenantSerializer
137
138        return TenantSerializer
139
140    def __str__(self) -> str:
141        return f"Tenant {self.name}"
142
143    class Meta:
144        verbose_name = _("Tenant")
145        verbose_name_plural = _("Tenants")

Tenant

def tenant_uuid(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def schema_name(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def name(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

auto_create_schema = True

Set this flag to false on a parent class if you don't want the schema to be automatically created upon save.

auto_drop_schema = True

USE THIS WITH CAUTION! Set this flag to true on a parent class if you want the schema to be automatically deleted if the tenant row gets deleted.

def ready(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def avatars(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def default_user_change_name(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def default_user_change_email(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def default_user_change_username(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def event_retention(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def reputation_lower_limit(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def reputation_upper_limit(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def gdpr_compliance(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def impersonation(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def impersonation_require_reason(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def default_token_duration(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def default_token_length(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def pagination_default_page_size(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def pagination_max_page_size(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def flags(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def save(self, *args, **kwargs): 
129    def save(self, *args, **kwargs):
130        if self.schema_name == get_tenant_base_schema() and not settings.TEST:
131            raise IntegrityError(f"Cannot create schema named {self.schema_name}")
132        super().save(*args, **kwargs)

Save the current instance. Override this in a subclass if you want to control the saving process.

The 'force_insert' and 'force_update' parameters can be used to insist that the "save" must be an SQL insert or update (or equivalent for non-SQL backends), respectively. Normally, they should not be set.

serializer: rest_framework.serializers.Serializer 
134    @property
135    def serializer(self) -> Serializer:
136        from authentik.tenants.api.tenants import TenantSerializer
137
138        return TenantSerializer

Get serializer for this model

def objects(unknown):

The type of the None singleton.

domains

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

task_set

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

class Tenant.DoesNotExist(django.core.exceptions.ObjectDoesNotExist):

The requested object does not exist

class Tenant.MultipleObjectsReturned(django.core.exceptions.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.

class Domain(django_tenants.models.DomainMixin, authentik.lib.models.SerializerModel): 
148class Domain(DomainMixin, SerializerModel):
149    """Tenant domain"""
150
151    tenant = models.ForeignKey(
152        Tenant, db_index=True, related_name="domains", on_delete=models.CASCADE
153    )
154
155    def __str__(self) -> str:
156        return f"Domain {self.domain}"
157
158    @property
159    def serializer(self) -> Serializer:
160        from authentik.tenants.api.domains import DomainSerializer
161
162        return DomainSerializer
163
164    class Meta:
165        verbose_name = _("Domain")
166        verbose_name_plural = _("Domains")

Tenant domain

tenant

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example::

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

serializer: rest_framework.serializers.Serializer 
158    @property
159    def serializer(self) -> Serializer:
160        from authentik.tenants.api.domains import DomainSerializer
161
162        return DomainSerializer

Get serializer for this model

def domain(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def is_primary(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

tenant_id
def id(unknown):

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

def objects(unknown):

The type of the None singleton.

class Domain.DoesNotExist(django.core.exceptions.ObjectDoesNotExist):

The requested object does not exist

class Domain.MultipleObjectsReturned(django.core.exceptions.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.

@receiver(post_schema_sync, sender=TenantMixin)
def tenant_needs_sync(sender, tenant, **kwargs): 
169@receiver(post_schema_sync, sender=TenantMixin)
170def tenant_needs_sync(sender, tenant, **kwargs):
171    """Reconcile apps for a specific tenant on creation"""
172    if tenant.ready:
173        return
174
175    with tenant:
176        for app in apps.get_app_configs():
177            if isinstance(app, ManagedAppConfig):
178                app._reconcile(ManagedAppConfig.RECONCILE_TENANT_CATEGORY)
179
180    tenant.ready = True
181    tenant.save()

Reconcile apps for a specific tenant on creation