Vulnerability Scanning

Vulnerability Scanning

Table of Contents

• Overview
• When to Use
• Quick Start
• Reference Guides
• Best Practices

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

Quick Start

Minimal working example:

// 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(
// ... (see reference guides for full implementation)

Reference Guides

Detailed implementations in the references/ directory:

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

CI/CD Integration - GitHub Actions

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

Node.js Vulnerability Scanner

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);

Python OWASP Scanner

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']}")