majiayu000 Vulnerability Scanning
Overview
Systematically identify security vulnerabilities in applications, dependencies, and infrastructure using automated scanning tools and manual security assessments.
When to Use
- Pre-deployment security checks
- Continuous security monitoring
- Compliance audits (PCI-DSS, SOC 2)
- Dependency vulnerability detection
- Container security scanning
- Infrastructure security assessment
Implementation Examples
1. Node.js Vulnerability Scanner
// 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);
2. Python OWASP Scanner
# 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']}")
3. CI/CD Integration - GitHub Actions
# .github/workflows/security-scan.yml
name: Security Vulnerability Scan
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
schedule:
- cron: '0 2 * * *' # Daily at 2 AM
jobs:
vulnerability-scan:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Run Trivy vulnerability scanner
uses: aquasecurity/trivy-action@master
with:
scan-type: 'fs'
scan-ref: '.'
format: 'sarif'
output: 'trivy-results.sarif'
severity: 'CRITICAL,HIGH,MEDIUM'
- name: Upload Trivy results to GitHub Security
uses: github/codeql-action/upload-sarif@v2
with:
sarif_file: 'trivy-results.sarif'
- name: Run Snyk Security Scan
uses: snyk/actions/node@master
env:
SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
with:
args: --severity-threshold=high
- name: OWASP Dependency Check
uses: dependency-check/Dependency-Check_Action@main
with:
project: 'myapp'
path: '.'
format: 'JSON'
- name: Fail on high severity
run: |
if [ -f trivy-results.sarif ]; then
HIGH_COUNT=$(jq '.runs[].results | length' trivy-results.sarif)
if [ $HIGH_COUNT -gt 0 ]; then
echo "Found $HIGH_COUNT high severity vulnerabilities"
exit 1
fi
fi
Best Practices
✅ DO
- Automate scans in CI/CD
- Scan dependencies regularly
- Use multiple scanning tools
- Set severity thresholds
- Track vulnerability trends
- Scan containers and images
- Monitor CVE databases
- Document false positives
❌ DON'T
- Skip vulnerability scanning
- Ignore low severity issues
- Trust single scanning tool
- Bypass security gates
- Commit secrets to repos
Common Vulnerability Types
- CVE: Known vulnerabilities in libraries
- CWE: Common weakness patterns
- OWASP Top 10: Web application risks
- Container vulnerabilities: Base image issues
- Dependency confusion: Supply chain attacks
- Secrets exposure: Hardcoded credentials
Scanning Tools
- Trivy: Container and filesystem scanner
- Snyk: Dependency and code scanning
- OWASP Dependency Check: Java, .NET, Node.js
- Safety: Python dependency checker
- npm audit / yarn audit: Node.js packages
- Bandit: Python security linter
- Semgrep: Static analysis tool