Automated vulnerability detection using OWASP tools, CVE databases, and security scanners. Use when performing security audits, compliance checks, or continuous security monitoring.
Systematically identify security vulnerabilities in applications, dependencies, and infrastructure using automated scanning tools and manual security assessments.
// scanner.js - Comprehensive vulnerability scanning
const { exec } = require("child_process");
const util = require("util");
const fs = require("fs").promises;
const execPromise = util.promisify(exec);
class VulnerabilityScanner {
constructor() {
this.results = {
dependencies: [],
code: [],
docker: [],
secrets: [],
};
}
async scanDependencies() {
console.log("Scanning dependencies with npm audit...");
try {
const { stdout } = await execPromise("npm audit --json");
const auditResults = JSON.parse(stdout);
for (const [name, advisory] of Object.entries(
auditResults.vulnerabilities || {},
)) {
this.results.dependencies.push({
package: name,
severity: advisory.severity,
cve: advisory.via
.filter((v) => typeof v === "object")
.map((v) => v.cve),
title: advisory.via
.filter((v) => typeof v === "object")
.map((v) => v.title),
vulnerableVersions: advisory.range,
recommendation: advisory.fixAvailable
? "Update available"
: "No fix available",
});
}
} catch (error) {
console.error("Dependency scan failed:", error.message);
}
}
async scanCode() {
console.log("Scanning code with ESLint security plugin...");
try {
const { stdout } = await execPromise(
"npx eslint . --format json --plugin security",
);
const eslintResults = JSON.parse(stdout);
for (const file of eslintResults) {
for (const message of file.messages) {
if (message.ruleId && message.ruleId.includes("security")) {
this.results.code.push({
file: file.filePath,
line: message.line,
column: message.column,
rule: message.ruleId,
severity: message.severity === 2 ? "high" : "medium",
message: message.message,
});
}
}
}
} catch (error) {
console.log("Code scan completed with findings");
}
}
async scanDockerfile() {
console.log("Scanning Dockerfile with hadolint...");
try {
const { stdout } = await execPromise("hadolint Dockerfile --format json");
const hadolintResults = JSON.parse(stdout);
for (const issue of hadolintResults) {
this.results.docker.push({
line: issue.line,
level: issue.level,
code: issue.code,
message: issue.message,
});
}
} catch (error) {
console.log("Dockerfile scan completed");
}
}
async scanSecrets() {
console.log("Scanning for secrets with truffleHog...");
try {
const { stdout } = await execPromise("trufflehog filesystem . --json");
const lines = stdout.trim().split("\n");
for (const line of lines) {
if (line) {
const finding = JSON.parse(line);
this.results.secrets.push({
file: finding.SourceMetadata?.Data?.Filesystem?.file,
line: finding.SourceMetadata?.Data?.Filesystem?.line,
type: finding.DetectorName,
verified: finding.Verified,
});
}
}
} catch (error) {
console.log("Secret scan completed");
}
}
async runFullScan() {
await this.scanDependencies();
await this.scanCode();
await this.scanDockerfile();
await this.scanSecrets();
return this.generateReport();
}
generateReport() {
const report = {
timestamp: new Date().toISOString(),
summary: {
critical: 0,
high: 0,
medium: 0,
low: 0,
},
findings: this.results,
totalIssues: 0,
};
// Count by severity
for (const dep of this.results.dependencies) {
report.summary[dep.severity]++;
report.totalIssues++;
}
for (const code of this.results.code) {
report.summary[code.severity]++;
report.totalIssues++;
}
// High severity for secrets
report.summary.high += this.results.secrets.filter(
(s) => s.verified,
).length;
report.totalIssues += this.results.secrets.length;
return report;
}
}
// Usage
async function main() {
const scanner = new VulnerabilityScanner();
const report = await scanner.runFullScan();
await fs.writeFile("security-report.json", JSON.stringify(report, null, 2));
console.log("\n=== Security Scan Summary ===");
console.log(`Total Issues: ${report.totalIssues}`);
console.log(`Critical: ${report.summary.critical}`);
console.log(`High: ${report.summary.high}`);
console.log(`Medium: ${report.summary.medium}`);
console.log(`Low: ${report.summary.low}`);
}
main().catch(console.error);
# owasp_scanner.py
import subprocess
import json
import os
from typing import Dict, List
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class Vulnerability:
severity: str
cve: str
package: str
version: str
description: str
fix: str
class OWASPScanner:
def __init__(self, project_path: str):
self.project_path = project_path
self.vulnerabilities: List[Vulnerability] = []
def scan_dependencies(self) -> None:
"""Scan Python dependencies using Safety"""
print("Scanning dependencies with Safety...")
try:
result = subprocess.run(
['safety', 'check', '--json', '--file', 'requirements.txt'],
cwd=self.project_path,
capture_output=True,
text=True
)
if result.stdout:
findings = json.loads(result.stdout)
for vuln in findings:
self.vulnerabilities.append(Vulnerability(
severity=self._map_severity(vuln.get('severity', 'unknown')),
cve=vuln.get('cve', 'N/A'),
package=vuln.get('package_name', ''),
version=vuln.get('analyzed_version', ''),
description=vuln.get('advisory', ''),
fix=f"Upgrade to {vuln.get('fixed_versions', 'latest')}"
))
except Exception as e:
print(f"Dependency scan error: {e}")
def scan_with_bandit(self) -> None:
"""Scan Python code with Bandit"""
print("Scanning code with Bandit...")
try:
result = subprocess.run(
['bandit', '-r', '.', '-f', 'json'],
cwd=self.project_path,
capture_output=True,
text=True
)
if result.stdout:
findings = json.loads(result.stdout)
for issue in findings.get('results', []):
self.vulnerabilities.append(Vulnerability(
severity=issue['issue_severity'].lower(),
cve='BANDIT-' + issue['test_id'],
package=os.path.basename(issue['filename']),
version=str(issue['line_number']),
description=f"{issue['issue_text']} - {issue['test_name']}",
fix=issue.get('more_info', 'Review code')
))
except Exception as e:
print(f"Bandit scan error: {e}")
def scan_with_trivy(self) -> None:
"""Scan container images with Trivy"""
print("Scanning container with Trivy...")
try:
result = subprocess.run(
['trivy', 'image', '--format', 'json', 'myapp:latest'],
capture_output=True,
text=True
)
if result.stdout:
findings = json.loads(result.stdout)
for result_item in findings.get('Results', []):
for vuln in result_item.get('Vulnerabilities', []):
self.vulnerabilities.append(Vulnerability(
severity=vuln['Severity'].lower(),
cve=vuln.get('VulnerabilityID', 'N/A'),
package=vuln['PkgName'],
version=vuln['InstalledVersion'],
description=vuln.get('Title', vuln.get('Description', '')),
fix=vuln.get('FixedVersion', 'No fix available')
))
except Exception as e:
print(f"Trivy scan error: {e}")
def _map_severity(self, severity: str) -> str:
mapping = {
'critical': 'critical',
'high': 'high',
'medium': 'medium',
'low': 'low'
}
return mapping.get(severity.lower(), 'medium')
def generate_report(self) -> Dict:
summary = {
'critical': 0,
'high': 0,
'medium': 0,
'low': 0
}
for vuln in self.vulnerabilities:
if vuln.severity in summary:
summary[vuln.severity] += 1
return {
'timestamp': datetime.now().isoformat(),
'total_vulnerabilities': len(self.vulnerabilities),
'summary': summary,
'vulnerabilities': [asdict(v) for v in self.vulnerabilities],
'compliance_status': 'FAIL' if summary['critical'] > 0 or summary['high'] > 0 else 'PASS'
}
def run_full_scan(self) -> Dict:
self.scan_dependencies()
self.scan_with_bandit()
self.scan_with_trivy()
report = self.generate_report()
with open('owasp-scan-report.json', 'w') as f:
json.dump(report, f, indent=2)
return report
# Usage
if __name__ == '__main__':
scanner = OWASPScanner('.')
report = scanner.run_full_scan()
print(f"\n=== OWASP Scan Complete ===")
print(f"Total Vulnerabilities: {report['total_vulnerabilities']}")
print(f"Critical: {report['summary']['critical']}")
print(f"High: {report['summary']['high']}")
print(f"Compliance Status: {report['compliance_status']}")
# .github/workflows/security-scan.yml