mirror of
https://github.com/FuzzingLabs/fuzzforge_ai.git
synced 2026-07-06 16:57:51 +02:00
feat: Add Python SAST workflow with three security analysis tools
Implements Issue #5 - Python SAST workflow that combines: - Dependency scanning (pip-audit) for CVE detection - Security linting (Bandit) for vulnerability patterns - Type checking (Mypy) for type safety issues ## Changes **New Modules:** - `DependencyScanner`: Scans Python dependencies for known CVEs using pip-audit - `BanditAnalyzer`: Analyzes Python code for security issues using Bandit - `MypyAnalyzer`: Checks Python code for type safety issues using Mypy **New Workflow:** - `python_sast`: Temporal workflow that orchestrates all three SAST tools - Runs tools in parallel for fast feedback (3-5 min vs hours for fuzzing) - Generates unified SARIF report with findings from all tools - Supports configurable severity/confidence thresholds **Updates:** - Added SAST dependencies to Python worker (bandit, pip-audit, mypy) - Updated module __init__.py files to export new analyzers - Added type_errors.py test file to vulnerable_app for Mypy validation ## Testing Workflow tested successfully on vulnerable_app: - ✅ Bandit: Detected 9 security issues (command injection, unsafe functions) - ✅ Mypy: Detected 5 type errors - ✅ DependencyScanner: Ran successfully (no CVEs in test dependencies) - ✅ SARIF export: Generated valid SARIF with 14 total findings
This commit is contained in:
tduhamel42
@@ -10,5 +10,7 @@
|
||||
# Additional attribution and requirements are provided in the NOTICE file.
|
||||
|
||||
from .security_analyzer import SecurityAnalyzer
|
||||
from .bandit_analyzer import BanditAnalyzer
|
||||
from .mypy_analyzer import MypyAnalyzer
|
||||
|
||||
__all__ = ["SecurityAnalyzer"]
|
||||
__all__ = ["SecurityAnalyzer", "BanditAnalyzer", "MypyAnalyzer"]
|
||||
@@ -0,0 +1,329 @@
|
||||
"""
|
||||
Bandit Analyzer Module - Analyzes Python code for security issues using Bandit
|
||||
"""
|
||||
|
||||
# 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 asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, List
|
||||
import uuid
|
||||
|
||||
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 BanditAnalyzer(BaseModule):
|
||||
"""
|
||||
Analyzes Python code for security issues using Bandit.
|
||||
|
||||
This module:
|
||||
- Runs Bandit security linter on Python files
|
||||
- Detects common security issues (SQL injection, hardcoded secrets, etc.)
|
||||
- Reports findings with severity levels
|
||||
"""
|
||||
|
||||
# Severity mapping from Bandit levels to our standard
|
||||
SEVERITY_MAP = {
|
||||
"LOW": "low",
|
||||
"MEDIUM": "medium",
|
||||
"HIGH": "high"
|
||||
}
|
||||
|
||||
def get_metadata(self) -> ModuleMetadata:
|
||||
"""Get module metadata"""
|
||||
return ModuleMetadata(
|
||||
name="bandit_analyzer",
|
||||
version="1.0.0",
|
||||
description="Analyzes Python code for security issues using Bandit",
|
||||
author="FuzzForge Team",
|
||||
category="analyzer",
|
||||
tags=["python", "security", "bandit", "sast"],
|
||||
input_schema={
|
||||
"severity_level": {
|
||||
"type": "string",
|
||||
"enum": ["low", "medium", "high"],
|
||||
"description": "Minimum severity level to report",
|
||||
"default": "low"
|
||||
},
|
||||
"confidence_level": {
|
||||
"type": "string",
|
||||
"enum": ["low", "medium", "high"],
|
||||
"description": "Minimum confidence level to report",
|
||||
"default": "medium"
|
||||
},
|
||||
"exclude_tests": {
|
||||
"type": "boolean",
|
||||
"description": "Exclude test files from analysis",
|
||||
"default": True
|
||||
},
|
||||
"skip_ids": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "List of Bandit test IDs to skip",
|
||||
"default": []
|
||||
}
|
||||
},
|
||||
output_schema={
|
||||
"findings": {
|
||||
"type": "array",
|
||||
"description": "List of security issues found by Bandit"
|
||||
}
|
||||
},
|
||||
requires_workspace=True
|
||||
)
|
||||
|
||||
def validate_config(self, config: Dict[str, Any]) -> bool:
|
||||
"""Validate module configuration"""
|
||||
severity = config.get("severity_level", "low")
|
||||
if severity not in ["low", "medium", "high"]:
|
||||
raise ValueError("severity_level must be one of: low, medium, high")
|
||||
|
||||
confidence = config.get("confidence_level", "medium")
|
||||
if confidence not in ["low", "medium", "high"]:
|
||||
raise ValueError("confidence_level must be one of: low, medium, high")
|
||||
|
||||
skip_ids = config.get("skip_ids", [])
|
||||
if not isinstance(skip_ids, list):
|
||||
raise ValueError("skip_ids must be a list")
|
||||
|
||||
return True
|
||||
|
||||
async def _run_bandit(
|
||||
self,
|
||||
workspace: Path,
|
||||
severity_level: str,
|
||||
confidence_level: str,
|
||||
exclude_tests: bool,
|
||||
skip_ids: List[str]
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Run Bandit on the workspace.
|
||||
|
||||
Args:
|
||||
workspace: Path to workspace
|
||||
severity_level: Minimum severity to report
|
||||
confidence_level: Minimum confidence to report
|
||||
exclude_tests: Whether to exclude test files
|
||||
skip_ids: List of test IDs to skip
|
||||
|
||||
Returns:
|
||||
Bandit JSON output as dict
|
||||
"""
|
||||
try:
|
||||
# Build bandit command
|
||||
cmd = [
|
||||
"bandit",
|
||||
"-r", str(workspace),
|
||||
"-f", "json",
|
||||
"-ll", # Report all findings (we'll filter later)
|
||||
]
|
||||
|
||||
# Add exclude patterns for test files
|
||||
if exclude_tests:
|
||||
cmd.extend(["-x", "*/test_*.py,*/tests/*,*_test.py"])
|
||||
|
||||
# Add skip IDs if specified
|
||||
if skip_ids:
|
||||
cmd.extend(["-s", ",".join(skip_ids)])
|
||||
|
||||
logger.info(f"Running Bandit on: {workspace}")
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE
|
||||
)
|
||||
|
||||
stdout, stderr = await process.communicate()
|
||||
|
||||
# Bandit returns non-zero if issues found, which is expected
|
||||
if process.returncode not in [0, 1]:
|
||||
logger.error(f"Bandit failed: {stderr.decode()}")
|
||||
return {"results": []}
|
||||
|
||||
# Parse JSON output
|
||||
result = json.loads(stdout.decode())
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error running Bandit: {e}")
|
||||
return {"results": []}
|
||||
|
||||
def _should_include_finding(
|
||||
self,
|
||||
issue: Dict[str, Any],
|
||||
min_severity: str,
|
||||
min_confidence: str
|
||||
) -> bool:
|
||||
"""
|
||||
Determine if a Bandit issue should be included based on severity/confidence.
|
||||
|
||||
Args:
|
||||
issue: Bandit issue dict
|
||||
min_severity: Minimum severity threshold
|
||||
min_confidence: Minimum confidence threshold
|
||||
|
||||
Returns:
|
||||
True if issue should be included
|
||||
"""
|
||||
severity_order = ["low", "medium", "high"]
|
||||
issue_severity = issue.get("issue_severity", "LOW").lower()
|
||||
issue_confidence = issue.get("issue_confidence", "LOW").lower()
|
||||
|
||||
severity_meets_threshold = severity_order.index(issue_severity) >= severity_order.index(min_severity)
|
||||
confidence_meets_threshold = severity_order.index(issue_confidence) >= severity_order.index(min_confidence)
|
||||
|
||||
return severity_meets_threshold and confidence_meets_threshold
|
||||
|
||||
def _convert_to_findings(
|
||||
self,
|
||||
bandit_result: Dict[str, Any],
|
||||
workspace: Path,
|
||||
min_severity: str,
|
||||
min_confidence: str
|
||||
) -> List[ModuleFinding]:
|
||||
"""
|
||||
Convert Bandit results to ModuleFindings.
|
||||
|
||||
Args:
|
||||
bandit_result: Bandit JSON output
|
||||
workspace: Workspace path for relative paths
|
||||
min_severity: Minimum severity to include
|
||||
min_confidence: Minimum confidence to include
|
||||
|
||||
Returns:
|
||||
List of ModuleFindings
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for issue in bandit_result.get("results", []):
|
||||
# Filter by severity and confidence
|
||||
if not self._should_include_finding(issue, min_severity, min_confidence):
|
||||
continue
|
||||
|
||||
# Extract issue details
|
||||
test_id = issue.get("test_id", "B000")
|
||||
test_name = issue.get("test_name", "unknown")
|
||||
issue_text = issue.get("issue_text", "No description")
|
||||
severity = self.SEVERITY_MAP.get(issue.get("issue_severity", "LOW"), "low")
|
||||
|
||||
# File location
|
||||
filename = issue.get("filename", "")
|
||||
line_number = issue.get("line_number", 0)
|
||||
code = issue.get("code", "")
|
||||
|
||||
# Try to get relative path
|
||||
try:
|
||||
file_path = Path(filename)
|
||||
rel_path = file_path.relative_to(workspace)
|
||||
except (ValueError, TypeError):
|
||||
rel_path = Path(filename).name
|
||||
|
||||
# Create finding
|
||||
finding = self.create_finding(
|
||||
title=f"{test_name} ({test_id})",
|
||||
description=issue_text,
|
||||
severity=severity,
|
||||
category="security-issue",
|
||||
file_path=str(rel_path),
|
||||
line_start=line_number,
|
||||
line_end=line_number,
|
||||
code_snippet=code.strip() if code else None,
|
||||
recommendation=f"Review and fix the security issue identified by Bandit test {test_id}",
|
||||
metadata={
|
||||
"test_id": test_id,
|
||||
"test_name": test_name,
|
||||
"confidence": issue.get("issue_confidence", "LOW").lower(),
|
||||
"cwe": issue.get("issue_cwe", {}).get("id") if issue.get("issue_cwe") else None,
|
||||
"more_info": issue.get("more_info", "")
|
||||
}
|
||||
)
|
||||
findings.append(finding)
|
||||
|
||||
return findings
|
||||
|
||||
async def execute(self, config: Dict[str, Any], workspace: Path) -> ModuleResult:
|
||||
"""
|
||||
Execute the Bandit analyzer module.
|
||||
|
||||
Args:
|
||||
config: Module configuration
|
||||
workspace: Path to workspace
|
||||
|
||||
Returns:
|
||||
ModuleResult with security findings
|
||||
"""
|
||||
start_time = time.time()
|
||||
metadata = self.get_metadata()
|
||||
|
||||
# Validate inputs
|
||||
self.validate_config(config)
|
||||
self.validate_workspace(workspace)
|
||||
|
||||
# Get configuration
|
||||
severity_level = config.get("severity_level", "low")
|
||||
confidence_level = config.get("confidence_level", "medium")
|
||||
exclude_tests = config.get("exclude_tests", True)
|
||||
skip_ids = config.get("skip_ids", [])
|
||||
|
||||
# Run Bandit
|
||||
logger.info("Starting Bandit analysis...")
|
||||
bandit_result = await self._run_bandit(
|
||||
workspace,
|
||||
severity_level,
|
||||
confidence_level,
|
||||
exclude_tests,
|
||||
skip_ids
|
||||
)
|
||||
|
||||
# Convert to findings
|
||||
findings = self._convert_to_findings(
|
||||
bandit_result,
|
||||
workspace,
|
||||
severity_level,
|
||||
confidence_level
|
||||
)
|
||||
|
||||
# Calculate summary
|
||||
severity_counts = {}
|
||||
for finding in findings:
|
||||
sev = finding.severity
|
||||
severity_counts[sev] = severity_counts.get(sev, 0) + 1
|
||||
|
||||
execution_time = time.time() - start_time
|
||||
|
||||
return ModuleResult(
|
||||
module=metadata.name,
|
||||
version=metadata.version,
|
||||
status="success",
|
||||
execution_time=execution_time,
|
||||
findings=findings,
|
||||
summary={
|
||||
"total_issues": len(findings),
|
||||
"by_severity": severity_counts,
|
||||
"files_analyzed": len(set(f.file_path for f in findings if f.file_path))
|
||||
},
|
||||
metadata={
|
||||
"bandit_version": bandit_result.get("generated_at", "unknown"),
|
||||
"metrics": bandit_result.get("metrics", {})
|
||||
}
|
||||
)
|
||||
@@ -0,0 +1,269 @@
|
||||
"""
|
||||
Mypy Analyzer Module - Analyzes Python code for type safety issues using Mypy
|
||||
"""
|
||||
|
||||
# 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 asyncio
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, List
|
||||
|
||||
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 MypyAnalyzer(BaseModule):
|
||||
"""
|
||||
Analyzes Python code for type safety issues using Mypy.
|
||||
|
||||
This module:
|
||||
- Runs Mypy type checker on Python files
|
||||
- Detects type errors and inconsistencies
|
||||
- Reports findings with configurable strictness
|
||||
"""
|
||||
|
||||
# Map Mypy error codes to severity
|
||||
ERROR_SEVERITY_MAP = {
|
||||
"error": "medium",
|
||||
"note": "info"
|
||||
}
|
||||
|
||||
def get_metadata(self) -> ModuleMetadata:
|
||||
"""Get module metadata"""
|
||||
return ModuleMetadata(
|
||||
name="mypy_analyzer",
|
||||
version="1.0.0",
|
||||
description="Analyzes Python code for type safety issues using Mypy",
|
||||
author="FuzzForge Team",
|
||||
category="analyzer",
|
||||
tags=["python", "type-checking", "mypy", "sast"],
|
||||
input_schema={
|
||||
"strict_mode": {
|
||||
"type": "boolean",
|
||||
"description": "Enable strict type checking",
|
||||
"default": False
|
||||
},
|
||||
"ignore_missing_imports": {
|
||||
"type": "boolean",
|
||||
"description": "Ignore errors about missing imports",
|
||||
"default": True
|
||||
},
|
||||
"follow_imports": {
|
||||
"type": "string",
|
||||
"enum": ["normal", "silent", "skip", "error"],
|
||||
"description": "How to handle imports",
|
||||
"default": "silent"
|
||||
}
|
||||
},
|
||||
output_schema={
|
||||
"findings": {
|
||||
"type": "array",
|
||||
"description": "List of type errors found by Mypy"
|
||||
}
|
||||
},
|
||||
requires_workspace=True
|
||||
)
|
||||
|
||||
def validate_config(self, config: Dict[str, Any]) -> bool:
|
||||
"""Validate module configuration"""
|
||||
follow_imports = config.get("follow_imports", "silent")
|
||||
if follow_imports not in ["normal", "silent", "skip", "error"]:
|
||||
raise ValueError("follow_imports must be one of: normal, silent, skip, error")
|
||||
|
||||
return True
|
||||
|
||||
async def _run_mypy(
|
||||
self,
|
||||
workspace: Path,
|
||||
strict_mode: bool,
|
||||
ignore_missing_imports: bool,
|
||||
follow_imports: str
|
||||
) -> str:
|
||||
"""
|
||||
Run Mypy on the workspace.
|
||||
|
||||
Args:
|
||||
workspace: Path to workspace
|
||||
strict_mode: Enable strict checking
|
||||
ignore_missing_imports: Ignore missing import errors
|
||||
follow_imports: How to handle imports
|
||||
|
||||
Returns:
|
||||
Mypy output as string
|
||||
"""
|
||||
try:
|
||||
# Build mypy command
|
||||
cmd = [
|
||||
"mypy",
|
||||
str(workspace),
|
||||
"--show-column-numbers",
|
||||
"--no-error-summary",
|
||||
f"--follow-imports={follow_imports}"
|
||||
]
|
||||
|
||||
if strict_mode:
|
||||
cmd.append("--strict")
|
||||
|
||||
if ignore_missing_imports:
|
||||
cmd.append("--ignore-missing-imports")
|
||||
|
||||
logger.info(f"Running Mypy on: {workspace}")
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE
|
||||
)
|
||||
|
||||
stdout, stderr = await process.communicate()
|
||||
|
||||
# Mypy returns non-zero if errors found, which is expected
|
||||
output = stdout.decode()
|
||||
return output
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error running Mypy: {e}")
|
||||
return ""
|
||||
|
||||
def _parse_mypy_output(self, output: str, workspace: Path) -> List[ModuleFinding]:
|
||||
"""
|
||||
Parse Mypy output and convert to findings.
|
||||
|
||||
Mypy output format:
|
||||
file.py:10:5: error: Incompatible return value type [return-value]
|
||||
file.py:15: note: See https://...
|
||||
|
||||
Args:
|
||||
output: Mypy stdout
|
||||
workspace: Workspace path for relative paths
|
||||
|
||||
Returns:
|
||||
List of ModuleFindings
|
||||
"""
|
||||
findings = []
|
||||
|
||||
# Regex to parse mypy output lines
|
||||
# Format: filename:line:column: level: message [error-code]
|
||||
pattern = r'^(.+?):(\d+)(?::(\d+))?: (error|note): (.+?)(?:\s+\[([^\]]+)\])?$'
|
||||
|
||||
for line in output.splitlines():
|
||||
match = re.match(pattern, line.strip())
|
||||
if not match:
|
||||
continue
|
||||
|
||||
filename, line_num, column, level, message, error_code = match.groups()
|
||||
|
||||
# Convert to relative path
|
||||
try:
|
||||
file_path = Path(filename)
|
||||
rel_path = file_path.relative_to(workspace)
|
||||
except (ValueError, TypeError):
|
||||
rel_path = Path(filename).name
|
||||
|
||||
# Skip if it's just a note (unless it's a standalone note)
|
||||
if level == "note" and not error_code:
|
||||
continue
|
||||
|
||||
# Map severity
|
||||
severity = self.ERROR_SEVERITY_MAP.get(level, "medium")
|
||||
|
||||
# Create finding
|
||||
title = f"Type error: {error_code or 'type-issue'}"
|
||||
description = message
|
||||
|
||||
finding = self.create_finding(
|
||||
title=title,
|
||||
description=description,
|
||||
severity=severity,
|
||||
category="type-error",
|
||||
file_path=str(rel_path),
|
||||
line_start=int(line_num),
|
||||
line_end=int(line_num),
|
||||
recommendation="Review and fix the type inconsistency or add appropriate type annotations",
|
||||
metadata={
|
||||
"error_code": error_code or "unknown",
|
||||
"column": int(column) if column else None,
|
||||
"level": level
|
||||
}
|
||||
)
|
||||
findings.append(finding)
|
||||
|
||||
return findings
|
||||
|
||||
async def execute(self, config: Dict[str, Any], workspace: Path) -> ModuleResult:
|
||||
"""
|
||||
Execute the Mypy analyzer module.
|
||||
|
||||
Args:
|
||||
config: Module configuration
|
||||
workspace: Path to workspace
|
||||
|
||||
Returns:
|
||||
ModuleResult with type checking findings
|
||||
"""
|
||||
start_time = time.time()
|
||||
metadata = self.get_metadata()
|
||||
|
||||
# Validate inputs
|
||||
self.validate_config(config)
|
||||
self.validate_workspace(workspace)
|
||||
|
||||
# Get configuration
|
||||
strict_mode = config.get("strict_mode", False)
|
||||
ignore_missing_imports = config.get("ignore_missing_imports", True)
|
||||
follow_imports = config.get("follow_imports", "silent")
|
||||
|
||||
# Run Mypy
|
||||
logger.info("Starting Mypy analysis...")
|
||||
mypy_output = await self._run_mypy(
|
||||
workspace,
|
||||
strict_mode,
|
||||
ignore_missing_imports,
|
||||
follow_imports
|
||||
)
|
||||
|
||||
# Parse output to findings
|
||||
findings = self._parse_mypy_output(mypy_output, workspace)
|
||||
|
||||
# Calculate summary
|
||||
error_code_counts = {}
|
||||
for finding in findings:
|
||||
code = finding.metadata.get("error_code", "unknown")
|
||||
error_code_counts[code] = error_code_counts.get(code, 0) + 1
|
||||
|
||||
execution_time = time.time() - start_time
|
||||
|
||||
return ModuleResult(
|
||||
module=metadata.name,
|
||||
version=metadata.version,
|
||||
status="success",
|
||||
execution_time=execution_time,
|
||||
findings=findings,
|
||||
summary={
|
||||
"total_errors": len(findings),
|
||||
"by_error_code": error_code_counts,
|
||||
"files_with_errors": len(set(f.file_path for f in findings if f.file_path))
|
||||
},
|
||||
metadata={
|
||||
"strict_mode": strict_mode,
|
||||
"ignore_missing_imports": ignore_missing_imports
|
||||
}
|
||||
)
|
||||
@@ -10,5 +10,6 @@
|
||||
# Additional attribution and requirements are provided in the NOTICE file.
|
||||
|
||||
from .file_scanner import FileScanner
|
||||
from .dependency_scanner import DependencyScanner
|
||||
|
||||
__all__ = ["FileScanner"]
|
||||
__all__ = ["FileScanner", "DependencyScanner"]
|
||||
@@ -0,0 +1,304 @@
|
||||
"""
|
||||
Dependency Scanner Module - Scans Python dependencies for known vulnerabilities using pip-audit
|
||||
"""
|
||||
|
||||
# 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 asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, List
|
||||
import uuid
|
||||
|
||||
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 DependencyScanner(BaseModule):
|
||||
"""
|
||||
Scans Python dependencies for known vulnerabilities using pip-audit.
|
||||
|
||||
This module:
|
||||
- Discovers dependency files (requirements.txt, pyproject.toml, setup.py, Pipfile)
|
||||
- Runs pip-audit to check for vulnerable dependencies
|
||||
- Reports CVEs with severity and affected versions
|
||||
"""
|
||||
|
||||
def get_metadata(self) -> ModuleMetadata:
|
||||
"""Get module metadata"""
|
||||
return ModuleMetadata(
|
||||
name="dependency_scanner",
|
||||
version="1.0.0",
|
||||
description="Scans Python dependencies for known vulnerabilities",
|
||||
author="FuzzForge Team",
|
||||
category="scanner",
|
||||
tags=["dependencies", "cve", "vulnerabilities", "pip-audit"],
|
||||
input_schema={
|
||||
"dependency_files": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "List of dependency files to scan (auto-discovered if empty)",
|
||||
"default": []
|
||||
},
|
||||
"ignore_vulns": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "List of vulnerability IDs to ignore",
|
||||
"default": []
|
||||
}
|
||||
},
|
||||
output_schema={
|
||||
"findings": {
|
||||
"type": "array",
|
||||
"description": "List of vulnerable dependencies with CVE information"
|
||||
}
|
||||
},
|
||||
requires_workspace=True
|
||||
)
|
||||
|
||||
def validate_config(self, config: Dict[str, Any]) -> bool:
|
||||
"""Validate module configuration"""
|
||||
dep_files = config.get("dependency_files", [])
|
||||
if not isinstance(dep_files, list):
|
||||
raise ValueError("dependency_files must be a list")
|
||||
|
||||
ignore_vulns = config.get("ignore_vulns", [])
|
||||
if not isinstance(ignore_vulns, list):
|
||||
raise ValueError("ignore_vulns must be a list")
|
||||
|
||||
return True
|
||||
|
||||
def _discover_dependency_files(self, workspace: Path) -> List[Path]:
|
||||
"""
|
||||
Discover Python dependency files in workspace.
|
||||
|
||||
Returns:
|
||||
List of discovered dependency file paths
|
||||
"""
|
||||
dependency_patterns = [
|
||||
"requirements.txt",
|
||||
"*requirements*.txt",
|
||||
"pyproject.toml",
|
||||
"setup.py",
|
||||
"Pipfile",
|
||||
"poetry.lock"
|
||||
]
|
||||
|
||||
found_files = []
|
||||
for pattern in dependency_patterns:
|
||||
found_files.extend(workspace.rglob(pattern))
|
||||
|
||||
# Deduplicate and return
|
||||
unique_files = list(set(found_files))
|
||||
logger.info(f"Discovered {len(unique_files)} dependency files")
|
||||
return unique_files
|
||||
|
||||
async def _run_pip_audit(self, file_path: Path) -> Dict[str, Any]:
|
||||
"""
|
||||
Run pip-audit on a specific dependency file.
|
||||
|
||||
Args:
|
||||
file_path: Path to dependency file
|
||||
|
||||
Returns:
|
||||
pip-audit JSON output as dict
|
||||
"""
|
||||
try:
|
||||
# Run pip-audit with JSON output
|
||||
cmd = [
|
||||
"pip-audit",
|
||||
"--requirement", str(file_path),
|
||||
"--format", "json",
|
||||
"--progress-spinner", "off"
|
||||
]
|
||||
|
||||
logger.info(f"Running pip-audit on: {file_path.name}")
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE
|
||||
)
|
||||
|
||||
stdout, stderr = await process.communicate()
|
||||
|
||||
# pip-audit returns 0 if no vulns, 1 if vulns found
|
||||
if process.returncode not in [0, 1]:
|
||||
logger.error(f"pip-audit failed: {stderr.decode()}")
|
||||
return {"dependencies": []}
|
||||
|
||||
# Parse JSON output
|
||||
result = json.loads(stdout.decode())
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error running pip-audit on {file_path}: {e}")
|
||||
return {"dependencies": []}
|
||||
|
||||
def _convert_to_findings(
|
||||
self,
|
||||
audit_result: Dict[str, Any],
|
||||
file_path: Path,
|
||||
workspace: Path,
|
||||
ignore_vulns: List[str]
|
||||
) -> List[ModuleFinding]:
|
||||
"""
|
||||
Convert pip-audit results to ModuleFindings.
|
||||
|
||||
Args:
|
||||
audit_result: pip-audit JSON output
|
||||
file_path: Path to scanned file
|
||||
workspace: Workspace path for relative path calculation
|
||||
ignore_vulns: List of vulnerability IDs to ignore
|
||||
|
||||
Returns:
|
||||
List of ModuleFindings
|
||||
"""
|
||||
findings = []
|
||||
|
||||
# pip-audit format: {"dependencies": [{package, version, vulns: []}]}
|
||||
for dep in audit_result.get("dependencies", []):
|
||||
package_name = dep.get("name", "unknown")
|
||||
package_version = dep.get("version", "unknown")
|
||||
vulnerabilities = dep.get("vulns", [])
|
||||
|
||||
for vuln in vulnerabilities:
|
||||
vuln_id = vuln.get("id", "UNKNOWN")
|
||||
|
||||
# Skip if in ignore list
|
||||
if vuln_id in ignore_vulns:
|
||||
logger.debug(f"Ignoring vulnerability: {vuln_id}")
|
||||
continue
|
||||
|
||||
description = vuln.get("description", "No description available")
|
||||
fix_versions = vuln.get("fix_versions", [])
|
||||
|
||||
# Map CVSS scores to severity
|
||||
# pip-audit doesn't always provide CVSS, so we default to medium
|
||||
severity = "medium"
|
||||
|
||||
# Try to get relative path
|
||||
try:
|
||||
rel_path = file_path.relative_to(workspace)
|
||||
except ValueError:
|
||||
rel_path = file_path
|
||||
|
||||
finding_id = str(uuid.uuid4())
|
||||
recommendation = f"Upgrade {package_name} to a fixed version: {', '.join(fix_versions)}" if fix_versions else f"Check for updates to {package_name}"
|
||||
|
||||
finding = self.create_finding(
|
||||
title=f"Vulnerable dependency: {package_name} ({vuln_id})",
|
||||
description=f"{description}\n\nAffected package: {package_name} {package_version}",
|
||||
severity=severity,
|
||||
category="vulnerable-dependency",
|
||||
file_path=str(rel_path),
|
||||
recommendation=recommendation,
|
||||
metadata={
|
||||
"cve_id": vuln_id,
|
||||
"package": package_name,
|
||||
"installed_version": package_version,
|
||||
"fix_versions": fix_versions,
|
||||
"aliases": vuln.get("aliases", []),
|
||||
"link": vuln.get("link", "")
|
||||
}
|
||||
)
|
||||
findings.append(finding)
|
||||
|
||||
return findings
|
||||
|
||||
async def execute(self, config: Dict[str, Any], workspace: Path) -> ModuleResult:
|
||||
"""
|
||||
Execute the dependency scanning module.
|
||||
|
||||
Args:
|
||||
config: Module configuration
|
||||
workspace: Path to workspace
|
||||
|
||||
Returns:
|
||||
ModuleResult with vulnerability findings
|
||||
"""
|
||||
start_time = time.time()
|
||||
metadata = self.get_metadata()
|
||||
|
||||
# Validate inputs
|
||||
self.validate_config(config)
|
||||
self.validate_workspace(workspace)
|
||||
|
||||
# Get configuration
|
||||
specified_files = config.get("dependency_files", [])
|
||||
ignore_vulns = config.get("ignore_vulns", [])
|
||||
|
||||
# Discover or use specified dependency files
|
||||
if specified_files:
|
||||
dep_files = [workspace / f for f in specified_files]
|
||||
else:
|
||||
dep_files = self._discover_dependency_files(workspace)
|
||||
|
||||
if not dep_files:
|
||||
logger.warning("No dependency files found in workspace")
|
||||
return ModuleResult(
|
||||
module=metadata.name,
|
||||
version=metadata.version,
|
||||
status="success",
|
||||
execution_time=time.time() - start_time,
|
||||
findings=[],
|
||||
summary={
|
||||
"total_files": 0,
|
||||
"total_vulnerabilities": 0,
|
||||
"vulnerable_packages": 0
|
||||
}
|
||||
)
|
||||
|
||||
# Scan each dependency file
|
||||
all_findings = []
|
||||
files_scanned = 0
|
||||
|
||||
for dep_file in dep_files:
|
||||
if not dep_file.exists():
|
||||
logger.warning(f"Dependency file not found: {dep_file}")
|
||||
continue
|
||||
|
||||
logger.info(f"Scanning dependencies in: {dep_file.name}")
|
||||
audit_result = await self._run_pip_audit(dep_file)
|
||||
findings = self._convert_to_findings(audit_result, dep_file, workspace, ignore_vulns)
|
||||
|
||||
all_findings.extend(findings)
|
||||
files_scanned += 1
|
||||
|
||||
# Calculate summary
|
||||
unique_packages = len(set(f.metadata.get("package") for f in all_findings))
|
||||
|
||||
execution_time = time.time() - start_time
|
||||
|
||||
return ModuleResult(
|
||||
module=metadata.name,
|
||||
version=metadata.version,
|
||||
status="success",
|
||||
execution_time=execution_time,
|
||||
findings=all_findings,
|
||||
summary={
|
||||
"total_files": files_scanned,
|
||||
"total_vulnerabilities": len(all_findings),
|
||||
"vulnerable_packages": unique_packages
|
||||
},
|
||||
metadata={
|
||||
"scanned_files": [str(f.name) for f in dep_files if f.exists()]
|
||||
}
|
||||
)
|
||||
Reference in New Issue
Block a user