fix(leads): Add file size validation to CSV import endpoint (LO-079)

backend/backend/settings.py

@@ -236,3 +236,8 @@ def _normalize_google_redirect_uri(raw_uri: str, backend_base_url: str) -> str:
 TWILIO_ACCOUNT_SID = os.getenv('TWILIO_ACCOUNT_SID', _read_local_env_value('TWILIO_ACCOUNT_SID', ''))
 TWILIO_AUTH_TOKEN = os.getenv('TWILIO_AUTH_TOKEN', _read_local_env_value('TWILIO_AUTH_TOKEN', ''))
 TWILIO_PHONE_NUMBER = os.getenv('TWILIO_PHONE_NUMBER', _read_local_env_value('TWILIO_PHONE_NUMBER', ''))
+
+# ─── File Upload Limits ─────────────────────────────
+# Maximum size allowed for CSV uploads (default 10MB)
+MAX_CSV_UPLOAD_SIZE = int(os.getenv('MAX_CSV_UPLOAD_SIZE', 10 * 1024 * 1024))
+

backend/config/settings.py

@@ -118,3 +118,7 @@
 # https://docs.djangoproject.com/en/6.0/howto/static-files/
 
 STATIC_URL = "static/"
+
+# Maximum size allowed for CSV uploads (default 10MB)
+MAX_CSV_UPLOAD_SIZE = int(os.getenv('MAX_CSV_UPLOAD_SIZE', 10 * 1024 * 1024))
+

backend/leads/tests.py

@@ -201,6 +201,39 @@ def test_manager_can_create_and_delete_organization_leads(self):
         self.assertFalse(Lead.objects.filter(organization=self.org_a).exists())
         self.assertTrue(Lead.objects.filter(id=self.lead_b.id).exists())
 
+    def test_import_csv_oversized_file_rejected(self):
+        self.client.force_authenticate(self.user_a)
+        with self.settings(MAX_CSV_UPLOAD_SIZE=50):
+            file_data = b"email,first_name,last_name\n" + b"a" * 100 + b"@example.com,John,Doe\n"
+            oversized_file = SimpleUploadedFile(
+                "too_large.csv",
+                file_data,
+                content_type="text/csv"
+            )
+            response = self.client.post(
+                '/api/v1/leads/import_csv/',
+                {'file': oversized_file},
+                format='multipart'
+            )
+            self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+            self.assertIn("File size exceeds the limit", response.data['error'])
+
+    def test_import_csv_normal_size_file_accepted(self):
+        self.client.force_authenticate(self.user_a)
+        with self.settings(MAX_CSV_UPLOAD_SIZE=1000):
+            file_data = b"email,first_name,last_name\n" + b"john@example.com,John,Doe\n"
+            normal_file = SimpleUploadedFile(
+                "normal.csv",
+                file_data,
+                content_type="text/csv"
+            )
+            response = self.client.post(
+                '/api/v1/leads/import_csv/',
+                {'file': normal_file},
+                format='multipart'
+            )
+            self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
     def test_blocked_domain_create_normalizes_domain_for_current_organization(self):
         self.client.force_authenticate(self.user_a)
 

backend/leads/views.py

@@ -97,6 +97,15 @@ def import_csv(self, request):
         if not file_obj:
             return Response({"error": "No file provided"}, status=status.HTTP_400_BAD_REQUEST)
 
+        from django.conf import settings
+        max_size = getattr(settings, 'MAX_CSV_UPLOAD_SIZE', 10 * 1024 * 1024)
+        if file_obj.size > max_size:
+            return Response(
+                {"error": f"File size exceeds the limit of {max_size / (1024 * 1024):.1f}MB."},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+
         job = LeadImportJob.objects.create(
             organization=request.user.organization,
             filename=file_obj.name or 'lead-import.csv',