fix (backend): Patching the SSRF vulnerability in Github/Web Search/R…

…equest related blocks (#8531)

autogpt_platform/backend/backend/blocks/ai_shortform_video_block.py

@@ -3,12 +3,12 @@
 from enum import Enum
 from typing import Literal
 
-import requests
 from autogpt_libs.supabase_integration_credentials_store.types import APIKeyCredentials
 from pydantic import SecretStr
 
 from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
 from backend.data.model import CredentialsField, CredentialsMetaInput, SchemaField
+from backend.util.request import requests
 
 TEST_CREDENTIALS = APIKeyCredentials(
     id="01234567-89ab-cdef-0123-456789abcdef",
@@ -217,7 +217,6 @@ def create_webhook(self):
         url = "https://webhook.site/token"
         headers = {"Accept": "application/json", "Content-Type": "application/json"}
         response = requests.post(url, headers=headers)
-        response.raise_for_status()
         webhook_data = response.json()
         return webhook_data["uuid"], f"https://webhook.site/{webhook_data['uuid']}"
 
@@ -228,14 +227,12 @@ def create_video(self, api_key: SecretStr, payload: dict) -> dict:
         logger.debug(
             f"API Response Status Code: {response.status_code}, Content: {response.text}"
         )
-        response.raise_for_status()
         return response.json()
 
     def check_video_status(self, api_key: SecretStr, pid: str) -> dict:
         url = f"https://www.revid.ai/api/public/v2/status?pid={pid}"
         headers = {"key": api_key.get_secret_value()}
         response = requests.get(url, headers=headers)
-        response.raise_for_status()
         return response.json()
 
     def wait_for_video(

autogpt_platform/backend/backend/blocks/github/_api.py

@@ -0,0 +1,43 @@
+from urllib.parse import urlparse
+
+from backend.blocks.github._auth import GithubCredentials
+from backend.util.request import Requests
+
+
+def _convert_to_api_url(url: str) -> str:
+    """
+    Converts a standard GitHub URL to the corresponding GitHub API URL.
+    Handles repository URLs, issue URLs, pull request URLs, and more.
+    """
+    parsed_url = urlparse(url)
+    path_parts = parsed_url.path.strip("/").split("/")
+
+    if len(path_parts) >= 2:
+        owner, repo = path_parts[0], path_parts[1]
+        api_base = f"https://api.github.com/repos/{owner}/{repo}"
+
+        if len(path_parts) > 2:
+            additional_path = "/".join(path_parts[2:])
+            api_url = f"{api_base}/{additional_path}"
+        else:
+            # Repository base URL
+            api_url = api_base
+    else:
+        raise ValueError("Invalid GitHub URL format.")
+
+    return api_url
+
+
+def _get_headers(credentials: GithubCredentials) -> dict[str, str]:
+    return {
+        "Authorization": credentials.bearer(),
+        "Accept": "application/vnd.github.v3+json",
+    }
+
+
+def get_api(credentials: GithubCredentials) -> Requests:
+    return Requests(
+        trusted_origins=["https://api.github.com", "https://github.com"],
+        extra_url_validator=_convert_to_api_url,
+        extra_headers=_get_headers(credentials),
+    )

autogpt_platform/backend/backend/blocks/github/issues.py

@@ -1,9 +1,11 @@
-import requests
+from urllib.parse import urlparse
+
 from typing_extensions import TypedDict
 
 from backend.data.block import Block, BlockCategory, BlockOutput, BlockSchema
 from backend.data.model import SchemaField
 
+from ._api import get_api
 from ._auth import (
     TEST_CREDENTIALS,
     TEST_CREDENTIALS_INPUT,
@@ -13,6 +15,10 @@
 )
 
 
+def is_github_url(url: str) -> bool:
+    return urlparse(url).netloc == "github.com"
+
+
 # --8<-- [start:GithubCommentBlockExample]
 class GithubCommentBlock(Block):
     class Input(BlockSchema):
@@ -62,27 +68,10 @@ def __init__(self):
     def post_comment(
         credentials: GithubCredentials, issue_url: str, body_text: str
     ) -> tuple[int, str]:
-        if "/pull/" in issue_url:
-            api_url = (
-                issue_url.replace("github.com", "api.github.com/repos").replace(
-                    "/pull/", "/issues/"
-                )
-                + "/comments"
-            )
-        else:
-            api_url = (
-                issue_url.replace("github.com", "api.github.com/repos") + "/comments"
-            )
-
-        headers = {
-            "Authorization": credentials.bearer(),
-            "Accept": "application/vnd.github.v3+json",
-        }
+        api = get_api(credentials)
         data = {"body": body_text}
-
-        response = requests.post(api_url, headers=headers, json=data)
-        response.raise_for_status()
-
+        comments_url = issue_url + "/comments"
+        response = api.post(comments_url, json=data)
         comment = response.json()
         return comment["id"], comment["html_url"]
 
@@ -156,16 +145,10 @@ def __init__(self):
     def create_issue(
         credentials: GithubCredentials, repo_url: str, title: str, body: str
     ) -> tuple[int, str]:
-        api_url = repo_url.replace("github.com", "api.github.com/repos") + "/issues"
-        headers = {
-            "Authorization": credentials.bearer(),
-            "Accept": "application/vnd.github.v3+json",
-        }
+        api = get_api(credentials)
         data = {"title": title, "body": body}
-
-        response = requests.post(api_url, headers=headers, json=data)
-        response.raise_for_status()
-
+        issues_url = repo_url + "/issues"
+        response = api.post(issues_url, json=data)
         issue = response.json()
         return issue["number"], issue["html_url"]
 
@@ -232,21 +215,12 @@ def __init__(self):
     def read_issue(
         credentials: GithubCredentials, issue_url: str
     ) -> tuple[str, str, str]:
-        api_url = issue_url.replace("github.com", "api.github.com/repos")
-
-        headers = {
-            "Authorization": credentials.bearer(),
-            "Accept": "application/vnd.github.v3+json",
-        }
-
-        response = requests.get(api_url, headers=headers)
-        response.raise_for_status()
-
+        api = get_api(credentials)
+        response = api.get(issue_url)
         data = response.json()
         title = data.get("title", "No title found")
         body = data.get("body", "No body content found")
         user = data.get("user", {}).get("login", "No user found")
-
         return title, body, user
 
     def run(
@@ -318,20 +292,13 @@ def __init__(self):
     def list_issues(
         credentials: GithubCredentials, repo_url: str
     ) -> list[Output.IssueItem]:
-        api_url = repo_url.replace("github.com", "api.github.com/repos") + "/issues"
-        headers = {
-            "Authorization": credentials.bearer(),
-            "Accept": "application/vnd.github.v3+json",
-        }
-
-        response = requests.get(api_url, headers=headers)
-        response.raise_for_status()
-
+        api = get_api(credentials)
+        issues_url = repo_url + "/issues"
+        response = api.get(issues_url)
         data = response.json()
         issues: list[GithubListIssuesBlock.Output.IssueItem] = [
             {"title": issue["title"], "url": issue["html_url"]} for issue in data
         ]
-
         return issues
 
     def run(
@@ -385,28 +352,10 @@ def __init__(self):
 
     @staticmethod
     def add_label(credentials: GithubCredentials, issue_url: str, label: str) -> str:
-        # Convert the provided GitHub URL to the API URL
-        if "/pull/" in issue_url:
-            api_url = (
-                issue_url.replace("github.com", "api.github.com/repos").replace(
-                    "/pull/", "/issues/"
-                )
-                + "/labels"
-            )
-        else:
-            api_url = (
-                issue_url.replace("github.com", "api.github.com/repos") + "/labels"
-            )
-
-        headers = {
-            "Authorization": credentials.bearer(),
-            "Accept": "application/vnd.github.v3+json",
-        }
+        api = get_api(credentials)
         data = {"labels": [label]}
-
-        response = requests.post(api_url, headers=headers, json=data)
-        response.raise_for_status()
-
+        labels_url = issue_url + "/labels"
+        api.post(labels_url, json=data)
         return "Label added successfully"
 
     def run(
@@ -463,31 +412,9 @@ def __init__(self):
 
     @staticmethod
     def remove_label(credentials: GithubCredentials, issue_url: str, label: str) -> str:
-        # Convert the provided GitHub URL to the API URL
-        if "/pull/" in issue_url:
-            api_url = (
-                issue_url.replace("github.com", "api.github.com/repos").replace(
-                    "/pull/", "/issues/"
-                )
-                + f"/labels/{label}"
-            )
-        else:
-            api_url = (
-                issue_url.replace("github.com", "api.github.com/repos")
-                + f"/labels/{label}"
-            )
-
-        # Log the constructed API URL for debugging
-        print(f"Constructed API URL: {api_url}")
-
-        headers = {
-            "Authorization": credentials.bearer(),
-            "Accept": "application/vnd.github.v3+json",
-        }
-
-        response = requests.delete(api_url, headers=headers)
-        response.raise_for_status()
-
+        api = get_api(credentials)
+        label_url = issue_url + f"/labels/{label}"
+        api.delete(label_url)
         return "Label removed successfully"
 
     def run(
@@ -550,23 +477,10 @@ def assign_issue(
         issue_url: str,
         assignee: str,
     ) -> str:
-        # Extracting repo path and issue number from the issue URL
-        repo_path, issue_number = issue_url.replace("https://github.com/", "").split(
-            "/issues/"
-        )
-        api_url = (
-            f"https://api.github.com/repos/{repo_path}/issues/{issue_number}/assignees"
-        )
-
-        headers = {
-            "Authorization": credentials.bearer(),
-            "Accept": "application/vnd.github.v3+json",
-        }
+        api = get_api(credentials)
+        assignees_url = issue_url + "/assignees"
         data = {"assignees": [assignee]}
-
-        response = requests.post(api_url, headers=headers, json=data)
-        response.raise_for_status()
-
+        api.post(assignees_url, json=data)
         return "Issue assigned successfully"
 
     def run(
@@ -629,23 +543,10 @@ def unassign_issue(
         issue_url: str,
         assignee: str,
     ) -> str:
-        # Extracting repo path and issue number from the issue URL
-        repo_path, issue_number = issue_url.replace("https://github.com/", "").split(
-            "/issues/"
-        )
-        api_url = (
-            f"https://api.github.com/repos/{repo_path}/issues/{issue_number}/assignees"
-        )
-
-        headers = {
-            "Authorization": credentials.bearer(),
-            "Accept": "application/vnd.github.v3+json",
-        }
+        api = get_api(credentials)
+        assignees_url = issue_url + "/assignees"
         data = {"assignees": [assignee]}
-
-        response = requests.delete(api_url, headers=headers, json=data)
-        response.raise_for_status()
-
+        api.delete(assignees_url, json=data)
         return "Issue unassigned successfully"
 
     def run(