habsi/authentik
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

email tasks migration

Browse Source 
Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>
This commit is contained in:
Marc 'risson' Schmitt
2025-03-28 18:55:05 +01:00
parent 1ed6cf7517
commit 04be734c49
5 changed files with 166 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
authentik/root/test_runner.py
View File

@ -6,6 +6,7 @@ from unittest import TestCase
from django.apps import apps
from django.test.testcases import apps
from django.test.utils import setup_test_environment, teardown_test_environment
import pytest
from django.conf import settings
from django.test.runner import DiscoverRunner

284
authentik/stages/authenticator_email/tests.py
View File

@ -1,10 +1,11 @@
"""Test Email Authenticator API"""
from datetime import timedelta
from unittest.mock import MagicMock, PropertyMock, patch
from unittest.mock import PropertyMock, patch
from django.core import mail
from django.core.mail.backends.smtp import EmailBackend
from django.core.mail.backends.locmem import EmailBackend
from django.core.mail.backends.smtp import EmailBackend as SMTPEmailBackend
from django.db.utils import IntegrityError
from django.template.exceptions import TemplateDoesNotExist
from django.urls import reverse
@ -83,24 +84,28 @@ class TestAuthenticatorEmailStage(FlowTestCase):
self.assertTrue(self.device.verify_token(token))
self.assertIsNone(self.device.token)
@patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_stage_no_prefill(self):
"""Test stage without prefilled email"""
self.client.force_login(self.user_noemail)
with patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
):
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
self.assertStageResponse(
response,
self.flow,
self.user_noemail,
component="ak-stage-authenticator-email",
email_required=True,
)
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
self.assertStageResponse(
response,
self.flow,
self.user_noemail,
component="ak-stage-authenticator-email",
email_required=True,
)
@patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_stage_submit(self):
"""Test stage email submission"""
# Initialize the flow
@ -115,34 +120,18 @@ class TestAuthenticatorEmailStage(FlowTestCase):
email_required=False,
)
# Test email submission with locmem backend
def mock_send_mails(stage, *messages):
"""Mock send_mails to send directly"""
for message in messages:
message.send()
with (
patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
return_value=EmailBackend,
),
patch(
"authentik.stages.authenticator_email.stage.send_mails",
side_effect=mock_send_mails,
),
):
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email", "email": "test@example.com"},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(mail.outbox), 1)
sent_mail = mail.outbox[0]
self.assertEqual(sent_mail.subject, self.stage.subject)
self.assertEqual(sent_mail.to, [f"{self.user} <test@example.com>"])
# Get from_address from global email config to test if global settings are being used
from_address_global = CONFIG.get("email.from")
self.assertEqual(sent_mail.from_email, from_address_global)
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email", "email": "test@example.com"},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(mail.outbox), 2)
sent_mail = mail.outbox[1]
self.assertEqual(sent_mail.subject, self.stage.subject)
self.assertEqual(sent_mail.to, [f"{self.user} <test@example.com>"])
# Get from_address from global email config to test if global settings are being used
from_address_global = CONFIG.get("email.from")
self.assertEqual(sent_mail.from_email, from_address_global)
self.assertStageResponse(
response,
@ -196,109 +185,109 @@ class TestAuthenticatorEmailStage(FlowTestCase):
with self.assertRaises(TemplateDoesNotExist):
self.stage.send(self.device)
@patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_challenge_response_validation(self):
"""Test challenge response validation"""
# Initialize the flow
self.client.force_login(self.user_noemail)
with patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
):
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
# Test missing code and email
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email"},
)
self.assertIn("email required", str(response.content))
# Test missing code and email
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email"},
)
self.assertIn("email required", str(response.content))
# Test invalid code
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email", "code": "000000"},
)
self.assertIn("Code does not match", str(response.content))
# Test invalid code
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email", "code": "000000"},
)
self.assertIn("Code does not match", str(response.content))
# Test valid code
self.client.force_login(self.user)
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
device = self.device
token = device.token
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email", "code": token},
)
self.assertEqual(response.status_code, 200)
self.assertTrue(device.confirmed)
# Test valid code
self.client.force_login(self.user)
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
device = self.device
token = device.token
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email", "code": token},
)
self.assertEqual(response.status_code, 200)
self.assertTrue(device.confirmed)
@patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_challenge_generation(self):
"""Test challenge generation"""
# Test with masked email
with patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
):
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
self.assertStageResponse(
response,
self.flow,
self.user,
component="ak-stage-authenticator-email",
email_required=False,
)
masked_email = mask_email(self.user.email)
self.assertEqual(masked_email, response.json()["email"])
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
self.assertStageResponse(
response,
self.flow,
self.user,
component="ak-stage-authenticator-email",
email_required=False,
)
masked_email = mask_email(self.user.email)
self.assertEqual(masked_email, response.json()["email"])
# Test without email
self.client.force_login(self.user_noemail)
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
self.assertStageResponse(
response,
self.flow,
self.user_noemail,
component="ak-stage-authenticator-email",
email_required=True,
)
self.assertIsNone(response.json()["email"])
# Test without email
self.client.force_login(self.user_noemail)
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
self.assertStageResponse(
response,
self.flow,
self.user_noemail,
component="ak-stage-authenticator-email",
email_required=True,
)
self.assertIsNone(response.json()["email"])
@patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_session_management(self):
"""Test session device management"""
# Test device creation in session
with patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
):
# Delete any existing devices for this test
EmailDevice.objects.filter(user=self.user).delete()
# Delete any existing devices for this test
EmailDevice.objects.filter(user=self.user).delete()
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
self.assertIn(SESSION_KEY_EMAIL_DEVICE, self.client.session)
device = self.client.session[SESSION_KEY_EMAIL_DEVICE]
self.assertIsInstance(device, EmailDevice)
self.assertFalse(device.confirmed)
self.assertEqual(device.user, self.user)
response = self.client.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
)
self.assertIn(SESSION_KEY_EMAIL_DEVICE, self.client.session)
device = self.client.session[SESSION_KEY_EMAIL_DEVICE]
self.assertIsInstance(device, EmailDevice)
self.assertFalse(device.confirmed)
self.assertEqual(device.user, self.user)
# Test device confirmation and cleanup
device.confirmed = True
device.email = "new_test@authentik.local" # Use a different email
self.client.session[SESSION_KEY_EMAIL_DEVICE] = device
self.client.session.save()
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email", "code": device.token},
)
self.assertEqual(response.status_code, 200)
# Test device confirmation and cleanup
device.confirmed = True
device.email = "new_test@authentik.local" # Use a different email
self.client.session[SESSION_KEY_EMAIL_DEVICE] = device
self.client.session.save()
response = self.client.post(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
data={"component": "ak-stage-authenticator-email", "code": device.token},
)
self.assertEqual(response.status_code, 200)
def test_model_properties_and_methods(self):
"""Test model properties"""
@ -306,30 +295,23 @@ class TestAuthenticatorEmailStage(FlowTestCase):
stage = self.stage
self.assertEqual(stage.serializer, AuthenticatorEmailStageSerializer)
self.assertIsInstance(stage.backend, EmailBackend)
self.assertIsInstance(stage.backend, SMTPEmailBackend)
self.assertEqual(device.serializer, EmailDeviceSerializer)
# Test AuthenticatorEmailStage send method
with patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
return_value=EmailBackend,
):
self.device.generate_token()
# Test EmailDevice _compose_email method
message = self.device._compose_email()
self.assertIsInstance(message, TemplateEmailMessage)
self.assertEqual(message.subject, self.stage.subject)
self.assertEqual(message.to, [f"{self.user.name} <{self.device.email}>"])
self.assertTrue(self.device.token in message.body)
# Test AuthenticatorEmailStage send method
self.stage.send(device)
self.device.generate_token()
# Test EmailDevice _compose_email method
message = self.device._compose_email()
self.assertIsInstance(message, TemplateEmailMessage)
self.assertEqual(message.subject, self.stage.subject)
self.assertEqual(message.to, [f"{self.user.name} <{self.device.email}>"])
self.assertTrue(self.device.token in message.body)
@patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_email_tasks(self):
email_send_mock = MagicMock()
with patch(
"authentik.stages.email.tasks.send_mails",
email_send_mock,
):
# Test AuthenticatorEmailStage send method
self.stage.send(self.device)
email_send_mock.assert_called_once()
# Test AuthenticatorEmailStage send method
self.stage.send(self.device)
self.assertEqual(len(mail.outbox), 1)

34
authentik/stages/email/tasks.py
View File

@ -1,19 +1,21 @@
"""email stage tasks"""
from dramatiq.actor import actor
from email.utils import make_msgid
from smtplib import SMTPException
from typing import Any
from celery import group
from django.core.mail import EmailMultiAlternatives
from django.core.mail.utils import DNS_NAME
from django.utils.text import slugify
from structlog.stdlib import get_logger
from authentik.tasks.models import Task, TaskStatus
from authentik.tasks.middleware import CurrentTask
from authentik.events.models import Event, EventAction, TaskStatus
from authentik.events.system_tasks import SystemTask
from dramatiq.composition import group
from authentik.events.models import Event, EventAction
from authentik.lib.utils.reflection import class_to_path, path_to_class
from authentik.root.celery import CELERY_APP
from authentik.stages.authenticator_email.models import AuthenticatorEmailStage
from authentik.stages.email.models import EmailStage
from authentik.stages.email.utils import logo_data
@ -30,16 +32,14 @@ def send_mails(
stage: Either an EmailStage or AuthenticatorEmailStage instance
messages: List of email messages to send
Returns:
Celery group promise for the email sending tasks
Dramatiq group promise for the email sending tasks
"""
tasks = []
# Use the class path instead of the class itself for serialization
stage_class_path = class_to_path(stage.__class__)
for message in messages:
tasks.append(send_mail.s(message.__dict__, stage_class_path, str(stage.pk)))
lazy_group = group(*tasks)
promise = lazy_group()
return promise
tasks.append(send_mail.message(message.__dict__, stage_class_path, str(stage.pk)))
return group(tasks).run()
def get_email_body(email: EmailMultiAlternatives) -> str:
@ -50,24 +50,16 @@ def get_email_body(email: EmailMultiAlternatives) -> str:
return email.body
@CELERY_APP.task(
bind=True,
autoretry_for=(
SMTPException,
ConnectionError,
OSError,
),
retry_backoff=True,
base=SystemTask,
)
@actor
def send_mail(
self: SystemTask,
message: dict[Any, Any],
stage_class_path: str | None = None,
email_stage_pk: str | None = None,
):
"""Send Email for Email Stage. Retries are scheduled automatically."""
self.save_on_success = False
self: Task = CurrentTask.get_task()
# TODO: fix me
# self.save_on_success = False
message_id = make_msgid(domain=DNS_NAME)
self.set_uid(slugify(message_id.replace(".", "_").replace("@", "_")))
try:

4
authentik/stages/email/tests/test_stage.py
View File

@ -35,6 +35,10 @@ class TestEmailStage(FlowTestCase):
)
self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
@patch(
"authentik.stages.email.models.EmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_rendering(self):
"""Test with pending user"""
plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])

27
authentik/stages/email/tests/test_tasks.py
View File

@ -1,12 +1,13 @@
"""Test email stage tasks"""
from unittest.mock import patch
from unittest.mock import PropertyMock, patch
from django.core import mail
from django.core.mail import EmailMultiAlternatives
from django.core.mail.backends.locmem import EmailBackend
from django.test import TestCase
from authentik.core.tests.utils import create_test_admin_user
from authentik.lib.utils.reflection import class_to_path
from authentik.stages.authenticator_email.models import AuthenticatorEmailStage
from authentik.stages.email.models import EmailStage
from authentik.stages.email.tasks import get_email_body, send_mails
@ -39,20 +40,22 @@ class TestEmailTasks(TestCase):
message.body = "plain text"
self.assertEqual(get_email_body(message), "plain text")
@patch(
"authentik.stages.email.models.EmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_send_mails_email_stage(self):
"""Test send_mails with EmailStage"""
message = EmailMultiAlternatives()
with patch("authentik.stages.email.tasks.send_mail") as mock_send:
send_mails(self.stage, message)
mock_send.s.assert_called_once_with(
message.__dict__, class_to_path(EmailStage), str(self.stage.pk)
)
send_mails(self.stage, message)
self.assertEqual(len(mail.outbox), 1)
@patch(
"authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
)
def test_send_mails_authenticator_stage(self):
"""Test send_mails with AuthenticatorEmailStage"""
message = EmailMultiAlternatives()
with patch("authentik.stages.email.tasks.send_mail") as mock_send:
send_mails(self.auth_stage, message)
mock_send.s.assert_called_once_with(
message.__dict__, class_to_path(AuthenticatorEmailStage), str(self.auth_stage.pk)
)
send_mails(self.auth_stage, message)
self.assertEqual(len(mail.outbox), 1)