VSGuard MCP

"""Semgrep scanner integration.""" import json import logging import subprocess import tempfile from pathlib import Path from typing import Optional from ..asvs.mapper import get_asvs_mapper from ..config import settings from ..models import Location, ScanResult, SeverityLevel from .base import BaseScanner logger = logging.getLogger(__name__) class SemgrepScanner(BaseScanner): """Security scanner using Semgrep.""" # Language support SUPPORTED_LANGUAGES = { "python": "py", "py": "py", "javascript": "js", "js": "js", "typescript": "ts", "ts": "ts", "java": "java", "go": "go", "ruby": "rb", "php": "php", "c": "c", "cpp": "cpp", "csharp": "cs", "rust": "rs", } # Map Semgrep severity to our severity levels SEVERITY_MAP = { "ERROR": SeverityLevel.HIGH, "WARNING": SeverityLevel.MEDIUM, "INFO": SeverityLevel.LOW, } def __init__(self, rules_dir: Optional[Path] = None) -> None: """ Initialize Semgrep scanner. Args: rules_dir: Directory containing custom Semgrep rules """ self.rules_dir = rules_dir or settings.rules_path self.mapper = get_asvs_mapper() async def scan( self, code: str, language: str, filename: Optional[str] = None, ) -> list[ScanResult]: """ Scan code using Semgrep. Args: code: Source code to scan language: Programming language filename: Optional filename for context Returns: List of security findings """ if not self.supports_language(language): logger.warning(f"Language not supported by Semgrep: {language}") return [] # Determine file extension extension = self._get_file_extension(language) if not filename: filename = f"input.{extension}" # Create temporary file with the code with tempfile.TemporaryDirectory() as tmpdir: tmpdir_path = Path(tmpdir) code_file = tmpdir_path / filename # Create parent directories if filename has path components code_file.parent.mkdir(parents=True, exist_ok=True) code_file.write_text(code, encoding="utf-8") # Run Semgrep results = await self._run_semgrep(code_file, tmpdir_path) return results def supports_language(self, language: str) -> bool: """Check if Semgrep supports this language.""" return language.lower() in self.SUPPORTED_LANGUAGES async def _run_semgrep(self, code_file: Path, working_dir: Path) -> list[ScanResult]: """ Run Semgrep on a file. Args: code_file: Path to code file working_dir: Working directory Returns: List of scan results """ try: # Build Semgrep command cmd = [ "semgrep", "--config=auto", # Use Semgrep Registry rules "--json", "--no-git-ignore", "--timeout", str(settings.scan_timeout), ] # Add custom rules if available if self.rules_dir.exists(): cmd.extend(["--config", str(self.rules_dir)]) logger.debug(f"Using custom rules from: {self.rules_dir}") cmd.append(str(code_file)) # Run Semgrep logger.debug(f"Running: {' '.join(cmd)}") result = subprocess.run( cmd, cwd=working_dir, capture_output=True, text=True, timeout=settings.scan_timeout, ) # Parse JSON output if result.stdout: output = json.loads(result.stdout) return self._parse_semgrep_output(output, code_file.name) else: logger.warning("Semgrep returned no output") return [] except subprocess.TimeoutExpired: logger.error(f"Semgrep scan timed out after {settings.scan_timeout}s") return [] except FileNotFoundError: logger.error("Semgrep not found - please install: pip install semgrep") return [] except json.JSONDecodeError as e: logger.error(f"Failed to parse Semgrep output: {e}") return [] except Exception as e: logger.error(f"Semgrep scan failed: {e}", exc_info=True) return [] def _parse_semgrep_output(self, output: dict, filename: str) -> list[ScanResult]: """ Parse Semgrep JSON output into ScanResult objects. Args: output: Semgrep JSON output filename: Filename that was scanned Returns: List of scan results """ results = [] for finding in output.get("results", []): try: # Extract rule metadata rule_id = finding.get("check_id", "") message = finding.get("extra", {}).get("message", finding.get("message", "")) severity = finding.get("extra", {}).get("severity", "WARNING") # Extract location info start = finding.get("start", {}) end = finding.get("end", {}) line = start.get("line") column = start.get("col") end_line = end.get("line") # Extract code snippet code_snippet = finding.get("extra", {}).get("lines", "") # Extract metadata metadata = finding.get("extra", {}).get("metadata", {}) vulnerability_type = metadata.get("vulnerability", rule_id.split(".")[-1]) cwe = metadata.get("cwe") asvs_id = metadata.get("asvs_id") # Map to ASVS requirements asvs_requirements = [] if asvs_id: asvs_requirements.append(asvs_id) else: # Try to map based on vulnerability type and CWE asvs_requirements = self.mapper.map_finding_to_asvs( vulnerability_type=vulnerability_type, cwe=cwe, ) # Build remediation message remediation = metadata.get("remediation", "Review and fix this security issue.") # Create scan result result = ScanResult( severity=self.SEVERITY_MAP.get(severity.upper(), SeverityLevel.MEDIUM), vulnerability_type=vulnerability_type, asvs_requirements=asvs_requirements, location=Location( line=line, column=column, end_line=end_line, filename=filename, code_snippet=code_snippet, ), message=message, remediation=remediation, scanner="semgrep", metadata={ "rule_id": rule_id, "cwe": cwe, }, ) results.append(result) except Exception as e: logger.error(f"Error parsing Semgrep finding: {e}", exc_info=True) logger.debug(f"Finding data: {finding}") continue return results def _get_file_extension(self, language: str) -> str: """Get file extension for a language.""" return self.SUPPORTED_LANGUAGES.get(language.lower(), "txt")