# 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.