🔒 security: Add XSS and URL injection protection

DevaOnBreaches · DevaOnBreaches · commit 3a1660ec6f71 · 2025-12-14T08:22:55.000+05:30
diff --git a/api/v1/alert.py b/api/v1/alert.py
@@ -29,6 +29,7 @@
 from utils.helpers import fetch_location_by_ip, get_preferred_ip_address
 from utils.token import confirm_token, generate_confirmation_token
 from utils.validation import validate_email_with_tld, validate_url, validate_variables
+from utils.safe_encoding import build_safe_url
 
 router = APIRouter()
 templates = Jinja2Templates(directory="templates")
@@ -192,11 +193,11 @@ async def alert_me_verification(verification_token: str, request: Request):
         if not has_exposure and not has_sensitive_exposure:
             return templates.TemplateResponse("email_verify.html", {"request": request})
 
-        # If exposures are found
-        base_url = "https://xposedornot.com/"
-        email_param = f"email={user_email}"
-        token_param = f"&token={verification_token}"
-        breaches_link = f"{base_url}data-breaches-risks.html?{email_param}{token_param}"
+        # If exposures are found, generate link with properly encoded parameters
+        breaches_link = build_safe_url(
+            "https://xposedornot.com/data-breaches-risks.html",
+            {"email": user_email, "token": verification_token},
+        )
         return templates.TemplateResponse(
             "email_success.html",
             {"request": request, "breaches_link": breaches_link},
diff --git a/api/v1/analytics.py b/api/v1/analytics.py
@@ -49,6 +49,11 @@
     validate_url,
     validate_variables,
 )
+from utils.safe_encoding import (
+    build_safe_url,
+    escape_html_attr,
+    escape_url_fragment,
+)
 
 router = APIRouter()
 templates = Jinja2Templates(directory="templates")
@@ -230,7 +235,8 @@ async def domain_verify(request: Request, verification_token: str) -> HTMLRespon
             )
 
         user_email = await confirm_token(verification_token)
-        if not user_email:
+        # Re-validate email from token for defense-in-depth
+        if not user_email or not validate_email_with_tld(user_email):
             return HTMLResponse(
                 content=templates.TemplateResponse(
                     "domain_dashboard_error.html", {"request": request}
@@ -254,11 +260,11 @@ async def domain_verify(request: Request, verification_token: str) -> HTMLRespon
         except Exception as e:
             raise
 
-        # Generate dashboard link
-        base_url = "https://xposedornot.com/"
-        email_param = f"email={user_email}"
-        token_param = f"token={verification_token}"
-        dashboard_link = f"{base_url}breach-dashboard.html?{email_param}&{token_param}"
+        # Generate dashboard link with properly encoded parameters
+        dashboard_link = build_safe_url(
+            "https://xposedornot.com/breach-dashboard.html",
+            {"email": user_email, "token": verification_token},
+        )
 
         return HTMLResponse(
             content=templates.TemplateResponse(
@@ -534,8 +540,8 @@ async def send_domain_breaches(
             for breach_name in breaches:
                 breach_logo = all_breaches_logo.get(breach_name, "")
                 details = (
-                    f"<img src='{breach_logo}' style='height:40px;width:65px;' />"
-                    f"<a target='_blank' href='https://xposedornot.com/xposed/#{breach_name}'>"
+                    f"<img src='{escape_html_attr(breach_logo)}' style='height:40px;width:65px;' />"
+                    f"<a target='_blank' href='https://xposedornot.com/xposed/#{escape_url_fragment(breach_name)}'>"
                     " &nbsp;Details</a>"
                 )
                 breach_node = {
@@ -890,8 +896,8 @@ async def get_breach_hierarchy_analytics(
                 logo = query.get("logo", "default_logo.jpg")
 
                 details = (
-                    f"<img src='{logo}' style='height:40px;width:65px;' />"
-                    f"<a target='_blank' href='https://xposedornot.com/xposed/#{bid}'>"
+                    f"<img src='{escape_html_attr(logo)}' style='height:40px;width:65px;' />"
+                    f"<a target='_blank' href='https://xposedornot.com/xposed/#{escape_url_fragment(bid)}'>"
                     " &nbsp;Details</a>"
                 )
 
@@ -927,8 +933,8 @@ async def get_breach_hierarchy_analytics(
                 logo = query.get("logo", "default_logo.jpg")
 
                 details = (
-                    f"<img src='{logo}' style='height:40px;width:65px;' />"
-                    f"<a target='_blank' href='https://xposedornot.com/xposed/#{bid}'>"
+                    f"<img src='{escape_html_attr(logo)}' style='height:40px;width:65px;' />"
+                    f"<a target='_blank' href='https://xposedornot.com/xposed/#{escape_url_fragment(bid)}'>"
                     " &nbsp;Details</a>"
                 )
 
diff --git a/api/v1/feeds.py b/api/v1/feeds.py
@@ -9,6 +9,7 @@
 from models.base import BaseResponse
 from services.send_email import send_exception_email
 from utils.custom_limiter import custom_rate_limiter
+from utils.safe_encoding import escape_rss_content, escape_url_fragment
 
 router = APIRouter()
 
@@ -87,12 +88,14 @@ async def rss_feed(request: Request):
 
             feed_entry.id(entity_key)
             feed_entry.title(entity_key)
-            feed_entry.link(href="https://xposedornot.com/xposed#" + entity_key)
+            feed_entry.link(
+                href="https://xposedornot.com/xposed#" + escape_url_fragment(entity_key)
+            )
 
             description = (
-                str(entity["xposure_desc"])
+                escape_rss_content(entity["xposure_desc"])
                 + ". Exposed data: "
-                + str(entity["xposed_data"])
+                + escape_rss_content(entity["xposed_data"])
             )
             feed_entry.description(description=description)
             feed_entry.pubDate(entity["timestamp"])
diff --git a/api/v1/monthly_digest_helpers.py b/api/v1/monthly_digest_helpers.py
@@ -5,6 +5,7 @@
 import time
 from datetime import datetime, timedelta, timezone
 from google.cloud import datastore
+from utils.safe_encoding import escape_html, build_safe_url
 
 logger = logging.getLogger(__name__)
 
@@ -259,14 +260,14 @@ def generate_mobile_exposure_cards(user_exposures: list) -> str:
         cards += f"""
                     <div class="mobile-card">
                         <div style="font-weight: bold; color: #e74c3c; font-size: 16px; margin-bottom: 8px;">
-                            🚨 {exposure.get('breach_name', 'Unknown')}
+                            🚨 {escape_html(exposure.get('breach_name', 'Unknown'))}
                         </div>
                         <div style="color: #495057; font-size: 14px; line-height: 1.4;">
-                            <div style="margin: 4px 0;"><strong>Breached:</strong> {exposure.get('breach_date', 'Unknown')}</div>
-                            <div style="margin: 4px 0;"><strong>Added:</strong> {exposure.get('added_date', 'Unknown')}</div>
+                            <div style="margin: 4px 0;"><strong>Breached:</strong> {escape_html(exposure.get('breach_date', 'Unknown'))}</div>
+                            <div style="margin: 4px 0;"><strong>Added:</strong> {escape_html(exposure.get('added_date', 'Unknown'))}</div>
                             <div style="margin: 4px 0;"><strong>Records:</strong> {exposure.get('records_count', 0):,}</div>
                             <div style="margin: 8px 0 0 0; padding: 8px; background-color: #f8f9fa; border-radius: 4px; font-size: 12px;">
-                                <strong>Data Exposed:</strong><br>{exposure.get('data_exposed', 'Unknown')}
+                                <strong>Data Exposed:</strong><br>{escape_html(exposure.get('data_exposed', 'Unknown'))}
                             </div>
                         </div>
                     </div>"""
@@ -286,14 +287,14 @@ def generate_mobile_breach_cards(new_breaches: list) -> str:
         cards += f"""
                     <div class="mobile-card">
                         <div style="font-weight: bold; color: #f39c12; font-size: 16px; margin-bottom: 8px;">
-                            🚨 {breach.get('breach_name', 'Unknown')}
+                            🚨 {escape_html(breach.get('breach_name', 'Unknown'))}
                         </div>
                         <div style="color: #495057; font-size: 14px; line-height: 1.4;">
-                            <div style="margin: 4px 0;"><strong>Breached:</strong> {breach.get('breach_date', 'Unknown')}</div>
-                            <div style="margin: 4px 0;"><strong>Added:</strong> {breach.get('added_date', 'Unknown')}</div>
+                            <div style="margin: 4px 0;"><strong>Breached:</strong> {escape_html(breach.get('breach_date', 'Unknown'))}</div>
+                            <div style="margin: 4px 0;"><strong>Added:</strong> {escape_html(breach.get('added_date', 'Unknown'))}</div>
                             <div style="margin: 4px 0;"><strong>Records:</strong> {breach.get('records_count', 0):,}</div>
                             <div style="margin: 8px 0 0 0; padding: 8px; background-color: #f8f9fa; border-radius: 4px; font-size: 12px;">
-                                <strong>Data Exposed:</strong><br>{breach.get('data_exposed', 'Unknown')}
+                                <strong>Data Exposed:</strong><br>{escape_html(breach.get('data_exposed', 'Unknown'))}
                             </div>
                         </div>
                     </div>"""
@@ -318,15 +319,22 @@ async def generate_html_template(
     # Use the verified domains passed to the function (not derived from exposures)
     # user_domains parameter already contains the verified domains for this user
 
-    # Summary info
-    domains_text = ", ".join(user_domains) if user_domains else "No verified domains"
+    # Summary info (escape domains for safe HTML display)
+    domains_text = (
+        ", ".join(escape_html(d) for d in user_domains)
+        if user_domains
+        else "No verified domains"
+    )
     summary_info = (
         "Your summary: <strong>{} verified domains</strong> ({}) • "
         "<strong>{} exposures</strong> • <strong>{} new breaches</strong> this month"
     ).format(len(user_domains), domains_text, len(user_exposures), len(new_breaches))
 
-    # Dashboard URL
-    dashboard_url = f"https://xposedornot.com/breach-dashboard?email={email}&token={dashboard_token}"
+    # Dashboard URL with properly encoded parameters
+    dashboard_url = build_safe_url(
+        "https://xposedornot.com/breach-dashboard",
+        {"email": email, "token": dashboard_token},
+    )
 
     # Build mobile-friendly cards instead of table rows
     exposure_cards = ""
@@ -335,14 +343,14 @@ async def generate_html_template(
             exposure_cards += f"""
                         <div style="border: 1px solid #dee2e6; border-radius: 6px; padding: 12px; margin: 8px 0; background-color: #fff;">
                             <div style="font-weight: bold; color: #e74c3c; font-size: 16px; margin-bottom: 8px;">
-                                🚨 {exposure.get('breach_name', 'Unknown')}
+                                🚨 {escape_html(exposure.get('breach_name', 'Unknown'))}
                             </div>
                             <div style="color: #495057; font-size: 14px; line-height: 1.4;">
-                                <div style="margin: 4px 0;"><strong>Breached:</strong> {exposure.get('breach_date', 'Unknown')}</div>
-                                <div style="margin: 4px 0;"><strong>Added:</strong> {exposure.get('added_date', 'Unknown')}</div>
+                                <div style="margin: 4px 0;"><strong>Breached:</strong> {escape_html(exposure.get('breach_date', 'Unknown'))}</div>
+                                <div style="margin: 4px 0;"><strong>Added:</strong> {escape_html(exposure.get('added_date', 'Unknown'))}</div>
                                 <div style="margin: 4px 0;"><strong>Records:</strong> {exposure.get('records_count', 0):,}</div>
                                 <div style="margin: 8px 0 0 0; padding: 8px; background-color: #f8f9fa; border-radius: 4px; font-size: 12px;">
-                                    <strong>Data Exposed:</strong><br>{exposure.get('data_exposed', 'Unknown')}
+                                    <strong>Data Exposed:</strong><br>{escape_html(exposure.get('data_exposed', 'Unknown'))}
                                 </div>
                             </div>
                         </div>"""
@@ -360,14 +368,14 @@ async def generate_html_template(
             breach_cards += f"""
                         <div style="border: 1px solid #dee2e6; border-radius: 6px; padding: 12px; margin: 8px 0; background-color: #fff;">
                             <div style="font-weight: bold; color: #f39c12; font-size: 16px; margin-bottom: 8px;">
-                                🚨 {breach.get('breach_name', 'Unknown')}
+                                🚨 {escape_html(breach.get('breach_name', 'Unknown'))}
                             </div>
                             <div style="color: #495057; font-size: 14px; line-height: 1.4;">
-                                <div style="margin: 4px 0;"><strong>Breached:</strong> {breach.get('breach_date', 'Unknown')}</div>
-                                <div style="margin: 4px 0;"><strong>Added:</strong> {breach.get('added_date', 'Unknown')}</div>
+                                <div style="margin: 4px 0;"><strong>Breached:</strong> {escape_html(breach.get('breach_date', 'Unknown'))}</div>
+                                <div style="margin: 4px 0;"><strong>Added:</strong> {escape_html(breach.get('added_date', 'Unknown'))}</div>
                                 <div style="margin: 4px 0;"><strong>Records:</strong> {breach.get('records_count', 0):,}</div>
                                 <div style="margin: 8px 0 0 0; padding: 8px; background-color: #f8f9fa; border-radius: 4px; font-size: 12px;">
-                                    <strong>Data Exposed:</strong><br>{breach.get('data_exposed', 'Unknown')}
+                                    <strong>Data Exposed:</strong><br>{escape_html(breach.get('data_exposed', 'Unknown'))}
                                 </div>
                             </div>
                         </div>"""
@@ -448,7 +456,7 @@ async def generate_html_template(
 
             <!-- Email Footer -->
             <div style="background-color: #f8f9fa; padding: 20px; margin-top: 30px; border-top: 1px solid #dee2e6; font-size: 12px; color: #6c757d; text-align: center;">
-                <p style="margin: 0 0 10px 0;">This email was sent to <strong>{email}</strong> because of monthly breach notifications in XposedOrNot.com.</p>
+                <p style="margin: 0 0 10px 0;">This email was sent to <strong>{escape_html(email)}</strong> because of monthly breach notifications in XposedOrNot.com.</p>
                 <p style="margin: 0;">© 2025 XposedOrNot. All rights reserved. | <a href="https://xposedornot.com/dashboard" style="color: #6c757d;">Visit Website</a></p>
             </div>
 
diff --git a/utils/safe_encoding.py b/utils/safe_encoding.py
@@ -0,0 +1,136 @@
+"""
+Safe encoding utilities for preventing XSS and URL injection vulnerabilities.
+
+This module provides standardized functions for:
+- HTML escaping (prevent XSS in displayed content)
+- URL encoding (prevent parameter injection in URLs)
+- RSS content escaping (prevent XSS in feed readers)
+
+Usage:
+    from utils.safe_encoding import escape_html, build_safe_url, escape_html_attr
+
+    # For displaying user data in HTML text:
+    f"Hello, {escape_html(username)}"
+
+    # For building URLs with parameters:
+    url = build_safe_url("https://example.com/page.html", {"email": email, "token": token})
+
+    # For HTML attributes (src, href with user data):
+    f"<img src='{escape_html_attr(logo_url)}' />"
+
+    # For fragment identifiers:
+    f"<a href='https://example.com/#{escape_url_fragment(breach_name)}'>"
+"""
+
+import html
+from urllib.parse import urlencode, quote
+from typing import Optional, Dict, Any
+
+
+def escape_html(value: Any) -> str:
+    """
+    Escape HTML special characters for safe display in HTML content.
+
+    Converts: < > & " ' to their HTML entity equivalents.
+    Use for: Text content displayed to users in HTML.
+
+    Args:
+        value: The value to escape (will be converted to string)
+
+    Returns:
+        HTML-escaped string safe for display
+
+    Example:
+        escape_html("<script>alert('xss')</script>")
+        → "&lt;script&gt;alert(&#x27;xss&#x27;)&lt;/script&gt;"
+    """
+    if value is None:
+        return ""
+    return html.escape(str(value), quote=True)
+
+
+def escape_html_attr(value: Any) -> str:
+    """
+    Escape value for use in HTML attributes (src, href, etc.).
+
+    Use for: Dynamic values in img src, anchor href, style attributes, etc.
+
+    Args:
+        value: The value to escape (will be converted to string)
+
+    Returns:
+        HTML-escaped string safe for use in attributes
+
+    Example:
+        f"<img src='{escape_html_attr(logo_url)}' />"
+    """
+    if value is None:
+        return ""
+    return html.escape(str(value), quote=True)
+
+
+def escape_url_fragment(value: Any) -> str:
+    """
+    Escape value for use in URL fragment identifiers (#anchor).
+
+    Use for: Anchor links with dynamic values like breach names.
+
+    Args:
+        value: The value to escape (will be converted to string)
+
+    Returns:
+        URL-encoded string safe for use in fragments
+
+    Example:
+        f"<a href='https://example.com/page#{escape_url_fragment(section_name)}'>"
+    """
+    if value is None:
+        return ""
+    # Quote everything except alphanumerics and safe chars
+    return quote(str(value), safe="")
+
+
+def build_safe_url(base_url: str, params: Optional[Dict[str, Any]] = None) -> str:
+    """
+    Build a URL with properly encoded query parameters.
+
+    Use for: Any URL construction with dynamic query parameters.
+
+    Args:
+        base_url: The base URL (e.g., "https://example.com/page.html")
+        params: Dictionary of query parameters to encode
+
+    Returns:
+        Complete URL with properly encoded query string
+
+    Example:
+        build_safe_url("https://example.com/dashboard", {"email": "user@test.com"})
+        → "https://example.com/dashboard?email=user%40test.com"
+    """
+    if not params:
+        return base_url
+
+    query_string = urlencode(params)
+    separator = "&" if "?" in base_url else "?"
+    return f"{base_url}{separator}{query_string}"
+
+
+def escape_rss_content(value: Any) -> str:
+    """
+    Escape content for RSS feed descriptions.
+
+    RSS readers may render HTML, so we escape to prevent XSS.
+    Uses quote=False to avoid escaping quotes in text content.
+
+    Args:
+        value: The value to escape (will be converted to string)
+
+    Returns:
+        XML-escaped string safe for RSS content
+
+    Example:
+        escape_rss_content(breach_description)
+    """
+    if value is None:
+        return ""
+    return html.escape(str(value), quote=False)
