Add DevSecOps CI validation workflow

Author: ZENLIK02

.github/workflows/heimdall-devsecops.yml

@@ -0,0 +1,50 @@
+name: Heimdall DevSecOps Validation
+
+on:
+  pull_request:
+  push:
+  workflow_dispatch:
+
+jobs:
+  validate:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      pull-requests: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install semgrep
+
+      - name: Run Semgrep
+        run: |
+          semgrep scan --config auto --json --output semgrep-results.json || true
+
+      - name: Run Heimdall policy validation
+        run: |
+          python -m heimdall.cli validate --semgrep semgrep-results.json --config heimdall.yml --output reports/
+
+      - name: Post PR comment when available
+        if: always()
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          python scripts/post_pr_comment.py || true
+
+      - name: Upload Heimdall reports
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: heimdall-devsecops-reports
+          path: reports/

README.md

@@ -155,6 +155,47 @@ Interpretation guidelines:
 
 The workflow `.github/workflows/heimdall.yml` installs dependencies, runs Semgrep when available, writes Semgrep JSON, executes Heimdall in dry-run mode with the sample dataset, uploads reports as artifacts, and fails only when the full pipeline confirms High or Critical True Positives.
 
+## DevSecOps Integration
+
+Heimdall V2 can run as a CI/CD validation backend for pull requests and pushes. The DevSecOps workflow consumes Semgrep JSON, converts findings into Heimdall alerts, runs the same safety-first validation pipeline, writes CI reports, and exits according to policy.
+
+Local commands:
+
+```bash
+python -m heimdall.cli check-config --config heimdall.yml
+python -m heimdall.cli validate --semgrep test_data/semgrep-results-sample.json --config heimdall.yml --output reports/
+python -m heimdall.cli experiment --dataset data/sample_alerts.jsonl --mode all --output reports/
+```
+
+GitHub Actions workflow:
+
+- `.github/workflows/heimdall-devsecops.yml` runs on `push` and `pull_request`.
+- Semgrep writes `semgrep-results.json`.
+- Heimdall writes `reports/ci_summary.md`, `reports/ci_results.json`, and `reports/ci_results.csv`.
+- Reports are uploaded as workflow artifacts.
+- `scripts/post_pr_comment.py` can post the Markdown summary to a pull request when `GITHUB_TOKEN` and PR context are available.
+
+Policy behavior:
+
+- Exit code `0` when no confirmed High/Critical True Positive is found.
+- Exit code `0` when findings are only False Positive or Needs Review.
+- Exit code `1` when a confirmed High/Critical True Positive is found and `heimdall.yml` says to fail.
+- Exit code `2` for unsafe or invalid config.
+- Exit code `3` for runtime errors.
+
+Needs Review does not fail by default because it indicates missing context, authentication, multi-step state, or safety restrictions rather than confirmed exploitability.
+
+Safety defaults:
+
+- Dry-run is enabled.
+- Mock LLM is enabled.
+- Active DAST is limited to `security.allowed_targets`.
+- Production-looking domains are blocked unless explicitly enabled.
+- Destructive payload markers are rejected.
+- All DAST attempts are logged.
+
+See `docs/devsecops_integration.md` for the full CI/CD architecture and local vulnerable app instructions.
+
 ## Safety Warning
 
 Do not run dynamic validation against production systems unless explicit authorization and target allowlisting are configured. Heimdall V2 defaults to dry-run validation, local targets, non-destructive payloads, request logging, rate limiting, timeouts, and a kill switch.

docs/devsecops_integration.md

@@ -0,0 +1,74 @@
+# Heimdall DevSecOps Integration
+
+Heimdall V2 can run as a CI/CD security validation backend. In this mode, Semgrep produces static findings, Heimdall converts those findings into alert objects, the validation pipeline classifies each alert, and the CLI returns a policy-based exit code.
+
+## CI/CD Workflow
+
+```mermaid
+flowchart TD
+    A[Developer opens PR] --> B[GitHub Actions starts]
+    B --> C[Semgrep scans code]
+    C --> D[Semgrep JSON saved]
+    D --> E[Heimdall validates findings]
+    E --> F[Heimdall classifies TP / FP / Needs Review]
+    F --> G[Report uploaded]
+    G --> H[Pipeline passes or fails based on policy]
+```
+
+## GitHub Actions Flow
+
+The workflow `.github/workflows/heimdall-devsecops.yml` installs Python dependencies, installs Semgrep, writes `semgrep-results.json`, runs `python -m heimdall.cli validate`, uploads reports, and optionally posts `reports/ci_summary.md` to pull requests.
+
+## Semgrep-To-Heimdall Data Flow
+
+1. Semgrep writes JSON results.
+2. `heimdall.semgrep_ingest` preserves rule ID, severity, file path, line number, message, snippet, and CWE metadata where available.
+3. Findings are converted into Heimdall `Alert` objects.
+4. The existing prompt guard, mock LLM provider, payload generator, DAST executor, response analyzer, and decision engine run in dry-run mode by default.
+5. CI reports are written to `reports/`.
+
+## Policy Decision Logic
+
+- Exit code `0`: no confirmed High/Critical True Positive was found.
+- Exit code `0`: only False Positive or Needs Review findings exist.
+- Exit code `1`: confirmed High/Critical True Positive exists and policy says to fail.
+- Exit code `2`: config is invalid or unsafe.
+- Exit code `3`: runtime error.
+
+Needs Review does not fail the pipeline by default because it means the system lacks enough context for safe automated validation.
+
+## Safety Model
+
+- Dry-run is enabled by default.
+- Mock LLM is enabled by default.
+- DAST refuses targets not present in `security.allowed_targets`.
+- Production-looking domains are rejected unless `security.allow_external_targets` is explicitly enabled.
+- Blocked targets cannot also be allowlisted.
+- Every DAST attempt is logged.
+- The kill switch stops DAST immediately.
+
+## Local Integration Test
+
+Start the local test app:
+
+```bash
+cd test_apps/flask_vulnerable_app
+python -m pip install flask
+python app.py
+```
+
+Run Heimdall with the sample Semgrep output:
+
+```bash
+python -m heimdall.cli check-config --config heimdall.yml
+python -m heimdall.cli validate --semgrep test_data/semgrep-results-sample.json --config heimdall.yml --output reports/
+```
+
+Read `reports/ci_summary.md` for the CI-friendly result.
+
+## Limitations
+
+- The default validation path is conservative and dry-run based.
+- Authenticated and multi-step workflows require additional context.
+- Real LLM providers should remain behind the structured output validator.
+- Production deployment requires audited target allowlists, secret management, and review of policy thresholds.

heimdall.yml

@@ -0,0 +1,24 @@
+security:
+  dry_run: true
+  allowed_targets:
+    - "http://localhost:5000"
+    - "http://127.0.0.1:5000"
+  blocked_targets:
+    - "https://production.example.com"
+  fail_on_confirmed_high: true
+  fail_on_confirmed_critical: true
+  needs_review_does_not_fail: true
+
+dast:
+  max_requests_per_scan: 50
+  request_timeout_seconds: 5
+
+llm:
+  provider: "mock"
+  use_mock_llm: true
+
+reports:
+  output_dir: "reports"
+
+semgrep:
+  output_path: "semgrep-results.json"

heimdall/ci_policy.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from heimdall.config import HeimdallConfig
+from heimdall.evaluation.models import EvaluationResult
+
+
+def determine_exit_code(results: list[EvaluationResult], config: HeimdallConfig) -> int:
+    for result in results:
+        severity = result.severity.lower()
+        confirmed = result.final_decision == "True Positive" or result.classification == "TP"
+        if confirmed and severity == "critical" and config.security.fail_on_confirmed_critical:
+            return 1
+        if confirmed and severity == "high" and config.security.fail_on_confirmed_high:
+            return 1
+    return 0
+
+
+def policy_summary(results: list[EvaluationResult], config: HeimdallConfig) -> dict:
+    high_critical = [
+        result
+        for result in results
+        if (result.final_decision == "True Positive" or result.classification == "TP")
+        and result.severity.lower() in {"high", "critical"}
+    ]
+    exit_code = determine_exit_code(results, config)
+    return {
+        "passed": exit_code == 0,
+        "exit_code": exit_code,
+        "high_critical_confirmed": len(high_critical),
+        "needs_review_does_not_fail": config.security.needs_review_does_not_fail,
+    }

heimdall/ci_reports.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import csv
+import json
+from dataclasses import asdict
+from pathlib import Path
+
+from heimdall.ci_policy import policy_summary
+from heimdall.config import HeimdallConfig
+from heimdall.evaluation.models import EvaluationResult
+
+
+def write_ci_reports(output_dir: str | Path, results: list[EvaluationResult], total_semgrep: int, config: HeimdallConfig) -> None:
+    output = Path(output_dir)
+    output.mkdir(parents=True, exist_ok=True)
+    rows = [asdict(result) for result in results]
+    (output / "ci_results.json").write_text(json.dumps(rows, indent=2, ensure_ascii=False), encoding="utf-8")
+    _write_csv(output / "ci_results.csv", rows)
+    _write_markdown(output / "ci_summary.md", results, total_semgrep, config)
+
+
+def _write_csv(path: Path, rows: list[dict]) -> None:
+    fields = [
+        "alert_id",
+        "severity",
+        "vulnerability_type",
+        "prediction",
+        "classification",
+        "final_decision",
+        "evidence",
+        "recommended_action",
+    ]
+    with path.open("w", encoding="utf-8", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=fields)
+        writer.writeheader()
+        for row in rows:
+            writer.writerow({field: _csv_value(row.get(field, "")) for field in fields})
+
+
+def _write_markdown(path: Path, results: list[EvaluationResult], total_semgrep: int, config: HeimdallConfig) -> None:
+    confirmed = [row for row in results if row.final_decision == "True Positive" or row.classification == "TP"]
+    false_positive = [row for row in results if row.final_decision == "False Positive" or row.classification in {"TN", "FN"}]
+    review = [row for row in results if row.final_decision == "Needs Review" or row.classification == "REVIEW"]
+    high_critical = [row for row in confirmed if row.severity.lower() in {"high", "critical"}]
+    policy = policy_summary(results, config)
+    lines = [
+        "# Heimdall CI/CD Summary",
+        "",
+        f"Pipeline status: {'PASSED' if policy['passed'] else 'FAILED'}",
+        "",
+        "| Metric | Count |",
+        "|---|---:|",
+        f"| Total Semgrep findings | {total_semgrep} |",
+        f"| Total validated findings | {len(results)} |",
+        f"| Confirmed True Positives | {len(confirmed)} |",
+        f"| False Positives | {len(false_positive)} |",
+        f"| Needs Review | {len(review)} |",
+        f"| High/Critical confirmed vulnerabilities | {len(high_critical)} |",
+        "",
+        "## Findings",
+        "",
+        "| Rule ID | Severity | File | Line | Heimdall Decision | Evidence | Recommended Action |",
+        "|---|---|---|---:|---|---|---|",
+    ]
+    for result in results:
+        metadata = result.metadata or {}
+        file_path = metadata.get("file_path", "")
+        line = metadata.get("line_number", "")
+        lines.append(
+            "| {rule} | {severity} | {file} | {line} | {decision} | {evidence} | {action} |".format(
+                rule=_md(result.alert_id),
+                severity=_md(result.severity),
+                file=_md(str(file_path)),
+                line=line,
+                decision=_md(result.final_decision or result.prediction),
+                evidence=_md((result.evidence or result.rationale)[:160]),
+                action=_md((result.recommended_action or "Review finding.")[:160]),
+            )
+        )
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def _csv_value(value: object) -> object:
+    if isinstance(value, (dict, list)):
+        return json.dumps(value, ensure_ascii=False)
+    return value
+
+
+def _md(value: str) -> str:
+    return value.replace("|", "\\|").replace("\n", " ")