pentest_ip.py

#!/usr/bin/env python3
"""
Single Target Pentesting - For IP addresses or single hostnames
No subdomain enumeration, direct target scanning
"""

import subprocess
import sys
import os
import json
import argparse
from pathlib import Path
from datetime import datetime
import defusedxml.ElementTree as ET  # Secure XML parsing
import shlex
from typing import Tuple, Optional, List, Dict, Set, Union
import logging
import ipaddress
import re
from filelock import FileLock, Timeout as LockTimeout

class Colors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKCYAN = '\033[96m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    PURPLE = '\033[35m'
    YELLOW = '\033[33m'

class SingleTargetPentest:
    def __init__(self, target, output_dir="results", threads=10, verbose=False, top_ports=1000):
        # Validate target
        self.target = self._validate_target(target)

        # Sanitize target for filesystem
        safe_target = re.sub(r'[^\w\.-]', '_', self.target)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.output_dir = (Path(output_dir) / safe_target / timestamp).resolve()

        # Path traversal protection
        base_dir = Path(output_dir).resolve()
        if not str(self.output_dir).startswith(str(base_dir)):
            raise ValueError(f"Path traversal detected: {self.output_dir}")

        self.threads = threads
        self.verbose = verbose
        self.top_ports = top_ports

        # Tool paths - verify existence (dynamically detect user home)
        self.go_bin = Path.home() / "go" / "bin"
        if not self.go_bin.exists():
            self.log(f"[!] Warning: Go bin directory not found: {self.go_bin}", Colors.WARNING)

        # Results
        self.open_ports = {}
        self.live_urls = set()
        self.vulnerabilities = []

        # Setup
        self.setup_directories()

    def _validate_target(self, target: str) -> str:
        """Validate target is a valid IP or hostname"""
        # Try IP address
        try:
            ipaddress.ip_address(target)
            return target
        except ValueError:
            pass

        # Try hostname (basic validation)
        hostname_pattern = r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)*[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?$'
        if re.match(hostname_pattern, target) and len(target) <= 253:
            return target

        raise ValueError(f"Invalid target: {target}. Must be IP address or valid hostname.")

    def setup_directories(self):
        """Create output directories with secure permissions"""
        dirs = ['ports', 'http', 'vulnerabilities', 'bruteforce', 'reports']
        for d in dirs:
            dir_path = self.output_dir / d
            dir_path.mkdir(parents=True, exist_ok=True)
            # Set restrictive permissions (700 = rwx------)
            dir_path.chmod(0o700)

    def log(self, message, color=Colors.ENDC):
        """Print colored messages"""
        print(f"{color}{message}{Colors.ENDC}")

    def run_command(self, cmd: Union[str, List[str]], shell: bool = False,
                    timeout: int = 600) -> Tuple[str, str, int]:
        """Execute command and return output"""
        try:
            if self.verbose:
                cmd_str = cmd if isinstance(cmd, str) else ' '.join(cmd)
                self.log(f"[*] Running: {cmd_str}", Colors.OKCYAN)

            result = subprocess.run(
                cmd,
                shell=shell,
                capture_output=True,
                text=True,
                timeout=timeout,
                check=False
            )
            return result.stdout, result.stderr, result.returncode
        except subprocess.TimeoutExpired:
            self.log(f"[!] Command timeout after {timeout}s", Colors.WARNING)
            return "", f"Timeout after {timeout}s", -1
        except Exception as e:
            self.log(f"[!] Error: {e}", Colors.FAIL)
            return "", str(e), -1

    def print_banner(self):
        """Print banner"""
        banner = f"""
{Colors.PURPLE}{Colors.BOLD}SINGLE TARGET PENTEST{Colors.ENDC} {Colors.YELLOW}Direct Scan Mode{Colors.ENDC}
{Colors.OKCYAN}🎯 {self.target}  {Colors.ENDC}|{Colors.OKCYAN}  📅 {datetime.now().strftime("%H:%M:%S")}  {Colors.ENDC}|{Colors.OKCYAN}  🧵 {self.threads} threads{Colors.ENDC}
"""
        print(banner)

    def print_phase_header(self, phase_num, phase_name, emoji):
        """Print phase header"""
        print(f"\n{Colors.BOLD}{Colors.PURPLE}╔{'═'*68}╗{Colors.ENDC}")
        print(f"{Colors.BOLD}{Colors.PURPLE}║{Colors.ENDC} {emoji}  {Colors.BOLD}{Colors.YELLOW}PHASE {phase_num}: {phase_name:<54}{Colors.PURPLE}║{Colors.ENDC}")
        print(f"{Colors.BOLD}{Colors.PURPLE}╚{'═'*68}╝{Colors.ENDC}\n")

    def phase1_port_scanning(self):
        """Comprehensive port scanning with rustscan→nmap cascade"""
        self.print_phase_header("1", "PORT SCANNING & SERVICE DETECTION", "🔌")

        base_name = self.target.replace('.', '_').replace('/', '_')
        discovered_ports = []

        # Check if target is an IP address
        is_ip = False
        try:
            ipaddress.ip_address(self.target)
            is_ip = True
        except ValueError:
            pass

        # Phase 1.1: Fast port discovery with rustscan (only for IPs)
        if is_ip:
            rustscan_bin = Path("/usr/bin/rustscan")
            if rustscan_bin.exists():
                self.log("[*] Quick port discovery with rustscan (all 65535 ports)...", Colors.OKCYAN)
                rustscan_output = self.output_dir / "ports" / f"rustscan_{base_name}.txt"

                # rustscan finds ports, then passes to nmap
                cmd = f"{rustscan_bin} -a {shlex.quote(self.target)} --ulimit 5000 -t 2000 -b 1000 --no-nmap > {shlex.quote(str(rustscan_output))} 2>&1"
                stdout, stderr, code = self.run_command(cmd, shell=True, timeout=300)

                # Parse rustscan output to get ports
                if rustscan_output.exists():
                    with open(rustscan_output, 'r') as f:
                        content = f.read()
                        # rustscan format: "Open 192.168.1.1:22" or just "22,80,443"
                        ports = re.findall(r'(?:Open\s+[\w\.:]+:(\d+)|(?:^|\s)(\d{1,5})(?:,|\s|$))', content)
                        # Flatten tuples from alternation groups
                        ports = [p for group in ports for p in group if p]

                        # Filter valid ports and remove duplicates
                        discovered_ports = list(set([p for p in ports if p.isdigit() and 1 <= int(p) <= 65535]))
                        discovered_ports.sort(key=int)

                        if discovered_ports:
                            self.log(f"[+] ✅ rustscan found {len(discovered_ports)} open port(s): {', '.join(discovered_ports[:10])}", Colors.OKGREEN)
                            if len(discovered_ports) > 10:
                                self.log(f"    ... and {len(discovered_ports) - 10} more", Colors.OKCYAN)

        # Phase 1.2: Detailed scan with nmap
        xml_output = self.output_dir / "ports" / f"nmap_{base_name}.xml"
        txt_output = self.output_dir / "ports" / f"nmap_{base_name}.txt"

        if discovered_ports and is_ip:
            # Scan only discovered ports
            self.log(f"[*] Running nmap on {len(discovered_ports)} discovered port(s) with service detection...", Colors.OKCYAN)
            ports_arg = ",".join(discovered_ports[:500])  # Limit to 500 ports max

            cmd = [
                "nmap",
                "-p", ports_arg,
                "-sV", "-sC", "-T4",
                "-oX", str(xml_output),
                "-oN", str(txt_output),
                self.target
            ]
        else:
            # Fallback: scan top ports (hostname or rustscan not available)
            self.log(f"[*] Running nmap on top {self.top_ports} ports with service detection...", Colors.OKCYAN)

            cmd = [
                "nmap",
                "--top-ports", str(self.top_ports),
                "-sV", "-sC", "-T4",
                "-oX", str(xml_output),
                "-oN", str(txt_output),
                self.target
            ]

        stdout, stderr, code = self.run_command(cmd, shell=False, timeout=900)

        if xml_output.exists():
            # Parse results
            self.parse_nmap_xml(xml_output)

            # Display results
            total_ports = sum(len(ports) for ports in self.open_ports.values())
            self.log(f"[+] ✅ Found {total_ports} open port(s)", Colors.OKGREEN)

            # Display open ports
            for ip, ports in self.open_ports.items():
                self.log(f"[+] {ip}: {', '.join(ports[:10])}", Colors.OKCYAN)
                if len(ports) > 10:
                    self.log(f"    ... and {len(ports) - 10} more", Colors.OKCYAN)
        else:
            self.log("[!] Nmap scan failed", Colors.FAIL)

    def parse_nmap_xml(self, xml_file):
        """Parse nmap XML to extract ports"""
        try:
            tree = ET.parse(xml_file)
            root = tree.getroot()

            for host in root.findall('host'):
                # Get IP/hostname
                addr_elem = host.find('address')
                if addr_elem is None:
                    continue
                ip = addr_elem.get('addr')

                # Get open ports
                ports_elem = host.find('ports')
                if ports_elem is None:
                    continue

                if ip not in self.open_ports:
                    self.open_ports[ip] = []

                for port in ports_elem.findall('port'):
                    state = port.find('state')
                    if state is not None and state.get('state') == 'open':
                        portid = port.get('portid')
                        if portid not in self.open_ports[ip]:
                            self.open_ports[ip].append(portid)

        except Exception as e:
            self.log(f"[!] Error parsing nmap XML: {e}", Colors.WARNING)

    def phase2_http_probing(self):
        """HTTP probing with httpx"""
        self.print_phase_header("2", "HTTP/HTTPS PROBING", "🌐")

        # Generate URLs to probe
        urls = []
        https_ports = {'443', '8443', '9443', '10443'}
        web_ports = {'80', '443', '8080', '8000', '8443', '8888', '9000', '3000', '5000', '8001'}

        for ip, ports in self.open_ports.items():
            for port in ports:
                if port in web_ports:
                    # Known web ports
                    if port in https_ports:
                        urls.append(f"https://{self.target}:{port}")
                    else:
                        urls.append(f"http://{self.target}:{port}")
                else:
                    # Unknown ports - try both
                    urls.append(f"http://{self.target}:{port}")
                    urls.append(f"https://{self.target}:{port}")

        if not urls:
            self.log("[!] No ports to probe", Colors.WARNING)
            return

        # Save URLs with file lock
        urls_file = self.output_dir / "http" / "urls_to_probe.txt"
        lock_file = str(urls_file) + ".lock"
        try:
            with FileLock(lock_file, timeout=10):
                with open(urls_file, 'w') as f:
                    f.write('\n'.join(urls))
        except LockTimeout:
            self.log(f"[!] Could not acquire lock for {urls_file}", Colors.WARNING)

        self.log(f"[*] Probing {len(urls)} URL(s) with httpx...", Colors.OKCYAN)

        # Run httpx with output file
        httpx_output = self.output_dir / "http" / "httpx_results.txt"
        cmd = f"{self.go_bin}/httpx -l {shlex.quote(str(urls_file))} -silent -tech-detect -sc -title -timeout 15 -threads 20 -o {shlex.quote(str(httpx_output))}"
        stdout, stderr, code = self.run_command(cmd, shell=True)

        # Parse results from stdout first, then from file if stdout is empty
        if stdout and stdout.strip():
            lines = stdout.strip().split('\n')
            for line in lines:
                if line.strip():
                    url = line.split()[0].strip()
                    self.live_urls.add(url)
        elif httpx_output.exists() and httpx_output.stat().st_size > 0:
            # Fallback: read from output file
            with open(httpx_output, 'r') as f:
                for line in f:
                    if line.strip():
                        url = line.split()[0].strip()
                        self.live_urls.add(url)

        if self.live_urls:
            # Save clean URL list (only URLs, no metadata)
            httpx_clean = self.output_dir / "http" / "live_urls.txt"
            with open(httpx_clean, 'w') as f:
                f.write('\n'.join(sorted(self.live_urls)))

            self.log(f"[+] ✅ Found {len(self.live_urls)} live web service(s)", Colors.OKGREEN)
            for url in list(self.live_urls)[:10]:
                self.log(f"    → {url}", Colors.OKCYAN)
            if len(self.live_urls) > 10:
                self.log(f"    ... and {len(self.live_urls) - 10} more", Colors.OKCYAN)
        else:
            self.log("[!] No live web services found", Colors.WARNING)

    def phase3_vulnerability_scanning(self):
        """Vulnerability scanning with nuclei, dalfox, nikto, whatweb"""
        self.print_phase_header("3", "VULNERABILITY SCANNING", "🛡️")

        if not self.live_urls:
            self.log("[!] No URLs to scan", Colors.WARNING)
            return

        # Save live URLs with file lock
        urls_file = self.output_dir / "http" / "live_urls.txt"
        lock_file = str(urls_file) + ".lock"
        try:
            with FileLock(lock_file, timeout=10):
                with open(urls_file, 'w') as f:
                    f.write('\n'.join(self.live_urls))
        except LockTimeout:
            self.log(f"[!] Could not acquire lock for {urls_file}", Colors.WARNING)

        # 3.1: Nuclei scan
        self.log(f"[*] Running nuclei on {len(self.live_urls)} URL(s)...", Colors.OKCYAN)
        nuclei_output = self.output_dir / "vulnerabilities" / "nuclei_results.txt"

        cmd = f"{self.go_bin}/nuclei -l {shlex.quote(str(urls_file))} -s critical,high,medium -jsonl -o {shlex.quote(str(nuclei_output))} -silent -c 10"
        self.run_command(cmd, shell=True, timeout=1200)

        if nuclei_output.exists() and nuclei_output.stat().st_size > 0:
            # Count findings
            with open(nuclei_output, 'r') as f:
                vulns = [json.loads(line) for line in f if line.strip()]
                self.vulnerabilities.extend(vulns)

            self.log(f"[+] ✅ Found {len(vulns)} vulnerability/vulnerabilities", Colors.OKGREEN)
        else:
            self.log("[*] No vulnerabilities found", Colors.OKCYAN)

        # 3.2: Dalfox XSS scan
        self.log("[*] Running dalfox for XSS scanning...", Colors.OKCYAN)
        dalfox_output = self.output_dir / "vulnerabilities" / "dalfox_results.txt"
        cmd = f"{self.go_bin}/dalfox file {shlex.quote(str(urls_file))} -S --format jsonl -o {shlex.quote(str(dalfox_output))} --worker 10"
        self.run_command(cmd, shell=True, timeout=600)

        if dalfox_output.exists() and dalfox_output.stat().st_size > 0:
            self.log("[+] ✅ Dalfox scan completed", Colors.OKGREEN)

        # 3.3: Nikto scan (first 3 URLs)
        self.log(f"[*] Running nikto on {min(3, len(self.live_urls))} URL(s)...", Colors.OKCYAN)
        for url in list(self.live_urls)[:3]:
            base_name = re.sub(r'[^\w\.-]', '_', url)
            nikto_output = self.output_dir / "vulnerabilities" / f"nikto_{base_name}.txt"
            cmd = f"nikto -h {shlex.quote(url)} -timeout 10 -maxtime 300 -output {shlex.quote(str(nikto_output))}"
            self.run_command(cmd, shell=True, timeout=360)

        # 3.4: Whatweb technology detection
        self.log("[*] Running whatweb for technology detection...", Colors.OKCYAN)
        whatweb_output = self.output_dir / "vulnerabilities" / "whatweb_results.txt"
        cmd = f"whatweb -i {shlex.quote(str(urls_file))} --color=never --log-brief={shlex.quote(str(whatweb_output))}"
        self.run_command(cmd, shell=True, timeout=300)

        if whatweb_output.exists():
            self.log("[+] ✅ Technology detection completed", Colors.OKGREEN)

    def phase4_directory_bruteforce(self):
        """Directory bruteforce with ffuf (optimized for speed)"""
        self.print_phase_header("4", "DIRECTORY & PARAMETER DISCOVERY", "📂")

        if not self.live_urls:
            self.log("[!] No URLs to bruteforce", Colors.WARNING)
            return

        # Use SMALL wordlist by default (common.txt = 4.6k lines - FAST)
        wordlist = Path("/usr/share/wordlists/dirb/common.txt")
        param_wordlist = Path("/usr/share/seclists/Discovery/Web-Content/burp-parameter-names.txt")

        if not wordlist.exists():
            self.log(f"[!] Wordlist not found: {wordlist}", Colors.WARNING)
            return

        urls_to_scan = list(self.live_urls)[:3]  # Limit to first 3 URLs

        # FFUF ONLY (fastest and most efficient)
        self.log(f"[*] Running ffuf on {len(urls_to_scan)} URL(s)...", Colors.OKCYAN)
        for url in urls_to_scan:
            base_name = re.sub(r'[^\w\.-]', '_', url)
            ffuf_output = self.output_dir / "bruteforce" / f"ffuf_{base_name}.json"
            cmd = f"{self.go_bin}/ffuf -u {shlex.quote(url)}/FUZZ -w {shlex.quote(str(wordlist))} -mc 200,201,301,302,403 -fc 404 -t 50 -timeout 5 -rate 100 -o {shlex.quote(str(ffuf_output))} -of json -s"
            self.run_command(cmd, shell=True, timeout=300)

        # Parameter discovery with arjun (if available)
        try:
            arjun_check = subprocess.run(['which', 'arjun'], capture_output=True, text=True)
            arjun_available = arjun_check.returncode == 0
        except:
            arjun_available = False

        if arjun_available and param_wordlist.exists():
            self.log(f"[*] Running arjun for parameter discovery...", Colors.OKCYAN)
            urls_file = self.output_dir / "http" / "live_urls.txt"
            arjun_output = self.output_dir / "bruteforce" / "arjun_params.txt"
            cmd = f"arjun -i {shlex.quote(str(urls_file))} -o {shlex.quote(str(arjun_output))} -t 10"
            self.run_command(cmd, shell=True, timeout=600)

        self.log("[+] ✅ Directory & parameter discovery completed", Colors.OKGREEN)

    def phase5_screenshots_and_extras(self):
        """Screenshots and additional analysis"""
        self.print_phase_header("5", "SCREENSHOTS & EXTRAS", "📸")

        if not self.live_urls:
            self.log("[!] No URLs for screenshots", Colors.WARNING)
            return

        # Save URLs for tools
        urls_file = self.output_dir / "http" / "live_urls.txt"

        # 5.1: Screenshots with gowitness
        gowitness_bin = self.go_bin / "gowitness"
        if gowitness_bin.exists():
            self.log(f"[*] Taking screenshots with gowitness ({len(self.live_urls)} URL(s))...", Colors.OKCYAN)
            screenshot_dir = self.output_dir / "screenshots"
            screenshot_dir.mkdir(exist_ok=True)
            cmd = f"{gowitness_bin} file -f {shlex.quote(str(urls_file))} --screenshot-path {shlex.quote(str(screenshot_dir))} --timeout 30 --threads 5"
            self.run_command(cmd, shell=True, timeout=600)
            self.log("[+] ✅ Screenshots captured", Colors.OKGREEN)

        # 5.2: JavaScript file discovery
        self.log("[*] Discovering JavaScript files...", Colors.OKCYAN)
        js_files = set()
        for url in list(self.live_urls)[:5]:
            cmd = f"echo {shlex.quote(url)} | {self.go_bin}/katana -silent -d 2 -jc -jsluice | grep -i '\\.js$'"
            stdout, stderr, code = self.run_command(cmd, shell=True, timeout=120)
            if stdout:
                js_files.update(stdout.strip().split('\n'))

        if js_files:
            js_file_list = self.output_dir / "vulnerabilities" / "javascript_files.txt"
            lock_file = str(js_file_list) + ".lock"
            try:
                with FileLock(lock_file, timeout=10):
                    with open(js_file_list, 'w') as f:
                        f.write('\n'.join(js_files))
            except LockTimeout:
                pass

            self.log(f"[+] Found {len(js_files)} JavaScript file(s)", Colors.OKGREEN)

            # Scan JS files with nuclei
            nuclei_js_output = self.output_dir / "vulnerabilities" / "nuclei_javascript.json"
            nuclei_templates = Path.home() / "nuclei-templates"
            if nuclei_templates.exists():
                cmd = f"{self.go_bin}/nuclei -l {shlex.quote(str(js_file_list))} -t {shlex.quote(str(nuclei_templates))}/http/exposures/ -jsonl -o {shlex.quote(str(nuclei_js_output))} -silent"
                self.run_command(cmd, shell=True, timeout=600)

        self.log("[+] ✅ Additional analysis completed", Colors.OKGREEN)

    def generate_report(self):
        """Generate simple JSON report"""
        self.log(f"\n{Colors.BOLD}{'='*70}", Colors.PURPLE)
        self.log(f"GENERATING REPORT", Colors.PURPLE)
        self.log(f"{'='*70}{Colors.ENDC}", Colors.PURPLE)

        report = {
            'target': self.target,
            'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            'open_ports': self.open_ports,
            'live_urls': list(self.live_urls),
            'vulnerabilities': self.vulnerabilities,
            'statistics': {
                'total_ports': sum(len(p) for p in self.open_ports.values()),
                'live_web_services': len(self.live_urls),
                'vulnerabilities_found': len(self.vulnerabilities)
            }
        }

        report_file = self.output_dir / "reports" / "scan_report.json"
        lock_file = str(report_file) + ".lock"
        try:
            with FileLock(lock_file, timeout=10):
                with open(report_file, 'w') as f:
                    json.dump(report, f, indent=2)
                # Set restrictive permissions
                report_file.chmod(0o600)
        except LockTimeout:
            self.log(f"[!] Could not acquire lock for report", Colors.WARNING)
            return

        self.log(f"[+] Report saved: {report_file}", Colors.OKGREEN)

    def run(self, skip_bruteforce=False, skip_extras=False):
        """Execute full scan"""
        self.print_banner()
        start_time = datetime.now()

        try:
            self.phase1_port_scanning()
            self.phase2_http_probing()
            self.phase3_vulnerability_scanning()

            if not skip_bruteforce:
                self.phase4_directory_bruteforce()

            if not skip_extras:
                self.phase5_screenshots_and_extras()

            self.generate_report()

        except KeyboardInterrupt:
            self.log("\n[!] Scan interrupted", Colors.WARNING)
        except Exception as e:
            self.log(f"\n[!] Error: {e}", Colors.FAIL)
            import traceback
            traceback.print_exc()

        elapsed = (datetime.now() - start_time).total_seconds() / 60
        self.log(f"\n[+] ✅ Scan completed in {elapsed:.2f} minutes", Colors.OKGREEN)
        self.log(f"[+] Results: {self.output_dir}", Colors.OKGREEN)


def main():
    parser = argparse.ArgumentParser(
        description="Single Target Pentesting - For IP or single hostname",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Scan an IP address
  python3 pentest_ip.py -t 192.168.1.1

  # Scan a specific hostname
  python3 pentest_ip.py -t app.example.com

  # Scan with more ports
  python3 pentest_ip.py -t 10.0.0.5 --top-ports 5000

  # Skip directory bruteforce
  python3 pentest_ip.py -t testphp.vulnweb.com --no-bruteforce

  # Verbose mode
  python3 pentest_ip.py -t 44.228.249.3 -v
        """
    )

    parser.add_argument('-t', '--target', required=True, help='Target IP address or hostname')
    parser.add_argument('-o', '--output', default='results', help='Output directory')
    parser.add_argument('-T', '--threads', type=int, default=10, help='Number of threads')
    parser.add_argument('--top-ports', type=int, default=1000, help='Number of top ports to scan')
    parser.add_argument('--no-bruteforce', action='store_true', help='Skip directory bruteforce')
    parser.add_argument('--no-extras', action='store_true', help='Skip screenshots and extras')
    parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')

    args = parser.parse_args()

    scanner = SingleTargetPentest(
        target=args.target,
        output_dir=args.output,
        threads=args.threads,
        verbose=args.verbose,
        top_ports=args.top_ports
    )

    scanner.run(skip_bruteforce=args.no_bruteforce, skip_extras=args.no_extras)


if __name__ == '__main__':
    main()