Initial commit

backend/toolbox/modules/analyzer/security_analyzer.py

@@ -0,0 +1,368 @@
+"""
+Security Analyzer Module - Analyzes code for security vulnerabilities
+"""
+
+# Copyright (c) 2025 FuzzingLabs
+#
+# Licensed under the Business Source License 1.1 (BSL). See the LICENSE file
+# at the root of this repository for details.
+#
+# After the Change Date (four years from publication), this version of the
+# Licensed Work will be made available under the Apache License, Version 2.0.
+# See the LICENSE-APACHE file or http://www.apache.org/licenses/LICENSE-2.0
+#
+# Additional attribution and requirements are provided in the NOTICE file.
+
+import logging
+import re
+from pathlib import Path
+from typing import Dict, Any, List, Optional
+
+try:
+    from toolbox.modules.base import BaseModule, ModuleMetadata, ModuleResult, ModuleFinding
+except ImportError:
+    try:
+        from modules.base import BaseModule, ModuleMetadata, ModuleResult, ModuleFinding
+    except ImportError:
+        from src.toolbox.modules.base import BaseModule, ModuleMetadata, ModuleResult, ModuleFinding
+
+logger = logging.getLogger(__name__)
+
+
+class SecurityAnalyzer(BaseModule):
+    """
+    Analyzes source code for common security vulnerabilities.
+
+    This module:
+    - Detects hardcoded secrets and credentials
+    - Identifies dangerous function calls
+    - Finds SQL injection vulnerabilities
+    - Detects insecure configurations
+    """
+
+    def get_metadata(self) -> ModuleMetadata:
+        """Get module metadata"""
+        return ModuleMetadata(
+            name="security_analyzer",
+            version="1.0.0",
+            description="Analyzes code for security vulnerabilities",
+            author="FuzzForge Team",
+            category="analyzer",
+            tags=["security", "vulnerabilities", "static-analysis"],
+            input_schema={
+                "file_extensions": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "File extensions to analyze",
+                    "default": [".py", ".js", ".java", ".php", ".rb", ".go"]
+                },
+                "check_secrets": {
+                    "type": "boolean",
+                    "description": "Check for hardcoded secrets",
+                    "default": True
+                },
+                "check_sql": {
+                    "type": "boolean",
+                    "description": "Check for SQL injection risks",
+                    "default": True
+                },
+                "check_dangerous_functions": {
+                    "type": "boolean",
+                    "description": "Check for dangerous function calls",
+                    "default": True
+                }
+            },
+            output_schema={
+                "findings": {
+                    "type": "array",
+                    "description": "List of security findings"
+                }
+            },
+            requires_workspace=True
+        )
+
+    def validate_config(self, config: Dict[str, Any]) -> bool:
+        """Validate module configuration"""
+        extensions = config.get("file_extensions", [])
+        if not isinstance(extensions, list):
+            raise ValueError("file_extensions must be a list")
+
+        return True
+
+    async def execute(self, config: Dict[str, Any], workspace: Path) -> ModuleResult:
+        """
+        Execute the security analysis module.
+
+        Args:
+            config: Module configuration
+            workspace: Path to the workspace directory
+
+        Returns:
+            ModuleResult with security findings
+        """
+        self.start_timer()
+        self.validate_workspace(workspace)
+        self.validate_config(config)
+
+        findings = []
+        files_analyzed = 0
+
+        # Get configuration
+        file_extensions = config.get("file_extensions", [".py", ".js", ".java", ".php", ".rb", ".go"])
+        check_secrets = config.get("check_secrets", True)
+        check_sql = config.get("check_sql", True)
+        check_dangerous = config.get("check_dangerous_functions", True)
+
+        logger.info(f"Analyzing files with extensions: {file_extensions}")
+
+        try:
+            # Analyze each file
+            for ext in file_extensions:
+                for file_path in workspace.rglob(f"*{ext}"):
+                    if not file_path.is_file():
+                        continue
+
+                    files_analyzed += 1
+                    relative_path = file_path.relative_to(workspace)
+
+                    try:
+                        content = file_path.read_text(encoding='utf-8', errors='ignore')
+                        lines = content.splitlines()
+
+                        # Check for secrets
+                        if check_secrets:
+                            secret_findings = self._check_hardcoded_secrets(
+                                content, lines, relative_path
+                            )
+                            findings.extend(secret_findings)
+
+                        # Check for SQL injection
+                        if check_sql and ext in [".py", ".php", ".java", ".js"]:
+                            sql_findings = self._check_sql_injection(
+                                content, lines, relative_path
+                            )
+                            findings.extend(sql_findings)
+
+                        # Check for dangerous functions
+                        if check_dangerous:
+                            dangerous_findings = self._check_dangerous_functions(
+                                content, lines, relative_path, ext
+                            )
+                            findings.extend(dangerous_findings)
+
+                    except Exception as e:
+                        logger.error(f"Error analyzing file {relative_path}: {e}")
+
+            # Create summary
+            summary = {
+                "files_analyzed": files_analyzed,
+                "total_findings": len(findings),
+                "extensions_scanned": file_extensions
+            }
+
+            return self.create_result(
+                findings=findings,
+                status="success" if files_analyzed > 0 else "partial",
+                summary=summary,
+                metadata={
+                    "workspace": str(workspace),
+                    "config": config
+                }
+            )
+
+        except Exception as e:
+            logger.error(f"Security analyzer failed: {e}")
+            return self.create_result(
+                findings=findings,
+                status="failed",
+                error=str(e)
+            )
+
+    def _check_hardcoded_secrets(
+        self, content: str, lines: List[str], file_path: Path
+    ) -> List[ModuleFinding]:
+        """
+        Check for hardcoded secrets in code.
+
+        Args:
+            content: File content
+            lines: File lines
+            file_path: Relative file path
+
+        Returns:
+            List of findings
+        """
+        findings = []
+
+        # Patterns for secrets
+        secret_patterns = [
+            (r'api[_-]?key\s*=\s*["\']([^"\']{20,})["\']', 'API Key'),
+            (r'api[_-]?secret\s*=\s*["\']([^"\']{20,})["\']', 'API Secret'),
+            (r'password\s*=\s*["\']([^"\']+)["\']', 'Hardcoded Password'),
+            (r'token\s*=\s*["\']([^"\']{20,})["\']', 'Authentication Token'),
+            (r'aws[_-]?access[_-]?key\s*=\s*["\']([^"\']+)["\']', 'AWS Access Key'),
+            (r'aws[_-]?secret[_-]?key\s*=\s*["\']([^"\']+)["\']', 'AWS Secret Key'),
+            (r'private[_-]?key\s*=\s*["\']([^"\']+)["\']', 'Private Key'),
+            (r'["\']([A-Za-z0-9]{32,})["\']', 'Potential Secret Hash'),
+            (r'Bearer\s+([A-Za-z0-9\-_]+\.[A-Za-z0-9\-_]+\.[A-Za-z0-9\-_]+)', 'JWT Token'),
+        ]
+
+        for pattern, secret_type in secret_patterns:
+            for match in re.finditer(pattern, content, re.IGNORECASE):
+                # Find line number
+                line_num = content[:match.start()].count('\n') + 1
+                line_content = lines[line_num - 1] if line_num <= len(lines) else ""
+
+                # Skip common false positives
+                if self._is_false_positive_secret(match.group(0)):
+                    continue
+
+                findings.append(self.create_finding(
+                    title=f"Hardcoded {secret_type} detected",
+                    description=f"Found potential hardcoded {secret_type} in {file_path}",
+                    severity="high" if "key" in secret_type.lower() else "medium",
+                    category="hardcoded_secret",
+                    file_path=str(file_path),
+                    line_start=line_num,
+                    code_snippet=line_content.strip()[:100],
+                    recommendation=f"Remove hardcoded {secret_type} and use environment variables or secure vault",
+                    metadata={"secret_type": secret_type}
+                ))
+
+        return findings
+
+    def _check_sql_injection(
+        self, content: str, lines: List[str], file_path: Path
+    ) -> List[ModuleFinding]:
+        """
+        Check for potential SQL injection vulnerabilities.
+
+        Args:
+            content: File content
+            lines: File lines
+            file_path: Relative file path
+
+        Returns:
+            List of findings
+        """
+        findings = []
+
+        # SQL injection patterns
+        sql_patterns = [
+            (r'(SELECT|INSERT|UPDATE|DELETE).*\+\s*[\'"]?\s*\+?\s*\w+', 'String concatenation in SQL'),
+            (r'(SELECT|INSERT|UPDATE|DELETE).*%\s*[\'"]?\s*%?\s*\w+', 'String formatting in SQL'),
+            (r'f[\'"].*?(SELECT|INSERT|UPDATE|DELETE).*?\{.*?\}', 'F-string in SQL query'),
+            (r'query\s*=.*?\+', 'Dynamic query building'),
+            (r'execute\s*\(.*?\+.*?\)', 'Dynamic execute statement'),
+        ]
+
+        for pattern, vuln_type in sql_patterns:
+            for match in re.finditer(pattern, content, re.IGNORECASE):
+                line_num = content[:match.start()].count('\n') + 1
+                line_content = lines[line_num - 1] if line_num <= len(lines) else ""
+
+                findings.append(self.create_finding(
+                    title=f"Potential SQL Injection: {vuln_type}",
+                    description=f"Detected potential SQL injection vulnerability via {vuln_type}",
+                    severity="high",
+                    category="sql_injection",
+                    file_path=str(file_path),
+                    line_start=line_num,
+                    code_snippet=line_content.strip()[:100],
+                    recommendation="Use parameterized queries or prepared statements instead",
+                    metadata={"vulnerability_type": vuln_type}
+                ))
+
+        return findings
+
+    def _check_dangerous_functions(
+        self, content: str, lines: List[str], file_path: Path, ext: str
+    ) -> List[ModuleFinding]:
+        """
+        Check for dangerous function calls.
+
+        Args:
+            content: File content
+            lines: File lines
+            file_path: Relative file path
+            ext: File extension
+
+        Returns:
+            List of findings
+        """
+        findings = []
+
+        # Language-specific dangerous functions
+        dangerous_functions = {
+            ".py": [
+                (r'eval\s*\(', 'eval()', 'Arbitrary code execution'),
+                (r'exec\s*\(', 'exec()', 'Arbitrary code execution'),
+                (r'os\.system\s*\(', 'os.system()', 'Command injection risk'),
+                (r'subprocess\.call\s*\(.*shell=True', 'subprocess with shell=True', 'Command injection risk'),
+                (r'pickle\.loads?\s*\(', 'pickle.load()', 'Deserialization vulnerability'),
+            ],
+            ".js": [
+                (r'eval\s*\(', 'eval()', 'Arbitrary code execution'),
+                (r'new\s+Function\s*\(', 'new Function()', 'Arbitrary code execution'),
+                (r'innerHTML\s*=', 'innerHTML', 'XSS vulnerability'),
+                (r'document\.write\s*\(', 'document.write()', 'XSS vulnerability'),
+            ],
+            ".php": [
+                (r'eval\s*\(', 'eval()', 'Arbitrary code execution'),
+                (r'exec\s*\(', 'exec()', 'Command execution'),
+                (r'system\s*\(', 'system()', 'Command execution'),
+                (r'shell_exec\s*\(', 'shell_exec()', 'Command execution'),
+                (r'\$_GET\[', 'Direct $_GET usage', 'Input validation missing'),
+                (r'\$_POST\[', 'Direct $_POST usage', 'Input validation missing'),
+            ]
+        }
+
+        if ext in dangerous_functions:
+            for pattern, func_name, risk_type in dangerous_functions[ext]:
+                for match in re.finditer(pattern, content):
+                    line_num = content[:match.start()].count('\n') + 1
+                    line_content = lines[line_num - 1] if line_num <= len(lines) else ""
+
+                    findings.append(self.create_finding(
+                        title=f"Dangerous function: {func_name}",
+                        description=f"Use of potentially dangerous function {func_name}: {risk_type}",
+                        severity="medium",
+                        category="dangerous_function",
+                        file_path=str(file_path),
+                        line_start=line_num,
+                        code_snippet=line_content.strip()[:100],
+                        recommendation=f"Consider safer alternatives to {func_name}",
+                        metadata={
+                            "function": func_name,
+                            "risk": risk_type
+                        }
+                    ))
+
+        return findings
+
+    def _is_false_positive_secret(self, value: str) -> bool:
+        """
+        Check if a potential secret is likely a false positive.
+
+        Args:
+            value: Potential secret value
+
+        Returns:
+            True if likely false positive
+        """
+        false_positive_patterns = [
+            'example',
+            'test',
+            'demo',
+            'sample',
+            'dummy',
+            'placeholder',
+            'xxx',
+            '123',
+            'change',
+            'your',
+            'here'
+        ]
+
+        value_lower = value.lower()
+        return any(pattern in value_lower for pattern in false_positive_patterns)