wpsentry.py

#!/usr/bin/env python3
"""
WPSentry — Enterprise-Grade WordPress Security & Exploitation Toolkit.
Main orchestrator entry point.
"""

import argparse
import sys
import os
import requests

# Core configuration and types
from modules.config import (
    DEFAULT_TIMEOUT, DEFAULT_THREADS, DEFAULT_UA,
    REPORT_FORMAT_CONSOLE, REPORT_FORMAT_HTML, REPORT_FORMAT_MD, VALID_REPORT_FORMATS
)
from modules.scanner_context import ScanContext, setup_context

# Core components
from modules.fingerprinter import WPFingerprinter
from modules.vuln_scanner import VulnerabilityScanner
from modules.checks import DeepScanner
from modules.exploits import Exploiter
from modules.reporter import Reporter

# Utilities
from modules.utils import (
    banner, save_json, set_verbose, print_info, print_success, print_warning, print_error
)

# Disable urllib3 warnings globally for clean output
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)


class WPSentry:
    """Orchestrates individual scanning operations against a single target."""

    def __init__(self, context: ScanContext, report_format: str = REPORT_FORMAT_CONSOLE):
        self.context = context
        self.report_format = report_format

        # Instantiate sub-modules with individual arguments matching their legacy signatures
        self.fingerprinter = WPFingerprinter(
            session=context.session,
            target=context.target,
            headers=context.headers,
            timeout=context.timeout,
            threads=context.threads
        )
        self.vuln_scanner = VulnerabilityScanner(
            session=context.session,
            target=context.target,
            headers=context.headers,
            timeout=context.timeout,
            threads=context.threads
        )
        self.deep_scanner = DeepScanner(
            session=context.session,
            target=context.target,
            headers=context.headers,
            timeout=context.timeout,
            threads=context.threads
        )
        self.exploiter = Exploiter(
            session=context.session,
            target=context.target,
            headers=context.headers,
            timeout=context.timeout
        )
        self.reporter = Reporter(
            output_dir=context.output_dir,
            target=context.target
        )

    def run(self) -> None:
        """Execute the entire standard WordPress scan pipeline."""
        banner()
        print_info(f"Starting scan against {self.context.target}")
        self.context.log(f"Scan started against {self.context.target}")

        try:
            # 1. Fingerprint
            wp_info = self.fingerprinter.fingerprint()
            save_json(self.context.output_dir, "wp_info.json", wp_info)
            if not wp_info.get("is_wordpress"):
                self.context.log("Target is not WordPress. Terminating scan.")
                return

            self.deep_scanner.wp_info = wp_info

            # 2. Database Vulnerability Check
            vulnerabilities = self.vuln_scanner.scan(wp_info)
            save_json(self.context.output_dir, "vulnerabilities.json", vulnerabilities)

            # 3. Active Deep Scan (always runs by default in merged pipeline)
            deep_findings = self.deep_scanner.scan()
            save_json(self.context.output_dir, "deep_scan_findings.json", deep_findings)

            # 4. Exploitation
            exploitation_results = []
            if self.context.exploit:
                vuln_list = self._collect_vulnerabilities(vulnerabilities, deep_findings)
                if vuln_list:
                    exploitation_results = self.exploiter.exploit(vuln_list)
                    save_json(self.context.output_dir, "exploitation_results.json", exploitation_results)

            # 5. Reporting
            self._generate_report(wp_info, vulnerabilities, exploitation_results, deep_findings)

        except KeyboardInterrupt:
            print_warning("Scan interrupted by user")
            self.context.log("Scan interrupted by user")
        except Exception as e:
            print_error(f"An error occurred: {e}")
            self.context.log(f"Error: {e}")

        print_info(f"Scan completed. Results saved to {self.context.output_dir}")
        self.context.log(f"Scan completed. Results saved to {self.context.output_dir}")

    def _collect_vulnerabilities(self, vulnerabilities: dict, deep_findings: list) -> list:
        """Consolidate vulnerabilities and deep findings into a uniform exploitable format."""
        vuln_list = []
        if vulnerabilities.get("core"):
            vuln_list.extend(vulnerabilities["core"])

        for plugin, data in vulnerabilities.get("plugins", {}).items():
            for v in data.get("vulns", []):
                vc = v.copy()
                vc["plugin"] = plugin
                vuln_list.append(vc)

        for theme, data in vulnerabilities.get("themes", {}).items():
            for v in data.get("vulns", []):
                vc = v.copy()
                vc["theme"] = theme
                vuln_list.append(vc)

        # Map exploitable active deep findings to exploit methods
        for finding in deep_findings:
            title = finding.get("title", "")
            # Prefer exploit_method already set by the check (e.g. CVE checks)
            method = finding.get("exploit_method")
            if not method:
                if "Username Enumeration via Author Archives" in title:
                    method = "user_enumeration"
                elif "REST API Exposes User List" in title:
                    method = "rest_api_user_enum"
                elif "XML-RPC" in title or "XMLRPC" in title:
                    method = "xmlrpc_multicall"
                elif "Contact Form 7" in title:
                    method = "cf7_file_upload"
                elif "WP Super Cache" in title:
                    method = "wp_super_cache_rce"
                elif "TimThumb" in title:
                    method = "timthumb_rce"

            if method:
                vuln_list.append({
                    "title": title,
                    "type": "Information Disclosure" if "Enumeration" in title or "Exposes" in title or "XML-RPC" in title else "Remote Code Execution",
                    "exploit_method": method,
                    "exploit_available": True,
                    "theme": "Impreza" if "Impreza" in title else ""
                })
        return vuln_list

    def _generate_report(self, wp_info, vulnerabilities, exploitation_results, deep_findings) -> None:
        """Direct report generation to the requested format output driver."""
        fmt = self.report_format
        if fmt == REPORT_FORMAT_HTML:
            self.reporter.generate_html_report(wp_info, vulnerabilities, exploitation_results, deep_findings)
        elif fmt == REPORT_FORMAT_MD:
            self.reporter.generate_markdown_report(wp_info, vulnerabilities, exploitation_results, deep_findings)
        else:
            self.reporter.generate_json_report(wp_info, vulnerabilities, exploitation_results, deep_findings)
            self.reporter.print_console_summary(wp_info, vulnerabilities, exploitation_results, deep_findings)


def main():
    try:
        parser = argparse.ArgumentParser(
            prog="wpsentry",
            description="WPSentry — WordPress Security & Exploitation Toolkit"
        )
        parser.add_argument("-t", "--target", help="Target WordPress site URL")
        parser.add_argument("-f", "--file", dest="file", help="File containing list of target URLs (one per line)")
        parser.add_argument("-o", "--output", help="Output directory for scan results")
        parser.add_argument("--threads", type=int, default=DEFAULT_THREADS, help=f"Number of threads (default: {DEFAULT_THREADS})")
        parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help=f"Request timeout in seconds (default: {DEFAULT_TIMEOUT})")
        parser.add_argument("--user-agent", help="Custom User-Agent string")
        parser.add_argument("--proxy", help="Proxy URL (e.g., http://127.0.0.1:8080)")
        parser.add_argument("--pwn", action="store_true", help="Attempt to exploit found vulnerabilities")
        parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose output")
        parser.add_argument("--batch-dir", dest="batch_dir", help="Base directory for batch scan results (default: wpsentry_results)")
        parser.add_argument("--db-update", action="store_true", help="Update the tool and vulnerability databases")
        parser.add_argument("--auto-sync", action="store_true", help="Automatically sync vulnerability databases before scanning")
        parser.add_argument("--format", dest="format", default=REPORT_FORMAT_CONSOLE, choices=VALID_REPORT_FORMATS,
                            help=f"Output report format (default: {REPORT_FORMAT_CONSOLE})")

        args = parser.parse_args()
        set_verbose(args.verbose)

        # Handle database updates
        if args.db_update or args.auto_sync:
            from modules.updater import Updater
            updater = Updater()
            if args.db_update:
                updater.update_all()
                sys.exit(0)
            else:
                updater.update_all()

        if not args.target and not args.file:
            parser.error("Either --target or --file must be specified")

        if args.target and args.file:
            parser.error("--target and --file cannot be used together")

        if args.target:
            context = setup_context(args.target, args.output, args)
            scanner = WPSentry(context, report_format=args.format)
            scanner.run()
        elif args.file:
            from modules.mass_scanner import MassScanner
            mass_scanner = MassScanner(args)
            mass_scanner.run()

    except Exception as e:
        print_error(f"An unexpected error occurred: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()