updated tests.py again

django_aws_ses/migrations/0001_initial.py

@@ -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')],
-            },
-        ),
-    ]

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):
+    def generate_unsubscribe_token(self):
-        """Generate a secure hash for unsubscribe verification."""
+        """Generate a signed token for unsubscribe verification."""
-        email = self.get_email()
+        signer = Signer()
-        message = f"{self.user.pk}{email}".encode()
+        value = f"{self.user.pk}:{self.get_email()}"
-        return hmac.new(
+        return signer.sign(value)
-            settings.SECRET_KEY.encode(),
-            message,
-            hashlib.sha256
-        ).hexdigest()
 
-    def check_unsubscribe_hash(self, hash_value):
+    def verify_unsubscribe_token(self, token):
-        """Verify an unsubscribe hash."""
+        """Verify a signed unsubscribe token.
-        return hmac.compare_digest(self.unsubscribe_hash_generator(), hash_value)
+
+        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."""
+        """Generate a secure unsubscribe URL with a signed token."""
-        uuid = urlsafe_base64_encode(force_bytes(str(self.user.pk)))
+        uuid = urlsafe_base64_encode(str(self.user.pk).encode())
-        hash_value = self.unsubscribe_hash_generator()
+        token = self.generate_unsubscribe_token()
-        return reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={"uuid": uuid, "hash": hash_value})
+        return reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={"uuid": uuid, "token": token})
 
 class SESStat(models.Model):
     """Daily statistics for AWS SES email sending."""

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()
+        token = self.ses_addon.generate_unsubscribe_token()
-        url = reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={'uuid': uuid_b64, 'hash': hash_value})
+        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()
+        token = self.ses_addon.generate_unsubscribe_token()
-        url = reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={'uuid': uuid_b64, 'hash': hash_value})
+        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')

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/<str:uuid>/<str:hash>/', HandleUnsubscribe.as_view(), name='aws_ses_unsubscribe')
+    path('unsubscribe/<str:uuid>/<str:token>/', HandleUnsubscribe.as_view(), name='aws_ses_unsubscribe')
 ]

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):
+        if not user or not ses.verify_unsubscribe_token(token):
-            logger.warning(f"Invalid unsubscribe hash for user: {user.email}")
+            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):
+        if not user or not ses.verify_unsubscribe_token(token):
-            logger.warning(f"Invalid unsubscribe hash for user: {user.email}")
+            logger.warning(f"Invalid token for user: {user.email}")
             return redirect(settings.HOME_URL)
 
         if action == 'unsubscribe':