vynte-infra/plane
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[SECUR-117] fix: update Ip address validation as common util function #6208

Browse Source
This commit is contained in:
sriram veeraghanta
2026-03-10 14:42:53 +05:30
committed by GitHub
parent 2b623bb72a
commit f5fe4a5cd8
4 changed files with 77 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/plane/app/serializers/webhook.py
+13 -64
View File
@@ -10,8 +10,6 @@
# NOTICE: Proprietary and confidential. Unauthorized use or distribution is prohibited.
# Python imports
import socket
import ipaddress
from urllib.parse import urlparse
# Django imports
@@ -24,87 +22,38 @@ from rest_framework import serializers
from .base import DynamicBaseSerializer
from plane.db.models import Webhook, WebhookLog
from plane.db.models.webhook import validate_domain, validate_schema
from plane.utils.ip_address import validate_url
class WebhookSerializer(DynamicBaseSerializer):
url = serializers.URLField(validators=[validate_schema, validate_domain])
def create(self, validated_data):
url = validated_data.get("url", None)
# Extract the hostname from the URL
hostname = urlparse(url).hostname
if not hostname:
raise serializers.ValidationError({"url": "Invalid URL: No hostname found."})
# Resolve the hostname to IP addresses
def _validate_webhook_url(self, url):
"""Validate a webhook URL against SSRF and disallowed domain rules."""
try:
ip_addresses = socket.getaddrinfo(hostname, None)
except socket.gaierror:
raise serializers.ValidationError({"url": "Hostname could not be resolved."})
validate_url(url, block_private=not settings.IS_SELF_MANAGED)
except ValueError as e:
raise serializers.ValidationError({"url": str(e)})
if not ip_addresses:
raise serializers.ValidationError({"url": "No IP addresses found for the hostname."})
for addr in ip_addresses:
ip = ipaddress.ip_address(addr[4][0])
if ip.is_loopback:
raise serializers.ValidationError({"url": "URL resolves to a blocked IP address."})
# if in cloud environment, private IP addresses are also not allowed
if not settings.IS_SELF_MANAGED and ip.is_private:
raise serializers.ValidationError({"url": "URL resolves to a blocked IP address."})
# Additional validation for multiple request domains and their subdomains
hostname = urlparse(url).hostname
request = self.context.get("request")
disallowed_domains = ["plane.so"] # Add your disallowed domains here
disallowed_domains = ["plane.so"]
if request:
request_host = request.get_host().split(":")[0] # Remove port if present
request_host = request.get_host().split(":")[0]
disallowed_domains.append(request_host)
# Check if hostname is a subdomain or exact match of any disallowed domain
if any(hostname == domain or hostname.endswith("." + domain) for domain in disallowed_domains):
raise serializers.ValidationError({"url": "URL domain or its subdomain is not allowed."})
def create(self, validated_data):
url = validated_data.get("url", None)
self._validate_webhook_url(url)
return Webhook.objects.create(**validated_data)
def update(self, instance, validated_data):
url = validated_data.get("url", None)
if url:
# Extract the hostname from the URL
hostname = urlparse(url).hostname
if not hostname:
raise serializers.ValidationError({"url": "Invalid URL: No hostname found."})
# Resolve the hostname to IP addresses
try:
ip_addresses = socket.getaddrinfo(hostname, None)
except socket.gaierror:
raise serializers.ValidationError({"url": "Hostname could not be resolved."})
if not ip_addresses:
raise serializers.ValidationError({"url": "No IP addresses found for the hostname."})
for addr in ip_addresses:
ip = ipaddress.ip_address(addr[4][0])
if ip.is_loopback:
raise serializers.ValidationError({"url": "URL resolves to a blocked IP address."})
# if in cloud environment, private IP addresses are also not allowed
if not settings.IS_SELF_MANAGED and ip.is_private:
raise serializers.ValidationError({"url": "URL resolves to a blocked IP address."})
# Additional validation for multiple request domains and their subdomains
request = self.context.get("request")
disallowed_domains = ["plane.so"] # Add your disallowed domains here
if request:
request_host = request.get_host().split(":")[0] # Remove port if present
disallowed_domains.append(request_host)
# Check if hostname is a subdomain or exact match of any disallowed domain
if any(hostname == domain or hostname.endswith("." + domain) for domain in disallowed_domains):
raise serializers.ValidationError({"url": "URL domain or its subdomain is not allowed."})
self._validate_webhook_url(url)
return super().update(instance, validated_data)
class Meta:
apps/api/plane/bgtasks/link_crawler_task.py
+6 -41
View File
@@ -11,9 +11,7 @@
# Python imports
import base64
import ipaddress
import logging
import socket
from typing import Any, Dict, Optional
from urllib.parse import urljoin, urlparse
@@ -24,6 +22,7 @@ from celery import shared_task
# Module imports
from plane.utils.exception_logger import log_exception
from plane.utils.ip_address import validate_url
from plane.utils.link_crawler import LinkCrawlerEntity, LinkCrawlerInput
logger = logging.getLogger("plane.worker")
@@ -33,40 +32,6 @@ DEFAULT_FAVICON = "PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoP
MAX_REDIRECTS = 5
def validate_url_ip(url: str) -> None:
"""
Validate that a URL doesn't point to a private/internal IP address.
Resolves hostnames to IPs before checking.
Args:
url: The URL to validate
Raises:
ValueError: If the URL points to a private/internal IP
"""
parsed = urlparse(url)
hostname = parsed.hostname
if not hostname:
raise ValueError("Invalid URL: No hostname found")
if parsed.scheme not in ("http", "https"):
raise ValueError("Invalid URL scheme. Only HTTP and HTTPS are allowed")
try:
addr_info = socket.getaddrinfo(hostname, None)
except socket.gaierror:
raise ValueError("Hostname could not be resolved")
if not addr_info:
raise ValueError("No IP addresses found for the hostname")
for addr in addr_info:
ip = ipaddress.ip_address(addr[4][0])
if ip.is_private or ip.is_loopback or ip.is_reserved or ip.is_link_local:
raise ValueError("Access to private/internal networks is not allowed")
def find_favicon_url(soup: Optional[BeautifulSoup], base_url: str) -> Optional[str]:
"""
Find the favicon URL from HTML soup.
@@ -91,7 +56,7 @@ def find_favicon_url(soup: Optional[BeautifulSoup], base_url: str) -> Optional[s
if favicon_tag and favicon_tag.get("href"):
favicon_href = urljoin(base_url, favicon_tag["href"])
try:
validate_url_ip(favicon_href)
validate_url(favicon_href)
except ValueError:
continue
return favicon_href
@@ -100,7 +65,7 @@ def find_favicon_url(soup: Optional[BeautifulSoup], base_url: str) -> Optional[s
fallback_url = f"{parsed_url.scheme}://{parsed_url.netloc}/favicon.ico"
try:
validate_url_ip(fallback_url)
validate_url(fallback_url)
response = requests.head(fallback_url, timeout=2, allow_redirects=False)
if response.status_code == 200:
return fallback_url
@@ -133,7 +98,7 @@ def fetch_and_encode_favicon(
"favicon_base64": f"data:image/svg+xml;base64,{DEFAULT_FAVICON}",
}
validate_url_ip(favicon_url)
validate_url(favicon_url)
response = requests.get(favicon_url, headers=headers, timeout=1)
content_type = response.headers.get("content-type", "image/x-icon")
@@ -172,7 +137,7 @@ def crawl_link_metadata(url: str) -> Dict[str, Any]:
final_url = url
try:
validate_url_ip(final_url)
validate_url(final_url)
except ValueError as e:
logger.warning(f"URL validation failed for {url}: {e}")
return {
@@ -191,7 +156,7 @@ def crawl_link_metadata(url: str) -> Dict[str, Any]:
if not redirect_url:
break
final_url = urljoin(final_url, redirect_url)
validate_url_ip(final_url)
validate_url(final_url)
redirect_count += 1
response = requests.get(final_url, headers=headers, timeout=1, allow_redirects=False)
apps/api/plane/ee/views/app/webhook/base.py
+15 -44
View File
@@ -10,11 +10,10 @@
# NOTICE: Proprietary and confidential. Unauthorized use or distribution is prohibited.
# Python imports
import socket
import ipaddress
from urllib.parse import urlparse
# Django imports
from django.conf import settings
from django.db import IntegrityError
# Third party imports
@@ -25,6 +24,7 @@ from rest_framework.response import Response
from plane.ee.permissions import WorkSpaceAdminPermission
from plane.db.models import Workspace, Webhook
from plane.ee.views.base import BaseAPIView
from plane.utils.ip_address import validate_url
class InternalWebhookEndpoint(BaseAPIView):
@@ -41,56 +41,27 @@ class InternalWebhookEndpoint(BaseAPIView):
status=status.HTTP_400_BAD_REQUEST,
)
# Extract the hostname from the URL
hostname = urlparse(url).hostname
if not hostname:
return Response(
{"url": "Invalid URL: No hostname found."},
status=status.HTTP_400_BAD_REQUEST,
)
# Resolve the hostname to IP addresses
# Validate URL against SSRF
try:
ip_addresses = socket.getaddrinfo(hostname, None)
except socket.gaierror:
validate_url(url, block_private=not settings.IS_SELF_MANAGED)
except ValueError as e:
return Response(
{"url": "Hostname could not be resolved."},
{"url": str(e)},
status=status.HTTP_400_BAD_REQUEST,
)
if not ip_addresses:
return Response(
{"url": "No IP addresses found for the hostname."},
status=status.HTTP_400_BAD_REQUEST,
)
for addr in ip_addresses:
_ip = ipaddress.ip_address(addr[4][0])
# if ip.is_loopback:
# return Response(
# {"url": "URL resolves to a blocked IP address."},
# status=status.HTTP_400_BAD_REQUEST,
# )
# if in cloud environment, private IP addresses are also not allowed
# if not settings.IS_SELF_MANAGED and ip.is_private:
# return Response(
# {"url": "URL resolves to a blocked IP address."},
# status=status.HTTP_400_BAD_REQUEST,
# )
# Additional validation for multiple request domains and their subdomains
disallowed_domains = ["plane.so"] # Add your disallowed domains here
# Additional validation for disallowed domains
hostname = urlparse(url).hostname
disallowed_domains = ["plane.so"]
if request:
request_host = request.get_host().split(":")[0] # Remove port if present
request_host = request.get_host().split(":")[0]
disallowed_domains.append(request_host)
# Check if hostname is a subdomain or exact match of any disallowed domain
# if any(
# hostname == domain or hostname.endswith("." + domain)
# for domain in disallowed_domains
# ):
# return Response({"url": "URL domain or its subdomain is not allowed."})
if any(hostname == domain or hostname.endswith("." + domain) for domain in disallowed_domains):
return Response(
{"url": "URL domain or its subdomain is not allowed."},
status=status.HTTP_400_BAD_REQUEST,
)
try:
existing_webhooks = Webhook.objects.filter(workspace_id=workspace.id, url=url).first()
apps/api/plane/utils/ip_address.py
+43
View File
@@ -10,6 +10,49 @@
# NOTICE: Proprietary and confidential. Unauthorized use or distribution is prohibited.
# Python imports
import ipaddress
import socket
from urllib.parse import urlparse
def validate_url(url, block_private=True):
"""
Validate that a URL doesn't resolve to a private/internal IP address (SSRF protection).
Args:
url: The URL to validate.
block_private: If True, also block private IPs. Set to False for self-managed
deployments where private IPs may be legitimate webhook targets.
Raises:
ValueError: If the URL is invalid or resolves to a blocked IP.
"""
parsed = urlparse(url)
hostname = parsed.hostname
if not hostname:
raise ValueError("Invalid URL: No hostname found")
if parsed.scheme not in ("http", "https"):
raise ValueError("Invalid URL scheme. Only HTTP and HTTPS are allowed")
try:
addr_info = socket.getaddrinfo(hostname, None)
except socket.gaierror:
raise ValueError("Hostname could not be resolved")
if not addr_info:
raise ValueError("No IP addresses found for the hostname")
for addr in addr_info:
ip = ipaddress.ip_address(addr[4][0])
if ip.is_loopback or ip.is_reserved or ip.is_link_local:
raise ValueError("Access to private/internal networks is not allowed")
if block_private and ip.is_private:
raise ValueError("Access to private/internal networks is not allowed")
def get_client_ip(request):
x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
if x_forwarded_for: