""" AUTARCH Workflow Module Automated pentest pipeline orchestration Run multi-step security assessments with automated data flow between tools. """ import os import sys import json import subprocess import re import time from pathlib import Path from datetime import datetime # Module metadata DESCRIPTION = "Automated pentest workflow" AUTHOR = "darkHal" VERSION = "1.0" CATEGORY = "offense" sys.path.insert(0, str(Path(__file__).parent.parent)) from core.banner import Colors, clear_screen, display_banner class WorkflowRunner: """Orchestrate multi-step pentest workflows.""" def __init__(self): self.results_dir = Path("results") self.results_dir.mkdir(exist_ok=True) def print_status(self, msg, level="info"): icons = {"info": f"{Colors.CYAN}[*]", "success": f"{Colors.GREEN}[+]", "warning": f"{Colors.YELLOW}[!]", "error": f"{Colors.RED}[-]"} icon = icons.get(level, icons["info"]) print(f" {icon} {msg}{Colors.RESET}") # ========================================================================= # MENU # ========================================================================= def show_menu(self): clear_screen() display_banner() print(f"{Colors.RED}{Colors.BOLD} Automated Workflow{Colors.RESET}") print(f"{Colors.DIM} Multi-step pentest pipeline orchestration{Colors.RESET}") print(f"{Colors.DIM} {'─' * 50}{Colors.RESET}") print() print(f" {Colors.RED}[1]{Colors.RESET} New Workflow {Colors.DIM}- Full automated pipeline{Colors.RESET}") print(f" {Colors.RED}[2]{Colors.RESET} Quick Scan {Colors.DIM}- Nmap → CVE → Report (no LLM){Colors.RESET}") print(f" {Colors.RED}[3]{Colors.RESET} Resume Workflow {Colors.DIM}- Load saved state{Colors.RESET}") print() print(f" {Colors.DIM}[0]{Colors.RESET} Back") print() # ========================================================================= # NMAP SCAN (shared helper) # ========================================================================= def _nmap_service_scan(self, target): """Run nmap service detection scan on target.""" self.print_status(f"Running nmap -sV -T4 on {target}...", "info") try: result = subprocess.run( f"nmap -sV --top-ports 20 -T4 {target}", shell=True, capture_output=True, text=True, timeout=300 ) if result.returncode != 0: self.print_status("nmap scan failed", "error") return [] services = [] port_re = re.compile(r'(\d+)/(tcp|udp)\s+open\s+(\S+)\s*(.*)') for line in result.stdout.split('\n'): m = port_re.match(line.strip()) if m: parts = m.group(4).strip().split() services.append({ 'port': int(m.group(1)), 'protocol': m.group(2), 'service': parts[0] if parts else m.group(3), 'version': ' '.join(parts[1:]) if len(parts) > 1 else '' }) self.print_status(f"Found {len(services)} open services", "success") return services except subprocess.TimeoutExpired: self.print_status("nmap timed out after 5 minutes", "error") return [] except Exception as e: self.print_status(f"Scan error: {e}", "error") return [] # ========================================================================= # CVE CORRELATION (shared helper) # ========================================================================= def _correlate_cves(self, services): """Correlate services with CVEs from the database.""" try: from core.cve import get_cve_db cve_db = get_cve_db() except Exception as e: self.print_status(f"CVE database unavailable: {e}", "warning") return [] SERVICE_TO_CPE = { 'apache': ('apache', 'http_server'), 'nginx': ('f5', 'nginx'), 'openssh': ('openbsd', 'openssh'), 'ssh': ('openbsd', 'openssh'), 'mysql': ('oracle', 'mysql'), 'postgresql': ('postgresql', 'postgresql'), 'samba': ('samba', 'samba'), 'smb': ('samba', 'samba'), 'vsftpd': ('vsftpd_project', 'vsftpd'), 'proftpd': ('proftpd', 'proftpd'), 'postfix': ('postfix', 'postfix'), 'dovecot': ('dovecot', 'dovecot'), 'php': ('php', 'php'), 'tomcat': ('apache', 'tomcat'), 'isc': ('isc', 'bind'), 'bind': ('isc', 'bind'), } correlations = [] for svc in services: self.print_status(f"Checking CVEs for {svc['service']}:{svc.get('version', '?')} on port {svc['port']}...", "info") cves = [] svc_lower = svc['service'].lower() version = svc.get('version', '').split()[0] if svc.get('version') else '' if svc_lower in SERVICE_TO_CPE and version: vendor, product = SERVICE_TO_CPE[svc_lower] cpe = f"cpe:2.3:a:{vendor}:{product}:{version}:*:*:*:*:*:*:*" try: cves = cve_db.search_cves(cpe_pattern=cpe) except Exception: pass if not cves and version: try: cves = cve_db.search_cves(keyword=f"{svc['service']} {version}") except Exception: pass if cves: self.print_status(f" Found {len(cves)} CVEs", "success") else: self.print_status(f" No CVEs found", "info") correlations.append({ 'service': svc, 'cves': cves[:20] # cap per service }) return correlations # ========================================================================= # FULL WORKFLOW # ========================================================================= def run_workflow(self, target): """Run full automated pentest workflow.""" clear_screen() display_banner() print(f"{Colors.RED}{Colors.BOLD} Full Workflow - {target}{Colors.RESET}") print(f"{Colors.DIM} {'─' * 50}{Colors.RESET}") print() state = { 'target': target, 'started': datetime.now().isoformat(), 'services': [], 'correlations': [], 'exploits': [], 'report': None, 'current_step': 1 } state_file = self.results_dir / f"workflow_{target.replace('.', '-').replace('/', '_')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" # Step 1: Nmap scan print(f"\n{Colors.CYAN}{Colors.BOLD} Step 1/4: Service Detection{Colors.RESET}") print(f"{Colors.DIM} {'─' * 40}{Colors.RESET}") services = self._nmap_service_scan(target) state['services'] = services state['current_step'] = 2 self._save_state(state, state_file) if not services: self.print_status("No services found. Workflow cannot continue.", "warning") input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") return cont = input(f"\n{Colors.WHITE} Continue to CVE correlation? [Y/n]: {Colors.RESET}").strip().lower() if cont == 'n': self.print_status(f"State saved to {state_file}", "info") input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") return # Step 2: CVE correlation print(f"\n{Colors.CYAN}{Colors.BOLD} Step 2/4: CVE Correlation{Colors.RESET}") print(f"{Colors.DIM} {'─' * 40}{Colors.RESET}") correlations = self._correlate_cves(services) state['correlations'] = correlations state['current_step'] = 3 self._save_state(state, state_file) total_cves = sum(len(c.get('cves', [])) for c in correlations) self.print_status(f"Total CVEs found: {total_cves}", "success" if total_cves > 0 else "info") cont = input(f"\n{Colors.WHITE} Continue to exploit suggestion? [Y/n]: {Colors.RESET}").strip().lower() if cont == 'n': # Skip to report self._generate_workflow_report(state) input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") return # Step 3: Exploit suggestion (LLM) print(f"\n{Colors.CYAN}{Colors.BOLD} Step 3/4: Exploit Suggestion{Colors.RESET}") print(f"{Colors.DIM} {'─' * 40}{Colors.RESET}") exploits = self._suggest_exploits(services, correlations) state['exploits'] = exploits state['current_step'] = 4 self._save_state(state, state_file) cont = input(f"\n{Colors.WHITE} Generate report? [Y/n]: {Colors.RESET}").strip().lower() if cont == 'n': self.print_status(f"State saved to {state_file}", "info") input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") return # Step 4: Report print(f"\n{Colors.CYAN}{Colors.BOLD} Step 4/4: Report Generation{Colors.RESET}") print(f"{Colors.DIM} {'─' * 40}{Colors.RESET}") self._generate_workflow_report(state) state['current_step'] = 5 self._save_state(state, state_file) input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") # ========================================================================= # QUICK SCAN # ========================================================================= def quick_scan(self, target): """Run quick scan: Nmap → CVE → Report (no LLM).""" clear_screen() display_banner() print(f"{Colors.RED}{Colors.BOLD} Quick Scan - {target}{Colors.RESET}") print(f"{Colors.DIM} {'─' * 50}{Colors.RESET}") print() start_time = time.time() # Step 1: Nmap print(f"\n{Colors.CYAN}{Colors.BOLD} Step 1/3: Service Detection{Colors.RESET}") print(f"{Colors.DIM} {'─' * 40}{Colors.RESET}") services = self._nmap_service_scan(target) if not services: self.print_status("No services found.", "warning") input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") return # Step 2: CVE correlation print(f"\n{Colors.CYAN}{Colors.BOLD} Step 2/3: CVE Correlation{Colors.RESET}") print(f"{Colors.DIM} {'─' * 40}{Colors.RESET}") correlations = self._correlate_cves(services) total_cves = sum(len(c.get('cves', [])) for c in correlations) self.print_status(f"Total CVEs found: {total_cves}", "success" if total_cves > 0 else "info") # Step 3: Report print(f"\n{Colors.CYAN}{Colors.BOLD} Step 3/3: Report Generation{Colors.RESET}") print(f"{Colors.DIM} {'─' * 40}{Colors.RESET}") scan_time = time.time() - start_time state = { 'target': target, 'services': services, 'correlations': correlations, 'exploits': [], 'scan_time': scan_time } self._generate_workflow_report(state) self.print_status(f"Quick scan completed in {scan_time:.1f}s", "success") input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") # ========================================================================= # RESUME WORKFLOW # ========================================================================= def resume_workflow(self): """Resume a saved workflow from JSON state.""" clear_screen() display_banner() print(f"{Colors.RED}{Colors.BOLD} Resume Workflow{Colors.RESET}") print(f"{Colors.DIM} {'─' * 50}{Colors.RESET}") print() state_files = sorted(self.results_dir.glob("workflow_*.json")) if not state_files: self.print_status("No saved workflows found.", "warning") input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") return for i, f in enumerate(state_files, 1): try: with open(f, 'r') as fh: data = json.load(fh) target = data.get('target', '?') step = data.get('current_step', '?') started = data.get('started', '?') print(f" {Colors.RED}[{i}]{Colors.RESET} {f.name}") print(f" {Colors.DIM}Target: {target} | Step: {step}/4 | Started: {started}{Colors.RESET}") except Exception: print(f" {Colors.RED}[{i}]{Colors.RESET} {f.name} {Colors.DIM}(corrupt){Colors.RESET}") print(f"\n {Colors.DIM}[0]{Colors.RESET} Back") sel = input(f"\n{Colors.WHITE} Select: {Colors.RESET}").strip() if sel == "0": return try: idx = int(sel) - 1 with open(state_files[idx], 'r') as f: state = json.load(f) except (ValueError, IndexError, json.JSONDecodeError) as e: self.print_status(f"Error: {e}", "error") input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") return target = state.get('target', '') current_step = state.get('current_step', 1) state_file = state_files[idx] self.print_status(f"Resuming workflow for {target} at step {current_step}/4", "info") if current_step <= 1: services = self._nmap_service_scan(target) state['services'] = services state['current_step'] = 2 self._save_state(state, state_file) else: services = state.get('services', []) if not services: self.print_status("No services available.", "warning") input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") return if current_step <= 2: print(f"\n{Colors.CYAN}{Colors.BOLD} Step 2/4: CVE Correlation{Colors.RESET}") correlations = self._correlate_cves(services) state['correlations'] = correlations state['current_step'] = 3 self._save_state(state, state_file) else: correlations = state.get('correlations', []) if current_step <= 3: cont = input(f"\n{Colors.WHITE} Run exploit suggestion? [Y/n]: {Colors.RESET}").strip().lower() if cont != 'n': print(f"\n{Colors.CYAN}{Colors.BOLD} Step 3/4: Exploit Suggestion{Colors.RESET}") exploits = self._suggest_exploits(services, correlations) state['exploits'] = exploits state['current_step'] = 4 self._save_state(state, state_file) if current_step <= 4: print(f"\n{Colors.CYAN}{Colors.BOLD} Step 4/4: Report Generation{Colors.RESET}") self._generate_workflow_report(state) state['current_step'] = 5 self._save_state(state, state_file) input(f"\n{Colors.WHITE}Press Enter to continue...{Colors.RESET}") # ========================================================================= # HELPERS # ========================================================================= def _suggest_exploits(self, services, correlations): """Try LLM-based exploit suggestion, fallback to CVE-MSF lookup.""" exploits = [] # Collect all CVEs all_cves = [] for corr in correlations: for cve in corr.get('cves', []): all_cves.append(cve) if not all_cves: self.print_status("No CVEs to suggest exploits for.", "info") return [] # Try LLM try: from core.llm import get_llm llm = get_llm() if llm and llm.is_loaded(): self.print_status("Using LLM for exploit suggestions...", "info") svc_text = "\n".join( f"- {s['service']}:{s.get('version', '?')} on port {s['port']}" for s in services ) cve_text = "\n".join( f"- {c.get('id', '?')} (CVSS {c.get('cvss', '?')}): {c.get('description', '')[:100]}" for c in all_cves[:20] ) prompt = f"""Given these services and vulnerabilities, suggest the top 5 attack paths. Services: {svc_text} CVEs: {cve_text} For each suggestion provide: rank, Metasploit module path (if known), target service, CVE, and reasoning. Format each as: N. MODULE | TARGET | CVE | REASONING""" response = llm.generate(prompt) if response: # Parse suggestions for line in response.split('\n'): line = line.strip() match = re.match(r'\d+\.\s*(.+?)\s*\|\s*(.+?)\s*\|\s*(.+?)\s*\|\s*(.+)', line) if match: exploits.append({ 'module': match.group(1).strip(), 'target': match.group(2).strip(), 'cve': match.group(3).strip(), 'reasoning': match.group(4).strip() }) if exploits: self.print_status(f"LLM suggested {len(exploits)} attack paths", "success") for i, exp in enumerate(exploits, 1): print(f" {Colors.RED}{i}.{Colors.RESET} {exp['module']} → {exp['target']} ({exp['cve']})") return exploits except Exception: pass # Fallback: CVE-to-MSF mapping self.print_status("LLM unavailable, using CVE-to-MSF module lookup...", "warning") try: from core.msf_modules import search_modules for cve in all_cves[:30]: cve_id = cve.get('id', '') if cve_id: matches = search_modules(cve_id) for mod_name, mod_info in matches: exploits.append({ 'module': mod_name, 'target': mod_info.get('description', '')[:60], 'cve': cve_id, 'reasoning': f"Direct CVE match (CVSS {cve.get('cvss', '?')})" }) except Exception as e: self.print_status(f"MSF module lookup failed: {e}", "warning") if exploits: self.print_status(f"Found {len(exploits)} exploit matches", "success") for i, exp in enumerate(exploits[:10], 1): print(f" {Colors.RED}{i}.{Colors.RESET} {exp['module']} ({exp['cve']})") else: self.print_status("No exploit matches found.", "info") return exploits def _generate_workflow_report(self, state): """Generate HTML report from workflow state.""" target = state.get('target', 'unknown') # Build network_data from services network_data = None services = state.get('services', []) if services: network_data = [{ 'ip': target, 'hostname': target, 'os_guess': '-', 'ports': services }] vuln_data = state.get('correlations') or None exploit_data = state.get('exploits') or None try: from core.report_generator import get_report_generator rg = get_report_generator() report_path = rg.generate_pentest_report( target=target, network_data=network_data, vuln_data=vuln_data, exploit_data=exploit_data ) self.print_status(f"Report saved to {report_path}", "success") state['report'] = report_path except Exception as e: self.print_status(f"Report generation failed: {e}", "error") def _save_state(self, state, state_file): """Save workflow state to JSON.""" try: # Make serializable - convert CVE objects if needed serializable = json.loads(json.dumps(state, default=str)) with open(state_file, 'w') as f: json.dump(serializable, f, indent=2) except Exception: pass # ========================================================================= # MAIN LOOP # ========================================================================= def run(self): """Main menu loop.""" while True: self.show_menu() try: choice = input(f"{Colors.WHITE} Select: {Colors.RESET}").strip() if choice == "0": break elif choice == "1": target = input(f"\n{Colors.WHITE} Target IP/hostname: {Colors.RESET}").strip() if target: self.run_workflow(target) elif choice == "2": target = input(f"\n{Colors.WHITE} Target IP/hostname: {Colors.RESET}").strip() if target: self.quick_scan(target) elif choice == "3": self.resume_workflow() except (EOFError, KeyboardInterrupt): print() break def run(): """Module entry point.""" runner = WorkflowRunner() runner.run() if __name__ == "__main__": run()