authentik.lib.models

Generic models

  1"""Generic models"""
  2
  3import re
  4
  5from django.core.validators import URLValidator
  6from django.db import models
  7from django.utils.regex_helper import _lazy_re_compile
  8from model_utils.managers import InheritanceManager
  9from rest_framework.serializers import BaseSerializer
 10
 11
 12class SerializerModel(models.Model):
 13    """Base Abstract Model which has a serializer"""
 14
 15    class Meta:
 16        abstract = True
 17
 18    @property
 19    def serializer(self) -> type[BaseSerializer]:
 20        """Get serializer for this model"""
 21        # Special handling for built-in source
 22        if (
 23            hasattr(self, "managed")
 24            and hasattr(self, "MANAGED_INBUILT")
 25            and self.managed == self.MANAGED_INBUILT
 26        ):
 27            from authentik.core.api.sources import SourceSerializer
 28
 29            return SourceSerializer
 30        raise NotImplementedError
 31
 32
 33class CreatedUpdatedModel(models.Model):
 34    """Base Abstract Model to save created and update"""
 35
 36    created = models.DateTimeField(auto_now_add=True)
 37    last_updated = models.DateTimeField(auto_now=True)
 38
 39    class Meta:
 40        abstract = True
 41
 42
 43class InheritanceAutoManager(InheritanceManager):
 44    """Object manager which automatically selects the subclass"""
 45
 46    def get_queryset(self):
 47        return super().get_queryset().select_subclasses()
 48
 49
 50class InheritanceForwardManyToOneDescriptor(models.fields.related.ForwardManyToOneDescriptor):
 51    """Forward ManyToOne Descriptor that selects subclass. Requires InheritanceAutoManager."""
 52
 53    def get_queryset(self, **hints):
 54        return self.field.remote_field.model.objects.db_manager(hints=hints).select_subclasses()
 55
 56
 57class InheritanceForeignKey(models.ForeignKey):
 58    """Custom ForeignKey that uses InheritanceForwardManyToOneDescriptor"""
 59
 60    forward_related_accessor_class = InheritanceForwardManyToOneDescriptor
 61
 62
 63class DeprecatedMixin:
 64    """Mixin for classes that are deprecated"""
 65
 66
 67class InternallyManagedMixin:
 68    """Mixin for models that should _not_ be manageable via blueprint."""
 69
 70
 71class DomainlessURLValidator(URLValidator):
 72    """Subclass of URLValidator which doesn't check the domain
 73    (to allow hostnames without domain)"""
 74
 75    def __init__(self, *args, **kwargs) -> None:
 76        super().__init__(*args, **kwargs)
 77        self.host_re = "(" + self.hostname_re + self.domain_re + "|localhost)"
 78        self.regex = _lazy_re_compile(
 79            r"^(?:[a-z0-9.+-]*)://"  # scheme is validated separately
 80            r"(?:[^\s:@/]+(?::[^\s:@/]*)?@)?"  # user:pass authentication
 81            r"(?:" + self.ipv4_re + "|" + self.ipv6_re + "|" + self.host_re + ")"
 82            r"(?::\d{1,5})?"  # port
 83            r"(?:[/?#][^\s]*)?"  # resource path
 84            r"\Z",
 85            re.IGNORECASE,
 86        )
 87        self.schemes = ["http", "https", "blank"] + list(self.schemes)
 88
 89    def __call__(self, value: str):
 90        # Check if the scheme is valid.
 91        scheme = value.split("://", maxsplit=1)[0].lower()
 92        if scheme not in self.schemes:
 93            value = "default" + value
 94        super().__call__(value)
 95
 96
 97class DomainlessFormattedURLValidator(DomainlessURLValidator):
 98    """URL validator which allows for python format strings"""
 99
100    def __init__(self, *args, **kwargs) -> None:
101        super().__init__(*args, **kwargs)
102        self.formatter_re = r"([%\(\)a-zA-Z])*"
103        self.host_re = "(" + self.formatter_re + self.hostname_re + self.domain_re + "|localhost)"
104        self.regex = _lazy_re_compile(
105            r"^(?:[a-z0-9.+-]*)://"  # scheme is validated separately
106            r"(?:[^\s:@/]+(?::[^\s:@/]*)?@)?"  # user:pass authentication
107            r"(?:" + self.ipv4_re + "|" + self.ipv6_re + "|" + self.host_re + ")"
108            r"(?::\d{1,5})?"  # port
109            r"(?:[/?#][^\s]*)?"  # resource path
110            r"\Z",
111            re.IGNORECASE,
112        )
113        self.schemes = ["http", "https", "blank", "ssh", "sftp"] + list(self.schemes)
class SerializerModel(django.db.models.base.Model):
13class SerializerModel(models.Model):
14    """Base Abstract Model which has a serializer"""
15
16    class Meta:
17        abstract = True
18
19    @property
20    def serializer(self) -> type[BaseSerializer]:
21        """Get serializer for this model"""
22        # Special handling for built-in source
23        if (
24            hasattr(self, "managed")
25            and hasattr(self, "MANAGED_INBUILT")
26            and self.managed == self.MANAGED_INBUILT
27        ):
28            from authentik.core.api.sources import SourceSerializer
29
30            return SourceSerializer
31        raise NotImplementedError

Base Abstract Model which has a serializer

serializer: type[rest_framework.serializers.BaseSerializer]
19    @property
20    def serializer(self) -> type[BaseSerializer]:
21        """Get serializer for this model"""
22        # Special handling for built-in source
23        if (
24            hasattr(self, "managed")
25            and hasattr(self, "MANAGED_INBUILT")
26            and self.managed == self.MANAGED_INBUILT
27        ):
28            from authentik.core.api.sources import SourceSerializer
29
30            return SourceSerializer
31        raise NotImplementedError

Get serializer for this model

class SerializerModel.Meta:
16    class Meta:
17        abstract = True
abstract = False
class CreatedUpdatedModel(django.db.models.base.Model):
34class CreatedUpdatedModel(models.Model):
35    """Base Abstract Model to save created and update"""
36
37    created = models.DateTimeField(auto_now_add=True)
38    last_updated = models.DateTimeField(auto_now=True)
39
40    class Meta:
41        abstract = True

Base Abstract Model to save created and update

def created(unknown):

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

def last_updated(unknown):

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

def get_next_by_created(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

def get_previous_by_created(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

def get_next_by_last_updated(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

def get_previous_by_last_updated(unknown):

Method descriptor with partial application of the given arguments and keywords.

Supports wrapping existing descriptors and handles non-descriptor callables as instance methods.

class CreatedUpdatedModel.Meta:
40    class Meta:
41        abstract = True
abstract = False
class InheritanceAutoManager(model_utils.managers.InheritanceManagerMixin[+ModelT], django.db.models.manager.Manager):
44class InheritanceAutoManager(InheritanceManager):
45    """Object manager which automatically selects the subclass"""
46
47    def get_queryset(self):
48        return super().get_queryset().select_subclasses()

Object manager which automatically selects the subclass

def get_queryset(self):
47    def get_queryset(self):
48        return super().get_queryset().select_subclasses()

Return a new QuerySet object. Subclasses can override this method to customize the behavior of the Manager.

class InheritanceForwardManyToOneDescriptor(django.db.models.fields.related_descriptors.ForwardManyToOneDescriptor):
51class InheritanceForwardManyToOneDescriptor(models.fields.related.ForwardManyToOneDescriptor):
52    """Forward ManyToOne Descriptor that selects subclass. Requires InheritanceAutoManager."""
53
54    def get_queryset(self, **hints):
55        return self.field.remote_field.model.objects.db_manager(hints=hints).select_subclasses()

Forward ManyToOne Descriptor that selects subclass. Requires InheritanceAutoManager.

def get_queryset(self, **hints):
54    def get_queryset(self, **hints):
55        return self.field.remote_field.model.objects.db_manager(hints=hints).select_subclasses()
class InheritanceForeignKey(django.db.models.fields.related.ForeignKey):
58class InheritanceForeignKey(models.ForeignKey):
59    """Custom ForeignKey that uses InheritanceForwardManyToOneDescriptor"""
60
61    forward_related_accessor_class = InheritanceForwardManyToOneDescriptor

Custom ForeignKey that uses InheritanceForwardManyToOneDescriptor

class DeprecatedMixin:
64class DeprecatedMixin:
65    """Mixin for classes that are deprecated"""

Mixin for classes that are deprecated

class InternallyManagedMixin:
68class InternallyManagedMixin:
69    """Mixin for models that should _not_ be manageable via blueprint."""

Mixin for models that should _not_ be manageable via blueprint.

class DomainlessURLValidator(django.core.validators.URLValidator):
72class DomainlessURLValidator(URLValidator):
73    """Subclass of URLValidator which doesn't check the domain
74    (to allow hostnames without domain)"""
75
76    def __init__(self, *args, **kwargs) -> None:
77        super().__init__(*args, **kwargs)
78        self.host_re = "(" + self.hostname_re + self.domain_re + "|localhost)"
79        self.regex = _lazy_re_compile(
80            r"^(?:[a-z0-9.+-]*)://"  # scheme is validated separately
81            r"(?:[^\s:@/]+(?::[^\s:@/]*)?@)?"  # user:pass authentication
82            r"(?:" + self.ipv4_re + "|" + self.ipv6_re + "|" + self.host_re + ")"
83            r"(?::\d{1,5})?"  # port
84            r"(?:[/?#][^\s]*)?"  # resource path
85            r"\Z",
86            re.IGNORECASE,
87        )
88        self.schemes = ["http", "https", "blank"] + list(self.schemes)
89
90    def __call__(self, value: str):
91        # Check if the scheme is valid.
92        scheme = value.split("://", maxsplit=1)[0].lower()
93        if scheme not in self.schemes:
94            value = "default" + value
95        super().__call__(value)

Subclass of URLValidator which doesn't check the domain (to allow hostnames without domain)

DomainlessURLValidator(*args, **kwargs)
76    def __init__(self, *args, **kwargs) -> None:
77        super().__init__(*args, **kwargs)
78        self.host_re = "(" + self.hostname_re + self.domain_re + "|localhost)"
79        self.regex = _lazy_re_compile(
80            r"^(?:[a-z0-9.+-]*)://"  # scheme is validated separately
81            r"(?:[^\s:@/]+(?::[^\s:@/]*)?@)?"  # user:pass authentication
82            r"(?:" + self.ipv4_re + "|" + self.ipv6_re + "|" + self.host_re + ")"
83            r"(?::\d{1,5})?"  # port
84            r"(?:[/?#][^\s]*)?"  # resource path
85            r"\Z",
86            re.IGNORECASE,
87        )
88        self.schemes = ["http", "https", "blank"] + list(self.schemes)
host_re = '([a-z¡-\uffff0-9](?:[a-z¡-\uffff0-9-]{0,61}[a-z¡-\uffff0-9])?(?:\\.(?!-)[a-z¡-\uffff0-9-]{1,63}(?<!-))*\\.(?!-)(?:[a-z¡-\uffff-]{2,63}|xn--[a-z0-9]{1,59})(?<!-)\\.?|localhost)'
regex = <SimpleLazyObject: re.compile('^(?:[a-z0-9.+-]*)://(?:[^\\s:@/]+(?::[^\\s:@/]*)?@)?(?:(?:0|25[0-5]|2[0-4][0-9]|1[0-9]?[0-9]?|[1-9][0-9]?)(?:\\.(?:0|25[0-5]|2[0-4][0-9]|1[0-9]?[0-9]?|[1-9][0-9]?)){3}|\\[[0-9a-f:.]+\\]|([a-z¡-\uffff, re.IGNORECASE)>
schemes = ['http', 'https', 'ftp', 'ftps']
class DomainlessFormattedURLValidator(DomainlessURLValidator):
 98class DomainlessFormattedURLValidator(DomainlessURLValidator):
 99    """URL validator which allows for python format strings"""
100
101    def __init__(self, *args, **kwargs) -> None:
102        super().__init__(*args, **kwargs)
103        self.formatter_re = r"([%\(\)a-zA-Z])*"
104        self.host_re = "(" + self.formatter_re + self.hostname_re + self.domain_re + "|localhost)"
105        self.regex = _lazy_re_compile(
106            r"^(?:[a-z0-9.+-]*)://"  # scheme is validated separately
107            r"(?:[^\s:@/]+(?::[^\s:@/]*)?@)?"  # user:pass authentication
108            r"(?:" + self.ipv4_re + "|" + self.ipv6_re + "|" + self.host_re + ")"
109            r"(?::\d{1,5})?"  # port
110            r"(?:[/?#][^\s]*)?"  # resource path
111            r"\Z",
112            re.IGNORECASE,
113        )
114        self.schemes = ["http", "https", "blank", "ssh", "sftp"] + list(self.schemes)

URL validator which allows for python format strings

DomainlessFormattedURLValidator(*args, **kwargs)
101    def __init__(self, *args, **kwargs) -> None:
102        super().__init__(*args, **kwargs)
103        self.formatter_re = r"([%\(\)a-zA-Z])*"
104        self.host_re = "(" + self.formatter_re + self.hostname_re + self.domain_re + "|localhost)"
105        self.regex = _lazy_re_compile(
106            r"^(?:[a-z0-9.+-]*)://"  # scheme is validated separately
107            r"(?:[^\s:@/]+(?::[^\s:@/]*)?@)?"  # user:pass authentication
108            r"(?:" + self.ipv4_re + "|" + self.ipv6_re + "|" + self.host_re + ")"
109            r"(?::\d{1,5})?"  # port
110            r"(?:[/?#][^\s]*)?"  # resource path
111            r"\Z",
112            re.IGNORECASE,
113        )
114        self.schemes = ["http", "https", "blank", "ssh", "sftp"] + list(self.schemes)
formatter_re
host_re = '([a-z¡-\uffff0-9](?:[a-z¡-\uffff0-9-]{0,61}[a-z¡-\uffff0-9])?(?:\\.(?!-)[a-z¡-\uffff0-9-]{1,63}(?<!-))*\\.(?!-)(?:[a-z¡-\uffff-]{2,63}|xn--[a-z0-9]{1,59})(?<!-)\\.?|localhost)'
regex = <SimpleLazyObject: re.compile('^(?:[a-z0-9.+-]*)://(?:[^\\s:@/]+(?::[^\\s:@/]*)?@)?(?:(?:0|25[0-5]|2[0-4][0-9]|1[0-9]?[0-9]?|[1-9][0-9]?)(?:\\.(?:0|25[0-5]|2[0-4][0-9]|1[0-9]?[0-9]?|[1-9][0-9]?)){3}|\\[[0-9a-f:.]+\\]|([a-z¡-\uffff, re.IGNORECASE)>
schemes = ['http', 'https', 'ftp', 'ftps']