Skip to content

Lab 06 β€” Vulnerability Data Analysis and Prioritization

Difficulty: Intermediate β†’ Advanced
Estimated Time: 120–180 minutes
Primary Language: Python
Security Domain: Vulnerability Management / Security Operations / Risk Management
Environment: Local controlled lab
Automation Type: Defensive Vulnerability Analytics

Your task is to build a Python-based vulnerability analysis and prioritization pipeline.

You will start with:

RAW SCANNER EXPORT

containing:

ASSETS
VULNERABILITIES
SEVERITIES
CVSS SCORES
DUPLICATE FINDINGS
MISSING OWNERS
INTERNET EXPOSURE
REMEDIATION DATES

Then transform it into:

RAW FINDINGS
↓
VALIDATE
↓
NORMALIZE
↓
DEDUPLICATE
↓
ENRICH WITH ASSET CONTEXT
↓
CALCULATE PRIORITY
↓
MAP OWNER
↓
IDENTIFY OVERDUE ITEMS
↓
BUILD REMEDIATION QUEUE
↓
REPORT

A vulnerability scanner may produce:

10 FINDINGS
1,000 FINDINGS
100,000 FINDINGS

The security problem is rarely:

HOW MANY VULNERABILITIES EXIST?

The more important questions are:

WHICH FINDINGS MATTER MOST?
WHICH ASSETS ARE MOST IMPORTANT?
WHICH SYSTEMS ARE INTERNET EXPOSED?
WHICH FINDINGS ARE OVERDUE?
WHO OWNS THE AFFECTED SYSTEM?
WHICH ITEMS SHOULD BE FIXED FIRST?

That is the difference between:

VULNERABILITY SCANNING

and:

VULNERABILITY MANAGEMENT

By completing this lab, you should be able to:

READ VULNERABILITY CSV DATA
VALIDATE REQUIRED FIELDS
NORMALIZE SEVERITY VALUES
NORMALIZE ASSET NAMES
PARSE CVSS SCORES
REMOVE DUPLICATE FINDINGS
LOAD ASSET CONTEXT
MAP ASSET OWNERS
IDENTIFY INTERNET-EXPOSED SYSTEMS
IDENTIFY HIGH-CRITICALITY SYSTEMS
IDENTIFY OVERDUE REMEDIATION
BUILD A TRAINING RISK MODEL
GENERATE A REMEDIATION QUEUE
EXPORT CSV
EXPORT JSON
GENERATE MARKDOWN REPORTS
DOCUMENT ANALYSIS LIMITATIONS
SCANNER EXPORT
↓
CSV INGEST
↓
VALIDATE
↓
NORMALIZE
↓
DEDUPLICATE
↓
ASSET INVENTORY JOIN
↓
β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”
↓ ↓ ↓
CVSS CRITICALITY EXPOSURE
↓ ↓ ↓
β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
↓
DUE DATE
↓
PRIORITY ENGINE
↓
OWNER MAPPING
↓
REMEDIATION QUEUE
β”Œβ”€β”€β”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β”€β”€β”
↓ ↓ ↓
CSV JSON MD

This lab uses:

SYNTHETIC VULNERABILITY DATA

Do not scan, assess, or prioritize systems outside your authorized environment.

The lab focuses on:

DATA ANALYSIS
RISK CONTEXT
REPORTING
REMEDIATION PLANNING

not exploitation.

Create:

vulnerability-prioritization/
|
+-- data/
|
+-- reports/
|
+-- src/
|
+-- tests/
|
+-- README.md

Linux/macOS:

Terminal window
mkdir -p vulnerability-prioritization/{data,reports,src,tests}
cd vulnerability-prioritization

PowerShell:

Terminal window
mkdir vulnerability-prioritization
cd vulnerability-prioritization
mkdir data
mkdir reports
mkdir src
mkdir tests

Run:

Terminal window
python --version

or:

Terminal window
python3 --version

Recommended:

Python 3.10+

Create:

data/assets.csv

Add:

asset_id,hostname,environment,criticality,owner,internet_exposed
1,WEB01,Production,Critical,Web Team,true
2,DB01,Production,Critical,Database Team,false
3,APP01,Production,High,Application Team,false
4,DEV01,Development,Medium,Development Team,false
5,OLD01,Production,High,,false
6,VPN01,Production,Critical,Network Team,true
7,JUMP01,Production,Critical,Security Team,false

Each asset includes:

HOSTNAME
ENVIRONMENT
CRITICALITY
OWNER
INTERNET EXPOSURE

These fields will influence remediation priority.

Create:

data/vulnerabilities.csv

Add:

finding_id,hostname,vulnerability_name,severity,cvss,status,discovered_date,remediation_due_date
VULN-001,WEB01,Outdated Web Framework,Critical,9.8,Open,2026-08-01,2026-08-15
VULN-002,WEB01,Weak TLS Configuration,HIGH,8.1,Open,2026-08-02,2026-08-20
VULN-003,DB01,Missing Security Update,critical,9.5,Open,2026-08-03,2026-08-17
VULN-004,APP01,Outdated Application Library,High,7.8,Open,2026-08-05,2026-08-25
VULN-005,DEV01,Development Package Finding,Medium,5.0,Closed,2026-08-07,2026-09-07
VULN-006,OLD01,Legacy Service Exposure,Critical,9.0,Open,2026-07-15,2026-07-30
VULN-007,VPN01,Remote Access Security Update,Critical,9.1,Open,2026-08-10,2026-08-22
VULN-008,JUMP01,Administrative Tool Update,High,7.5,Open,2026-08-14,2026-09-05
VULN-009,WEB01,Outdated Web Framework,Critical,9.8,Open,2026-08-01,2026-08-15
VULN-010,APP01,Missing Application Header,medium,5.4,Open,2026-08-20,2026-09-20
VULN-011,UNKNOWN01,Unmapped Host Finding,High,8.0,Open,2026-08-21,2026-09-10
VULN-012,DB01,Invalid Score Example,High,not-a-score,Open,2026-08-22,2026-09-12

It includes:

DUPLICATE FINDING
INCONSISTENT SEVERITY CASE
OPEN AND CLOSED FINDINGS
CRITICAL ASSETS
INTERNET-EXPOSED ASSETS
MISSING OWNER
UNKNOWN ASSET
INVALID CVSS VALUE
OVERDUE REMEDIATION

Create:

src/vulnerability_analyzer.py

Start with:

from pathlib import Path
from datetime import datetime, date
from collections import Counter, defaultdict
import csv
import json

Add:

BASE_DIR = Path(__file__).resolve().parent.parent
ASSET_FILE = BASE_DIR / "data" / "assets.csv"
VULNERABILITY_FILE = (
BASE_DIR /
"data" /
"vulnerabilities.csv"
)
REPORT_DIR = BASE_DIR / "reports"
REPORT_DIR.mkdir(
parents=True,
exist_ok=True
)

For reproducible results, use:

REFERENCE_DATE = date(
2026,
8,
29
)

If you use:

date.today()

your results will change over time.

A fixed lab date provides:

REPRODUCIBLE OUTPUT

Create:

def parse_boolean(value):
return (
str(value)
.strip()
.lower()
in {
"true",
"1",
"yes"
}
)

Create:

SEVERITIES = {
"critical",
"high",
"medium",
"low",
"informational"
}

Then:

def normalize_severity(value):
normalized = (
str(value)
.strip()
.lower()
)
if normalized not in SEVERITIES:
return "unknown"
return normalized

Without normalization:

Critical
CRITICAL
critical

may behave like three different values.

After normalization:

critical

Create:

def parse_cvss(value):
try:
score = float(value)
if 0.0 <= score <= 10.0:
return score
except (
TypeError,
ValueError
):
pass
return None

A scanner export might contain:

9.8
N/A
Unknown
not-a-score

Your script should not crash because one row contains bad data.

Create:

def parse_date(value):
try:
return datetime.strptime(
value.strip(),
"%Y-%m-%d"
).date()
except (
AttributeError,
ValueError
):
return None

Create:

def normalize_hostname(value):
return (
str(value)
.strip()
.upper()
)

Therefore:

web01
WEB01
Web01

become:

WEB01

Create:

def load_assets():
assets = {}
with ASSET_FILE.open(
"r",
encoding="utf-8",
newline=""
) as file:
reader = csv.DictReader(
file
)
for row in reader:
hostname = normalize_hostname(
row["hostname"]
)
assets[hostname] = {
"asset_id":
row["asset_id"],
"hostname":
hostname,
"environment":
row["environment"].strip(),
"criticality":
row["criticality"]
.strip()
.lower(),
"owner":
row["owner"].strip(),
"internet_exposed":
parse_boolean(
row["internet_exposed"]
)
}
return assets

Temporarily:

assets = load_assets()
print(
json.dumps(
assets,
indent=2
)
)

Run:

Terminal window
python src/vulnerability_analyzer.py

Create:

REQUIRED_FIELDS = {
"finding_id",
"hostname",
"vulnerability_name",
"severity",
"cvss",
"status",
"discovered_date",
"remediation_due_date"
}

Create:

def validate_headers(fieldnames):
if not fieldnames:
return False
return REQUIRED_FIELDS.issubset(
set(fieldnames)
)

Create:

def load_vulnerabilities():
valid = []
invalid = []
with VULNERABILITY_FILE.open(
"r",
encoding="utf-8",
newline=""
) as file:
reader = csv.DictReader(
file
)
if not validate_headers(
reader.fieldnames
):
raise ValueError(
"Vulnerability CSV is missing required columns."
)
for line_number, row in enumerate(
reader,
start=2
):
normalized = normalize_finding(
row,
line_number
)
if normalized["valid"]:
valid.append(
normalized["record"]
)
else:
invalid.append(
normalized["record"]
)
return valid, invalid

Add:

def normalize_finding(
row,
line_number
):
hostname = normalize_hostname(
row.get(
"hostname",
""
)
)
vulnerability_name = (
row.get(
"vulnerability_name",
""
)
.strip()
)
severity = normalize_severity(
row.get(
"severity",
""
)
)
cvss = parse_cvss(
row.get(
"cvss",
""
)
)
discovered_date = parse_date(
row.get(
"discovered_date",
""
)
)
due_date = parse_date(
row.get(
"remediation_due_date",
""
)
)
record = {
"line_number":
line_number,
"finding_id":
row.get(
"finding_id",
""
).strip(),
"hostname":
hostname,
"vulnerability_name":
vulnerability_name,
"severity":
severity,
"cvss":
cvss,
"status":
row.get(
"status",
""
).strip().lower(),
"discovered_date":
discovered_date,
"remediation_due_date":
due_date
}
reasons = []
if not hostname:
reasons.append(
"Missing hostname"
)
if not vulnerability_name:
reasons.append(
"Missing vulnerability name"
)
if severity == "unknown":
reasons.append(
"Unknown severity"
)
if cvss is None:
reasons.append(
"Invalid CVSS score"
)
if discovered_date is None:
reasons.append(
"Invalid discovered date"
)
if due_date is None:
reasons.append(
"Invalid remediation due date"
)
if reasons:
record["reason"] = "; ".join(
reasons
)
return {
"valid": False,
"record": record
}
return {
"valid": True,
"record": record
}

Temporarily:

valid, invalid = load_vulnerabilities()
print(
f"Valid findings: {len(valid)}"
)
print(
f"Invalid findings: {len(invalid)}"
)

Expected:

Valid findings: 11
Invalid findings: 1

because:

VULN-012

contains:

not-a-score

Two findings can refer to the same underlying vulnerability.

For this lab, use:

HOSTNAME
+
VULNERABILITY NAME
+
STATUS

as a simple deduplication key.

Create:

def get_deduplication_key(
record
):
return (
record["hostname"],
record[
"vulnerability_name"
].strip().lower(),
record["status"]
)

Create:

def deduplicate_findings(
records
):
unique = []
duplicates = []
seen = set()
for record in records:
key = get_deduplication_key(
record
)
if key in seen:
duplicates.append(
record
)
continue
seen.add(key)
unique.append(
record
)
return unique, duplicates

Scanner pipelines may produce duplicates because of:

MULTIPLE SCAN JOBS
MULTIPLE SCANNERS
REPEATED IMPORTS
AGENT + NETWORK SCANS
DATA PIPELINE ISSUES

Without deduplication:

10 REAL FINDINGS

might appear as:

25 FINDINGS

The simple key:

HOSTNAME + FINDING NAME + STATUS

is not universally reliable.

Production systems may need:

PLUGIN ID
CVE
PORT
PROTOCOL
SCANNER ID
ASSET ID
FIRST SEEN
FINGERPRINT

Create:

def enrich_with_assets(
findings,
assets
):
enriched = []
for finding in findings:
item = finding.copy()
asset = assets.get(
item["hostname"]
)
if asset:
item[
"asset_mapped"
] = True
item.update({
"environment":
asset[
"environment"
],
"criticality":
asset[
"criticality"
],
"owner":
asset[
"owner"
],
"internet_exposed":
asset[
"internet_exposed"
]
})
else:
item[
"asset_mapped"
] = False
item.update({
"environment":
"unknown",
"criticality":
"unknown",
"owner":
"",
"internet_exposed":
False
})
enriched.append(
item
)
return enriched

Before enrichment:

Vulnerability:
Outdated Web Framework
CVSS:
9.8

After enrichment:

Vulnerability:
Outdated Web Framework
CVSS:
9.8
Asset:
WEB01
Environment:
Production
Criticality:
Critical
Internet Exposed:
Yes
Owner:
Web Team

That is much more actionable.

Create:

def get_unmapped_findings(
findings
):
return [
item
for item in findings
if not item[
"asset_mapped"
]
]

You should identify:

UNKNOWN01

An unmapped vulnerability may mean:

ASSET INVENTORY GAP
HOSTNAME MISMATCH
STALE SCANNER RECORD
UNMANAGED ASSET
DECOMMISSIONING ISSUE

Do not silently ignore it.

Create:

def is_overdue(
finding,
reference_date=REFERENCE_DATE
):
if finding[
"status"
] != "open":
return False
due_date = finding[
"remediation_due_date"
]
if due_date is None:
return False
return (
due_date
<
reference_date
)

Create:

def calculate_days_overdue(
finding,
reference_date=REFERENCE_DATE
):
if not is_overdue(
finding,
reference_date
):
return 0
return (
reference_date
-
finding[
"remediation_due_date"
]
).days

Create:

def calculate_finding_age(
finding,
reference_date=REFERENCE_DATE
):
discovered = finding[
"discovered_date"
]
if discovered is None:
return None
return (
reference_date
-
discovered
).days

A critical vulnerability open for:

1 DAY

and a critical vulnerability open for:

120 DAYS

may require different management attention.

Create:

def add_time_context(
findings
):
results = []
for finding in findings:
item = finding.copy()
item["overdue"] = (
is_overdue(
item
)
)
item["days_overdue"] = (
calculate_days_overdue(
item
)
)
item["finding_age_days"] = (
calculate_finding_age(
item
)
)
results.append(
item
)
return results

For training, create a score based on:

CVSS
ASSET CRITICALITY
INTERNET EXPOSURE
OVERDUE STATUS
MISSING OWNER

This scoring system is:

A TRAINING MODEL

It is not a universal enterprise risk formula.

Real prioritization may include:

KNOWN EXPLOITATION
EPSS
THREAT INTELLIGENCE
BUSINESS IMPACT
DATA CLASSIFICATION
EXPOSURE
COMPENSATING CONTROLS
ASSET FUNCTION
ATTACK PATH CONTEXT

Create:

CRITICALITY_WEIGHT = {
"critical": 30,
"high": 20,
"medium": 10,
"low": 5,
"unknown": 0
}

Create:

SEVERITY_WEIGHT = {
"critical": 30,
"high": 20,
"medium": 10,
"low": 5,
"informational": 0,
"unknown": 0
}

Create:

def calculate_priority_score(
finding
):
if finding[
"status"
] != "open":
return 0
score = 0
score += SEVERITY_WEIGHT.get(
finding["severity"],
0
)
score += CRITICALITY_WEIGHT.get(
finding["criticality"],
0
)
if finding["cvss"] is not None:
score += int(
finding["cvss"] * 2
)
if finding[
"internet_exposed"
]:
score += 20
if finding[
"overdue"
]:
score += 15
if not finding[
"owner"
]:
score += 10
return score

Suppose:

Severity:
Critical
β†’ 30
Asset Criticality:
Critical
β†’ 30
CVSS:
9.8
β†’ 19
Internet Exposed:
Yes
β†’ 20
Overdue:
Yes
β†’ 15

Total:

114

This is:

A PRIORITIZATION SCORE

not a standardized risk value.

Create:

def determine_priority(
score
):
if score >= 100:
return "P1"
if score >= 75:
return "P2"
if score >= 50:
return "P3"
if score > 0:
return "P4"
return "Closed"

For this lab:

P1
Immediate remediation review
P2
High-priority remediation
P3
Planned remediation
P4
Standard remediation
Closed
No active remediation queue

Create:

def add_priority_context(
findings
):
results = []
for finding in findings:
item = finding.copy()
score = (
calculate_priority_score(
item
)
)
item[
"priority_score"
] = score
item[
"priority"
] = determine_priority(
score
)
results.append(
item
)
return results

Create:

def process_findings():
assets = load_assets()
valid, invalid = (
load_vulnerabilities()
)
unique, duplicates = (
deduplicate_findings(
valid
)
)
enriched = (
enrich_with_assets(
unique,
assets
)
)
enriched = (
add_time_context(
enriched
)
)
enriched = (
add_priority_context(
enriched
)
)
return {
"raw_valid_count":
len(valid),
"invalid_count":
len(invalid),
"unique_count":
len(unique),
"duplicate_count":
len(duplicates),
"findings":
enriched,
"invalid":
invalid,
"duplicates":
duplicates
}

Temporarily:

results = process_findings()
print(
json.dumps(
results,
indent=2,
default=str
)
)

Run:

Terminal window
python src/vulnerability_analyzer.py

Create:

def sort_findings(
findings
):
return sorted(
findings,
key=lambda item:
item[
"priority_score"
],
reverse=True
)

Create:

def build_remediation_queue(
findings
):
return [
item
for item in sort_findings(
findings
)
if item[
"status"
] == "open"
]

Your highest-priority findings should likely include:

WEB01
Outdated Web Framework
VPN01
Remote Access Security Update
DB01
Missing Security Update
OLD01
Legacy Service Exposure

Exact ordering depends on your scoring logic.

Context:

SEVERITY
Critical
CVSS
9.8
ASSET CRITICALITY
Critical
ENVIRONMENT
Production
INTERNET EXPOSED
Yes
OVERDUE
Yes

That combination creates stronger remediation urgency.

Context:

HIGH-CRITICALITY ASSET
CRITICAL VULNERABILITY
OVERDUE
NO OWNER

Even though it is not internet exposed, the ownership gap adds operational risk.

Create:

def findings_without_owner(
findings
):
return [
item
for item in findings
if (
item[
"status"
] == "open"
and
not item[
"owner"
]
)
]

Create:

def internet_exposed_findings(
findings
):
return [
item
for item in findings
if (
item[
"status"
] == "open"
and
item[
"internet_exposed"
]
)
]

Create:

def overdue_findings(
findings
):
return [
item
for item in findings
if item[
"overdue"
]
]

Create:

def critical_open_findings(
findings
):
return [
item
for item in findings
if (
item[
"status"
] == "open"
and
item[
"severity"
] == "critical"
)
]

Create:

def count_by_severity(
findings
):
return Counter(
item["severity"]
for item in findings
if item[
"status"
] == "open"
)

Create:

def count_by_owner(
findings
):
counter = Counter()
for item in findings:
if item[
"status"
] != "open":
continue
owner = (
item["owner"]
or "Unassigned"
)
counter[owner] += 1
return counter

Create:

def count_by_priority(
findings
):
return Counter(
item["priority"]
for item in findings
if item[
"status"
] == "open"
)

Create:

def build_summary(
results
):
findings = results[
"findings"
]
open_findings = [
item
for item in findings
if item[
"status"
] == "open"
]
return {
"valid_raw_findings":
results[
"raw_valid_count"
],
"unique_findings":
results[
"unique_count"
],
"duplicates_removed":
results[
"duplicate_count"
],
"invalid_findings":
results[
"invalid_count"
],
"open_findings":
len(
open_findings
),
"critical_open":
len(
critical_open_findings(
findings
)
),
"overdue":
len(
overdue_findings(
findings
)
),
"internet_exposed":
len(
internet_exposed_findings(
findings
)
),
"unowned_open":
len(
findings_without_owner(
findings
)
),
"severity_counts":
dict(
count_by_severity(
findings
)
),
"priority_counts":
dict(
count_by_priority(
findings
)
)
}

Create:

def export_remediation_queue(
findings
):
output = (
REPORT_DIR /
"remediation-queue.csv"
)
fields = [
"priority",
"priority_score",
"hostname",
"environment",
"criticality",
"owner",
"internet_exposed",
"vulnerability_name",
"severity",
"cvss",
"discovered_date",
"remediation_due_date",
"overdue",
"days_overdue"
]
with output.open(
"w",
encoding="utf-8",
newline=""
) as file:
writer = csv.DictWriter(
file,
fieldnames=fields,
extrasaction="ignore"
)
writer.writeheader()
for item in (
build_remediation_queue(
findings
)
):
writer.writerow(
item
)

Create:

def export_invalid_findings(
records
):
output = (
REPORT_DIR /
"invalid-findings.csv"
)
fields = [
"line_number",
"finding_id",
"hostname",
"vulnerability_name",
"severity",
"cvss",
"reason"
]
with output.open(
"w",
encoding="utf-8",
newline=""
) as file:
writer = csv.DictWriter(
file,
fieldnames=fields,
extrasaction="ignore"
)
writer.writeheader()
writer.writerows(
records
)

Create:

def export_duplicates(
records
):
output = (
REPORT_DIR /
"duplicate-findings.csv"
)
fields = [
"finding_id",
"hostname",
"vulnerability_name",
"severity",
"cvss",
"status"
]
with output.open(
"w",
encoding="utf-8",
newline=""
) as file:
writer = csv.DictWriter(
file,
fieldnames=fields,
extrasaction="ignore"
)
writer.writeheader()
writer.writerows(
records
)

Create:

def export_unmapped_findings(
findings
):
output = (
REPORT_DIR /
"unmapped-assets.csv"
)
records = get_unmapped_findings(
findings
)
fields = [
"finding_id",
"hostname",
"vulnerability_name",
"severity",
"cvss"
]
with output.open(
"w",
encoding="utf-8",
newline=""
) as file:
writer = csv.DictWriter(
file,
fieldnames=fields,
extrasaction="ignore"
)
writer.writeheader()
writer.writerows(
records
)

Create:

def export_owner_report(
findings
):
output = (
REPORT_DIR /
"owner-remediation-summary.csv"
)
owner_data = defaultdict(
lambda: {
"open_findings": 0,
"critical": 0,
"p1": 0,
"overdue": 0
}
)
for item in findings:
if item[
"status"
] != "open":
continue
owner = (
item["owner"]
or "Unassigned"
)
owner_data[
owner
][
"open_findings"
] += 1
if item[
"severity"
] == "critical":
owner_data[
owner
][
"critical"
] += 1
if item[
"priority"
] == "P1":
owner_data[
owner
][
"p1"
] += 1
if item[
"overdue"
]:
owner_data[
owner
][
"overdue"
] += 1
with output.open(
"w",
encoding="utf-8",
newline=""
) as file:
writer = csv.writer(
file
)
writer.writerow([
"owner",
"open_findings",
"critical_findings",
"p1_findings",
"overdue_findings"
])
for owner, data in sorted(
owner_data.items()
):
writer.writerow([
owner,
data[
"open_findings"
],
data[
"critical"
],
data[
"p1"
],
data[
"overdue"
]
])

Create:

def export_json(
results,
summary
):
output = (
REPORT_DIR /
"vulnerability-analysis.json"
)
data = {
"reference_date":
str(
REFERENCE_DATE
),
"summary":
summary,
"findings":
results[
"findings"
],
"invalid":
results[
"invalid"
],
"duplicates":
results[
"duplicates"
]
}
with output.open(
"w",
encoding="utf-8"
) as file:
json.dump(
data,
file,
indent=2,
default=str
)

Create:

def generate_markdown_report(
results,
summary
):
findings = results[
"findings"
]
queue = (
build_remediation_queue(
findings
)
)
lines = []
lines.append(
"# Vulnerability Prioritization Report"
)
lines.append("")
lines.append(
f"Reference Date: {REFERENCE_DATE}"
)
lines.append("")
lines.append(
"## Executive Summary"
)
lines.append("")
lines.append(
f"- Open findings: "
f"{summary['open_findings']}"
)
lines.append(
f"- Critical open findings: "
f"{summary['critical_open']}"
)
lines.append(
f"- Overdue findings: "
f"{summary['overdue']}"
)
lines.append(
f"- Internet-exposed findings: "
f"{summary['internet_exposed']}"
)
lines.append(
f"- Open findings without owner: "
f"{summary['unowned_open']}"
)
lines.append(
f"- Duplicate findings removed: "
f"{summary['duplicates_removed']}"
)
lines.append(
f"- Invalid findings rejected: "
f"{summary['invalid_findings']}"
)
lines.append("")
lines.append(
"## Highest Priority Remediation"
)
lines.append("")
for item in queue[:10]:
lines.append(
f"- {item['priority']} | "
f"{item['hostname']} | "
f"{item['vulnerability_name']} | "
f"Score {item['priority_score']} | "
f"Owner: {item['owner'] or 'Unassigned'}"
)
lines.append("")
lines.append(
"## Overdue Findings"
)
lines.append("")
overdue = overdue_findings(
findings
)
if overdue:
for item in sort_findings(
overdue
):
lines.append(
f"- {item['hostname']} | "
f"{item['vulnerability_name']} | "
f"{item['days_overdue']} days overdue"
)
else:
lines.append(
"- No overdue findings."
)
lines.append("")
lines.append(
"## Unmapped Assets"
)
lines.append("")
unmapped = get_unmapped_findings(
findings
)
if unmapped:
for item in unmapped:
lines.append(
f"- {item['hostname']} | "
f"{item['vulnerability_name']}"
)
else:
lines.append(
"- No unmapped assets."
)
lines.append("")
lines.append(
"## Analyst Guidance"
)
lines.append("")
lines.append(
"Priority scores in this lab are training values. "
"Validate remediation urgency using asset criticality, "
"exposure, threat intelligence, exploitability, business "
"impact, compensating controls, and change-management "
"requirements before making production decisions."
)
output = (
REPORT_DIR /
"vulnerability-prioritization-report.md"
)
output.write_text(
"\n".join(
lines
),
encoding="utf-8"
)

Add:

def main():
results = process_findings()
summary = build_summary(
results
)
export_remediation_queue(
results[
"findings"
]
)
export_invalid_findings(
results[
"invalid"
]
)
export_duplicates(
results[
"duplicates"
]
)
export_unmapped_findings(
results[
"findings"
]
)
export_owner_report(
results[
"findings"
]
)
export_json(
results,
summary
)
generate_markdown_report(
results,
summary
)
print(
"Vulnerability analysis complete."
)
print(
f"Reports saved to: "
f"{REPORT_DIR}"
)

Add:

if __name__ == "__main__":
main()

Run:

Terminal window
python src/vulnerability_analyzer.py

Expected:

Vulnerability analysis complete.
Reports saved to: ...

You should now have:

reports/
|
+-- remediation-queue.csv
|
+-- invalid-findings.csv
|
+-- duplicate-findings.csv
|
+-- unmapped-assets.csv
|
+-- owner-remediation-summary.csv
|
+-- vulnerability-analysis.json
|
+-- vulnerability-prioritization-report.md

The queue should place the most important open findings near the top.

Example:

P1 WEB01 Outdated Web Framework
P1 VPN01 Remote Access Security Update
P1 DB01 Missing Security Update
P1/P2 OLD01 Legacy Service Exposure

Exact priority may depend on your calculated score.

Do not prioritize only by:

CVSS

Compare:

CVSS 9.8
on development system

with:

CVSS 8.1
on internet-facing critical production system

Risk context can change urgency.

TECHNICAL SEVERITY
+
ASSET CRITICALITY
+
EXPOSURE
+
THREAT CONTEXT
+
AGE
+
OWNERSHIP
=
REMEDIATION PRIORITY

Example:

summary = count_by_severity(
results["findings"]
)
print(summary)

Expected structure:

{
"critical": 4,
"high": 3,
"medium": 1
}

Exact counts depend on deduplication and open status.

Example:

print(
count_by_priority(
results["findings"]
)
)

This helps produce:

P1 COUNT
P2 COUNT
P3 COUNT
P4 COUNT

A real organization may define:

CRITICAL
15 days
HIGH
30 days
MEDIUM
60 days
LOW
90 days

For the lab, treat existing:

remediation_due_date

as the approved SLA-derived deadline.

Create:

def sla_compliance(
findings
):
open_items = [
item
for item in findings
if item[
"status"
] == "open"
]
if not open_items:
return 100.0
compliant = [
item
for item in open_items
if not item[
"overdue"
]
]
return round(
100.0 *
len(compliant)
/
len(open_items),
2
)

Leadership may ask:

WHAT PERCENTAGE OF OPEN FINDINGS
ARE WITHIN REMEDIATION SLA?

This is different from:

HOW MANY FINDINGS EXIST?

Include:

"open_sla_compliance_percentage":
sla_compliance(
findings
)

Create:

def age_bucket(
days
):
if days is None:
return "unknown"
if days <= 30:
return "0-30"
if days <= 60:
return "31-60"
if days <= 90:
return "61-90"
return "90+"

A vulnerability management dashboard might show:

0–30 DAYS
31–60 DAYS
61–90 DAYS
90+ DAYS

Large numbers in:

90+

may indicate remediation backlog.

Inside time enrichment:

item[
"age_bucket"
] = age_bucket(
item[
"finding_age_days"
]
)

Create:

def count_by_age_bucket(
findings
):
return Counter(
item[
"age_bucket"
]
for item in findings
if item[
"status"
] == "open"
)

A mature remediation queue should answer:

WHO NEEDS TO ACT?

not only:

WHAT IS VULNERABLE?
FINDING
↓
ASSET
↓
OWNER
↓
PRIORITY
↓
DUE DATE
↓
REMEDIATION ACTION

If:

OWNER = EMPTY

do not simply discard the finding.

Create:

OWNERSHIP REVIEW

because unowned systems can become unmanaged risk.

Your pipeline already detects:

INVALID CVSS
UNMAPPED ASSET
DUPLICATE FINDING

A mature program should also detect:

MISSING HOSTNAME
INVALID DATE
UNKNOWN SEVERITY
MISSING FINDING ID
MISSING OWNER
UNKNOWN ENVIRONMENT

Import:

import logging

Configure:

logging.basicConfig(
level=logging.INFO,
format=(
"%(asctime)s "
"%(levelname)s "
"%(message)s"
)
)

Example:

logging.info(
"Loading asset inventory"
)
logging.info(
"Loading vulnerability data"
)
logging.info(
"Deduplicating findings"
)
logging.info(
"Generating remediation reports"
)

Real vulnerability records may reveal:

INTERNAL HOSTNAMES
IP ADDRESSES
SYSTEM OWNERS
SECURITY WEAKNESSES
BUSINESS SYSTEMS

Treat vulnerability reports as sensitive security information.

Import:

import argparse

Create:

def get_arguments():
parser = argparse.ArgumentParser(
description=(
"Analyze vulnerability scanner "
"exports and build a remediation queue."
)
)
parser.add_argument(
"--vulnerabilities",
default=str(
VULNERABILITY_FILE
),
help="Path to vulnerability CSV"
)
parser.add_argument(
"--assets",
default=str(
ASSET_FILE
),
help="Path to asset inventory CSV"
)
return parser.parse_args()

Eventually you could run:

Terminal window
python src/vulnerability_analyzer.py \
--vulnerabilities data/weekly-scan.csv \
--assets data/cmdb-export.csv

The tool should:

ANALYZE
PRIORITIZE
REPORT

It should not automatically:

PATCH SYSTEMS
SHUT DOWN SERVERS
DISABLE SERVICES
CHANGE FIREWALL RULES

without approved remediation workflows.

DISCOVER
↓
VALIDATE
↓
NORMALIZE
↓
DEDUPLICATE
↓
ENRICH
↓
PRIORITIZE
↓
ASSIGN OWNER
↓
REMEDIATE
↓
VERIFY
↓
CLOSE

A vulnerability should not be considered closed because someone says:

PATCH APPLIED

You should ideally:

RESCAN
VERIFY
CONFIRM FINDING NO LONGER EXISTS

A production schema might contain:

status
remediation_status
verification_status

Example:

Open
β†’ Remediation In Progress
β†’ Pending Verification
β†’ Verified Closed

Not every vulnerability can be immediately remediated.

Possible handling includes:

REMEDIATE
MITIGATE
ACCEPT RISK
DECOMMISSION
DEFER WITH APPROVAL

Risk acceptance should include:

OWNER
BUSINESS JUSTIFICATION
EXPIRY DATE
APPROVER
COMPENSATING CONTROLS
REVIEW DATE

not simply:

WE CANNOT FIX IT

Some scanner findings may be inaccurate.

Workflow:

FINDING
↓
VALIDATE
↓
FALSE POSITIVE?
↓
DOCUMENT EVIDENCE
↓
APPROVED EXCEPTION

Instead retain:

WHY IT WAS CLOSED
WHO APPROVED IT
WHEN IT WAS REVIEWED

This supports auditability.

A more advanced prioritization model could add:

KNOWN EXPLOITED?
ACTIVE CAMPAIGN?
EXPLOIT AVAILABLE?
EPSS?

Then:

VULNERABILITY
+
THREAT CONTEXT
=
BETTER PRIORITY

Even if a vulnerability is known to be actively exploited, you still need:

IS THE AFFECTED PRODUCT PRESENT?
IS THE VULNERABLE VERSION PRESENT?
IS THE SYSTEM EXPOSED?
DO COMPENSATING CONTROLS EXIST?

Instead of only:

Critical
High
Medium
Low

some enterprises may use:

Tier 0
Identity-critical
Tier 1
Business-critical
Tier 2
Important
Tier 3
Standard
VULNERABILITY
↓
ASSET
↓
BUSINESS SERVICE
↓
DATA SENSITIVITY
↓
EXPOSURE
↓
THREAT
↓
OWNER
↓
PRIORITY

Create:

tests/test_vulnerability_analyzer.py
def test_normalize_severity():
assert (
normalize_severity(
"CRITICAL"
)
== "critical"
)
def test_valid_cvss():
assert (
parse_cvss(
"9.8"
)
== 9.8
)
def test_invalid_cvss():
assert (
parse_cvss(
"not-a-score"
)
is None
)
def test_hostname_normalization():
assert (
normalize_hostname(
" web01 "
)
== "WEB01"
)
def test_boolean():
assert parse_boolean(
"true"
)
assert not parse_boolean(
"false"
)

Example:

def test_overdue():
finding = {
"status": "open",
"remediation_due_date":
date(
2026,
8,
15
)
}
assert is_overdue(
finding
)
def test_closed_not_overdue():
finding = {
"status": "closed",
"remediation_due_date":
date(
2026,
7,
1
)
}
assert not is_overdue(
finding
)

Provide two:

WEB01
+
Outdated Web Framework
+
Open

records.

Verify:

UNIQUE = 1
DUPLICATE = 1

Ensure:

UNKNOWN01

receives:

asset_mapped = False

Create a synthetic finding with:

critical severity
critical asset
CVSS 9.8
internet exposed
overdue

Verify that it receives:

P1

under your training model.

Add a field:

cve

to the scanner export.

Then use it as part of your deduplication logic.

Add:

port

and:

protocol

This helps distinguish the same finding across different services.

Update:

assets.csv

with:

business_service

Examples:

Customer Portal
Payments
Identity
Corporate VPN

121 β€” Challenge 04 β€” Add Data Classification

Section titled β€œ121 β€” Challenge 04 β€” Add Data Classification”

Add:

data_classification

such as:

Public
Internal
Confidential
Restricted

Then adjust priority.

122 β€” Challenge 05 β€” Add Exploitability Context

Section titled β€œ122 β€” Challenge 05 β€” Add Exploitability Context”

Use synthetic data:

known_exploited
exploit_available

Do not use this as proof of compromise.

Use it only as additional prioritization context.

123 β€” Challenge 06 β€” Add Compensating Controls

Section titled β€œ123 β€” Challenge 06 β€” Add Compensating Controls”

Example fields:

network_isolated
waf_protected
edr_enabled

Then ask:

SHOULD THESE LOWER PRIORITY?
BY HOW MUCH?
WHO APPROVES THAT LOGIC?

Instead of supplying due dates manually, calculate them from:

SEVERITY
DISCOVERED DATE
SLA POLICY

Example training values:

Critical
15 days
High
30 days
Medium
60 days
Low
90 days

125 β€” Challenge 08 β€” Add Owner-Specific Reports

Section titled β€œ125 β€” Challenge 08 β€” Add Owner-Specific Reports”

Generate:

Web-Team.csv
Database-Team.csv
Network-Team.csv
Unassigned.csv

Create a simple HTML report showing:

TOTAL OPEN
P1
P2
OVERDUE
INTERNET EXPOSED
UNOWNED
TOP OWNERS

127 β€” Challenge 10 β€” Store Results in SQLite

Section titled β€œ127 β€” Challenge 10 β€” Store Results in SQLite”

Architecture:

SCANNER CSV
↓
PYTHON
↓
NORMALIZED FINDINGS
↓
SQLITE
↓
SQL REPORTING

This connects directly to:

Lab 05 β€” SQL Security Analytics

Create:

scan-week-1.csv
scan-week-2.csv

Identify:

NEW FINDINGS
REMEDIATED FINDINGS
PERSISTENT FINDINGS
PREVIOUS SCAN
↓
COMPARE
↓
CURRENT SCAN
↓
NEW
REMEDIATED
UNCHANGED

Management should not only ask:

HOW MANY VULNERABILITIES
DO WE HAVE?

Also ask:

ARE WE GETTING BETTER?
ARE FINDINGS AGING?
ARE P1 ITEMS CLOSING?
IS THE BACKLOG GROWING?

If you add:

closed_date

you can calculate:

MTTR

Conceptually:

CLOSED DATE
-
DISCOVERED DATE
=
REMEDIATION TIME

Useful metrics may include:

OPEN FINDINGS
CRITICAL OPEN
P1 OPEN
OVERDUE
SLA COMPLIANCE
UNOWNED FINDINGS
INTERNET-EXPOSED FINDINGS
90+ DAY FINDINGS
MEAN TIME TO REMEDIATE
REOPENED FINDINGS

Metrics can drive bad behavior if poorly designed.

Example:

GOAL:
REDUCE VULNERABILITY COUNT

could encourage teams to close:

LOW-RISK EASY FINDINGS

while ignoring:

HIGH-RISK DIFFICULT FINDINGS

Measure:

RISK REDUCTION

not only:

TICKET VOLUME

You find:

WEB01
Critical vulnerability
CVSS 9.8
Internet exposed
14 days overdue

What should happen?

Review:

VALIDATE FINDING
CONFIRM AFFECTED VERSION
CHECK EXPOSURE
CHECK COMPENSATING CONTROLS
CONTACT OWNER
PRIORITIZE CHANGE
REMEDIATE
RESCAN

You find:

OLD01
Critical vulnerability
No asset owner
Production
High criticality

The vulnerability is important.

But there is also a:

GOVERNANCE FAILURE

because nobody owns the system.

You find:

UNKNOWN01
High severity
CVSS 8.0
Asset not in inventory

Possible response:

VERIFY HOSTNAME
SEARCH CMDB
CHECK SCANNER ASSET ID
CHECK WHETHER SYSTEM WAS DECOMMISSIONED
IDENTIFY OWNER
DO NOT DISCARD THE FINDING

A finding has:

CVSS 9.8

but the asset is:

Development
Not internet exposed
Short-lived lab system

Should it automatically be the organization’s top risk?

Not necessarily.

Context matters.

A finding has:

CVSS 7.5

on:

JUMP01
Critical administrative host

It may deserve greater attention than raw CVSS suggests because of:

ASSET ROLE
DO NOT ASK ONLY:
"HOW SEVERE IS
THE VULNERABILITY?"
ALSO ASK:
"WHERE IS IT?"
"WHAT DOES THE ASSET DO?"
"WHO OWNS IT?"
"IS IT EXPOSED?"
"HOW OLD IS IT?"
"IS IT BEING EXPLOITED?"
"WHAT CONTROLS EXIST?"

Your final repository should look like:

vulnerability-prioritization/
|
+-- data/
| +-- assets.csv
| +-- vulnerabilities.csv
|
+-- src/
| +-- vulnerability_analyzer.py
|
+-- tests/
| +-- test_vulnerability_analyzer.py
|
+-- reports/
| +-- remediation-queue.csv
| +-- invalid-findings.csv
| +-- duplicate-findings.csv
| +-- unmapped-assets.csv
| +-- owner-remediation-summary.csv
| +-- vulnerability-analysis.json
| +-- vulnerability-prioritization-report.md
|
+-- README.md
|
+-- architecture.md

Include:

PROJECT OVERVIEW
SECURITY PROBLEM
ARCHITECTURE
INPUT DATA
NORMALIZATION
DEDUPLICATION
ASSET ENRICHMENT
RISK MODEL
OUTPUT REPORTS
HOW TO RUN
TESTING
SECURITY CONSIDERATIONS
LIMITATIONS
FUTURE IMPROVEMENTS

Be transparent.

Do not simply output:

Risk Score = 114

without explaining where it came from.

Document:

Severity weight
Asset criticality weight
CVSS contribution
Internet exposure weight
Overdue weight
Missing-owner weight

Security teams need to understand:

WHY DID THIS FINDING
BECOME P1?

A transparent model supports:

AUDITABILITY
TRUST
TUNING
GOVERNANCE

Avoid:

PRIORITIZING ONLY BY CVSS
NO ASSET CONTEXT
NO OWNERSHIP DATA
NO DEDUPLICATION
NO SLA TRACKING
IGNORING OLD FINDINGS
IGNORING UNKNOWN ASSETS
COUNTING CLOSED ITEMS AS ACTIVE RISK
NO FALSE-POSITIVE PROCESS
NO VERIFICATION AFTER REMEDIATION
AUTOMATICALLY PATCHING WITHOUT CHANGE CONTROL
NO DOCUMENTED RISK MODEL

Avoid:

TRUSTING EVERY CSV FIELD
CRASHING ON ONE BAD RECORD
SILENTLY DISCARDING INVALID DATA
CASE-SENSITIVE HOSTNAME MISMATCHES
BAD DATE PARSING
NO DUPLICATE HANDLING
NO REPORT VERSIONING
NO LOGGING

Confirm:

  • Python environment prepared
  • Asset CSV created
  • Vulnerability CSV created
  • Asset inventory loads
  • Vulnerability data loads
  • Required columns validated
  • Hostnames normalized
  • Severity normalized
  • CVSS validated
  • Dates parsed
  • Invalid findings recorded
  • Duplicate findings identified
  • Duplicate findings removed
  • Findings mapped to assets
  • Unknown assets identified
  • Asset criticality added
  • Asset owner added
  • Internet exposure added
  • Overdue status calculated
  • Days overdue calculated
  • Finding age calculated
  • Priority score calculated
  • Priority category calculated
  • Remediation queue generated
  • P1 findings reviewed
  • Open critical findings reviewed
  • Internet-exposed findings reviewed
  • Unowned findings reviewed
  • Owner summary generated
  • CSV reports generated
  • JSON report generated
  • Markdown report generated
  • Risk formula documented
  • Bad-data behavior tested
  • Deduplication tested
  • Priority logic tested
  • No automatic remediation performed
  • Limitations documented

You started with:

RAW VULNERABILITY
SCANNER DATA

containing:

DUPLICATES
BAD DATA
SEVERITY
CVSS
DATES
HOSTNAMES

You transformed it into:

SCANNER FINDING
↓
VALIDATE
↓
NORMALIZE
↓
DEDUPLICATE
↓
MAP ASSET
↓
ADD BUSINESS CONTEXT
↓
CHECK EXPOSURE
↓
CHECK AGE / SLA
↓
CALCULATE PRIORITY
↓
ASSIGN OWNER
↓
REMEDIATION QUEUE

You now have a vulnerability management analytics workflow capable of answering:

WHAT SHOULD WE FIX FIRST?
WHICH FINDINGS ARE OVERDUE?
WHICH CRITICAL ASSETS ARE AFFECTED?
WHICH INTERNET-EXPOSED ASSETS ARE VULNERABLE?
WHICH FINDINGS HAVE NO OWNER?
WHICH SCANNER RECORDS ARE DUPLICATES?
WHICH ASSETS ARE MISSING FROM INVENTORY?

The central lesson is:

VULNERABILITY
SEVERITY
IS NOT THE SAME AS
BUSINESS RISK

A useful remediation decision requires:

TECHNICAL SEVERITY
+
ASSET CRITICALITY
+
EXPOSURE
+
THREAT CONTEXT
+
AGE
+
OWNERSHIP
+
BUSINESS IMPACT
=
REMEDIATION PRIORITY

Whenever vulnerability data arrives, think:

IS THE DATA VALID?
↓
IS IT A DUPLICATE?
↓
WHICH ASSET IS AFFECTED?
↓
HOW CRITICAL IS THE ASSET?
↓
IS IT EXPOSED?
↓
HOW SEVERE IS THE FINDING?
↓
HOW OLD IS IT?
↓
IS IT OVERDUE?
↓
WHO OWNS IT?
↓
WHAT SHOULD BE FIXED FIRST?
↓
HOW WILL WE VERIFY CLOSURE?

The goal is not:

GENERATE A LONGER
VULNERABILITY REPORT

The goal is:

TURN SCANNER DATA
INTO A PRIORITIZED,
OWNED,
TRACKABLE
REMEDIATION PROGRAM

➑️ Lab 07 β€” Security API Integration

The next lab moves from local CSV-based vulnerability analysis into external security-system integration.

You will build:

SECURITY AUTOMATION
↓
API CLIENT
↓
AUTHENTICATION
↓
REQUEST
↓
TIMEOUT
↓
RESPONSE VALIDATION
↓
JSON PARSING
↓
ERROR HANDLING
↓
RETRY
↓
RATE LIMIT AWARENESS
↓
NORMALIZED SECURITY DATA

You will use a controlled training API or local mock service to learn how security tools exchange data safely without hard-coding secrets or depending on uncontrolled external systems.