fix: harden webhook/link/OAuth-avatar SSRF (advisory clusters A/B/C/E)

apps/api/plane/app/views/webhook/base.py

@@ -81,7 +81,7 @@ def patch(self, request, slug, pk):
         serializer = WebhookSerializer(
             webhook,
             data=request.data,
-            context={request: request},
+            context={"request": request},
             partial=True,
             fields=(
                 "id",

apps/api/plane/bgtasks/webhook_task.py

@@ -52,7 +52,7 @@
 from plane.license.utils.instance_value import get_email_configuration
 from plane.utils.email import generate_plain_text_from_html
 from plane.utils.exception_logger import log_exception
-from plane.utils.ip_address import validate_url
+from plane.utils.url_security import pinned_fetch
 
 
 SERIALIZER_MAPPER = {
@@ -307,22 +307,20 @@ def webhook_send_task(
         return
 
     try:
-        # Re-validate the webhook URL at send time to prevent DNS-rebinding attacks
-        validate_url(
+        # Resolve + validate the webhook URL and pin the connection to the
+        # validated IP. Pinning closes the DNS-rebinding TOCTOU (validating the
+        # name then letting requests re-resolve it lets an attacker swap in an
+        # internal IP between the two lookups). Redirects are never followed, so
+        # a 3xx Location cannot bounce the request to an internal address
+        # (GHSA-mq87-52pf-hm3h / cluster C).
+        response = pinned_fetch(
+            "POST",
             webhook.url,
             allowed_ips=settings.WEBHOOK_ALLOWED_IPS,
             allowed_hosts=settings.WEBHOOK_ALLOWED_HOSTS,
-        )
-
-        # Send the webhook event
-        # allow_redirects=False prevents SSRF via 3xx hops to internal addresses
-        # bypassing the validate_url() check above (GHSA-mq87-52pf-hm3h).
-        response = requests.post(
-            webhook.url,
             headers=headers,
             json=payload,
             timeout=30,
-            allow_redirects=False,
         )
 
         # Log the webhook request
@@ -366,6 +364,25 @@ def webhook_send_task(
             return
         raise requests.RequestException()
 
+    except ValueError as e:
+        # SSRF validation failure (blocked/internal target or unresolvable host).
+        # Not retryable — record it so the failure is visible to the admin, but
+        # do not raise (no Celery retry) and do not auto-deactivate (the cause
+        # may be transient DNS).
+        save_webhook_log(
+            webhook=webhook,
+            request_method=action,
+            request_headers=headers,
+            request_body=payload,
+            response_status=400,
+            response_headers="",
+            response_body=f"Webhook URL rejected: {e}",
+            retry_count=self.request.retries,
+            event_type=event,
+        )
+        logger.warning(f"Webhook {webhook.id} URL rejected: {e}")
+        return
+
     except Exception as e:
         log_exception(e)
         return

apps/api/plane/bgtasks/work_item_link_task.py

@@ -17,6 +17,8 @@
 from typing import Optional
 from plane.db.models import IssueLink
 from plane.utils.exception_logger import log_exception
+from plane.utils.ip_address import is_blocked_ip
+from plane.utils.url_security import pinned_fetch, pinned_fetch_following_redirects
 
 logger = logging.getLogger("plane.worker")
 
@@ -56,10 +58,12 @@ def validate_url_ip(url: str) -> None:
     if not addr_info:
         raise ValueError("No IP addresses found for the hostname")
 
-    # Check every resolved IP against blocked ranges to prevent SSRF
+    # Check every resolved IP against blocked ranges to prevent SSRF. The
+    # actual fetch is pinned to the validated IP (see safe_get), so this acts
+    # as an early, fail-closed pre-filter.
     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:
+        ip = ipaddress.ip_address(addr[4][0].split("%")[0])
+        if is_blocked_ip(ip):
             raise ValueError("Access to private/internal networks is not allowed")
 
 
@@ -72,8 +76,9 @@ def safe_get(
     timeout: int = 1,
 ) -> Tuple[requests.Response, str]:
     """
-    Perform a GET request that validates every redirect hop against private IPs.
-    Prevents SSRF by ensuring no redirect lands on a private/internal address.
+    Perform a GET request that resolves, validates and pins every hop to its
+    validated IP. Prevents SSRF via private/internal targets, DNS rebinding
+    (TOCTOU) and redirects that bounce to internal addresses.
 
     Args:
         url: The URL to fetch
@@ -85,32 +90,16 @@ def safe_get(
 
     Raises:
         ValueError: If any URL in the redirect chain points to a private IP
-        requests.RequestException: On network errors
-        RuntimeError: If max redirects exceeded
+        requests.RequestException: On network errors (incl. TooManyRedirects)
     """
-    validate_url_ip(url)
-
-    current_url = url
-    response = requests.get(
-        current_url, headers=headers, timeout=timeout, allow_redirects=False
+    return pinned_fetch_following_redirects(
+        "GET",
+        url,
+        headers=headers,
+        timeout=timeout,
+        max_redirects=MAX_REDIRECTS,
     )
 
-    redirect_count = 0
-    while response.is_redirect:
-        if redirect_count >= MAX_REDIRECTS:
-            raise RuntimeError(f"Too many redirects for URL: {url}")
-        redirect_url = response.headers.get("Location")
-        if not redirect_url:
-            break
-        current_url = urljoin(current_url, redirect_url)
-        validate_url_ip(current_url)
-        redirect_count += 1
-        response = requests.get(
-            current_url, headers=headers, timeout=timeout, allow_redirects=False
-        )
-
-    return response, current_url
-
 
 def crawl_work_item_link_title_and_favicon(url: str) -> Dict[str, Any]:
     """
@@ -199,14 +188,13 @@ def find_favicon_url(soup: Optional[BeautifulSoup], base_url: str) -> Optional[s
     parsed_url = urlparse(base_url)
     fallback_url = f"{parsed_url.scheme}://{parsed_url.netloc}/favicon.ico"
 
-    # Check if fallback exists
+    # Check if fallback exists (pinned to the validated IP).
     try:
-        validate_url_ip(fallback_url)
-        response = requests.head(fallback_url, timeout=2, allow_redirects=False)
+        response = pinned_fetch("HEAD", fallback_url, timeout=2)
 
         if response.status_code == 200:
             return fallback_url
-    except requests.RequestException as e:
+    except (requests.RequestException, ValueError) as e:
         log_exception(e, warning=True)
         return None