feat: improve error handling and lifecycle management for tabCapture

backend/backend/settings.py

@@ -57,10 +57,14 @@ def _normalize_google_redirect_uri(raw_uri: str, backend_base_url: str) -> str:
     'fallback-insecure-key-for-local-dev-and-testing' if (DEBUG or TESTING) else '',
 )
 ALLOWED_HOSTS = ['*']
-CORS_ALLOWED_ORIGINS = [
-    "http://localhost:3000",
-    "http://127.0.0.1:3000",
-]
+ADMIN_PATH = os.getenv('ADMIN_PATH', 'secure-admin-portal')
+
+# Cookie Security
+SESSION_COOKIE_SECURE = not DEBUG
+SESSION_COOKIE_HTTPONLY = True
+SESSION_COOKIE_SAMESITE = 'Lax'
+CSRF_COOKIE_SECURE = not DEBUG
+CSRF_COOKIE_SAMESITE = 'Lax'
 
 INSTALLED_APPS = [
     'django.contrib.admin',
@@ -71,6 +75,7 @@ def _normalize_google_redirect_uri(raw_uri: str, backend_base_url: str) -> str:
     'django.contrib.staticfiles',
     'rest_framework',
     'rest_framework_simplejwt',
+    'rest_framework_simplejwt.token_blacklist',
     'corsheaders',
     'django_celery_beat',
     'tenants',
@@ -124,9 +129,13 @@ def _normalize_google_redirect_uri(raw_uri: str, backend_base_url: str) -> str:
 
 AUTH_PASSWORD_VALIDATORS = [
     {'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator'},
-    {'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator'},
+    {
+        'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
+        'OPTIONS': {'min_length': 10},
+    },
     {'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'},
     {'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'},
+    {'NAME': 'users.validators.ComplexityPasswordValidator'},
 ]
 
 LANGUAGE_CODE = 'en-us'
@@ -142,7 +151,16 @@ def _normalize_google_redirect_uri(raw_uri: str, backend_base_url: str) -> str:
     ),
     'DEFAULT_PERMISSION_CLASSES': (
         'rest_framework.permissions.IsAuthenticated',
-    )
+    ),
+    'DEFAULT_THROTTLE_CLASSES': [
+        'rest_framework.throttling.AnonRateThrottle',
+        'rest_framework.throttling.UserRateThrottle',
+    ],
+    'DEFAULT_THROTTLE_RATES': {
+        'anon': '100/day',
+        'user': '1000/day',
+        'login': '5/minute',
+    }
 }
 
 # Celery Configuration
@@ -179,10 +197,19 @@ def _normalize_google_redirect_uri(raw_uri: str, backend_base_url: str) -> str:
     'REFRESH_TOKEN_LIFETIME': timedelta(days=7),
     'USER_ID_FIELD': 'id',
     'USER_ID_CLAIM': 'user_id',
+    'SIGNING_KEY': os.getenv('JWT_SIGNING_KEY', SECRET_KEY),
 }
 
-# Allow all origins in development
-CORS_ALLOW_ALL_ORIGINS = True
+# Restrict CORS to trusted origins
+CORS_ALLOW_ALL_ORIGINS = False
+CORS_ALLOWED_ORIGINS = [
+    "http://localhost:3000",
+    "http://127.0.0.1:3000",
+    "http://localhost:8080",
+    "http://127.0.0.1:8080",
+]
+if os.getenv('FRONTEND_BASE_URL'):
+    CORS_ALLOWED_ORIGINS.append(os.getenv('FRONTEND_BASE_URL').rstrip('/'))
 
 # Gemini API Key
 GEMINI_API_KEY = os.getenv('GEMINI_API_KEY', _read_local_env_value('GEMINI_API_KEY', ''))

backend/backend/urls.py

@@ -6,8 +6,12 @@
 from rest_framework_simplejwt.views import TokenObtainPairView as BaseTokenObtainPairView
 from users.jwt import CustomTokenObtainSerializer
 
+from django.conf import settings
+from users.throttling import LoginRateThrottle
+
 class CustomTokenObtainPairView(BaseTokenObtainPairView):
     serializer_class = CustomTokenObtainSerializer
+    throttle_classes = [LoginRateThrottle]
 
 from users.views import AuthViewSet
 from leads.views import BlockedDomainViewSet, LeadImportJobViewSet, LeadViewSet, TagViewSet
@@ -49,7 +53,7 @@ def api_root(_request):
 
 urlpatterns = [
     path('', api_root, name='api_root'),
-    path('admin/', admin.site.urls),
+    path(f'{settings.ADMIN_PATH}/', admin.site.urls),
     path('api/v1/token/', CustomTokenObtainPairView.as_view(), name='token_obtain_pair'),
     path('api/v1/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
     path('api/v1/webhooks/email/', WebhookView.as_view(), name='email_webhook'),

backend/tenants/security.py

@@ -58,4 +58,5 @@ def process_response(self, request, response):
         response['X-XSS-Protection'] = '1; mode=block'
         response['Referrer-Policy'] = 'strict-origin-when-cross-origin'
         response['Permissions-Policy'] = 'camera=(), microphone=(), geolocation=()'
+        response['Content-Security-Policy'] = "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; frame-ancestors 'none';"
         return response

backend/users/jwt.py

@@ -1,4 +1,35 @@
+import jwt
+from datetime import datetime, timedelta, timezone
+from django.conf import settings
+from rest_framework import serializers
 from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
 
 class CustomTokenObtainSerializer(TokenObtainPairSerializer):
     username_field = 'email'
+
+    def validate(self, attrs):
+        # We need to authenticate first. SimpleJWT validation handles active user check.
+        data = super().validate(attrs)
+
+        # Enforce email verification
+        if not self.user.is_email_verified:
+            raise serializers.ValidationError('Email not verified. Please verify your email first.')
+
+        # Enforce MFA/2FA check
+        if self.user.mfa_enabled:
+            temp_token = jwt.encode(
+                {
+                    'user_id': str(self.user.id),
+                    'mfa_pending': True,
+                    'exp': datetime.now(timezone.utc) + timedelta(minutes=5)
+                },
+                settings.SECRET_KEY,
+                algorithm='HS256'
+            )
+            return {
+                'mfa_required': True,
+                'mfa_token': temp_token,
+                'email': self.user.email
+            }
+
+        return data

backend/users/migrations/0003_user_is_email_verified_user_mfa_enabled_and_more.py

@@ -0,0 +1,28 @@
+# Generated by Django 5.0.14 on 2026-07-12 06:12
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0002_user_member_role_default'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='user',
+            name='is_email_verified',
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name='user',
+            name='mfa_enabled',
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name='user',
+            name='mfa_secret',
+            field=models.CharField(blank=True, max_length=32, null=True),
+        ),
+    ]

backend/users/models.py

@@ -33,6 +33,9 @@ class User(AbstractBaseUser, PermissionsMixin, TenantModel):
     role = models.CharField(max_length=20, choices=ROLE_CHOICES, default=ROLE_MEMBER)
     is_active = models.BooleanField(default=True)
     is_staff = models.BooleanField(default=False)
+    is_email_verified = models.BooleanField(default=False)
+    mfa_enabled = models.BooleanField(default=False)
+    mfa_secret = models.CharField(max_length=32, blank=True, null=True)
 
     objects = UserManager()
 

backend/users/serializers.py

@@ -25,6 +25,16 @@ def validate_email(self, value):
             raise serializers.ValidationError('A user with this email already exists.')
         return value
 
+    def validate(self, attrs):
+        from django.contrib.auth.password_validation import validate_password
+        from django.core.exceptions import ValidationError as DjangoValidationError
+        user = User(email=attrs.get('email'))
+        try:
+            validate_password(attrs.get('password'), user)
+        except DjangoValidationError as exc:
+            raise serializers.ValidationError({'password': list(exc.messages)})
+        return attrs
+
     def create(self, validated_data):
         org = Organization.objects.create(name=validated_data['organization_name'])
         user = User.objects.create_user(

backend/users/tests.py

@@ -1,5 +1,9 @@
 from rest_framework import status
 from rest_framework.test import APITestCase
+from django.core import mail
+from django.utils.http import urlsafe_base64_encode
+from django.utils.encoding import force_bytes
+from django.contrib.auth.tokens import default_token_generator
 
 from tenants.models import Organization
 from users.models import User
@@ -104,3 +108,152 @@ def test_member_cannot_delete_organization(self):
         self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
         self.assertTrue(Organization.objects.filter(id=self.organization.id).exists())
         self.assertTrue(User.objects.filter(id=self.user.id).exists())
+
+
+class SecurityHardeningTests(APITestCase):
+    def setUp(self):
+        self.organization = Organization.objects.create(name='Security Org')
+
+    def test_password_complexity_validator(self):
+        # Register with simple password (missing special char and digit)
+        response = self.client.post(
+            '/api/v1/auth/register/',
+            {
+                'email': 'test_weak@example.com',
+                'password': 'password',
+                'organization_name': 'New Org',
+            },
+            format='json',
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('password', response.data)
+
+    def test_email_verification_flow(self):
+        # 1. Register user
+        response = self.client.post(
+            '/api/v1/auth/register/',
+            {
+                'email': 'unverified@example.com',
+                'password': 'StrongPass123!',
+                'organization_name': 'New Org',
+            },
+            format='json',
+        )
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertIn('user', response.data)
+        user = User.objects.get(email='unverified@example.com')
+        self.assertFalse(user.is_email_verified)
+
+        # Ensure verification email was sent to outbox
+        self.assertEqual(len(mail.outbox), 1)
+        self.assertIn('Verify Your LeadOrbit Email', mail.outbox[0].subject)
+
+        # 2. Try logging in (should fail since email is unverified)
+        login_response = self.client.post(
+            '/api/v1/token/',
+            {
+                'email': 'unverified@example.com',
+                'password': 'StrongPass123!',
+            },
+            format='json',
+        )
+        self.assertEqual(login_response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('non_field_errors', login_response.data)
+
+        # 3. Get verification token/uid
+        uidb64 = urlsafe_base64_encode(force_bytes(user.pk))
+        token = default_token_generator.make_token(user)
+
+        # 4. Verify email
+        verify_response = self.client.post(
+            '/api/v1/auth/verify-email/',
+            {
+                'uid': uidb64,
+                'token': token,
+            },
+            format='json',
+        )
+        self.assertEqual(verify_response.status_code, status.HTTP_200_OK)
+        user.refresh_from_db()
+        self.assertTrue(user.is_email_verified)
+
+        # 5. Log in after verification (should succeed)
+        login_response2 = self.client.post(
+            '/api/v1/token/',
+            {
+                'email': 'unverified@example.com',
+                'password': 'StrongPass123!',
+            },
+            format='json',
+        )
+        self.assertEqual(login_response2.status_code, status.HTTP_200_OK)
+        self.assertIn('access', login_response2.data)
+
+    def test_mfa_flow(self):
+        import pyotp
+        # Create a verified user
+        user = User.objects.create_user(
+            email='mfa_user@example.com',
+            password='StrongPass123!',
+            organization=self.organization,
+            role=User.ROLE_ADMIN,
+        )
+        user.is_email_verified = True
+        user.save()
+
+        # 1. Enable MFA setup
+        self.client.force_authenticate(user)
+        enable_response = self.client.post('/api/v1/auth/enable-mfa/', format='json')
+        self.assertEqual(enable_response.status_code, status.HTTP_200_OK)
+        self.assertIn('secret', enable_response.data)
+        secret = enable_response.data['secret']
+
+        # 2. Confirm MFA setup with valid TOTP code
+        totp = pyotp.TOTP(secret)
+        confirm_response = self.client.post(
+            '/api/v1/auth/confirm-mfa/',
+            {'code': totp.now()},
+            format='json',
+        )
+        self.assertEqual(confirm_response.status_code, status.HTTP_200_OK)
+        user.refresh_from_db()
+        self.assertTrue(user.mfa_enabled)
+
+        # 3. Log in (should return MFA challenge/token)
+        self.client.logout()
+        login_response = self.client.post(
+            '/api/v1/token/',
+            {
+                'email': 'mfa_user@example.com',
+                'password': 'StrongPass123!',
+            },
+            format='json',
+        )
+        self.assertEqual(login_response.status_code, status.HTTP_200_OK)
+        self.assertTrue(login_response.data.get('mfa_required'))
+        mfa_token = login_response.data.get('mfa_token')
+        self.assertIsNotNone(mfa_token)
+
+        # 4. Verify MFA with invalid code (should fail)
+        verify_response_fail = self.client.post(
+            '/api/v1/auth/verify-mfa/',
+            {
+                'mfa_token': mfa_token,
+                'code': '000000',
+            },
+            format='json',
+        )
+        self.assertEqual(verify_response_fail.status_code, status.HTTP_400_BAD_REQUEST)
+
+        # 5. Verify MFA with valid code (should succeed and return JWT tokens)
+        verify_response_success = self.client.post(
+            '/api/v1/auth/verify-mfa/',
+            {
+                'mfa_token': mfa_token,
+                'code': totp.now(),
+            },
+            format='json',
+        )
+        self.assertEqual(verify_response_success.status_code, status.HTTP_200_OK)
+        self.assertIn('access', verify_response_success.data)
+

backend/users/throttling.py

@@ -0,0 +1,24 @@
+from rest_framework.throttling import SimpleRateThrottle
+
+class LoginRateThrottle(SimpleRateThrottle):
+    scope = 'login'
+
+    def get_cache_key(self, request, view):
+        if request.user and request.user.is_authenticated:
+            ident = request.user.pk
+        else:
+            ident = self.get_ident(request)
+
+        username = None
+        if hasattr(request, 'data') and isinstance(request.data, dict):
+            username = request.data.get('email')
+
+        if username:
+            return self.cache_format % {
+                'scope': self.scope,
+                'ident': f"{ident}_{username}"
+            }
+        return self.cache_format % {
+            'scope': self.scope,
+            'ident': ident
+        }