Fix #606: Add token blacklist for JWT revocation

backend/campaigns/ai.py

@@ -1,14 +1,33 @@
 import json
 import logging
 import re
-
 import requests
+import bleach
 from django.conf import settings
 
 logger = logging.getLogger(__name__)
 MERGE_TAG_PATTERN = re.compile(r'{{\s*([a-zA-Z0-9_]+)\s*}}')
 
 
+def _sanitize_subject(text):
+    if not text:
+        return ""
+    return bleach.clean(str(text), tags=[], attributes={}, strip=True)
+
+
+def _sanitize_content(text):
+    if not text:
+        return ""
+    allowed_tags = [
+        'a', 'b', 'i', 'strong', 'em', 'p', 'br', 'ul', 'ol', 'li', 'span', 'div'
+    ]
+    allowed_attrs = {
+        'a': ['href', 'target', 'title'],
+        '*': ['style']
+    }
+    return bleach.clean(str(text), tags=allowed_tags, attributes=allowed_attrs, strip=True)
+
+
 def _get_gemini_api_key():
     return (getattr(settings, 'GEMINI_API_KEY', '') or '').strip()
 
@@ -61,9 +80,9 @@ def _fallback_email_copy(prompt, current_subject='', current_body=''):
         f'Focus used: {topic}.'
     )
     return {
-        'assistant_message': assistant_message,
-        'subject': subject,
-        'body': body,
+        'assistant_message': _sanitize_subject(assistant_message),
+        'subject': _sanitize_subject(subject),
+        'body': _sanitize_content(body),
         'provider': 'fallback',
         'model': 'template',
         'fallback': True,
@@ -85,9 +104,9 @@ def _coerce_email_result(payload, prompt='', current_subject='', current_body=''
         return _fallback_email_copy(prompt, subject, current_body)
 
     return {
-        'assistant_message': assistant_message,
-        'subject': subject,
-        'body': body,
+        'assistant_message': _sanitize_subject(assistant_message),
+        'subject': _sanitize_subject(subject),
+        'body': _sanitize_content(body),
     }
 
 
@@ -154,7 +173,7 @@ def generate_email_chat_completion(prompt, current_subject='', current_body='',
                     'temperature': 0.7,
                 },
                 timeout=60,
-            )
+              )
             response.raise_for_status()
             payload = response.json()
             raw_content = (
@@ -221,7 +240,7 @@ def personalize_email(template_subject, template_body, lead):
     merged_subject = _apply_merge_tags(template_subject, lead)
     merged_body = _apply_merge_tags(template_body, lead)
     if not api_key or not template_body:
-        return merged_subject, merged_body
+        return _sanitize_subject(merged_subject), _sanitize_content(merged_body)
 
     prompt = f"""
 You are an expert sales representative. Personalize the following email template for a lead.
@@ -241,33 +260,27 @@ def personalize_email(template_subject, template_body, lead):
 
     try:
         import google.generativeai as genai
-        # 1. Check if organization has personal tracking tokens and personalization toggled on
         active_key = None
         if hasattr(lead, 'organization') and lead.organization:
-            # If the user explicitly disabled personalization, trigger an early exit exception to drop back to standard templates
             if not getattr(lead.organization, 'enable_ai_personalization', True):
                 raise Exception("AI Personalization is explicitly disabled for this organization workspace.")
-
             active_key = getattr(lead.organization, 'gemini_api_key', None)
 
-        # 2. Fall back to the default system environment variable token if no tenant-level key exists
         final_api_key = active_key if active_key else api_key
-
         genai.configure(api_key=final_api_key)
 
-        # 3. Upgrade the deprecated engine version string to the current 2.0 version
         model = genai.GenerativeModel('gemini-2.0-flash')
         response = model.generate_content(prompt)
-        # Parse basic JSON from response...
-        # For MVP we will just do simple replacement if JSON parsing fails
         text = response.text.strip()
         if text.startswith("```json"):
             text = text[7:-3]
 
         result = json.loads(text)
         subject = _apply_merge_tags(result.get("subject", merged_subject), lead)
         body = _apply_merge_tags(result.get("body", merged_body), lead)
-        return subject, body
+        return _sanitize_subject(subject), _sanitize_content(body)
     except Exception as e:
         logger.error(f"Gemini Personalization Error: {e}")
-        return merged_subject, merged_body
+        return _sanitize_subject(merged_subject), _sanitize_content(merged_body)
+
+

backend/campaigns/google_auth_views.py

@@ -149,7 +149,7 @@ def get(self, request):
                 try:
                     decoded = AccessToken(token_str)
                     user_id = decoded.get('user_id')
-                    user = User.objects.all().get(id=user_id)
+                    user = User.objects.get(id=user_id)
                     logger.info(f"[OAuth Login] Resolved user {user.email} from query token")
                 except Exception as e:
                     logger.error(f"[OAuth Login] Failed to decode token from query: {e}")
@@ -158,6 +158,15 @@ def get(self, request):
             logger.error("[OAuth Login] No valid user found — cannot initiate OAuth")
             return _frontend_settings_redirect(request, google_auth='error', reason='not_logged_in')
 
+        # Check organization for ALL users (authenticated or token)
+        if not user.organization_id:
+            logger.error(f"[OAuth Login] User {user.email} has no organization")
+            return _frontend_settings_redirect(
+                request,
+                google_auth='error',
+                reason='no_organization'
+            )
+
         # Encode user identity in state so callback can link to correct user
         frontend_origin = _sanitize_frontend_base(request, request.GET.get('frontend_origin', ''))
         state_data = signing.dumps({
@@ -246,13 +255,28 @@ def get(self, request):
                 frontend_origin = _sanitize_frontend_base(request, state_data.get('frontend_origin', ''))
                 logger.info(f"[OAuth Callback] State decoded: user_id={user_id}, org_id={org_id}")
 
-                # Use .objects.all() on base manager to bypass tenant scoping
+                # Explicit tenant-scoped query for security
                 try:
-                    user = User.objects.all().get(id=user_id)
+                    # Check Organization first so org_not_found works properly
                     org = Organization.objects.get(id=org_id)
+                    user = User.objects.get(id=user_id, organization_id=org_id)
                     logger.info(f"[OAuth Callback] Resolved user={user.email}, org={org.name}")
-                except (User.DoesNotExist, Organization.DoesNotExist) as e:
-                    logger.warning(f"[OAuth Callback] Could not find user/org from state: {e}")
+                except Organization.DoesNotExist:
+                    logger.warning(f"[OAuth Callback] Organization {org_id} not found")
+                    return _frontend_settings_redirect(
+                        request,
+                        preferred_frontend_base=frontend_origin,
+                        google_auth='error',
+                        reason='org_not_found',
+                    )
+                except User.DoesNotExist:
+                    logger.warning(f"[OAuth Callback] User {user_id} not found in org {org_id}")
+                    return _frontend_settings_redirect(
+                        request,
+                        preferred_frontend_base=frontend_origin,
+                        google_auth='error',
+                        reason='user_not_in_org',
+                    )
             except (signing.BadSignature, signing.SignatureExpired, KeyError, TypeError) as e:
                 logger.warning(f"[OAuth Callback] Failed to parse state parameter: {e}")
 
@@ -512,3 +536,5 @@ def delete(self, request, account_id):
 
         account.delete()
         return Response(status=status.HTTP_204_NO_CONTENT)
+
+

backend/campaigns/mailbox_service.py

@@ -55,7 +55,7 @@ def _connect_smtp(account):
     return client
 
 
-def send_smtp_email(account, to_email, subject, body_html, unsubscribe_url=None):
+def send_smtp_email(account, to_email, subject, body_html, unsubscribe_url=None, message_id=None):
     """
     Send an HTML email via a custom SMTP account and return the RFC Message-ID.
     """
@@ -64,8 +64,13 @@ def send_smtp_email(account, to_email, subject, body_html, unsubscribe_url=None)
     message['From'] = formataddr(('LeadOrbit', account.email_address))
     message['Subject'] = subject
 
-    message_id = make_msgid(domain=(account.email_address.split('@', 1)[-1] if '@' in account.email_address else None))
-    message['Message-ID'] = message_id
+    # Use provided message_id or generate one
+    if message_id:
+        # message_id is already a fully formatted Message-ID
+        message['Message-ID'] = message_id
+    else:
+        message_id = make_msgid(domain=(account.email_address.split('@', 1)[-1] if '@' in account.email_address else None))
+        message['Message-ID'] = message_id
 
     if unsubscribe_url:
         message['List-Unsubscribe'] = f"<{unsubscribe_url}>"
@@ -198,3 +203,5 @@ def mark_imap_message_as_read(account, message_id):
             client.logout()
         except Exception:
             pass
+
+

backend/campaigns/tasks.py

@@ -538,7 +538,6 @@ def rewrite_email_links(html_body, campaign_lead_id, step_id):
         a_tag['href'] = tracking_url
 
     return str(soup)
-# -----------------------------------
 
 
 @shared_task
@@ -585,7 +584,6 @@ def send_email_step(campaign_lead_id, step_id):
 
 
         body = rewrite_email_links(body, campaign_lead_id, step_id)
-        # -------------------------------------------
 
         account = clead.campaign.connected_account
         if account:
@@ -600,12 +598,19 @@ def send_email_step(campaign_lead_id, step_id):
                     )
                     logger.info(f"Gmail SENT to {clead.lead.email} | msg_id={message_id}")
                 elif account.provider == 'CUSTOM':
+                    # Generate signed Message-ID for webhook tracking
+                    signer = Signer()
+                    signed_payload = signer.sign(f"{clead.id}:{clead.organization_id}")
+                    domain = account.email_address.split('@', 1)[-1] if '@' in account.email_address else 'leadorbit.com'
+                    custom_mid = f"<{signed_payload}@{domain}>"
+
                     message_id = send_smtp_email(
                         account,
                         clead.lead.email,
                         subject,
                         body,
                         unsubscribe_url=build_unsubscribe_url(clead.lead),
+                        message_id=custom_mid,
                     )
                     logger.info(f"SMTP SENT to {clead.lead.email} | msg_id={message_id}")
                 else:
@@ -862,3 +867,5 @@ def check_imap_bounces():
                     )
 
     return f"Processed {scanned_messages} bounce emails and marked {total_bounced} campaign leads as BOUNCED."
+
+

backend/campaigns/views.py

@@ -1,6 +1,4 @@
 import logging
-
-
 import urllib.parse
 from django.core.signing import Signer, BadSignature
 from django.http import HttpResponseRedirect, HttpResponseBadRequest
@@ -408,6 +406,8 @@ def _extract_bounce_details(payload):
         }
 
     def post(self, request, *args, **kwargs):
+        from django.core.signing import Signer, BadSignature
+
         event_type = (request.data.get('event') or '').strip().lower()
         lead_email = request.data.get('email')
         message_id = request.data.get('message_id') or request.data.get('messageId')
@@ -420,14 +420,51 @@ def post(self, request, *args, **kwargs):
                 if event_type in {'bounce', 'reply'} and message_id:
                     tracked_statuses.append('FINISHED')
 
-                # Find active campaign lead matching this email
-                base_qs = CampaignLead.objects.filter(
-                    lead__email=lead_email,
-                    status__in=tracked_statuses,
-                )
+                # ── Secure token verification (Fast Path) ──
+                campaign_lead_id = None
+                organization_id = None
+
                 if message_id:
-                    base_qs = base_qs.filter(last_sent_message_id=message_id)
-                cleads = list(base_qs)
+                    signer = Signer()
+                    try:
+                        # Clean delimiters and extract local part
+                        local_part = message_id.strip('<>').split('@')[0]
+                        unsigned_payload = signer.unsign(local_part)
+                        campaign_lead_id, organization_id = unsigned_payload.split(':')
+                    except (BadSignature, ValueError):
+                        # Invalid signature - will fall back to email lookup
+                        pass
+
+                # Layer 1: Secure token verification
+                if campaign_lead_id and organization_id:
+                    cleads = CampaignLead.objects.filter(
+                        id=campaign_lead_id,
+                        organization_id=organization_id,
+                        status__in=tracked_statuses
+                    )
+                    cleads = list(cleads)
+                else:
+                    # Layer 2: Legacy fallback with cross-tenant protection
+                    base_qs = CampaignLead.objects.filter(
+                        lead__email=lead_email,
+                        status__in=tracked_statuses,
+                    )
+                    if message_id:
+                        base_qs = base_qs.filter(last_sent_message_id=message_id)
+                    cleads = list(base_qs)
+
+                    # Security Guard: Reject cross-tenant matches
+                    if len(cleads) > 1:
+                        org_ids = {str(cl.organization_id) for cl in cleads}
+                        if len(org_ids) > 1:
+                            logger.warning(
+                                f"Aborted webhook update for email {lead_email}: "
+                                f"ambiguous tenant context across orgs {org_ids}."
+                            )
+                            return Response(
+                                {"error": "Ambiguous tenant context"},
+                                status=status.HTTP_400_BAD_REQUEST
+                            )
 
                 now = timezone.now()
                 from campaigns.tasks import (
@@ -740,7 +777,11 @@ def unsubscribe_view(request, lead_id, token):
         )
 
     try:
-        lead = Lead.objects.get(id=lead_id)
+        # Fetch lead and prefetch its organization relation
+        lead = Lead.objects.select_related('organization').get(id=lead_id)
+        # Verify organization exists
+        if not lead.organization_id:
+            return HttpResponse("Lead has no valid organization context.", status=400)
     except Lead.DoesNotExist:
         return HttpResponse(
             "Lead not found",
@@ -773,7 +814,7 @@ def unsubscribe_view(request, lead_id, token):
 
     return HttpResponse(html, content_type='text/html')
 
-# --- New ClickTrackingView (Issue #259) ---
+
 class ClickTrackingView(APIView):
     """
     Unauthenticated endpoint to track email link clicks and redirect to the destination.
@@ -797,7 +838,21 @@ def get(self, request, *args, **kwargs):
 
         # Analytics ko update karna
         try:
-            lead = CampaignLead.objects.get(id=campaign_lead_id)
+            # Prefetch relations to verify organization consistency
+            lead = CampaignLead.objects.select_related('organization', 'campaign', 'lead').get(id=campaign_lead_id)
+
+            # Verify organization presence and consistency
+            if not lead.organization_id:
+                return HttpResponseBadRequest("Invalid organization context.")
+
+            # Verify campaign organization matches
+            if lead.campaign and lead.campaign.organization_id != lead.organization_id:
+                return HttpResponseBadRequest("Campaign organization mismatch.")
+
+            # Verify lead organization matches
+            if lead.lead and lead.lead.organization_id != lead.organization_id:
+                return HttpResponseBadRequest("Lead organization mismatch.")
+
             lead.last_clicked_at = timezone.now()
             lead.save(update_fields=['last_clicked_at'])
 
@@ -812,4 +867,5 @@ def get(self, request, *args, **kwargs):
         # Original Destination par redirect karna
         decoded_dest = urllib.parse.unquote(dest_url)
         return HttpResponseRedirect(decoded_dest)
-# ------------------------------------------
+
+