Vulnerability Assessment

Vulnerability Assessment

What I do

I enable systematic vulnerability identification, classification, prioritization, and tracking. I provide frameworks for scanning, risk scoring, remediation guidance, and vulnerability management program development.

When to use me

• Setting up vulnerability scanning programs
• Analyzing and prioritizing vulnerability findings
• Creating risk-based remediation strategies
• Tracking vulnerability remediation progress
• Compliance vulnerability assessments
• Third-party/vendor security assessments
• Continuous vulnerability monitoring
• Metrics and reporting for stakeholders

Core Concepts

• CVE/NVD: Common Vulnerabilities and Exposures database
• CVSS: Common Vulnerability Scoring System for severity
• Asset Discovery: Identifying all systems and software
• Vulnerability Scanning: Automated detection of weaknesses
• Risk Prioritization: Combining CVSS with asset criticality
• Remediation Tracking: Managing vulnerability fixes
• Compliance Mapping: Linking vulns to compliance frameworks
• False Positive Analysis: Validating scanner findings
• Vulnerability Aging: Tracking time to remediate
• Metrics and KPIs: Measuring program effectiveness

Code Examples

CVSS Calculator

from enum import Enum
from dataclasses import dataclass
from typing import Dict, Optional
import json

class CVSSVersion(Enum):
    CVSS30 = "3.0"
    CVSS31 = "3.1"
    CVSS40 = "4.0"

class AttackVector(Enum):
    NETWORK = "N"
    ADJACENT_NETWORK = "A"
    LOCAL = "L"
    PHYSICAL = "P"

class AttackComplexity(Enum):
    LOW = "L"
    HIGH = "H"

class PrivilegesRequired(Enum):
    NONE = "N"
    LOW = "L"
    HIGH = "H"

class UserInteraction(Enum):
    NONE = "N"
    REQUIRED = "R"

class Scope(Enum):
    UNCHANGED = "U"
    CHANGED = "C"

class Integrity(Enum):
    NONE = "N"
    LOW = "L"
    HIGH = "H"

class Availability(Enum):
    NONE = "N"
    LOW = "L"
    HIGH = "H"

@dataclass
class CVSSVector:
    version: CVSSVersion
    attack_vector: AttackVector
    attack_complexity: AttackComplexity
    privileges_required: PrivilegesRequired
    user_interaction: UserInteraction
    scope: Scope
    confidentiality: Integrity
    integrity: Integrity
    availability: Availability

class CVSSCalculator:
    @staticmethod
    def calculate_base_score(vector: CVSSVector) -> float:
        if vector.version == CVSSVersion.CVSS30 or vector.version == CVSSVersion.CVSS31:
            return CVSSCalculator._calculate_v31(vector)
        return 0.0
    
    @staticmethod
    def _calculate_v31(vector: CVSSVector) -> float:
        if vector.scope == Scope.CHANGED:
            if vector.privileges_required == PrivilegesRequired.NONE:
                iss = 1 - ((1 - vector.confidentiality.value * 0.56) * 
                          (1 - vector.integrity.value * 0.56) * 
                          (1 - vector.availability.value * 0.56))
            elif vector.privileges_required == PrivilegesRequired.LOW:
                iss = 1 - ((1 - vector.confidentiality.value * 0.56) * 
                          (1 - vector.integrity.value * 0.56) * 
                          (1 - vector.availability.value * 0.56))
            else:
                iss = 1 - ((1 - vector.confidentiality.value * 0.22) * 
                          (1 - vector.integrity.value * 0.22) * 
                          (1 - vector.availability.value * 0.22))
        else:
            if vector.privileges_required == PrivilegesRequired.NONE:
                iss = 1 - ((1 - vector.confidentiality.value * 0.56) * 
                          (1 - vector.integrity.value * 0.56) * 
                          (1 - vector.availability.value * 0.56))
            elif vector.privileges_required == PrivilegesRequired.LOW:
                iss = 1 - ((1 - vector.confidentiality.value * 0.56) * 
                          (1 - vector.integrity.value * 0.56) * 
                          (1 - vector.availability.value * 0.56))
            else:
                iss = 1 - ((1 - vector.confidentiality.value * 0.27) * 
                          (1 - vector.integrity.value * 0.27) * 
                          (1 - vector.availability.value * 0.27))
        
        if vector.attack_vector == AttackVector.NETWORK:
            if vector.scope == Scope.CHANGED:
                impact = 6.42 * iss
            else:
                impact = 8.22 * iss
        elif vector.attack_vector == AttackVector.ADJACENT_NETWORK:
            if vector.scope == Scope.CHANGED:
                impact = 4.87 * iss
            else:
                impact = 6.12 * iss
        elif vector.attack_vector == AttackVector.LOCAL:
            if vector.scope == Scope.CHANGED:
                impact = 3.25 * iss
            else:
                impact = 4.53 * iss
        else:
            if vector.scope == Scope.CHANGED:
                impact = 1.22 * iss
            else:
                impact = 2.01 * iss
        
        if vector.attack_complexity == AttackComplexity.LOW:
            exploitability = 0.85
        else:
            exploitability = 0.62
        
        if vector.privileges_required == PrivilegesRequired.NONE:
            exploitability *= 0.85
        elif vector.privileges_required == PrivilegesRequired.LOW:
            exploitability *= 0.62
        else:
            exploitability *= 0.27
        
        if vector.user_interaction == UserInteraction.NONE:
            exploitability *= 0.85
        else:
            exploitability *= 0.62
        
        score = min((1 - impact) * exploitability + impact, 0.915)
        
        if iss == 0:
            return 0.0
        
        if vector.scope == Scope.CHANGED:
            if score <= 0:
                return 0.0
            elif score < 0.1:
                return 0.0
            else:
                return roundup(score * 10) / 10
        else:
            return roundup(score * 10) / 10
    
    @staticmethod
    def get_severity(score: float) -> str:
        if score == 0:
            return "NONE"
        elif score < 4.0:
            return "LOW"
        elif score < 7.0:
            return "MEDIUM"
        elif score < 9.0:
            return "HIGH"
        else:
            return "CRITICAL"
    
    @staticmethod
    def parse_vector_string(vector_string: str) -> CVSSVector:
        parts = vector_string.split('/')
        values = {p.split(':')[0]: p.split(':')[1] for p in parts}
        
        return CVSSVector(
            version=CVSSVersion.CVSS31,
            attack_vector=AttackVector(values.get('AV', 'N')),
            attack_complexity=AttackComplexity(values.get('AC', 'L')),
            privileges_required=PrivilegesRequired(values.get('PR', 'N')),
            user_interaction=UserInteraction(values.get('UI', 'N')),
            scope=Scope(values.get('S', 'U')),
            confidentiality=Integrity(values.get('C', 'N')),
            integrity=Integrity(values.get('I', 'N')),
            availability=Availability(values.get('A', 'N'))
        )

def roundup(x):
    return int(x + 0.9) if x % 1 != 0 else int(x)

Vulnerability Database Manager

import json
import sqlite3
from typing import List, Optional, Dict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class VulnerabilityStatus(Enum):
    NEW = "new"
    IN_PROGRESS = "in_progress"
    RISK_ACCEPTED = "risk_accepted"
    REMEDIATED = "remediated"
    FALSE_POSITIVE = "false_positive"

@dataclass
class Vulnerability:
    cve_id: str
    title: str
    description: str
    cvss_score: float
    cvss_vector: str
    severity: str
    cvss_version: str = "3.1"
    published_date: datetime = None
    modified_date: datetime = None
    affected_products: List[str] = field(default_factory=list)
    references: List[str] = field(default_factory=list)
    status: VulnerabilityStatus = VulnerabilityStatus.NEW
    asset_id: str = ""
    remediation: str = ""
    notes: str = ""

@dataclass
class Asset:
    asset_id: str
    hostname: str
    ip_address: str
    os: str
    services: List[Dict] = field(default_factory=list)
    criticality: str = "medium"
    owner: str = ""
    last_scanned: datetime = None

class VulnerabilityDatabase:
    def __init__(self, db_path: str = "vuln_manager.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS vulnerabilities (
                    cve_id TEXT PRIMARY KEY,
                    title TEXT NOT NULL,
                    description TEXT,
                    cvss_score REAL,
                    cvss_vector TEXT,
                    severity TEXT,
                    cvss_version TEXT,
                    published_date TEXT,
                    modified_date TEXT,
                    affected_products TEXT,
                    references TEXT,
                    remediation TEXT
                )
            """)
            
            conn.execute("""
                CREATE TABLE IF NOT EXISTS assets (
                    asset_id TEXT PRIMARY KEY,
                    hostname TEXT,
                    ip_address TEXT,
                    os TEXT,
                    services TEXT,
                    criticality TEXT,
                    owner TEXT,
                    last_scanned TEXT
                )
            """)
            
            conn.execute("""
                CREATE TABLE IF NOT EXISTS findings (
                    finding_id INTEGER PRIMARY KEY AUTOINCREMENT,
                    cve_id TEXT,
                    asset_id TEXT,
                    status TEXT,
                    detected_date TEXT,
                    remediation TEXT,
                    notes TEXT,
                    FOREIGN KEY (cve_id) REFERENCES vulnerabilities(cve_id),
                    FOREIGN KEY (asset_id) REFERENCES assets(asset_id)
                )
            """)
    
    def add_vulnerability(self, vuln: Vulnerability):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT OR REPLACE INTO vulnerabilities 
                (cve_id, title, description, cvss_score, cvss_vector, severity,
                 cvss_version, published_date, modified_date, affected_products,
                 references, remediation)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                vuln.cve_id, vuln.title, vuln.description, vuln.cvss_score,
                vuln.cvss_vector, vuln.severity, vuln.cvss_version,
                vuln.published_date, vuln.modified_date,
                json.dumps(vuln.affected_products),
                json.dumps(vuln.references), vuln.remediation
            ))
    
    def add_asset(self, asset: Asset):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT OR REPLACE INTO assets 
                (asset_id, hostname, ip_address, os, services, criticality, owner, last_scanned)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                asset.asset_id, asset.hostname, asset.ip_address, asset.os,
                json.dumps(asset.services), asset.criticality, asset.owner,
                asset.last_scanned
            ))
    
    def record_finding(self, cve_id: str, asset_id: str) -> int:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                INSERT INTO findings (cve_id, asset_id, status, detected_date)
                VALUES (?, ?, 'new', ?)
            """, (cve_id, asset_id, datetime.now().isoformat()))
            return cursor.lastrowid
    
    def get_open_findings(self) -> List[Dict]:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT f.finding_id, v.cve_id, v.title, v.severity, v.cvss_score,
                       a.hostname, a.ip_address, a.criticality, f.detected_date
                FROM findings f
                JOIN vulnerabilities v ON f.cve_id = v.cve_id
                JOIN assets a ON f.asset_id = a.asset_id
                WHERE f.status NOT IN ('remediated', 'false_positive', 'risk_accepted')
                ORDER BY v.cvss_score DESC
            """)
            
            return [dict(zip([c[0] for c in cursor.description], row)) 
                   for row in cursor.fetchall()]
    
    def get_vulnerability_summary(self) -> Dict:
        with sqlite3.connect(self.db_path) as conn:
            summary = {
                "total": 0,
                "by_severity": {"critical": 0, "high": 0, "medium": 0, "low": 0, "none": 0},
                "avg_age_days": 0,
                "by_status": {}
            }
            
            cursor = conn.execute("SELECT COUNT(*) FROM findings WHERE status NOT IN ('remediated', 'false_positive')")
            summary["total"] = cursor.fetchone()[0]
            
            for severity in summary["by_severity"]:
                cursor = conn.execute("""
                    SELECT COUNT(*) FROM findings f
                    JOIN vulnerabilities v ON f.cve_id = v.cve_id
                    WHERE v.severity = ? AND f.status NOT IN ('remediated', 'false_positive')
                """, (severity.upper(),))
                summary["by_severity"][severity] = cursor.fetchone()[0]
            
            return summary

Vulnerability Prioritizer

from typing import List, Dict
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class PrioritizedVulnerability:
    cve_id: str
    title: str
    cvss_score: float
    asset_criticality: str
    exploit_available: bool
    exposure_factor: str
    calculated_risk: float
    priority: int

class VulnerabilityPrioritizer:
    CVSS_WEIGHT = 0.4
    ASSET_CRITICALITY_WEIGHT = 0.3
    EXPLOIT_WEIGHT = 0.2
    EXPOSURE_WEIGHT = 0.1
    
    def __init__(self):
        self.known_exploits = set()
    
    def add_known_exploit(self, cve_id: str):
        self.known_exploits.add(cve_id)
    
    def calculate_risk_score(self, cvss_score: float, asset_criticality: str,
                            has_exploit: bool, is_exposed: bool) -> float:
        normalized_cvss = cvss_score / 10.0
        
        criticality_map = {"critical": 1.0, "high": 0.8, "medium": 0.5, "low": 0.3}
        normalized_criticality = criticality_map.get(asset_criticality, 0.5)
        
        exploit_bonus = 0.2 if has_exploit else 0.0
        
        exposure_score = 0.1 if is_exposed else 0.0
        
        risk = (normalized_cvss * self.CVSS_WEIGHT +
               normalized_criticality * self.ASSET_CRITICALITY_WEIGHT +
               exploit_bonus * self.EXPLOIT_WEIGHT +
               exposure_score * self.EXPOSURE_WEIGHT)
        
        return min(risk, 1.0)
    
    def prioritize(self, vulnerabilities: List[Dict], 
                   assets_criticality: Dict[str, str],
                   exposed_assets: set) -> List[PrioritizedVulnerability]:
        prioritized = []
        
        for vuln in vulnerabilities:
            asset_crit = assets_criticality.get(vuln.get('asset_id', ''), 'medium')
            has_exploit = vuln.get('cve_id') in self.known_exploits
            is_exposed = vuln.get('asset_id', '') in exposed_assets
            
            risk_score = self.calculate_risk_score(
                vuln.get('cvss_score', 0),
                asset_crit,
                has_exploit,
                is_exposed
            )
            
            priority = self._calculate_priority(risk_score)
            
            prioritized.append(PrioritizedVulnerability(
                cve_id=vuln.get('cve_id', ''),
                title=vuln.get('title', ''),
                cvss_score=vuln.get('cvss_score', 0),
                asset_criticality=asset_crit,
                exploit_available=has_exploit,
                exposure_factor="exposed" if is_exposed else "internal",
                calculated_risk=round(risk_score, 3),
                priority=priority
            ))
        
        return sorted(prioritized, key=lambda x: (-x.priority, -x.calculated_risk))
    
    def _calculate_priority(self, risk_score: float) -> int:
        if risk_score >= 0.8:
            return 1
        elif risk_score >= 0.6:
            return 2
        elif risk_score >= 0.4:
            return 3
        elif risk_score >= 0.2:
            return 4
        else:
            return 5

Report Generator

from typing import Dict, List
from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
 VulnerabilityReport:
    report_id: str
    generated_date: datetime
    scan_period: tuple
    summary: Dict
    critical_findings: List[Dict]
    recommendations: List[str]
    metrics: Dict

class VulnerabilityReportGenerator:
    def __init__(self, db_path: str = "vuln_manager.db"):
        self.db = VulnerabilityDatabase(db_path)
    
    def generate_report(self, start_date: datetime = None,
                        end_date: datetime = None) -> VulnerabilityReport:
        if start_date is None:
            start_date = datetime.now() - timedelta(days=30)
        if end_date is None:
            end_date = datetime.now()
        
        summary = self.db.get_vulnerability_summary()
        open_findings = self.db.get_open_findings()
        
        critical_findings = [
            f for f in open_findings 
            if f.get('severity') in ['CRITICAL', 'HIGH']
        ][:20]
        
        recommendations = self._generate_recommendations(summary, critical_findings)
        
        metrics = self._calculate_metrics(open_findings, start_date, end_date)
        
        return VulnerabilityReport(
            report_id=f"VULN-{datetime.now().strftime('%Y%m%d%H%M%S')}",
            generated_date=datetime.now(),
            scan_period=(start_date, end_date),
            summary=summary,
            critical_findings=critical_findings,
            recommendations=recommendations,
            metrics=metrics
        )
    
    def _generate_recommendations(self, summary: Dict, 
                                 critical_findings: List[Dict]) -> List[str]:
        recommendations = []
        
        if summary["by_severity"].get("critical", 0) > 0:
            recommendations.append(
                "CRITICAL: Remediate all critical severity vulnerabilities within 24 hours"
            )
        
        if summary["by_severity"].get("high", 0) > 0:
            recommendations.append(
                "HIGH: Address high severity vulnerabilities within 7 days"
            )
        
        exploit_vulns = [
            f for f in critical_findings 
            if any(exp in f.get('cve_id', '') for exp in ['CVE-2024', 'CVE-2023'])
        ]
        
        if exploit_vulns:
            recommendations.append(
                f"URGENT: {len(exploit_vulns)} vulnerabilities have known exploits - patch immediately"
            )
        
        recommendations.append(
            "Implement a continuous vulnerability management program"
        )
        recommendations.append(
            "Regularly update and patch all systems and applications"
        )
        recommendations.append(
            "Review and update asset criticality classifications"
        )
        
        return recommendations[:10]
    
    def _calculate_metrics(self, findings: List[Dict],
                          start_date: datetime,
                          end_date: datetime) -> Dict:
        total = len(findings)
        
        severity_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
        for f in findings:
            sev = f.get('severity', '').lower()
            if sev in severity_counts:
                severity_counts[sev] += 1
        
        mttd = 0
        if findings:
            avg_age = sum(1 for f in findings) / max(len(findings), 1)
            mttd = avg_age
        
        return {
            "total_vulnerabilities": total,
            "by_severity": severity_counts,
            "remediation_rate": 0.0,
            "avg_days_to_remediate": mttd,
            "scan_coverage": 0.0,
            "risk_score": self._calculate_risk_score(severity_counts)
        }
    
    def _calculate_risk_score(self, severity_counts: Dict) -> float:
        weights = {"critical": 10, "high": 7, "medium": 4, "low": 1}
        
        total_weighted = sum(
            severity_counts.get(s, 0) * weights.get(s, 0) 
            for s in weights
        )
        
        max_possible = (10 * weights["critical"] + 10 * weights["high"] + 
                       20 * weights["medium"] + 30 * weights["low"])
        
        return min(total_weighted / max_possible * 100, 100)
    
    def export_json(self, report: VulnerabilityReport) -> str:
        return json.dumps({
            "report_id": report.report_id,
            "generated_date": report.generated_date.isoformat(),
            "scan_period": [d.isoformat() for d in report.scan_period],
            "summary": report.summary,
            "critical_findings": report.critical_findings,
            "recommendations": report.recommendations,
            "metrics": report.metrics
        }, indent=2)

Best Practices

• Maintain accurate asset inventory before scanning
• Use authenticated scanning for better visibility
• Combine multiple scanning tools for comprehensive coverage
• Establish SLAs based on severity and asset criticality
• Track remediation progress with metrics and KPIs
• Validate remediation through rescanning
• Manage false positives through tuning
• Integrate vulnerability data with SIEM and ticketing
• Prioritize by risk, not just CVSS score
• Regular scanning cadence (weekly/monthly depending on environment)
• Automate ticket creation and escalation

Common Patterns

• Risk-Based Prioritization: CVSS × Asset Criticality × Exposure
• Patch Tuesday Workflow: Monthly patch cycle with prioritization
• Continuous Scanning: Agent-based scanning for real-time visibility
• Exception Management: Documenting accepted risks
• Metrics Dashboard: Executive and technical reporting
• Third-Party Assessment: Vendor vulnerability management