From 42fd9353f56a6e600a5b1ff24c1e5e4bd607adee Mon Sep 17 00:00:00 2001 From: Raymond Jessop Date: Fri, 18 Apr 2025 20:07:09 -0500 Subject: [PATCH] updated tests.py again --- django_aws_ses/migrations/0001_initial.py | 143 ---------------------- django_aws_ses/migrations/__init__.py | 0 django_aws_ses/models.py | 42 ++++--- django_aws_ses/tests.py | 13 +- django_aws_ses/urls.py | 2 +- django_aws_ses/views.py | 15 ++- 6 files changed, 40 insertions(+), 175 deletions(-) delete mode 100644 django_aws_ses/migrations/0001_initial.py delete mode 100644 django_aws_ses/migrations/__init__.py diff --git a/django_aws_ses/migrations/0001_initial.py b/django_aws_ses/migrations/0001_initial.py deleted file mode 100644 index 23f64b2..0000000 --- a/django_aws_ses/migrations/0001_initial.py +++ /dev/null @@ -1,143 +0,0 @@ -# Generated by Django 5.2 on 2025-04-18 22:19 - -import django.db.models.deletion -from django.conf import settings -from django.db import migrations, models - - -class Migration(migrations.Migration): - - initial = True - - dependencies = [ - ('sites', '0002_alter_domain_unique'), - migrations.swappable_dependency(settings.AUTH_USER_MODEL), - ] - - operations = [ - migrations.CreateModel( - name='BlackListedDomains', - fields=[ - ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('domain', models.CharField(db_index=True, max_length=255, unique=True)), - ('timestamp', models.DateTimeField(auto_now_add=True, db_index=True)), - ], - options={ - 'verbose_name': 'Blacklisted Domain', - 'verbose_name_plural': 'Blacklisted Domains', - }, - ), - migrations.CreateModel( - name='SESStat', - fields=[ - ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('date', models.DateField(db_index=True, unique=True)), - ('delivery_attempts', models.PositiveIntegerField()), - ('bounces', models.PositiveIntegerField()), - ('complaints', models.PositiveIntegerField()), - ('rejects', models.PositiveIntegerField()), - ], - options={ - 'verbose_name': 'SES Statistic', - 'verbose_name_plural': 'SES Statistics', - 'ordering': ['-date'], - }, - ), - migrations.CreateModel( - name='AwsSesSettings', - fields=[ - ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('access_key', models.CharField(blank=True, max_length=255, null=True)), - ('secret_key', models.CharField(blank=True, max_length=255, null=True)), - ('region_name', models.CharField(blank=True, max_length=255, null=True)), - ('region_endpoint', models.CharField(blank=True, max_length=255, null=True)), - ('site', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='awssessettings', to='sites.site')), - ], - options={ - 'verbose_name': 'AWS SES Settings', - 'verbose_name_plural': 'AWS SES Settings', - }, - ), - migrations.CreateModel( - name='AwsSesUserAddon', - fields=[ - ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('unsubscribe', models.BooleanField(default=False)), - ('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='aws_ses', to=settings.AUTH_USER_MODEL)), - ], - options={ - 'verbose_name': 'AWS SES User Addon', - 'verbose_name_plural': 'AWS SES User Addons', - }, - ), - migrations.CreateModel( - name='BounceRecord', - fields=[ - ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('timestamp', models.DateTimeField(auto_now_add=True, db_index=True)), - ('email', models.EmailField(db_index=True, max_length=254)), - ('bounce_type', models.CharField(blank=True, max_length=255, null=True)), - ('bounce_sub_type', models.CharField(blank=True, max_length=255, null=True)), - ('reporting_mta', models.CharField(blank=True, max_length=255, null=True)), - ('status', models.CharField(blank=True, max_length=255, null=True)), - ('action', models.CharField(blank=True, max_length=255, null=True)), - ('feedback_id', models.TextField(blank=True, null=True)), - ('diagnostic_code', models.CharField(blank=True, max_length=2048, null=True)), - ('cleared', models.BooleanField(default=False)), - ], - options={ - 'verbose_name': 'Bounce Record', - 'verbose_name_plural': 'Bounce Records', - 'indexes': [models.Index(fields=['email', 'timestamp'], name='django_aws__email_6bb737_idx')], - }, - ), - migrations.CreateModel( - name='ComplaintRecord', - fields=[ - ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('timestamp', models.DateTimeField(auto_now_add=True, db_index=True)), - ('email', models.EmailField(db_index=True, max_length=254)), - ('sub_type', models.CharField(blank=True, max_length=255, null=True)), - ('feedback_id', models.TextField(blank=True, null=True)), - ('feedback_type', models.CharField(blank=True, max_length=255, null=True)), - ], - options={ - 'verbose_name': 'Complaint Record', - 'verbose_name_plural': 'Complaint Records', - 'indexes': [models.Index(fields=['email', 'timestamp'], name='django_aws__email_36ac11_idx')], - }, - ), - migrations.CreateModel( - name='SendRecord', - fields=[ - ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('timestamp', models.DateTimeField(auto_now_add=True, db_index=True)), - ('source', models.EmailField(max_length=254)), - ('destination', models.EmailField(db_index=True, max_length=254)), - ('subject', models.TextField(blank=True, max_length=998, null=True)), - ('message_id', models.TextField(blank=True, max_length=255, null=True)), - ('aws_process_time', models.IntegerField(default=0)), - ('smtp_response', models.CharField(blank=True, max_length=255, null=True)), - ('status', models.CharField(blank=True, choices=[('Send', 'Send'), ('Delivery', 'Delivery')], max_length=20, null=True)), - ], - options={ - 'verbose_name': 'Send Record', - 'verbose_name_plural': 'Send Records', - 'indexes': [models.Index(fields=['destination', 'timestamp'], name='django_aws__destina_e0db33_idx')], - }, - ), - migrations.CreateModel( - name='UnknownRecord', - fields=[ - ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), - ('timestamp', models.DateTimeField(auto_now_add=True, db_index=True)), - ('event_type', models.CharField(blank=True, max_length=255, null=True)), - ('aws_data', models.TextField(blank=True, null=True)), - ], - options={ - 'verbose_name': 'Unknown Record', - 'verbose_name_plural': 'Unknown Records', - 'indexes': [models.Index(fields=['event_type', 'timestamp'], name='django_aws__event_t_ca3517_idx')], - }, - ), - ] diff --git a/django_aws_ses/migrations/__init__.py b/django_aws_ses/migrations/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/django_aws_ses/models.py b/django_aws_ses/models.py index ec87f03..854e001 100644 --- a/django_aws_ses/models.py +++ b/django_aws_ses/models.py @@ -11,6 +11,7 @@ from django.dispatch import receiver from django.urls import reverse from django.utils.encoding import force_bytes from django.utils.http import urlsafe_base64_encode +from django.core.signing import Signer, BadSignature User = get_user_model() logger = logging.getLogger(__name__) @@ -67,25 +68,34 @@ class AwsSesUserAddon(models.Model): email_field = self.user.get_email_field_name() return getattr(self.user, email_field, '') or '' - def unsubscribe_hash_generator(self): - """Generate a secure hash for unsubscribe verification.""" - email = self.get_email() - message = f"{self.user.pk}{email}".encode() - return hmac.new( - settings.SECRET_KEY.encode(), - message, - hashlib.sha256 - ).hexdigest() + def generate_unsubscribe_token(self): + """Generate a signed token for unsubscribe verification.""" + signer = Signer() + value = f"{self.user.pk}:{self.get_email()}" + return signer.sign(value) - def check_unsubscribe_hash(self, hash_value): - """Verify an unsubscribe hash.""" - return hmac.compare_digest(self.unsubscribe_hash_generator(), hash_value) + def verify_unsubscribe_token(self, token): + """Verify a signed unsubscribe token. + + Args: + token (str): The signed token to verify. + + Returns: + bool: True if the token is valid, False otherwise. + """ + signer = Signer() + try: + value = signer.unsign(token) + pk, email = value.split(':') + return str(self.user.pk) == pk and self.get_email() == email + except BadSignature: + return False def unsubscribe_url_generator(self): - """Generate a secure unsubscribe URL.""" - uuid = urlsafe_base64_encode(force_bytes(str(self.user.pk))) - hash_value = self.unsubscribe_hash_generator() - return reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={"uuid": uuid, "hash": hash_value}) + """Generate a secure unsubscribe URL with a signed token.""" + uuid = urlsafe_base64_encode(str(self.user.pk).encode()) + token = self.generate_unsubscribe_token() + return reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={"uuid": uuid, "token": token}) class SESStat(models.Model): """Daily statistics for AWS SES email sending.""" diff --git a/django_aws_ses/tests.py b/django_aws_ses/tests.py index 33ce9e6..f7af3d7 100644 --- a/django_aws_ses/tests.py +++ b/django_aws_ses/tests.py @@ -7,7 +7,6 @@ from django.utils.http import urlsafe_base64_encode from django.utils.encoding import force_bytes from unittest.mock import patch, Mock import json -import uuid from .backends import SESBackend from .views import handle_bounce, HandleUnsubscribe @@ -131,8 +130,8 @@ class DjangoAwsSesTests(TestCase): def test_unsubscribe_confirmation(self): """Test unsubscribe confirmation page and action.""" uuid_b64 = urlsafe_base64_encode(str(self.user.pk).encode()) - hash_value = self.ses_addon.unsubscribe_hash_generator() - url = reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={'uuid': uuid_b64, 'hash': hash_value}) + token = self.ses_addon.generate_unsubscribe_token() + url = reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={'uuid': uuid_b64, 'token': token}) # Test GET (confirmation page) response = self.client.get(url) @@ -141,7 +140,7 @@ class DjangoAwsSesTests(TestCase): self.assertFalse(self.ses_addon.unsubscribe) # Test POST (unsubscribe) - response = self.client.post(url, {'action': 'unsubscribe'}, follow=True) + response = self.client.post(url, {'action': 'unsubscribe'}) self.ses_addon.refresh_from_db() self.assertEqual(response.status_code, 200) self.assertContains(response, 'You have been unsubscribed') @@ -152,10 +151,10 @@ class DjangoAwsSesTests(TestCase): self.ses_addon.unsubscribe = True self.ses_addon.save() uuid_b64 = urlsafe_base64_encode(str(self.user.pk).encode()) - hash_value = self.ses_addon.unsubscribe_hash_generator() - url = reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={'uuid': uuid_b64, 'hash': hash_value}) + token = self.ses_addon.generate_unsubscribe_token() + url = reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={'uuid': uuid_b64, 'token': token}) - response = self.client.post(url, {'action': 'resubscribe'}, follow=True) + response = self.client.post(url, {'action': 'resubscribe'}) self.ses_addon.refresh_from_db() self.assertEqual(response.status_code, 200) self.assertContains(response, 'You have been re-subscribed') diff --git a/django_aws_ses/urls.py b/django_aws_ses/urls.py index 0080bf9..d1207ce 100644 --- a/django_aws_ses/urls.py +++ b/django_aws_ses/urls.py @@ -12,5 +12,5 @@ app_name = "django_aws_ses" urlpatterns = [ path('status/', dashboard, name='aws_ses_status'), path('bounce/', csrf_exempt(handle_bounce),name='aws_ses_bounce'), - path('unsubscribe///', HandleUnsubscribe.as_view(), name='aws_ses_unsubscribe') + path('unsubscribe///', HandleUnsubscribe.as_view(), name='aws_ses_unsubscribe') ] \ No newline at end of file diff --git a/django_aws_ses/views.py b/django_aws_ses/views.py index 731b782..94c226a 100644 --- a/django_aws_ses/views.py +++ b/django_aws_ses/views.py @@ -13,7 +13,7 @@ from django.utils.encoding import force_bytes, force_str from django.utils.http import urlsafe_base64_decode, urlsafe_base64_encode from django.views.decorators.http import require_POST from django.views.generic.base import TemplateView -from django.views.decorators.csrf import csrf_protect +from django.core.signing import Signer, BadSignature from . import settings from . import signals @@ -308,7 +308,7 @@ class HandleUnsubscribe(TemplateView): def get(self, request, *args, **kwargs): """Show confirmation page for unsubscribe or re-subscribe.""" uuid = self.kwargs['uuid'] - hash_value = self.kwargs['hash'] + token = self.kwargs['token'] try: uuid = force_str(urlsafe_base64_decode(uuid)) @@ -322,18 +322,17 @@ class HandleUnsubscribe(TemplateView): except AwsSesUserAddon.DoesNotExist: ses = AwsSesUserAddon.objects.create(user=user) - if not user or not ses.check_unsubscribe_hash(hash_value): - logger.warning(f"Invalid unsubscribe hash for user: {user.email}") + if not user or not ses.verify_unsubscribe_token(token): + logger.warning(f"Invalid token for user: {user.email}") return redirect(settings.HOME_URL) self.user_email = user.email return super().get(request, *args, **kwargs) - @csrf_protect def post(self, request, *args, **kwargs): """Process unsubscribe or re-subscribe request.""" uuid = self.kwargs['uuid'] - hash_value = self.kwargs['hash'] + token = self.kwargs['token'] action = request.POST.get('action') try: @@ -348,8 +347,8 @@ class HandleUnsubscribe(TemplateView): except AwsSesUserAddon.DoesNotExist: ses = AwsSesUserAddon.objects.create(user=user) - if not user or not ses.check_unsubscribe_hash(hash_value): - logger.warning(f"Invalid unsubscribe hash for user: {user.email}") + if not user or not ses.verify_unsubscribe_token(token): + logger.warning(f"Invalid token for user: {user.email}") return redirect(settings.HOME_URL) if action == 'unsubscribe':