CVE-2024-1727: Cross-Site Request Forgery (CSRF) (CWE-352)
value: A Cross-Site Request Forgery (CSRF) vulnerability in gradio-app/gradio allows attackers to upload multiple large files to a victim's system if they are running Gradio locally. By crafting a malicious HTML page that triggers an unauthorized file upload to the victim's server, an attacker can deplete the system's disk space, potentially leading to a denial of service. This issue affects the file upload functionality as implemented in gradio/routes.py.
CWE description: The web application does not, or can not, sufficiently verify whether a well-formed, valid, consistent request was intentionally provided by the user who submitted the request.
Published: 2024-03-21T20:15Z
Repository: https://github.com/gradio-app/gradio
Language: Python
Files changed: .changeset/olive-symbols-heal.md, gradio/route_utils.py, gradio/routes.py, test/test_routes.py
Commit date: 2024-02-22 11:54:51 -0800
Commit message: Tighten CORS rules (#7503)

* tighten cors rules

* add changeset

* cors policy

* cors

* add changeset

* lint

* changes

* changes

* changes

* logging

* add null

* changes

* changes

* options

* options

* safe changes

* let browser enforce cors

* clean

* route utils

* fix

* fix test

* fix

---------

Co-authored-by: gradio-pr-bot <gradio-pr-bot@users.noreply.github.com>
Diff: diff --git a/.changeset/olive-symbols-heal.md b/.changeset/olive-symbols-heal.md
new file mode 100644
index 000000000..20c1c05f5
--- /dev/null
+++ b/.changeset/olive-symbols-heal.md
@@ -0,0 +1,5 @@
+---
+"gradio": patch
+---
+
+feat:Tighten CORS rules
diff --git a/gradio/route_utils.py b/gradio/route_utils.py
index 7ab8d0a58..2dccb5900 100644
--- a/gradio/route_utils.py
+++ b/gradio/route_utils.py
@@ -7,18 +7,20 @@ import re
 import shutil
 from collections import deque
 from dataclasses import dataclass as python_dataclass
 from tempfile import NamedTemporaryFile, _TemporaryFileWrapper
 from typing import TYPE_CHECKING, AsyncGenerator, BinaryIO, List, Optional, Tuple, Union
+from urllib.parse import urlparse
 
 import fastapi
 import httpx
 import multipart
 from gradio_client.documentation import document
 from multipart.multipart import parse_options_header
 from starlette.datastructures import FormData, Headers, UploadFile
 from starlette.formparsers import MultiPartException, MultipartPart
+from starlette.middleware.base import BaseHTTPMiddleware
 
 from gradio import processing_utils, utils
 from gradio.data_classes import PredictBody
 from gradio.exceptions import Error
 from gradio.helpers import EventData
@@ -581,5 +583,60 @@ def starts_with_protocol(string: str) -> bool:
     """This regex matches strings that start with a scheme (one or more characters not including colon, slash, or space)
     followed by ://
     """
     pattern = r"^[a-zA-Z][a-zA-Z0-9+\-.]*://"
     return re.match(pattern, string) is not None
+
+
+def get_hostname(url: str) -> str:
+    """
+    Returns the hostname of a given url, or an empty string if the url cannot be parsed.
+    Examples:
+        get_hostname("https://www.gradio.app") -> "www.gradio.app"
+        get_hostname("localhost:7860") -> "localhost"
+        get_hostname("127.0.0.1") -> "127.0.0.1"
+    """
+    if not url:
+        return ""
+    if "://" not in url:
+        url = "http://" + url
+    try:
+        return urlparse(url).hostname or ""
+    except Exception:
+        return ""
+
+
+class CustomCORSMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: fastapi.Request, call_next):
+        host: str = request.headers.get("host", "")
+        origin: str = request.headers.get("origin", "")
+        host_name = get_hostname(host)
+        origin_name = get_hostname(origin)
+
+        # Any of these hosts suggests that the Gradio app is running locally.
+        # Note: "null" is a special case that happens if a Gradio app is running
+        # as an embedded web component in a local static webpage.
+        localhost_aliases = ["localhost", "127.0.0.1", "0.0.0.0", "null"]
+        is_preflight = (
+            request.method == "OPTIONS"
+            and "access-control-request-method" in request.headers
+        )
+
+        if host_name in localhost_aliases and origin_name not in localhost_aliases:
+            allow_origin_header = None
+        else:
+            allow_origin_header = origin
+
+        if is_preflight:
+            response = fastapi.Response()
+        else:
+            response = await call_next(request)
+
+        if allow_origin_header:
+            response.headers["Access-Control-Allow-Origin"] = allow_origin_header
+        response.headers[
+            "Access-Control-Allow-Methods"
+        ] = "GET, POST, PUT, DELETE, OPTIONS"
+        response.headers[
+            "Access-Control-Allow-Headers"
+        ] = "Origin, Content-Type, Accept"
+        return response
diff --git a/gradio/routes.py b/gradio/routes.py
index 29386e59b..591a30b0b 100644
--- a/gradio/routes.py
+++ b/gradio/routes.py
@@ -27,11 +27,10 @@ from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional, Type
 import fastapi
 import httpx
 import markupsafe
 import orjson
 from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException, status
-from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import (
     FileResponse,
     HTMLResponse,
     JSONResponse,
     PlainTextResponse,
@@ -53,10 +52,11 @@ from gradio.data_classes import ComponentServerBody, PredictBody, ResetBody
 from gradio.exceptions import Error
 from gradio.oauth import attach_oauth
 from gradio.processing_utils import add_root_url
 from gradio.queueing import Estimation
 from gradio.route_utils import (  # noqa: F401
+    CustomCORSMiddleware,
     FileUploadProgress,
     FileUploadProgressNotQueuedError,
     FileUploadProgressNotTrackedError,
     GradioMultiPartParser,
     GradioUploadFile,
@@ -194,16 +194,11 @@ class App(FastAPI):
         app_kwargs.setdefault("default_response_class", ORJSONResponse)
         app = App(**app_kwargs)
         app.configure_app(blocks)
 
         if not wasm_utils.IS_WASM:
-            app.add_middleware(
-                CORSMiddleware,
-                allow_origins=["*"],
-                allow_methods=["*"],
-                allow_headers=["*"],
-            )
+            app.add_middleware(CustomCORSMiddleware)
 
         @app.get("/user")
         @app.get("/user/")
         def get_current_user(request: fastapi.Request) -> Optional[str]:
             token = request.cookies.get(
diff --git a/test/test_routes.py b/test/test_routes.py
index 239befb25..26f321663 100644
--- a/test/test_routes.py
+++ b/test/test_routes.py
@@ -460,10 +460,28 @@ class TestRoutes:
         response = client.get("/")
         assert response.is_success
         response = client.get("/config/")
         assert response.is_success
 
+    def test_cors_restrictions(self):
+        io = gr.Interface(lambda s: s.name, gr.File(), gr.File())
+        app, _, _ = io.launch(prevent_thread_lock=True)
+        client = TestClient(app)
+        custom_headers = {
+            "host": "localhost:7860",
+            "origin": "https://example.com",
+        }
+        file_response = client.get("/config", headers=custom_headers)
+        assert "access-control-allow-origin" not in file_response.headers
+        custom_headers = {
+            "host": "localhost:7860",
+            "origin": "127.0.0.1",
+        }
+        file_response = client.get("/config", headers=custom_headers)
+        assert file_response.headers["access-control-allow-origin"] == "127.0.0.1"
+        io.close()
+
 
 class TestApp:
     def test_create_app(self):
         app = routes.App.create_app(Interface(lambda x: x, "text", "text"))
         assert isinstance(app, FastAPI)

Vulnerable code: from fastapi.middleware.cors import CORSMiddleware
            app.add_middleware(
                CORSMiddleware,
                allow_origins=["*"],
                allow_methods=["*"],
                allow_headers=["*"],
            )
Fixed code: ---
"gradio": patch
---

feat:Tighten CORS rules
from urllib.parse import urlparse
from starlette.middleware.base import BaseHTTPMiddleware


def get_hostname(url: str) -> str:
    """
    Returns the hostname of a given url, or an empty string if the url cannot be parsed.
    Examples:
        get_hostname("https://www.gradio.app") -> "www.gradio.app"
        get_hostname("localhost:7860") -> "localhost"
        get_hostname("127.0.0.1") -> "127.0.0.1"
    """
    if not url:
        return ""
    if "://" not in url:
        url = "http://" + url
    try:
        return urlparse(url).hostname or ""
    except Exception:
        return ""


class CustomCORSMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: fastapi.Request, call_next):
        host: str = request.headers.get("host", "")
        origin: str = request.headers.get("origin", "")
        host_name = get_hostname(host)
        origin_name = get_hostname(origin)

        # Any of these hosts suggests that the Gradio app is running locally.
        # Note: "null" is a special case that happens if a Gradio app is running
        # as an embedded web component in a local static webpage.
        localhost_aliases = ["localhost", "127.0.0.1", "0.0.0.0", "null"]
        is_preflight = (
            request.method == "OPTIONS"
            and "access-control-request-method" in request.headers
        )

        if host_name in localhost_aliases and origin_name not in localhost_aliases:
            allow_origin_header = None
        else:
            allow_origin_header = origin

        if is_preflight:
            response = fastapi.Response()
        else:
            response = await call_next(request)

        if allow_origin_header:
            response.headers["Access-Control-Allow-Origin"] = allow_origin_header
        response.headers[
            "Access-Control-Allow-Methods"
        ] = "GET, POST, PUT, DELETE, OPTIONS"
        response.headers[
            "Access-Control-Allow-Headers"
        ] = "Origin, Content-Type, Accept"
        return response
    CustomCORSMiddleware,
            app.add_middleware(CustomCORSMiddleware)
    def test_cors_restrictions(self):
        io = gr.Interface(lambda s: s.name, gr.File(), gr.File())
        app, _, _ = io.launch(prevent_thread_lock=True)
        client = TestClient(app)
        custom_headers = {
            "host": "localhost:7860",
            "origin": "https://example.com",
        }
        file_response = client.get("/config", headers=custom_headers)
        assert "access-control-allow-origin" not in file_response.headers
        custom_headers = {
            "host": "localhost:7860",
            "origin": "127.0.0.1",
        }
        file_response = client.get("/config", headers=custom_headers)
        assert file_response.headers["access-control-allow-origin"] == "127.0.0.1"
        io.close()