Lab Specification — Module S08: SDLC Gate Harnesses

Course: 2A — Building AI Harnesses for Cybersecurity Module: S08 — SDLC Gate Harnesses Duration: 120–150 minutes (four labs, one per sub-section) Environment: Python 3.11+, Pydantic. Semgrep, Gitleaks, Checkov, and Snyk/OWASP Dependency-Check CLIs. A deliberately vulnerable sample repo (provided). An LLM API key for cross-scanner triage. Sample NVD/EPSS/KEV data (provided JSON).

Learning objectives

  1. Design a hard-gate vs soft-gate decision matrix per scan type and justify each threshold.
  2. Orchestrate SAST, SCA, secrets, and IaC scanners in parallel; aggregate and deduplicate into a unified finding list.
  3. Build a trend and risk-scoring system over simulated build history; implement a PR gate from the risk score.
  4. Build a CVE triage harness using NVD, EPSS, and CISA KEV to produce a prioritized remediation queue.

Phase 1 — Gate Decision Matrix (25 min)

1.1 Define the matrix

from pydantic import BaseModel
from typing import Literal

class GateDecision(BaseModel):
    scan_type: Literal["secrets", "sast", "sca", "iac", "license"]
    condition: str           # e.g. "validated + high confidence"
    action: Literal["hard_block", "soft_warn"]
    rationale: str

GATE_MATRIX = [
    GateDecision(scan_type="secrets", condition="validated + high confidence",
                 action="hard_block", rationale="Active breach path"),
    GateDecision(scan_type="sast", condition="critical + high confidence",
                 action="hard_block", rationale="Exploit path ships"),
    GateDecision(scan_type="sca", condition="CISA KEV or EPSS > 0.7",
                 action="hard_block", rationale="Known/likely exploited"),
    GateDecision(scan_type="iac", condition="public exposure (0.0.0.0/0, public S3)",
                 action="hard_block", rationale="Immediate risk"),
    GateDecision(scan_type="sast", condition="medium/low",
                 action="soft_warn", rationale="Surface for review, don't block"),
    GateDecision(scan_type="sca", condition="low EPSS, fix available",
                 action="soft_warn", rationale="Triage and schedule"),
    GateDecision(scan_type="license", condition="unknown/copyleft",
                 action="soft_warn", rationale="Legal review, not security block"),
]

def evaluate_gate(finding: UnifiedFinding, epss: float = 0.0, in_kev: bool = False) -> str:
    """Apply the matrix to a finding; return hard_block or soft_warn."""
    for rule in GATE_MATRIX:
        if matches(finding, rule, epss, in_kev):
            return rule.action
    return "soft_warn"  # default: warn, don't block

1.2 Document and defend each threshold

For each rule in the matrix, write a one-paragraph rationale defending the gate choice. Specifically address: why does this finding hard-block (immediately dangerous) vs soft-warn (needs attention, not immediate)?

Deliverable

Phase 2 — Multi-Scanner Orchestration (40 min)

2.1 Run the four scanners in parallel

import asyncio, subprocess, json

async def run_sast(repo_path: str) -> list[dict]:
    proc = await asyncio.create_subprocess_exec(
        "semgrep", "--config", "auto", "--json", repo_path,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, _ = await proc.communicate()
    return json.loads(stdout).get("results", [])

async def run_secrets(repo_path: str) -> list[dict]:
    proc = await asyncio.create_subprocess_exec(
        "gitleaks", "detect", "--source", repo_path, "--report-format", "json",
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, _ = await proc.communicate()
    return json.loads(stdout) if stdout.strip() else []

async def run_iac(repo_path: str) -> list[dict]:
    proc = await asyncio.create_subprocess_exec(
        "checkov", "-d", repo_path, "--output", "json",
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, _ = await proc.communicate()
    return json.loads(stdout).get("results", {}).get("failed_checks", [])

async def run_sca(manifest_path: str) -> list[dict]:
    # Snyk or OWASP Dependency-Check
    ...

async def run_all_scanners(repo_path: str, manifest_path: str) -> dict:
    """Run all 4 scanner types concurrently."""
    results = await asyncio.gather(
        run_sast(repo_path), run_sca(manifest_path),
        run_secrets(repo_path), run_iac(repo_path)
    )
    return {"sast": results[0], "sca": results[1], "secrets": results[2], "iac": results[3]}

2.2 Normalize to the unified schema

class UnifiedFinding(BaseModel):
    scanner: Literal["sast", "sca", "secrets", "iac"]
    tool: str
    severity: Literal["critical", "high", "medium", "low"]
    cwe: str | None = None
    cve: str | None = None
    location: dict
    description: str
    dedup_key: str
    confidence: Literal["high", "medium", "low"]

SEVERITY_MAP = {
    "semgrep": {"ERROR": "high", "WARNING": "medium", "INFO": "low"},
    "snyk": {"critical": "critical", "high": "high", "medium": "medium", "low": "low"},
    "gitleaks": {"high": "critical", "medium": "high", "low": "medium"},
    "checkov": {"FAILED": "high", "PASSED": "low"},
}

def normalize(raw: dict, scanner: str, tool: str) -> UnifiedFinding:
    raw_sev = raw.get("extra", {}).get("severity", raw.get("severity", "medium"))
    severity = SEVERITY_MAP[tool].get(raw_sev, "medium")
    # Build dedup_key from scanner-agnostic identity
    dedup = f"{scanner}:{raw.get('cwe', raw.get('cve', ''))}:{raw.get('path', '')}:{raw.get('start_line', raw.get('line', ''))}"
    return UnifiedFinding(scanner=scanner, tool=tool, severity=severity, ...)

2.3 Cross-scanner dedup

def dedup_findings(findings: list[UnifiedFinding]) -> list[UnifiedFinding]:
    """Dedup via dedup_key; merge cross-scanner duplicates."""
    seen = {}
    for f in findings:
        if f.dedup_key in seen:
            # Retain highest severity
            if SEV_ORDER[f.severity] > SEV_ORDER[seen[f.dedup_key].severity]:
                seen[f.dedup_key] = f
        else:
            seen[f.dedup_key] = f
    return list(seen.values())

2.4 Run against the deliberately vulnerable repo

  1. Run all four scanners (parallel) against the provided sample repo.
  2. Normalize all results to UnifiedFinding.
  3. Dedup. Count raw vs deduplicated findings.

Deliverable

Phase 3 — Trend Analysis and Risk Scoring (35 min)

3.1 Simulate build history

import random

def simulate_build_history(num_builds: int = 10) -> list[BuildRecord]:
    """Simulate 10 builds with varying finding counts."""
    history = []
    base_counts = {"critical": 2, "high": 5, "medium": 12, "low": 20}
    for i in range(num_builds):
        # Add some drift: rising trend in first half, falling in second
        drift = 1 + (0.05 * (i - num_builds/2))
        counts = {sev: max(0, int(base * drift + random.randint(-2, 2)))
                  for sev, base in base_counts.items()}
        history.append(BuildRecord(
            build_id=f"build-{i}", timestamp=f"2026-07-0{i+1}T10:00:00Z",
            pr_id=f"pr-{100+i}", branch="main",
            finding_counts={"sast": counts, "sca": {}, "secrets": {}, "iac": {}},
            risk_score=0, author=f"dev-{i % 3}"
        ))
    return history

3.2 Compute risk score per PR

SEVERITY_WEIGHTS = {"critical": 25, "high": 10, "medium": 3, "low": 1}

def compute_risk_score(build: BuildRecord, history: list[BuildRecord]) -> float:
    baseline = average_counts(history[:-1]) if len(history) > 1 else {}
    new_findings = diff_counts(build.finding_counts, baseline)
    score = sum(
        count * SEVERITY_WEIGHTS[sev]
        for scan_type in new_findings.values()
        for sev, count in scan_type.items()
    )
    if trend_is_rising(history, window=10):
        score *= 1.2
    return min(score, 100.0)

def trend_is_rising(history: list[BuildRecord], window: int = 10) -> bool:
    """Is the total critical+high count rising over the last N builds?"""
    recent = history[-window:]
    totals = [sum(b.finding_counts.get("sast", {}).get(s, 0) for s in ["critical", "high"]) for b in recent]
    return len(totals) >= 2 and totals[-1] > totals[0]

3.3 Implement the PR gate

def gate_from_score(score: float) -> str:
    if score < 30:
        return "PASS"
    elif score <= 60:
        return "REQUIRES_SECURITY_REVIEW"
    else:
        return "BLOCKED"

3.4 Run and visualize

  1. Simulate 10 builds.
  2. Compute the risk score for each.
  3. Plot the trend chart (finding counts + risk score over builds).
  4. For a simulated current PR, compute the score and apply the gate.

Deliverable

Phase 4 — CVE Triage Harness (35 min)

4.1 Ingest the CVE feed and match dependencies

def match_cves(manifest: dict, nvd_feed: list[dict]) -> list[dict]:
    """Match dependencies in the manifest against the NVD feed."""
    matches = []
    for pkg, version in manifest.items():
        for cve in nvd_feed:
            if affects(cve, pkg, version):
                matches.append({**cve, "matched_package": pkg, "matched_version": version})
    return matches

4.2 Score with EPSS and check CISA KEV

def enrich_with_exploitability(matches: list[dict], epss_data: dict, kev_set: set) -> list[dict]:
    for m in matches:
        cve_id = m["cve_id"]
        m["epss"] = epss_data.get(cve_id, 0.0)
        m["in_kev"] = cve_id in kev_set
    return matches

4.3 Produce the prioritized remediation queue

def prioritize(findings: list[dict]) -> list[dict]:
    """Sort into the 5-tier priority queue."""
    def tier(f):
        if f["in_kev"]:
            return 0  # CISA KEV → fix in HOURS
        if f["epss"] > 0.7 and f.get("reachable", True):
            return 1  # EPSS > 0.7, reachable → fix in DAYS
        if f["epss"] > 0.7:
            return 2  # EPSS > 0.7, not confirmed → verify reachability
        if f["cvss"] >= 7.0:
            return 3  # High CVSS, low EPSS → backlog
        return 4    # Low CVSS, low EPSS → accept risk / batch-fix

    return sorted(findings, key=lambda f: (tier(f), -f.get("cvss", 0)))

4.4 Run against the provided data

  1. Load the provided mock-manifest.json (10 dependencies), mock-nvd.json (50 CVEs), mock-epss.json, and mock-kev.json.
  2. Match dependencies to CVEs.
  3. Enrich with EPSS and KEV status.
  4. Produce the prioritized queue.

Deliverable

Stretch goals

  1. Add environment-relevance filtering: given a reachability report from a SAST tool (which functions are called), filter out CVEs whose vulnerable code path is not reachable. Measure how much the queue shrinks.
  2. Implement bug report intake: write a parser for a mock HackerOne report JSON. Dedup it against the scanner findings via semantic similarity. Auto-route a P1 report to immediate response.
  3. Add cross-scanner LLM triage: group related findings (e.g. an SCA vuln and a SAST finding in the same file) and ask the LLM to correlate — upgrade or downgrade severity based on combined context.
# Lab Specification — Module S08: SDLC Gate Harnesses

**Course**: 2A — Building AI Harnesses for Cybersecurity
**Module**: S08 — SDLC Gate Harnesses
**Duration**: 120–150 minutes (four labs, one per sub-section)
**Environment**: Python 3.11+, Pydantic. Semgrep, Gitleaks, Checkov, and Snyk/OWASP Dependency-Check CLIs. A deliberately vulnerable sample repo (provided). An LLM API key for cross-scanner triage. Sample NVD/EPSS/KEV data (provided JSON).

---

## Learning objectives

1. Design a hard-gate vs soft-gate decision matrix per scan type and justify each threshold.
2. Orchestrate SAST, SCA, secrets, and IaC scanners in parallel; aggregate and deduplicate into a unified finding list.
3. Build a trend and risk-scoring system over simulated build history; implement a PR gate from the risk score.
4. Build a CVE triage harness using NVD, EPSS, and CISA KEV to produce a prioritized remediation queue.

---

## Phase 1 — Gate Decision Matrix (25 min)

### 1.1 Define the matrix

```python
from pydantic import BaseModel
from typing import Literal

class GateDecision(BaseModel):
    scan_type: Literal["secrets", "sast", "sca", "iac", "license"]
    condition: str           # e.g. "validated + high confidence"
    action: Literal["hard_block", "soft_warn"]
    rationale: str

GATE_MATRIX = [
    GateDecision(scan_type="secrets", condition="validated + high confidence",
                 action="hard_block", rationale="Active breach path"),
    GateDecision(scan_type="sast", condition="critical + high confidence",
                 action="hard_block", rationale="Exploit path ships"),
    GateDecision(scan_type="sca", condition="CISA KEV or EPSS > 0.7",
                 action="hard_block", rationale="Known/likely exploited"),
    GateDecision(scan_type="iac", condition="public exposure (0.0.0.0/0, public S3)",
                 action="hard_block", rationale="Immediate risk"),
    GateDecision(scan_type="sast", condition="medium/low",
                 action="soft_warn", rationale="Surface for review, don't block"),
    GateDecision(scan_type="sca", condition="low EPSS, fix available",
                 action="soft_warn", rationale="Triage and schedule"),
    GateDecision(scan_type="license", condition="unknown/copyleft",
                 action="soft_warn", rationale="Legal review, not security block"),
]

def evaluate_gate(finding: UnifiedFinding, epss: float = 0.0, in_kev: bool = False) -> str:
    """Apply the matrix to a finding; return hard_block or soft_warn."""
    for rule in GATE_MATRIX:
        if matches(finding, rule, epss, in_kev):
            return rule.action
    return "soft_warn"  # default: warn, don't block
```

### 1.2 Document and defend each threshold

For each rule in the matrix, write a one-paragraph rationale defending the gate choice. Specifically address: why does this finding hard-block (immediately dangerous) vs soft-warn (needs attention, not immediate)?

### Deliverable
- [ ] Gate decision matrix implemented with all scan types
- [ ] `evaluate_gate` correctly routes findings to hard_block or soft_warn
- [ ] Each threshold defended with a written rationale

---

## Phase 2 — Multi-Scanner Orchestration (40 min)

### 2.1 Run the four scanners in parallel

```python
import asyncio, subprocess, json

async def run_sast(repo_path: str) -> list[dict]:
    proc = await asyncio.create_subprocess_exec(
        "semgrep", "--config", "auto", "--json", repo_path,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, _ = await proc.communicate()
    return json.loads(stdout).get("results", [])

async def run_secrets(repo_path: str) -> list[dict]:
    proc = await asyncio.create_subprocess_exec(
        "gitleaks", "detect", "--source", repo_path, "--report-format", "json",
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, _ = await proc.communicate()
    return json.loads(stdout) if stdout.strip() else []

async def run_iac(repo_path: str) -> list[dict]:
    proc = await asyncio.create_subprocess_exec(
        "checkov", "-d", repo_path, "--output", "json",
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, _ = await proc.communicate()
    return json.loads(stdout).get("results", {}).get("failed_checks", [])

async def run_sca(manifest_path: str) -> list[dict]:
    # Snyk or OWASP Dependency-Check
    ...

async def run_all_scanners(repo_path: str, manifest_path: str) -> dict:
    """Run all 4 scanner types concurrently."""
    results = await asyncio.gather(
        run_sast(repo_path), run_sca(manifest_path),
        run_secrets(repo_path), run_iac(repo_path)
    )
    return {"sast": results[0], "sca": results[1], "secrets": results[2], "iac": results[3]}
```

### 2.2 Normalize to the unified schema

```python
class UnifiedFinding(BaseModel):
    scanner: Literal["sast", "sca", "secrets", "iac"]
    tool: str
    severity: Literal["critical", "high", "medium", "low"]
    cwe: str | None = None
    cve: str | None = None
    location: dict
    description: str
    dedup_key: str
    confidence: Literal["high", "medium", "low"]

SEVERITY_MAP = {
    "semgrep": {"ERROR": "high", "WARNING": "medium", "INFO": "low"},
    "snyk": {"critical": "critical", "high": "high", "medium": "medium", "low": "low"},
    "gitleaks": {"high": "critical", "medium": "high", "low": "medium"},
    "checkov": {"FAILED": "high", "PASSED": "low"},
}

def normalize(raw: dict, scanner: str, tool: str) -> UnifiedFinding:
    raw_sev = raw.get("extra", {}).get("severity", raw.get("severity", "medium"))
    severity = SEVERITY_MAP[tool].get(raw_sev, "medium")
    # Build dedup_key from scanner-agnostic identity
    dedup = f"{scanner}:{raw.get('cwe', raw.get('cve', ''))}:{raw.get('path', '')}:{raw.get('start_line', raw.get('line', ''))}"
    return UnifiedFinding(scanner=scanner, tool=tool, severity=severity, ...)
```

### 2.3 Cross-scanner dedup

```python
def dedup_findings(findings: list[UnifiedFinding]) -> list[UnifiedFinding]:
    """Dedup via dedup_key; merge cross-scanner duplicates."""
    seen = {}
    for f in findings:
        if f.dedup_key in seen:
            # Retain highest severity
            if SEV_ORDER[f.severity] > SEV_ORDER[seen[f.dedup_key].severity]:
                seen[f.dedup_key] = f
        else:
            seen[f.dedup_key] = f
    return list(seen.values())
```

### 2.4 Run against the deliberately vulnerable repo

1. Run all four scanners (parallel) against the provided sample repo.
2. Normalize all results to `UnifiedFinding`.
3. Dedup. Count raw vs deduplicated findings.

### Deliverable
- [ ] Four scanners run concurrently (verify parallel timing)
- [ ] All results normalized to the unified schema with severity mapping
- [ ] Cross-scanner dedup collapses duplicates (raw count vs dedup count documented)

---

## Phase 3 — Trend Analysis and Risk Scoring (35 min)

### 3.1 Simulate build history

```python
import random

def simulate_build_history(num_builds: int = 10) -> list[BuildRecord]:
    """Simulate 10 builds with varying finding counts."""
    history = []
    base_counts = {"critical": 2, "high": 5, "medium": 12, "low": 20}
    for i in range(num_builds):
        # Add some drift: rising trend in first half, falling in second
        drift = 1 + (0.05 * (i - num_builds/2))
        counts = {sev: max(0, int(base * drift + random.randint(-2, 2)))
                  for sev, base in base_counts.items()}
        history.append(BuildRecord(
            build_id=f"build-{i}", timestamp=f"2026-07-0{i+1}T10:00:00Z",
            pr_id=f"pr-{100+i}", branch="main",
            finding_counts={"sast": counts, "sca": {}, "secrets": {}, "iac": {}},
            risk_score=0, author=f"dev-{i % 3}"
        ))
    return history
```

### 3.2 Compute risk score per PR

```python
SEVERITY_WEIGHTS = {"critical": 25, "high": 10, "medium": 3, "low": 1}

def compute_risk_score(build: BuildRecord, history: list[BuildRecord]) -> float:
    baseline = average_counts(history[:-1]) if len(history) > 1 else {}
    new_findings = diff_counts(build.finding_counts, baseline)
    score = sum(
        count * SEVERITY_WEIGHTS[sev]
        for scan_type in new_findings.values()
        for sev, count in scan_type.items()
    )
    if trend_is_rising(history, window=10):
        score *= 1.2
    return min(score, 100.0)

def trend_is_rising(history: list[BuildRecord], window: int = 10) -> bool:
    """Is the total critical+high count rising over the last N builds?"""
    recent = history[-window:]
    totals = [sum(b.finding_counts.get("sast", {}).get(s, 0) for s in ["critical", "high"]) for b in recent]
    return len(totals) >= 2 and totals[-1] > totals[0]
```

### 3.3 Implement the PR gate

```python
def gate_from_score(score: float) -> str:
    if score < 30:
        return "PASS"
    elif score <= 60:
        return "REQUIRES_SECURITY_REVIEW"
    else:
        return "BLOCKED"
```

### 3.4 Run and visualize

1. Simulate 10 builds.
2. Compute the risk score for each.
3. Plot the trend chart (finding counts + risk score over builds).
4. For a simulated current PR, compute the score and apply the gate.

### Deliverable
- [ ] 10-build history simulated with visible trend
- [ ] Risk score computed per build with severity weights + trend modifier
- [ ] PR gate implemented (pass / requires review / blocked)
- [ ] Trend chart produced (finding counts + risk score over time)

---

## Phase 4 — CVE Triage Harness (35 min)

### 4.1 Ingest the CVE feed and match dependencies

```python
def match_cves(manifest: dict, nvd_feed: list[dict]) -> list[dict]:
    """Match dependencies in the manifest against the NVD feed."""
    matches = []
    for pkg, version in manifest.items():
        for cve in nvd_feed:
            if affects(cve, pkg, version):
                matches.append({**cve, "matched_package": pkg, "matched_version": version})
    return matches
```

### 4.2 Score with EPSS and check CISA KEV

```python
def enrich_with_exploitability(matches: list[dict], epss_data: dict, kev_set: set) -> list[dict]:
    for m in matches:
        cve_id = m["cve_id"]
        m["epss"] = epss_data.get(cve_id, 0.0)
        m["in_kev"] = cve_id in kev_set
    return matches
```

### 4.3 Produce the prioritized remediation queue

```python
def prioritize(findings: list[dict]) -> list[dict]:
    """Sort into the 5-tier priority queue."""
    def tier(f):
        if f["in_kev"]:
            return 0  # CISA KEV → fix in HOURS
        if f["epss"] > 0.7 and f.get("reachable", True):
            return 1  # EPSS > 0.7, reachable → fix in DAYS
        if f["epss"] > 0.7:
            return 2  # EPSS > 0.7, not confirmed → verify reachability
        if f["cvss"] >= 7.0:
            return 3  # High CVSS, low EPSS → backlog
        return 4    # Low CVSS, low EPSS → accept risk / batch-fix

    return sorted(findings, key=lambda f: (tier(f), -f.get("cvss", 0)))
```

### 4.4 Run against the provided data

1. Load the provided `mock-manifest.json` (10 dependencies), `mock-nvd.json` (50 CVEs), `mock-epss.json`, and `mock-kev.json`.
2. Match dependencies to CVEs.
3. Enrich with EPSS and KEV status.
4. Produce the prioritized queue.

### Deliverable
- [ ] Dependencies matched to CVEs from the NVD feed
- [ ] Each match enriched with EPSS score and KEV status
- [ ] Prioritized remediation queue (5 tiers) produced
- [ ] Top 3 items (the "fix this week" list) identified and justified

---

## Stretch goals

1. **Add environment-relevance filtering**: given a reachability report from a SAST tool (which functions are called), filter out CVEs whose vulnerable code path is not reachable. Measure how much the queue shrinks.
2. **Implement bug report intake**: write a parser for a mock HackerOne report JSON. Dedup it against the scanner findings via semantic similarity. Auto-route a P1 report to immediate response.
3. **Add cross-scanner LLM triage**: group related findings (e.g. an SCA vuln and a SAST finding in the same file) and ask the LLM to correlate — upgrade or downgrade severity based on combined context.