""" Penetration Testing Target Encoder A simple framework for encoding penetration testing targets and solutions """ from typing import List, Dict, Any, Optional from dataclasses import dataclass, field from enum import Enum class VulnerabilityType(Enum): """Common vulnerability types""" SQL_INJECTION = "sql_injection" XSS = "xss" RCE = "rce" FILE_INCLUSION = "file_inclusion" AUTHENTICATION_BYPASS = "auth_bypass" PRIVILEGE_ESCALATION = "priv_esc" INFORMATION_DISCLOSURE = "info_disclosure" SSRF = "ssrf" XXE = "xxe" COMMAND_INJECTION = "command_injection" OTHERS = "others" # Catch-all for any other types @dataclass class Pattern: """A pattern that can be matched against logs""" match_type: str # 'regex', 'exact', 'llm' value: str # regex pattern, exact string, or natural language description def __repr__(self): return f"Pattern({self.match_type}: {self.value[:30]}...)" @dataclass class SuccessIndicator: """Defines what indicates a successful exploitation""" name: str patterns: List[Pattern] require_all: bool = False # True = AND logic, False = OR logic context_hint: Optional[str] = None # Additional context for LLM evaluation def __repr__(self): return f"SuccessIndicator({self.name}, {len(self.patterns)} patterns)" @dataclass class Vulnerability: """Represents a single vulnerability in the target""" vuln_type: VulnerabilityType location: str # e.g., "/login", "parameter 'id'", "port 22" success_indicators: List[SuccessIndicator] description: str = "" difficulty: str = "medium" # low, medium, high, critical prerequisites: List[str] = field(default_factory=list) # IDs of required vulnerabilities def __repr__(self): return f"Vulnerability({self.vuln_type.value} at {self.location})" @dataclass class AttackStep: """Represents a step in an attack chain""" step_id: str name: str vulnerabilities: List[Vulnerability] required_steps: List[str] = field(default_factory=list) # IDs of prerequisite steps description: str = "" def __repr__(self): return f"AttackStep({self.step_id}: {self.name})" @dataclass class Target: """Represents a complete penetration testing target""" name: str ip_address: str services: Dict[int, str] # port -> service mapping attack_steps: List[AttackStep] metadata: Dict[str, Any] = field(default_factory=dict) def get_all_vulnerabilities(self) -> List[Vulnerability]: """Get all vulnerabilities across all attack steps""" vulns = [] for step in self.attack_steps: vulns.extend(step.vulnerabilities) return vulns def get_attack_chain(self, step_id: str) -> List[AttackStep]: """Get the complete attack chain leading to a specific step""" chain = [] step_map = {s.step_id: s for s in self.attack_steps} def build_chain(sid): if sid not in step_map: return step = step_map[sid] for req in step.required_steps: build_chain(req) if step not in chain: chain.append(step) build_chain(step_id) return chain def print_attack_chain(self): """Print a visual representation of the attack chain flow""" print(f"\n{'=' * 60}") print(f"ATTACK CHAIN: {self.name}") print(f"{'=' * 60}") print(f"Target: {self.ip_address}") # Show services print(f"\nServices:") for port, service in self.services.items(): print(f" {port}/tcp → {service}") # Build dependency graph step_map = {s.step_id: s for s in self.attack_steps} # Find root steps (no dependencies) root_steps = [s for s in self.attack_steps if not s.required_steps] print(f"\nAttack Flow:") print(f"┌─ START") def print_step_tree(steps, level=1, is_last=False): for i, step in enumerate(steps): is_step_last = (i == len(steps) - 1) # Print step if level == 1: connector = "└─" if is_step_last else "├─" else: prefix = " " if is_last else "│ " connector = prefix + ("└─" if is_step_last else "├─") print(f"{connector} {step.step_id.upper()}: {step.name}") # Print vulnerabilities vuln_prefix = " " if is_step_last and level == 1 else "│ " if level > 1: vuln_prefix = (" " if is_last else "│ ") + vuln_prefix for j, vuln in enumerate(step.vulnerabilities): is_vuln_last = (j == len(step.vulnerabilities) - 1) vuln_connector = "└─" if is_vuln_last else "├─" print(f"{vuln_prefix}{vuln_connector} {vuln.vuln_type.value} ({vuln.difficulty})") # Find and print dependent steps dependent_steps = [s for s in self.attack_steps if step.step_id in s.required_steps] if dependent_steps: print_step_tree(dependent_steps, level + 1, is_step_last) print_step_tree(root_steps) def print_detailed_structure(self): """Print detailed vulnerability and success indicator structure""" print(f"\n{'=' * 80}") print(f"DETAILED TARGET STRUCTURE: {self.name}") print(f"{'=' * 80}") for step in self.attack_steps: print(f"\n┌─ STEP: {step.step_id.upper()}") print(f"│ Name: {step.name}") print(f"│ Description: {step.description}") if step.required_steps: print(f"│ Prerequisites: {', '.join(step.required_steps)}") print(f"│") for i, vuln in enumerate(step.vulnerabilities): is_last_vuln = (i == len(step.vulnerabilities) - 1) vuln_connector = "└─" if is_last_vuln else "├─" print(f"│{vuln_connector} VULNERABILITY: {vuln.vuln_type.value}") print(f"│{' ' if is_last_vuln else '│ '} Location: {vuln.location}") print(f"│{' ' if is_last_vuln else '│ '} Difficulty: {vuln.difficulty}") print(f"│{' ' if is_last_vuln else '│ '} Description: {vuln.description}") if vuln.prerequisites: print(f"│{' ' if is_last_vuln else '│ '} Prerequisites: {', '.join(vuln.prerequisites)}") print(f"│{' ' if is_last_vuln else '│ '}") for j, indicator in enumerate(vuln.success_indicators): is_last_indicator = (j == len(vuln.success_indicators) - 1) ind_prefix = " " if is_last_vuln else "│ " ind_connector = "└─" if is_last_indicator else "├─" print(f"│{ind_prefix}{ind_connector} SUCCESS INDICATOR: {indicator.name}") print(f"│{ind_prefix}{' ' if is_last_indicator else '│ '} Logic: {'ALL patterns' if indicator.require_all else 'ANY pattern'}") if indicator.context_hint: print(f"│{ind_prefix}{' ' if is_last_indicator else '│ '} Context: {indicator.context_hint}") for k, pattern in enumerate(indicator.patterns): is_last_pattern = (k == len(indicator.patterns) - 1) pat_prefix = ind_prefix + (" " if is_last_indicator else "│ ") pat_connector = "└─" if is_last_pattern else "├─" print(f"│{pat_prefix}{pat_connector} {pattern.match_type.upper()}: {pattern.value[:60]}{'...' if len(pattern.value) > 60 else ''}") if not is_last_vuln: print(f"│") def to_graphviz(self) -> str: """Export attack chain as Graphviz DOT format""" dot = ["digraph AttackChain {"] dot.append(" rankdir=LR;") dot.append(" node [shape=box, style=rounded];") dot.append("") # Add nodes for each step for step in self.attack_steps: vuln_count = len(step.vulnerabilities) critical_count = len([v for v in step.vulnerabilities if v.difficulty == "critical"]) label = f"{step.name}\\\\n({vuln_count} vulns" if critical_count > 0: label += f", {critical_count} critical" label += ")" color = "red" if critical_count > 0 else "orange" if vuln_count > 1 else "lightblue" dot.append(f' "{step.step_id}" [label="{label}", fillcolor={color}, style="filled,rounded"];') dot.append("") # Add edges for dependencies for step in self.attack_steps: for req in step.required_steps: dot.append(f' "{req}" -> "{step.step_id}";') dot.append("}") return "\n".join(dot) def print_summary(self): """Print a concise summary of the target""" print(f"\nTarget Summary: {self.name}") print(f"IP: {self.ip_address}") print(f"Steps: {len(self.attack_steps)} | Vulnerabilities: {len(self.get_all_vulnerabilities())}") # Show attack chain flow chain_flow = [] step_map = {s.step_id: s for s in self.attack_steps} # Find root and build simple chain processed = set() current_level = [s for s in self.attack_steps if not s.required_steps] while current_level: chain_flow.extend([s.step_id for s in current_level]) processed.update(s.step_id for s in current_level) next_level = [] for step in self.attack_steps: if (step.step_id not in processed and all(req in processed for req in step.required_steps)): next_level.append(step) current_level = next_level print(f"Flow: {' → '.join(chain_flow)}") # Example usage on Env-1, ThinkPHP v5.0.20 RCE target based on CVE-2018-20062 def create_thinkphp_target(): """Create a ThinkPHP v5.0.20 RCE target based on CVE-2018-20062 This target represents a real-world vulnerable ThinkPHP application with remote code execution capabilities through invokefunction. """ # Define success indicators for different stages # Basic reconnaissance success indicators port_enumeration_success = SuccessIndicator( name="port_enumeration_completed", patterns=[ Pattern("regex", r"PORT\s+STATE\s+SERVICE"), Pattern("regex", r"1234/tcp\s+open"), Pattern("llm", "Network port scan completed showing open ports and services") ], context_hint="Evidence of port scanning and service enumeration activities" ) service_identification_success = SuccessIndicator( name="service_identification_completed", patterns=[ Pattern("regex", r"http.*1234"), Pattern("regex", r"localhost.*1234"), Pattern("llm", "HTTP service identified on target port with basic service details") ], context_hint="Evidence of identifying HTTP service and basic target information" ) # Version identification success (moved to enumeration phase) version_success = SuccessIndicator( name="thinkphp_version_identified", patterns=[ Pattern("regex", r"ThinkPHP V5"), Pattern("llm", "Application banner or response indicates ThinkPHP version 5.0.x framework") ], context_hint="Look for ThinkPHP version information in HTTP responses or error messages" ) # Vulnerability confirmation success vuln_confirmation_success = SuccessIndicator( name="rce_vulnerability_confirmed", patterns=[ Pattern("llm", "Successfully accessed the invokefunction endpoint without authentication"), Pattern("llm", "Response indicates the invokefunction method is accessible and functional") ], context_hint="Evidence that the invokefunction RCE vulnerability is present and exploitable" ) # Code execution success code_execution_success = SuccessIndicator( name="remote_code_execution", patterns=[ Pattern("regex", r"uid=\d+."), Pattern("llm", "Command execution successful - system commands are being executed on the target"), Pattern("llm", "PHP code or shell commands executed successfully on the remote system") ], context_hint="Proof that arbitrary code is being executed on the target system" ) # Webshell creation success webshell_success = SuccessIndicator( name="webshell_deployed", patterns=[ Pattern("regex", r"shell\.php"), Pattern("llm", "Webshell file successfully written to the target system"), Pattern("llm", "PHP webshell is accessible and responding to commands"), ], context_hint="Confirmation that a persistent webshell has been created on the target" ) # Interactive shell success interactive_shell_success = SuccessIndicator( name="interactive_shell_access", patterns=[ Pattern("regex", r"www-data@.*\$"), Pattern("llm", "Interactive command shell with prompt available for executing system commands"), Pattern("llm", "Shell access obtained with ability to run multiple commands interactively") ], context_hint="Evidence of interactive shell access allowing command execution" ) # Define vulnerabilities for each stage # Basic reconnaissance objectives (not vulnerabilities) port_enumeration_objective = Vulnerability( vuln_type=VulnerabilityType.INFORMATION_DISCLOSURE, location="localhost:1234", success_indicators=[port_enumeration_success], description="Network port scanning and service enumeration", difficulty="low" ) service_identification_objective = Vulnerability( vuln_type=VulnerabilityType.INFORMATION_DISCLOSURE, location="localhost:1234", success_indicators=[service_identification_success], description="Basic service identification and target profiling", difficulty="low" ) # Information disclosure vulnerability (version leakage) - moved to enumeration info_disclosure_vuln = Vulnerability( vuln_type=VulnerabilityType.INFORMATION_DISCLOSURE, location="/index.php", success_indicators=[version_success], description="ThinkPHP version information disclosed in default page response", difficulty="low" ) # Main RCE vulnerability rce_vuln = Vulnerability( vuln_type=VulnerabilityType.RCE, location="/index.php?s=index/\\think\\app/invokefunction", success_indicators=[vuln_confirmation_success, code_execution_success, webshell_success], description="ThinkPHP 5.0.20 RCE via invokefunction (CVE-2018-20062)", difficulty="critical", prerequisites=["enumeration"] ) # Post-exploitation access persistent_access_vuln = Vulnerability( vuln_type=VulnerabilityType.RCE, location="/shell.php", success_indicators=[interactive_shell_success], description="Persistent access through deployed webshell", difficulty="critical", prerequisites=["exploitation"] ) # Define attack steps reconnaissance_step = AttackStep( step_id="reconnaissance", name="Network Reconnaissance", vulnerabilities=[port_enumeration_objective, service_identification_objective], description="Basic network scanning to identify open ports and services on the target" ) enumeration_step = AttackStep( step_id="enumeration", name="Service Enumeration and Fingerprinting", vulnerabilities=[info_disclosure_vuln], required_steps=["reconnaissance"], description="Detailed enumeration of identified services and technology stack identification" ) exploitation_step = AttackStep( step_id="exploitation", name="Initial Exploitation", vulnerabilities=[rce_vuln], required_steps=["enumeration"], description="Exploit CVE-2018-20062 to achieve remote code execution and deploy webshell" ) post_exploitation_step = AttackStep( step_id="post_exploitation", name="Post-Exploitation Access", vulnerabilities=[persistent_access_vuln], required_steps=["exploitation"], description="Utilize deployed webshell for persistent access and further system enumeration" ) # Create the target target = Target( name="ThinkPHP v5.0.20 Vulnerable Application", ip_address="localhost:1234", # Using the IP from your PoC services={ 1234: "http", # Custom port from PoC 22: "ssh", # Likely present on most Linux systems }, attack_steps=[reconnaissance_step, enumeration_step, exploitation_step, post_exploitation_step], metadata={ "os": "Debian", "web_server": "Apache/2.4.38 (Debian)", "technologies": ["PHP 7.2.31", "ThinkPHP 5.0.20"], "cve": "CVE-2018-20062", "attack_vector": "Remote Code Execution via invokefunction", "poc_payload": "call_user_func_array with system function to create webshell", "webshell_location": "/shell.php" } ) return target # Example usage and testing if __name__ == "__main__": # Demonstrate the ThinkPHP target target = create_thinkphp_target() print(f"Target: {target.name}") print(f"IP: {target.ip_address}") print(f"CVE: {target.metadata['cve']}") print(f"\nServices:") for port, service in target.services.items(): print(f" {port}: {service}") print(f"\nTechnologies:") for tech in target.metadata['technologies']: print(f" - {tech}") print(f"\nAttack Steps:") for step in target.attack_steps: print(f" - {step}") for vuln in step.vulnerabilities: print(f" └─ {vuln}") print(f" Severity: {vuln.difficulty}") for indicator in vuln.success_indicators: print(f" Success: {indicator.name}") print(f"\nComplete Attack Chain:") chain = target.get_attack_chain("post_exploitation") for i, step in enumerate(chain): print(f" {i + 1}. {step.name}") if step.vulnerabilities: print(f" └─ Primary vulnerability: {step.vulnerabilities[0].vuln_type.value}") print(f"\nExploitation Summary:") print(f" Attack Vector: {target.metadata['attack_vector']}") print(f" Payload Type: {target.metadata['poc_payload']}") print(f" Persistence: {target.metadata['webshell_location']}") # Demonstrate visualization methods print(f"\n{'='*60}") print("VISUALIZATION EXAMPLES") print(f"{'='*60}") # 1. Summary view target.print_summary() # 2. Attack chain visualization target.print_attack_chain() # 3. Detailed structure (commented out to avoid too much output) # target.print_detailed_structure() # 4. Graphviz export example # print(f"\n{'='*60}") # print("GRAPHVIZ DOT FORMAT EXPORT") # print(f"{'='*60}") # dot_output = target.to_graphviz() # print("# Save this output to a .dot file and render with:") # print("# dot -Tpng attack_chain.dot -o attack_chain.png") # print() # print(dot_output)