authentik.stages.email.models

email stage models

  1"""email stage models"""
  2
  3from os import R_OK, access
  4from pathlib import Path
  5
  6from django.conf import settings
  7from django.core.mail.backends.base import BaseEmailBackend
  8from django.core.mail.backends.smtp import EmailBackend
  9from django.db import models
 10from django.utils.translation import gettext as _
 11from django.views import View
 12from rest_framework.serializers import BaseSerializer
 13from structlog.stdlib import get_logger
 14
 15from authentik.flows.models import Stage
 16from authentik.lib.config import CONFIG
 17from authentik.lib.utils.time import timedelta_string_validator
 18
 19EMAIL_RECOVERY_MAX_ATTEMPTS = 5
 20
 21LOGGER = get_logger()
 22
 23
 24class EmailTemplates(models.TextChoices):
 25    """Templates used for rendering the Email"""
 26
 27    PASSWORD_RESET = (
 28        "email/password_reset.html",
 29        _("Password Reset"),
 30    )  # nosec
 31    ACCOUNT_CONFIRM = (
 32        "email/account_confirmation.html",
 33        _("Account Confirmation"),
 34    )
 35    EMAIL_OTP = (
 36        "email/email_otp.html",
 37        _("Email OTP"),
 38    )  # nosec
 39    EVENT_NOTIFICATION = (
 40        "email/event_notification.html",
 41        _("Event Notification"),
 42    )
 43    INVITATION = (
 44        "email/invitation.html",
 45        _("Invitation"),
 46    )
 47
 48
 49def get_template_choices():
 50    """Get all available Email templates, including dynamically mounted ones.
 51    Directories are taken from TEMPLATES.DIR setting"""
 52    static_choices = EmailTemplates.choices
 53
 54    dirs = [Path(x) for x in settings.TEMPLATES[0]["DIRS"]]
 55    for template_dir in dirs:
 56        if not template_dir.exists() or not template_dir.is_dir():
 57            continue
 58        for template in template_dir.glob("**/*.html"):
 59            path = str(template)
 60            if not access(path, R_OK):
 61                LOGGER.warning("Custom template file is not readable, check permissions", path=path)
 62                continue
 63            rel_path = template.relative_to(template_dir)
 64            static_choices.append((str(rel_path), f"Custom Template: {rel_path}"))
 65    return static_choices
 66
 67
 68class EmailStage(Stage):
 69    """Send an Email to the user with a token to confirm their Email address."""
 70
 71    use_global_settings = models.BooleanField(
 72        default=False,
 73        help_text=_(
 74            "When enabled, global Email connection settings will be used and "
 75            "connection settings below will be ignored."
 76        ),
 77    )
 78
 79    host = models.TextField(default="localhost")
 80    port = models.IntegerField(default=25)
 81    username = models.TextField(default="", blank=True)
 82    password = models.TextField(default="", blank=True)
 83    use_tls = models.BooleanField(default=False)
 84    use_ssl = models.BooleanField(default=False)
 85    timeout = models.IntegerField(default=10)
 86    from_address = models.EmailField(default="system@authentik.local")
 87    recovery_max_attempts = models.PositiveIntegerField(default=EMAIL_RECOVERY_MAX_ATTEMPTS)
 88    recovery_cache_timeout = models.TextField(
 89        default="minutes=5",
 90        validators=[timedelta_string_validator],
 91        help_text=_(
 92            "The time window used to count recent account recovery attempts. "
 93            "If the number of attempts exceed recovery_max_attempts within "
 94            "this period, further attempts will be rate-limited. "
 95            "(Format: hours=1;minutes=2;seconds=3)."
 96        ),
 97    )
 98
 99    activate_user_on_success = models.BooleanField(
100        default=False, help_text=_("Activate users upon completion of stage.")
101    )
102
103    token_expiry = models.TextField(
104        default="minutes=30",
105        validators=[timedelta_string_validator],
106        help_text=_("Time the token sent is valid (Format: hours=3,minutes=17,seconds=300)."),
107    )
108    subject = models.TextField(default="authentik")
109    template = models.TextField(default=EmailTemplates.PASSWORD_RESET)
110
111    @property
112    def serializer(self) -> type[BaseSerializer]:
113        from authentik.stages.email.api import EmailStageSerializer
114
115        return EmailStageSerializer
116
117    @property
118    def view(self) -> type[View]:
119        from authentik.stages.email.stage import EmailStageView
120
121        return EmailStageView
122
123    @property
124    def component(self) -> str:
125        return "ak-stage-email-form"
126
127    @property
128    def backend_class(self) -> type[BaseEmailBackend]:
129        """Get the email backend class to use"""
130        return EmailBackend
131
132    @property
133    def backend(self) -> BaseEmailBackend:
134        """Get fully configured Email Backend instance"""
135        if self.use_global_settings:
136            CONFIG.refresh("email.password")
137            return self.backend_class(
138                host=CONFIG.get("email.host"),
139                port=CONFIG.get_int("email.port"),
140                username=CONFIG.get("email.username"),
141                password=CONFIG.get("email.password"),
142                use_tls=CONFIG.get_bool("email.use_tls", False),
143                use_ssl=CONFIG.get_bool("email.use_ssl", False),
144                timeout=CONFIG.get_int("email.timeout"),
145            )
146        return self.backend_class(
147            host=self.host,
148            port=self.port,
149            username=self.username,
150            password=self.password,
151            use_tls=self.use_tls,
152            use_ssl=self.use_ssl,
153            timeout=self.timeout,
154        )
155
156    def __str__(self):
157        return f"Email Stage {self.name}"
158
159    class Meta:
160        verbose_name = _("Email Stage")
161        verbose_name_plural = _("Email Stages")
EMAIL_RECOVERY_MAX_ATTEMPTS = 5
LOGGER = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
class EmailTemplates(django.db.models.enums.TextChoices):
25class EmailTemplates(models.TextChoices):
26    """Templates used for rendering the Email"""
27
28    PASSWORD_RESET = (
29        "email/password_reset.html",
30        _("Password Reset"),
31    )  # nosec
32    ACCOUNT_CONFIRM = (
33        "email/account_confirmation.html",
34        _("Account Confirmation"),
35    )
36    EMAIL_OTP = (
37        "email/email_otp.html",
38        _("Email OTP"),
39    )  # nosec
40    EVENT_NOTIFICATION = (
41        "email/event_notification.html",
42        _("Event Notification"),
43    )
44    INVITATION = (
45        "email/invitation.html",
46        _("Invitation"),
47    )

Templates used for rendering the Email

ACCOUNT_CONFIRM = EmailTemplates.ACCOUNT_CONFIRM
EVENT_NOTIFICATION = EmailTemplates.EVENT_NOTIFICATION
def get_template_choices():
50def get_template_choices():
51    """Get all available Email templates, including dynamically mounted ones.
52    Directories are taken from TEMPLATES.DIR setting"""
53    static_choices = EmailTemplates.choices
54
55    dirs = [Path(x) for x in settings.TEMPLATES[0]["DIRS"]]
56    for template_dir in dirs:
57        if not template_dir.exists() or not template_dir.is_dir():
58            continue
59        for template in template_dir.glob("**/*.html"):
60            path = str(template)
61            if not access(path, R_OK):
62                LOGGER.warning("Custom template file is not readable, check permissions", path=path)
63                continue
64            rel_path = template.relative_to(template_dir)
65            static_choices.append((str(rel_path), f"Custom Template: {rel_path}"))
66    return static_choices

Get all available Email templates, including dynamically mounted ones. Directories are taken from TEMPLATES.DIR setting

class EmailStage(authentik.flows.models.Stage):
 69class EmailStage(Stage):
 70    """Send an Email to the user with a token to confirm their Email address."""
 71
 72    use_global_settings = models.BooleanField(
 73        default=False,
 74        help_text=_(
 75            "When enabled, global Email connection settings will be used and "
 76            "connection settings below will be ignored."
 77        ),
 78    )
 79
 80    host = models.TextField(default="localhost")
 81    port = models.IntegerField(default=25)
 82    username = models.TextField(default="", blank=True)
 83    password = models.TextField(default="", blank=True)
 84    use_tls = models.BooleanField(default=False)
 85    use_ssl = models.BooleanField(default=False)
 86    timeout = models.IntegerField(default=10)
 87    from_address = models.EmailField(default="system@authentik.local")
 88    recovery_max_attempts = models.PositiveIntegerField(default=EMAIL_RECOVERY_MAX_ATTEMPTS)
 89    recovery_cache_timeout = models.TextField(
 90        default="minutes=5",
 91        validators=[timedelta_string_validator],
 92        help_text=_(
 93            "The time window used to count recent account recovery attempts. "
 94            "If the number of attempts exceed recovery_max_attempts within "
 95            "this period, further attempts will be rate-limited. "
 96            "(Format: hours=1;minutes=2;seconds=3)."
 97        ),
 98    )
 99
100    activate_user_on_success = models.BooleanField(
101        default=False, help_text=_("Activate users upon completion of stage.")
102    )
103
104    token_expiry = models.TextField(
105        default="minutes=30",
106        validators=[timedelta_string_validator],
107        help_text=_("Time the token sent is valid (Format: hours=3,minutes=17,seconds=300)."),
108    )
109    subject = models.TextField(default="authentik")
110    template = models.TextField(default=EmailTemplates.PASSWORD_RESET)
111
112    @property
113    def serializer(self) -> type[BaseSerializer]:
114        from authentik.stages.email.api import EmailStageSerializer
115
116        return EmailStageSerializer
117
118    @property
119    def view(self) -> type[View]:
120        from authentik.stages.email.stage import EmailStageView
121
122        return EmailStageView
123
124    @property
125    def component(self) -> str:
126        return "ak-stage-email-form"
127
128    @property
129    def backend_class(self) -> type[BaseEmailBackend]:
130        """Get the email backend class to use"""
131        return EmailBackend
132
133    @property
134    def backend(self) -> BaseEmailBackend:
135        """Get fully configured Email Backend instance"""
136        if self.use_global_settings:
137            CONFIG.refresh("email.password")
138            return self.backend_class(
139                host=CONFIG.get("email.host"),
140                port=CONFIG.get_int("email.port"),
141                username=CONFIG.get("email.username"),
142                password=CONFIG.get("email.password"),
143                use_tls=CONFIG.get_bool("email.use_tls", False),
144                use_ssl=CONFIG.get_bool("email.use_ssl", False),
145                timeout=CONFIG.get_int("email.timeout"),
146            )
147        return self.backend_class(
148            host=self.host,
149            port=self.port,
150            username=self.username,
151            password=self.password,
152            use_tls=self.use_tls,
153            use_ssl=self.use_ssl,
154            timeout=self.timeout,
155        )
156
157    def __str__(self):
158        return f"Email Stage {self.name}"
159
160    class Meta:
161        verbose_name = _("Email Stage")
162        verbose_name_plural = _("Email Stages")

Send an Email to the user with a token to confirm their Email address.

def use_global_settings(unknown):

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

def host(unknown):

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

def port(unknown):

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

def 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 password(unknown):

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

def use_tls(unknown):

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

def use_ssl(unknown):

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

def timeout(unknown):

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

def from_address(unknown):

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

def recovery_max_attempts(unknown):

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

def recovery_cache_timeout(unknown):

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

def activate_user_on_success(unknown):

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

def token_expiry(unknown):

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

def subject(unknown):

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

def template(unknown):

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

serializer: type[rest_framework.serializers.BaseSerializer]
112    @property
113    def serializer(self) -> type[BaseSerializer]:
114        from authentik.stages.email.api import EmailStageSerializer
115
116        return EmailStageSerializer

Get serializer for this model

view: type[django.views.generic.base.View]
118    @property
119    def view(self) -> type[View]:
120        from authentik.stages.email.stage import EmailStageView
121
122        return EmailStageView

Return StageView class that implements logic for this stage

component: str
124    @property
125    def component(self) -> str:
126        return "ak-stage-email-form"

Return component used to edit this object

backend_class: type[django.core.mail.backends.base.BaseEmailBackend]
128    @property
129    def backend_class(self) -> type[BaseEmailBackend]:
130        """Get the email backend class to use"""
131        return EmailBackend

Get the email backend class to use

backend: django.core.mail.backends.base.BaseEmailBackend
133    @property
134    def backend(self) -> BaseEmailBackend:
135        """Get fully configured Email Backend instance"""
136        if self.use_global_settings:
137            CONFIG.refresh("email.password")
138            return self.backend_class(
139                host=CONFIG.get("email.host"),
140                port=CONFIG.get_int("email.port"),
141                username=CONFIG.get("email.username"),
142                password=CONFIG.get("email.password"),
143                use_tls=CONFIG.get_bool("email.use_tls", False),
144                use_ssl=CONFIG.get_bool("email.use_ssl", False),
145                timeout=CONFIG.get_int("email.timeout"),
146            )
147        return self.backend_class(
148            host=self.host,
149            port=self.port,
150            username=self.username,
151            password=self.password,
152            use_tls=self.use_tls,
153            use_ssl=self.use_ssl,
154            timeout=self.timeout,
155        )

Get fully configured Email Backend instance

stage_ptr_id
stage_ptr

Accessor to the related object on the forward side of a one-to-one relation.

In the example::

class Restaurant(Model):
    place = OneToOneField(Place, related_name='restaurant')

Restaurant.place is a ForwardOneToOneDescriptor instance.

class EmailStage.DoesNotExist(authentik.flows.models.Stage.DoesNotExist):

The requested object does not exist

class EmailStage.MultipleObjectsReturned(authentik.flows.models.Stage.MultipleObjectsReturned):

The query returned multiple objects when only one was expected.