Hands-On: Security Testing Pipeline

# GitLab CI: security gates -- every merge request runs all stages

stages: [sast, dependency_check, secret_scan, binary_hardening, misra, fuzzing]

# Stage 1: SAST -- Coverity on changed C files; block on CVSS >= 7.0 new finding
coverity_scan:
  stage: sast
  script:
    - cov-build --dir cov-int make
    - cov-analyze --dir cov-int --all
    - cov-format-errors --dir cov-int --json-output-v7 coverity.json
    - python3 ci/check_coverity.py --max-cvss 7.0 coverity.json
  rules: [{changes: ["src/**/*.c","src/**/*.h"]}]

# Stage 2: SBOM vs NVD CVE database
sbom_cve_check:
  stage: dependency_check
  script:
    - dependency-check --project ECU --scan src/ --format JSON --out dep.json
    - python3 ci/check_cves.py dep.json --max-cvss 7.0 --fail-on-new

# Stage 3: Secret scanning -- no hardcoded keys/credentials
secret_scan:
  stage: secret_scan
  script:
    - trufflehog git file://. --json --no-history > secrets.json
    - python3 ci/fail_on_secrets.py secrets.json

# Stage 4: Binary hardening -- fail on missing stack canary or NX
binary_hardening:
  stage: binary_hardening
  script:
    - make all
    - checksec --file build/ecu.elf --json | python3 ci/check_hardening.py

# Stage 5: MISRA check -- fail on new Required violations
misra_check:
  stage: misra
  script:
    - pclint-plus -u src/ -o lint.xml
    - python3 ci/check_misra.py lint.xml --fail-on-required

# Stage 6: Automated UDS fuzzer -- zero new crashes required
uds_fuzzing:
  stage: fuzzing
  script:
    - python3 ci/uds_fuzzer_ci.py --iterations 5000 --interface vcan0
    - python3 ci/assert_zero_crashes.py fuzzer_results.json

#!/usr/bin/env python3
# CI UDS fuzzing: 5000 iterations, crash detection via CAN heartbeat
import can, random, time, json, sys

def run_ci_fuzzing(iface="vcan0", iterations=5000, heartbeat_id=0x3C5):
    bus = can.interface.Bus(iface, bustype="socketcan")
    crashes = []

    for i in range(iterations):
        sid = random.randint(0x00,0xFF)
        length = random.choice([0,1,7,8,255,random.randint(0,4095)])
        payload = bytes([sid]) + random.randbytes(min(length,7))
        bus.send(can.Message(arbitration_id=0x7DF, data=payload[:8], is_extended_id=False))
        time.sleep(0.005)

        if i % 100 == 0:
            # Check ECU heartbeat (100 ms expected period)
            deadline = time.time() + 0.5
            alive = False
            while time.time() < deadline:
                msg = bus.recv(timeout=0.05)
                if msg and msg.arbitration_id == heartbeat_id:
                    alive = True; break
            if not alive:
                crashes.append({"iteration": i, "payload": payload.hex()})
                print(f"CRASH detected at iteration {i}")

    results = {"iterations": iterations, "crashes": crashes}
    with open("fuzzer_results.json","w") as f: json.dump(results, f, indent=2)
    print(f"Fuzzing complete: {iterations} iterations, {len(crashes)} crashes")
    bus.shutdown()
    return crashes

if __name__ == "__main__":
    crashes = run_ci_fuzzing()
    if crashes: sys.exit(1)  # fail CI on any crash

#!/usr/bin/env python3
# SBOM vs NVD CVE correlation -- nightly job; new CVSS >= 7.0 creates Jira ticket
import json, requests
from datetime import date

def load_sbom(spdx_path: str) -> list:
    with open(spdx_path) as f: spdx = json.load(f)
    return [{"name": p["name"], "version": p["versionInfo"],
             "supplier": p.get("supplier","")}
            for p in spdx.get("packages",[])]

def check_nvd(package_name: str, version: str) -> list:
    # Query OSV.dev API for known CVEs in package + version
    resp = requests.post("https://api.osv.dev/v1/query",
        json={"package": {"name": package_name, "ecosystem": "CRAN"},
              "version": version}, timeout=10)
    if resp.status_code == 200:
        return resp.json().get("vulns", [])
    return []

def create_jira_ticket(cve_id: str, component: str, cvss: float, sla_days: int):
    due_date = date.today()
    print(f"JIRA: {cve_id} [{component}] CVSS {cvss} -- due {due_date} (+{sla_days}d)")

# Main: load SBOM, check each component, file tickets for CVSS >= 7.0
# sbom = load_sbom("build/sbom.spdx.json")
# for component in sbom:
#     vulns = check_nvd(component["name"], component["version"])
#     for v in vulns:
#         cvss = v.get("database_specific",{}).get("severity_cvss_v3",0)
#         if cvss >= 7.0:
#             create_jira_ticket(v["id"], component["name"], cvss,
#                                30 if cvss>=9.0 else 90)

#!/usr/bin/env python3
# Security regression: replay known-crash payloads against new firmware build
# If a previously-fixed payload now causes no crash → regression test PASSES
# If a previously-fixed payload still causes a crash → regression FAILS (re-introduced bug)

import can, json, time, sys

KNOWN_CRASH_DB = "security_crashes.json"  # built up from pentest + fuzzing history

def run_regression(iface="vcan0") -> dict:
    with open(KNOWN_CRASH_DB) as f:
        known_crashes = json.load(f)

    bus = can.interface.Bus(iface, bustype="socketcan")
    results = {"passed":[], "failed":[]}

    for crash in known_crashes:
        payload = bytes.fromhex(crash["payload"])
        frame = can.Message(arbitration_id=0x7DF, data=payload[:8], is_extended_id=False)
        bus.send(frame)
        time.sleep(0.05)

        # Check if ECU still responds (heartbeat within 500ms)
        deadline = time.time() + 0.5; alive = False
        while time.time() < deadline:
            msg = bus.recv(timeout=0.05)
            if msg and msg.arbitration_id == 0x3C5: alive = True; break

        test_id = crash.get("id", crash["payload"][:8])
        if alive:
            results["passed"].append(test_id)  # bug fixed → ECU survives
        else:
            results["failed"].append(test_id)  # regression! bug re-introduced
            print(f"REGRESSION: {test_id} -- previously-fixed crash re-introduced!")

    bus.shutdown()
    print(f"Regression: {len(results['passed'])} passed, {len(results['failed'])} failed")
    return results

if __name__ == "__main__":
    r = run_regression()
    if r["failed"]: sys.exit(1)  # block release on regression