diff --git a/.github/workflows/strix.yml b/.github/workflows/strix.yml index 071927b7..8ab7d77a 100644 --- a/.github/workflows/strix.yml +++ b/.github/workflows/strix.yml @@ -577,6 +577,7 @@ jobs: echo "STRIX_LLM_FILE=$strix_llm_file" >> "$GITHUB_ENV" - name: Run Strix (quick) + id: run_strix if: steps.gate.outputs.enabled == 'true' timeout-minutes: 30 # Security invariant for pull_request_target: execute only from the @@ -665,6 +666,70 @@ jobs: if-no-files-found: error retention-days: 5 + # Best-effort source-side emission of per-finding security issues into the + # ContextualWisdomLab/appguardrail tracker. This step must never change the + # Strix pass/fail gate: it is continue-on-error and the emitter itself is + # internally best-effort (always exits 0). + # + # Auth reuses the EXISTING OpenCode GitHub App token that this job already + # OIDC-exchanges for cross-repo reads/status (steps.target_app_token) -- no + # new GitHub App and no new org secrets are provisioned. If that token is + # unavailable (OIDC exchange failed) or lacks Issues: write on appguardrail, + # the emitter degrades to DRY-RUN / swallows the write and only logs the + # intended issue operations. See docs/strix-appguardrail-issues.md for the + # one-time OpenCode App permission/installation step this emitter relies on. + - name: Emit per-finding Strix issues to appguardrail + if: ${{ always() && steps.gate.outputs.enabled == 'true' }} + continue-on-error: true + env: + STRIX_ISSUE_APP_TOKEN: ${{ steps.target_app_token.outputs.token || '' }} + SOURCE_REPOSITORY: ${{ github.event.pull_request.base.repo.full_name || github.event.inputs.target_repository || github.repository }} + SOURCE_PR_NUMBER: ${{ github.event_name == 'pull_request_target' && github.event.pull_request.number || github.event.inputs.pr_number }} + SOURCE_HEAD_SHA: ${{ github.event_name == 'pull_request_target' && github.event.pull_request.head.sha || github.event.inputs.pr_head_sha || github.sha }} + # Close-on-fix runs only when the scan completed cleanly (gate passed). + # A failed or skipped Strix step leaves this false so a partial or + # infra-broken scan never closes issues. + STRIX_SCAN_COMPLETE: ${{ steps.run_strix.outcome == 'success' && 'true' || 'false' }} + # Scan scope mirrors the PR-scoping decision used by the Run Strix step + # (STRIX_DISABLE_PR_SCOPING / IS_PR_EVIDENCE_RUN / __PR_SCOPE__): PR + # events and workflow_dispatch with a pr_number scan only the PR's + # changed files, so their scope is 'pr' and close-on-fix MUST NOT run + # (an absent finding just means "outside this PR", not "fixed"). The + # push/schedule full-repo path scans everything, so its scope is 'full' + # and close-on-fix is allowed. Absence of this signal defaults to 'pr' + # (safe) in the emitter. + STRIX_SCAN_SCOPE: ${{ (github.event_name == 'pull_request_target' || github.event.inputs.pr_number != '') && 'pr' || 'full' }} + STRIX_RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + run: | + set -uo pipefail + emitter="$TRUSTED_STRIX_SOURCE/scripts/ci/strix_emit_appguardrail_issues.py" + if [ ! -f "$emitter" ]; then + echo "::notice::Strix issue emitter not present in trusted source; skipping." + exit 0 + fi + # Default to the safe 'pr' scope: close-on-fix only runs on a full scan. + scan_scope="${STRIX_SCAN_SCOPE:-pr}" + if [ "$scan_scope" != "full" ]; then + scan_scope="pr" + fi + set -- \ + --run-dir "$GITHUB_WORKSPACE/strix_runs" \ + --source-repo "$SOURCE_REPOSITORY" \ + --issues-repo "ContextualWisdomLab/appguardrail" \ + --run-url "$STRIX_RUN_URL" \ + --scope "$scan_scope" + if [ -n "${SOURCE_PR_NUMBER:-}" ]; then + set -- "$@" --pr-number "$SOURCE_PR_NUMBER" + fi + if [ -n "${SOURCE_HEAD_SHA:-}" ]; then + set -- "$@" --head-sha "$SOURCE_HEAD_SHA" + fi + if [ "${STRIX_SCAN_COMPLETE:-false}" = "true" ]; then + set -- "$@" --scan-complete + fi + # Never fail the Strix gate on issue-emission problems. + python3 "$emitter" "$@" || echo "::warning::Strix issue emission encountered an error; continuing." + - name: Publish same-head manual Strix status if: ${{ always() && !cancelled() && github.event_name == 'workflow_dispatch' && github.event.inputs.pr_head_sha != '' }} env: diff --git a/docs/strix-appguardrail-issues.md b/docs/strix-appguardrail-issues.md new file mode 100644 index 00000000..f13041ee --- /dev/null +++ b/docs/strix-appguardrail-issues.md @@ -0,0 +1,132 @@ +# Source-side Strix → appguardrail issue emitter + +The central Strix security workflow (`.github/workflows/strix.yml`) runs as a +required org workflow across (nearly) all repositories. Each run writes one +Markdown report per vulnerability under `strix_runs//vulnerabilities/*.md`. +This system turns those reports into **per-finding GitHub issues** in the +`ContextualWisdomLab/appguardrail` tracker, deduplicated and lifecycle-managed +so the tracker always reflects the current state of each repository's findings. + +It **replaces** the previous approach, where `appguardrail` polled failed runs +and opened one coarse issue per failed run with no close-on-fix. That collector +also never worked because its GitHub App identity was never provisioned. + +## Components + +| File | Role | +| ---- | ---- | +| `scripts/ci/strix_emit_appguardrail_issues.py` | Parses reports, plans and applies issue operations. Pure parsing/planning + a thin `gh api` client. | +| `tests/test_strix_emit_appguardrail_issues.py` | Unit tests: parsing, dedup hashing, op planning, close-on-fix set difference, incomplete-scan guard, dry-run and live execution. | +| `.github/workflows/strix.yml` (final steps) | Runs the emitter after the scan/gate, best-effort, reusing the OpenCode App token the job already OIDC-exchanges. | + +## How it works + +### Parsing + +Each `vulnerabilities/*.md` report is parsed into a normalized `Finding`: +`title`, `severity` (CRITICAL/HIGH/MEDIUM/LOW/NONE), `cvss` + vector, `target`, +`endpoint`, `method`, `model`, `code_location` (`path:line[-range]`), +`description`, `impact`, `remediation`. Field lines tolerate plain +(`Severity: HIGH`) and Markdown-bold (`- **Severity:** HIGH`) styles, and code +locations are recovered from a `Code Locations` section, a labelled line, or a +prose reference. `/workspace//` and PR-scope sandbox prefixes are stripped +so a file hashes identically across runs. + +### Deduplication + +The stable identity of a finding is: + +``` +dedup_key = sha256(source_repo + "\n" + finding_title + "\n" + normalized_code_location) +``` + +Titles are whitespace-collapsed. The full hash is stored in a hidden issue-body +marker ``; the first 12 hex chars also become a +label `strix-finding:`. Existing issues (open **or** closed) are +looked up by this hash before anything is created. Because the location is part +of the key, a finding that moves to a different line is a **new identity**: the +new location gets a fresh issue and the stale one is closed on fix. + +### Issue shape + +- **Title**: `[strix] : (<path>:<line>)` +- **Labels**: `strix`, `security`, `repo:<name>`, `severity:<level>`, + `strix-finding:<shorthash>` +- **Body**: full finding details, source repo/PR/head/run links, and three + hidden markers (`strix-finding`, `strix-severity`, `strix-location`) that drive + reconciliation. + +### Lifecycle + +| Situation | Action | +| --------- | ------ | +| Finding has no existing issue | **Create** an issue. | +| Finding matches an open issue | **Update** the body (refresh run/head/PR, CVSS, remediation). **Comment** only if severity changed. | +| Finding matches a closed issue | **Reopen** (update to `state: open`) and refresh. | +| Open issue's finding absent from a **complete** scan | **Close** with a `Resolved on <head_sha>` comment. | + +### Close-on-fix guard + +Close-on-fix runs **only when the scan completed cleanly**. In the workflow this +is gated on `steps.run_strix.outcome == 'success'` (`STRIX_SCAN_COMPLETE=true`). +A failed or skipped Strix step — which includes provider/infra errors and +incomplete scans — leaves the flag `false`, so issues are created/updated but +**never closed** on a partial scan. The emitter defaults `--scan-complete` off, +so the safe behaviour is the default. + +### Authentication and DRY-RUN + +The emitter needs a token with **Issues: write** on `appguardrail` (the org +`github.token` cannot create issues cross-repo). Instead of provisioning a new +GitHub App, the workflow **reuses the OpenCode GitHub App token it already +obtains** — the same token minted by the `Exchange OpenCode app token for target +repository reads` step (`steps.target_app_token`), which the job already uses for +cross-repo reads and status writes. That token is exchanged over OIDC +(`POST https://api.opencode.ai/exchange_github_app_token`) with **no stored +secret** and passed to the emitter via the `STRIX_ISSUE_APP_TOKEN` environment +variable. **No new GitHub App and no new org secrets are created.** + +If that variable is empty — because the OIDC exchange was unavailable — the +emitter runs in **DRY-RUN**: it parses, plans, and logs every intended +create/update/close operation, mutates nothing, and exits 0. If the token is +present but lacks Issues: write on `appguardrail`, the emitter degrades the same +way: a failed issue *list* read plans in DRY-RUN, and any individual write that +is rejected is logged as a warning and swallowed. Either way the Strix gate +result is never affected. `--dry-run` forces DRY-RUN for local use and tests, and +anything resembling a token is redacted from log output. + +## One-time enablement (manual, tiny — no new App or secret) + +The emitter reuses the **existing** OpenCode GitHub App, so there is nothing to +provision. It writes real issues as soon as that App can write issues to +`appguardrail`. Verify the two conditions below once; each is a couple of clicks +and neither creates a new App or secret: + +1. **Grant the OpenCode App `Issues: Read and write`.** + `github.com/organizations/ContextualWisdomLab/settings/apps` → open the + **OpenCode** app → **Permissions & events** → **Repository permissions** → + set **Issues** to **Read and write** → **Save changes**. (If the App is owned + by OpenCode rather than the org, this permission is already part of the + `opencode-github-action` App and no action is needed; an org-owned install + just needs the permission accepted.) +2. **Ensure the App is installed on `appguardrail`.** + `github.com/organizations/ContextualWisdomLab/settings/installations` → open + the **OpenCode** installation → **Repository access** → confirm **appguardrail** + is included (either **All repositories** or **Only select repositories** with + `appguardrail` checked) → **Save**. If a permission change from step 1 is + pending, an org owner accepts it on this same screen. + +Until both hold, the emitter stays in DRY-RUN and only logs intended operations — +the Strix gate is unaffected. No code change or redeploy is needed to switch out +of DRY-RUN; the next scan after the App can write issues begins emitting for real. + +## Retiring the old collector (do this after this lands) + +The previous per-failed-run collector in **`ContextualWisdomLab/appguardrail`**, +`.github/workflows/org-security-failure-collector.yml`, must be **disabled or +removed** once this source-side emitter is live, otherwise the two systems will +file duplicate/overlapping issues. That change lives in the `appguardrail` repo +and cannot be made from this repository — remove the workflow (or set it to +`workflow_dispatch`-only / delete it) as a follow-up PR there. The +`ORG_SECURITY_FAILURE_APP_ID` / `ORG_SECURITY_FAILURE_APP_PRIVATE_KEY` secrets it +depended on can also be retired. diff --git a/scripts/ci/noema_review_gate.py b/scripts/ci/noema_review_gate.py index 621e4506..51c96c2a 100644 --- a/scripts/ci/noema_review_gate.py +++ b/scripts/ci/noema_review_gate.py @@ -299,7 +299,7 @@ def call_llm(repo: str, number: int, pr: dict[str, Any], diff: str, truncated: b if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_unspecified: raise ValueError("URL cannot target internal IP addresses") - if not (api_url.startswith("http://") or api_url.startswith("https://")): + if not (api_url.lower().startswith("http://") or api_url.lower().startswith("https://")): raise ValueError(f"NOEMA_LLM_API_URL must start with http:// or https:// to prevent SSRF vulnerabilities, got: {api_url}") prompt = { diff --git a/scripts/ci/strix_emit_appguardrail_issues.py b/scripts/ci/strix_emit_appguardrail_issues.py new file mode 100755 index 00000000..be2d7aec --- /dev/null +++ b/scripts/ci/strix_emit_appguardrail_issues.py @@ -0,0 +1,808 @@ +#!/usr/bin/env python3 +"""Emit per-finding Strix security issues into the appguardrail tracker. + +The central Strix workflow (``.github/workflows/strix.yml``) writes one Markdown +report per vulnerability under ``strix_runs/<run>/vulnerabilities/*.md``. This +module parses those reports into normalized finding records and reconciles them +against issues in ``ContextualWisdomLab/appguardrail``: + +* one open issue per distinct finding (deduplicated by a stable content hash), +* body refreshed on every run, a comment added only when severity or location + changed, +* close-on-fix for issues whose finding disappeared — but only from a *complete + full-repo* scan. A PR-scoped scan (``--scope pr``) inspects just the PR's + changed files, so an absent finding means "outside this PR", never "fixed"; + such scans create/update/reopen findings but never close, preventing a clean + PR from wiping out every open finding in files it never touched. + +When the GitHub App token is absent the module runs in DRY-RUN: every intended +create/update/close operation is logged and nothing is mutated, so the Strix job +never fails because issue emission could not authenticate. A ``--dry-run`` flag +forces the same behaviour for tests. + +GitHub access is funneled through :class:`GitHubIssueClient`, a thin ``gh api`` +wrapper that is injected so tests can substitute an in-memory fake. All parsing, +hashing, content rendering and operation planning are pure functions. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import subprocess +import sys +from collections.abc import Iterable, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +DEFAULT_ISSUES_REPO = "ContextualWisdomLab/appguardrail" +DEFAULT_TOKEN_ENV = "STRIX_ISSUE_APP_TOKEN" +# Scan scope values. Only a FULL-repo scan sees every finding, so only a full +# scan may conclude that an absent finding is fixed and close its issue. A +# PR-scoped scan only inspects the PR's changed files, so a finding's absence +# means "not in this PR" — never "fixed" — and must not close anything. +SCOPE_FULL = "full" +SCOPE_PR = "pr" +FINDING_MARKER_PREFIX = "<!-- strix-finding:" +SEVERITY_MARKER_PREFIX = "<!-- strix-severity:" +LOCATION_MARKER_PREFIX = "<!-- strix-location:" +SHORT_HASH_LENGTH = 12 + +SEVERITY_ORDER = { + "CRITICAL": 4, + "HIGH": 3, + "MEDIUM": 2, + "LOW": 1, + "NONE": 0, + "INFO": 0, + "INFORMATIONAL": 0, +} + +# Field-line regexes tolerate optional Markdown bullets/bold, e.g. +# "- **Severity:** HIGH" or "Severity: HIGH". +_FIELD_PREFIX = r"^[\s>*_`-]*\**\s*" + + +def _field_pattern(names: str) -> re.Pattern[str]: + """Compile a case-insensitive field-line regex for the given field names.""" + return re.compile( + _FIELD_PREFIX + rf"(?:{names})\**\s*:\**\s*(.+?)\s*$", + re.IGNORECASE, + ) + + +TITLE_RE = _field_pattern("Title") +SEVERITY_RE = _field_pattern("Severity") +CVSS_SCORE_RE = _field_pattern("CVSS Score|CVSS") +CVSS_VECTOR_RE = _field_pattern("CVSS Vector|Vector") +TARGET_RE = _field_pattern("Target") +ENDPOINT_RE = _field_pattern("Endpoint") +METHOD_RE = _field_pattern("Method") +MODEL_RE = _field_pattern("Model") +HEADING_RE = re.compile(r"^#{1,6}\s+(.+?)\s*#*\s*$") +SECTION_RE = re.compile(_FIELD_PREFIX + r"(Description|Impact|Remediation)\**\s*:?\s*$", re.IGNORECASE) +CODE_LOCATIONS_HEADER_RE = re.compile(_FIELD_PREFIX + r"Code Locations?\**\s*:?\s*$", re.IGNORECASE) +# path:line or path:line-range, path allows repo-relative and /workspace forms. +LOCATION_RE = re.compile( + r"(?P<path>(?:/workspace/|/tmp/strix-pr-scope\.[^\s:`]+/)?[A-Za-z0-9_][A-Za-z0-9_./\[\]-]*\.[A-Za-z0-9_]+)" + r":(?P<start>\d+)(?:-(?P<end>\d+))?" +) + + +@dataclass +class Finding: + """A single normalized Strix vulnerability finding.""" + + source_repo: str + title: str + severity: str + code_location: str + cvss: str = "" + cvss_vector: str = "" + target: str = "" + endpoint: str = "" + method: str = "" + model: str = "" + description: str = "" + impact: str = "" + remediation: str = "" + source_file: str = "" + + @property + def normalized_location(self) -> str: + """Return the code location used for dedup (empty locations collapse).""" + return normalize_location(self.code_location) + + @property + def finding_hash(self) -> str: + """Return the stable dedup hash for this finding.""" + return finding_dedup_hash(self.source_repo, self.title, self.code_location) + + @property + def short_hash(self) -> str: + """Return the shortened dedup hash used in labels.""" + return self.finding_hash[:SHORT_HASH_LENGTH] + + +@dataclass +class Operation: + """A planned issue mutation (create/update/comment/close).""" + + action: str + finding_hash: str + short_hash: str + title: str + issue_number: int | None = None + reason: str = "" + labels: list[str] = field(default_factory=list) + body: str = "" + comment: str = "" + + def describe(self) -> str: + """Return a single-line human-readable summary of the operation.""" + target = f"#{self.issue_number}" if self.issue_number is not None else "new" + detail = f" ({self.reason})" if self.reason else "" + return f"{self.action.upper()} {target} [{self.short_hash}] {self.title}{detail}" + + +def severity_rank(severity: str) -> int: + """Return a numeric rank for a severity label (unknown severities rank -1).""" + return SEVERITY_ORDER.get((severity or "").strip().upper(), -1) + + +def normalize_location(location: str) -> str: + """Normalize a code location string for stable hashing/comparison. + + Strips ``/workspace/<repo>/`` and PR-scope sandbox prefixes so the same file + hashes identically across runs, and collapses whitespace. + """ + value = (location or "").strip() + if not value: + return "" + value = re.sub(r"^/tmp/strix-pr-scope\.[^/]+/", "", value) + value = re.sub(r"^/workspace/[^/]+/", "", value) + value = value.lstrip("/") + return value + + +def finding_dedup_hash(source_repo: str, title: str, code_location: str) -> str: + """Return the SHA-256 dedup key for a finding. + + Key = sha256(source_repo + '\\n' + title + '\\n' + normalized_code_location). + Titles are whitespace-collapsed so cosmetic wrapping differences do not fork + the identity of a finding. + """ + normalized_title = re.sub(r"\s+", " ", (title or "").strip()) + payload = "\n".join( + [ + (source_repo or "").strip(), + normalized_title, + normalize_location(code_location), + ] + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _match_field(pattern: re.Pattern[str], line: str) -> str | None: + """Return the captured field value for ``line`` or ``None``.""" + match = pattern.match(line) + if match: + return match.group(1).strip().strip("*").strip("`").strip() + return None + + +def parse_finding_markdown(text: str, source_repo: str, source_file: str = "") -> Finding | None: + """Parse one Strix vulnerability Markdown report into a :class:`Finding`. + + Returns ``None`` when the report has no title (i.e. is not a real finding). + Missing fields are tolerated; a finding with no code location keeps an empty + location (still deduplicated by repo+title). + """ + lines = text.splitlines() + title = severity = cvss = cvss_vector = target = endpoint = method = model = "" + location = "" + explicit_title = False + section_text: dict[str, list[str]] = {"description": [], "impact": [], "remediation": []} + current_section: str | None = None + in_code_locations = False + + for raw_line in lines: + line = raw_line.rstrip() + + heading = HEADING_RE.match(line) + if heading and not title: + candidate = heading.group(1).strip() + # Ignore generic report banners as titles. + if candidate.lower() not in {"vulnerability report", "strix", "findings"}: + title = candidate + current_section = None + in_code_locations = False + continue + + for pattern, setter in ( + (TITLE_RE, "title"), + (SEVERITY_RE, "severity"), + (CVSS_VECTOR_RE, "cvss_vector"), + (CVSS_SCORE_RE, "cvss"), + (TARGET_RE, "target"), + (ENDPOINT_RE, "endpoint"), + (METHOD_RE, "method"), + (MODEL_RE, "model"), + ): + value = _match_field(pattern, line) + if value is not None: + if setter == "title": + title = value + explicit_title = True + elif setter == "severity": + severity = value.upper() + elif setter == "cvss_vector": + cvss_vector = value + elif setter == "cvss": + cvss = value + elif setter == "target": + target = value + elif setter == "endpoint": + endpoint = value + elif setter == "method": + method = value + elif setter == "model": + model = value + current_section = None + in_code_locations = False + break + else: + section = SECTION_RE.match(line) + if section: + current_section = section.group(1).lower() + in_code_locations = False + continue + if CODE_LOCATIONS_HEADER_RE.match(line): + in_code_locations = True + current_section = None + continue + if in_code_locations or not location: + loc_match = LOCATION_RE.search(line) + if loc_match and (in_code_locations or _looks_like_location_line(line)): + location = _format_location(loc_match) + in_code_locations = False + continue + if current_section and line.strip(): + section_text[current_section].append(line.strip()) + + # Fall back to any location anywhere in the report. + if not location: + loc_match = LOCATION_RE.search(text) + if loc_match: + location = _format_location(loc_match) + + # A heading alone is only a finding when it carries a real finding signal. + has_finding_signal = explicit_title or bool(severity) or bool(location) or bool(cvss) + if not title or not has_finding_signal: + return None + + severity = severity if severity_rank(severity) >= 0 else severity.upper() + + return Finding( + source_repo=source_repo, + title=re.sub(r"\s+", " ", title).strip(), + severity=severity, + code_location=location, + cvss=cvss, + cvss_vector=cvss_vector, + target=target, + endpoint=endpoint, + method=method, + model=model, + description=" ".join(section_text["description"]).strip(), + impact=" ".join(section_text["impact"]).strip(), + remediation=" ".join(section_text["remediation"]).strip(), + source_file=source_file, + ) + + +def _looks_like_location_line(line: str) -> bool: + """Return whether a non-section line plausibly introduces a code location.""" + lowered = line.lower() + return any(key in lowered for key in ("location", "file", "path", "line")) + + +def _format_location(match: re.Match[str]) -> str: + """Render a location regex match as ``path:start[-end]``.""" + path = match.group("path").strip() + start = match.group("start") + end = match.group("end") + if end and end != start: + return f"{path}:{start}-{end}" + return f"{path}:{start}" + + +def iter_vulnerability_files(run_dir: Path) -> list[Path]: + """Return sorted ``vulnerabilities/*.md`` report paths beneath ``run_dir``. + + Accepts either a single run directory or a parent containing multiple runs; + symlinked report files/directories are skipped for safety. + """ + if not run_dir.is_dir(): + return [] + results: list[Path] = [] + for vuln_dir in sorted(run_dir.rglob("vulnerabilities")): + if not vuln_dir.is_dir() or vuln_dir.is_symlink(): + continue + for report in sorted(vuln_dir.glob("*.md")): + if report.is_file() and not report.is_symlink(): + results.append(report) + return results + + +def parse_run_dir(run_dir: Path, source_repo: str) -> list[Finding]: + """Parse every vulnerability report under ``run_dir`` into findings. + + Findings are deduplicated by hash within the run so repeated model reports of + the same vulnerability collapse to a single record. + """ + findings: dict[str, Finding] = {} + for report in iter_vulnerability_files(run_dir): + try: + text = report.read_text(encoding="utf-8", errors="replace") + except OSError: + continue + finding = parse_finding_markdown(text, source_repo, source_file=report.name) + if finding is None: + continue + existing = findings.get(finding.finding_hash) + # Prefer the higher-severity variant when duplicates disagree. + if existing is None or severity_rank(finding.severity) > severity_rank(existing.severity): + findings[finding.finding_hash] = finding + return list(findings.values()) + + +def _source_links(context: "EmitContext") -> list[str]: + """Return Markdown links back to the scanned source repo/PR/commit.""" + links: list[str] = [f"- Source repository: `{context.source_repo}`"] + base = f"https://github.com/{context.source_repo}" + if context.pr_number: + links.append(f"- Pull request: {base}/pull/{context.pr_number}") + if context.head_sha: + links.append(f"- Head commit: {base}/commit/{context.head_sha}") + if context.run_url: + links.append(f"- Strix run: {context.run_url}") + return links + + +def build_issue_title(finding: Finding) -> str: + """Return the issue title for a finding.""" + repo_short = finding.source_repo.split("/")[-1] + location = finding.code_location or "no-location" + severity = finding.severity or "UNKNOWN" + return f"[strix] {repo_short} {severity}: {finding.title} ({location})" + + +def build_issue_labels(finding: Finding) -> list[str]: + """Return the label set for a finding's issue.""" + repo_short = finding.source_repo.split("/")[-1] + severity = (finding.severity or "unknown").lower() + return [ + "strix", + "security", + f"repo:{repo_short}", + f"severity:{severity}", + f"strix-finding:{finding.short_hash}", + ] + + +def build_issue_body(finding: Finding, context: "EmitContext") -> str: + """Return the full issue body Markdown, including hidden reconciliation markers.""" + location = finding.code_location or "(no code location reported)" + parts: list[str] = [ + f"## {finding.title}", + "", + f"- Severity: **{finding.severity or 'UNKNOWN'}**", + f"- Code location: `{location}`", + ] + if finding.cvss: + parts.append(f"- CVSS score: {finding.cvss}") + if finding.cvss_vector: + parts.append(f"- CVSS vector: `{finding.cvss_vector}`") + if finding.target: + parts.append(f"- Target: `{finding.target}`") + if finding.endpoint: + parts.append(f"- Endpoint: `{finding.endpoint}`") + if finding.method: + parts.append(f"- Method: `{finding.method}`") + if finding.model: + parts.append(f"- Detected by model: `{finding.model}`") + parts.append("") + parts.extend(_source_links(context)) + if finding.description: + parts += ["", "### Description", "", finding.description] + if finding.impact: + parts += ["", "### Impact", "", finding.impact] + if finding.remediation: + parts += ["", "### Remediation", "", finding.remediation] + parts += [ + "", + "---", + "_Filed automatically by the source-side Strix issue emitter. " + "Do not edit the markers below; they drive deduplication and close-on-fix._", + "", + f"{FINDING_MARKER_PREFIX} {finding.finding_hash} -->", + f"{SEVERITY_MARKER_PREFIX} {finding.severity or 'UNKNOWN'} -->", + f"{LOCATION_MARKER_PREFIX} {finding.normalized_location} -->", + ] + return "\n".join(parts) + + +def marker_value(body: str, prefix: str) -> str: + """Return the value stored in a hidden ``<!-- prefix VALUE -->`` marker.""" + escaped = re.escape(prefix) + match = re.search(escaped + r"\s*(.+?)\s*-->", body or "") + return match.group(1).strip() if match else "" + + +def issue_finding_hash(issue: dict[str, Any]) -> str: + """Return the finding hash recorded on an existing issue (label or marker).""" + marker = marker_value(str(issue.get("body") or ""), FINDING_MARKER_PREFIX) + if marker: + return marker + for label in _issue_label_names(issue): + if label.startswith("strix-finding:"): + return label.split(":", 1)[1] + return "" + + +def _issue_label_names(issue: dict[str, Any]) -> list[str]: + """Return label names for an issue, tolerating string or object labels.""" + names: list[str] = [] + for label in issue.get("labels") or []: + if isinstance(label, dict): + name = label.get("name") + else: + name = label + if name: + names.append(str(name)) + return names + + +@dataclass +class EmitContext: + """Immutable context describing the scanned source and run.""" + + source_repo: str + issues_repo: str = DEFAULT_ISSUES_REPO + pr_number: str = "" + head_sha: str = "" + run_url: str = "" + scan_complete: bool = False + # Scan scope: SCOPE_FULL only for full-repo push/schedule runs. Defaults to + # SCOPE_PR (the safe value) so any unset/unknown scope disables close-on-fix. + scan_scope: str = SCOPE_PR + + @property + def close_on_fix_enabled(self) -> bool: + """Return whether close-on-fix may run for this scan. + + Both guards must hold: the scan finished cleanly (``scan_complete``) and + it covered the whole repository (``scan_scope == SCOPE_FULL``). A + PR-scoped or unknown-scope scan only inspects the PR's changed files, so + a finding's absence never proves it was fixed and must not close issues. + """ + return self.scan_complete and self.scan_scope == SCOPE_FULL + + +def plan_operations( + findings: Sequence[Finding], + existing_issues: Sequence[dict[str, Any]], + context: EmitContext, +) -> list[Operation]: + """Compute the create/update/comment/close plan for a scan. + + ``existing_issues`` are the issues already present in the tracker for this + source repo scope (any state). Pure function: performs no I/O so the + reconciliation logic — including the close-on-fix set difference and both + the incomplete-scan and PR-scope guards — is directly testable. Close-on-fix + runs only when ``context.close_on_fix_enabled`` (clean full-repo scan); a + PR-scoped scan creates/updates/reopens findings but never closes. + """ + by_hash: dict[str, dict[str, Any]] = {} + for issue in existing_issues: + found_hash = issue_finding_hash(issue) + if found_hash: + by_hash.setdefault(found_hash, issue) + + operations: list[Operation] = [] + current_hashes: set[str] = set() + + for finding in findings: + current_hashes.add(finding.finding_hash) + body = build_issue_body(finding, context) + labels = build_issue_labels(finding) + existing = by_hash.get(finding.finding_hash) + if existing is None: + operations.append( + Operation( + action="create", + finding_hash=finding.finding_hash, + short_hash=finding.short_hash, + title=build_issue_title(finding), + reason="new finding", + labels=labels, + body=body, + ) + ) + continue + + number = _issue_number(existing) + old_body = str(existing.get("body") or "") + old_severity = marker_value(old_body, SEVERITY_MARKER_PREFIX) + # The dedup key pins repo+title+location, so a hash match implies the same + # location; only severity (and refreshable body fields) can move here. A + # relocated finding forks a new hash -> a fresh create plus close-on-fix. + severity_changed = bool(old_severity) and old_severity.upper() != (finding.severity or "UNKNOWN").upper() + reopened = str(existing.get("state") or "").lower() == "closed" + + reason = "reopen (finding still present)" if reopened else "refresh finding" + operations.append( + Operation( + action="update", + finding_hash=finding.finding_hash, + short_hash=finding.short_hash, + title=build_issue_title(finding), + issue_number=number, + reason=reason, + labels=labels, + body=body, + ) + ) + if severity_changed: + change = f"severity {old_severity} -> {finding.severity or 'UNKNOWN'}" + operations.append( + Operation( + action="comment", + finding_hash=finding.finding_hash, + short_hash=finding.short_hash, + title=build_issue_title(finding), + issue_number=number, + reason=change, + comment=f"Strix finding changed: {change}.", + ) + ) + + if context.close_on_fix_enabled: + for issue in existing_issues: + if str(issue.get("state") or "").lower() != "open": + continue + found_hash = issue_finding_hash(issue) + if not found_hash or found_hash in current_hashes: + continue + resolved_ref = context.head_sha or "the latest scan" + operations.append( + Operation( + action="close", + finding_hash=found_hash, + short_hash=found_hash[:SHORT_HASH_LENGTH], + title=str(issue.get("title") or ""), + issue_number=_issue_number(issue), + reason="finding no longer present", + comment=f"Resolved on {resolved_ref}: this Strix finding is no longer " + f"reported for `{context.source_repo}`. Closing automatically.", + ) + ) + + return operations + + +def _issue_number(issue: dict[str, Any]) -> int | None: + """Return the integer issue number, or ``None`` when unparseable.""" + try: + return int(issue["number"]) + except (KeyError, TypeError, ValueError): + return None + + +class GitHubIssueClient: + """Thin ``gh api`` wrapper for issue reads/writes in the tracker repo.""" + + def __init__(self, repo: str, token: str) -> None: + """Store the target repo and the GitHub App token used for ``gh``.""" + self.repo = repo + self._token = token + + def _run(self, args: Sequence[str], *, stdin: str | None = None) -> str: + """Invoke ``gh`` with the App token in the environment and return stdout.""" + env = dict(os.environ) + env["GH_TOKEN"] = self._token + result = subprocess.run( # noqa: S603 - fixed gh argv, no shell + ["gh", *args], + input=stdin, + capture_output=True, + text=True, + env=env, + check=False, + ) + if result.returncode != 0: + raise RuntimeError(_scrub(result.stderr.strip() or "gh command failed")) + return result.stdout + + def list_scope_issues(self, repo_short: str) -> list[dict[str, Any]]: + """Return every ``strix``+``repo:<name>`` issue (any state) in the tracker.""" + raw = self._run( + [ + "api", + "--paginate", + "--slurp", + "-X", + "GET", + f"repos/{self.repo}/issues", + "-f", + "state=all", + "-f", + f"labels=strix,repo:{repo_short}", + "-f", + "per_page=100", + ] + ) + pages = json.loads(raw or "[]") + return [issue for page in pages for issue in page if "pull_request" not in issue] + + def create_issue(self, title: str, body: str, labels: Sequence[str]) -> None: + """Create a new issue with the given title/body/labels.""" + payload = json.dumps({"title": title, "body": body, "labels": list(labels)}) + self._run(["api", "-X", "POST", f"repos/{self.repo}/issues", "--input", "-"], stdin=payload) + + def update_issue(self, number: int, body: str, labels: Sequence[str]) -> None: + """Refresh an issue body/labels and ensure it is open.""" + payload = json.dumps({"body": body, "labels": list(labels), "state": "open"}) + self._run(["api", "-X", "PATCH", f"repos/{self.repo}/issues/{number}", "--input", "-"], stdin=payload) + + def comment_issue(self, number: int, comment: str) -> None: + """Post a comment on an issue.""" + payload = json.dumps({"body": comment}) + self._run( + ["api", "-X", "POST", f"repos/{self.repo}/issues/{number}/comments", "--input", "-"], + stdin=payload, + ) + + def close_issue(self, number: int, comment: str) -> None: + """Comment on and close an issue.""" + self.comment_issue(number, comment) + payload = json.dumps({"state": "closed", "state_reason": "completed"}) + self._run(["api", "-X", "PATCH", f"repos/{self.repo}/issues/{number}", "--input", "-"], stdin=payload) + + +_TOKEN_RE = re.compile(r"gh[opsu]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}") + + +def _scrub(text: str) -> str: + """Redact anything resembling a GitHub token from a message.""" + return _TOKEN_RE.sub("***", text or "") + + +def execute_plan( + operations: Iterable[Operation], + client: GitHubIssueClient | None, + *, + dry_run: bool, + log: Any = None, +) -> dict[str, int]: + """Apply (or, in dry-run, log) the planned operations. + + Returns a per-action count summary. Individual operation failures are logged + and swallowed so best-effort emission never fails the Strix job. + """ + emit = log or print + counts: dict[str, int] = {"create": 0, "update": 0, "comment": 0, "close": 0, "error": 0} + for op in operations: + if dry_run or client is None: + emit(f"DRY-RUN: would {op.describe()}") + counts[op.action] = counts.get(op.action, 0) + 1 + continue + try: + if op.action == "create": + client.create_issue(op.title, op.body, op.labels) + elif op.action == "update": + client.update_issue(int(op.issue_number), op.body, op.labels) + elif op.action == "comment": + client.comment_issue(int(op.issue_number), op.comment) + elif op.action == "close": + client.close_issue(int(op.issue_number), op.comment) + counts[op.action] = counts.get(op.action, 0) + 1 + emit(f"OK: {op.describe()}") + except Exception as exc: # noqa: BLE001 - best-effort emission + counts["error"] += 1 + emit(f"::warning::Strix issue emit failed for {op.describe()}: {_scrub(str(exc))}") + return counts + + +def build_arg_parser() -> argparse.ArgumentParser: + """Return the command-line argument parser.""" + parser = argparse.ArgumentParser(description="Emit per-finding Strix issues into appguardrail.") + parser.add_argument("--run-dir", required=True, help="Directory containing strix_runs vulnerability reports.") + parser.add_argument("--source-repo", required=True, help="Scanned repository in owner/name form.") + parser.add_argument("--issues-repo", default=DEFAULT_ISSUES_REPO, help="Tracker repository in owner/name form.") + parser.add_argument("--pr-number", default="", help="Pull request number, if the scan was PR-scoped.") + parser.add_argument("--head-sha", default="", help="Scanned head commit SHA.") + parser.add_argument("--run-url", default="", help="URL of the Strix workflow run.") + parser.add_argument( + "--scan-complete", + action="store_true", + help="Set only when the scan finished cleanly; required to enable close-on-fix.", + ) + parser.add_argument( + "--scope", + choices=(SCOPE_FULL, SCOPE_PR), + default=SCOPE_PR, + help=( + "Scan scope: 'full' for a whole-repo push/schedule scan (enables " + "close-on-fix), 'pr' for a PR-scoped scan of changed files only " + "(never closes issues). Defaults to the safe 'pr' value." + ), + ) + parser.add_argument("--token-env", default=DEFAULT_TOKEN_ENV, help="Env var holding the GitHub App token.") + parser.add_argument("--dry-run", action="store_true", help="Plan and log operations without mutating GitHub.") + return parser + + +def run(argv: Sequence[str] | None = None) -> int: + """CLI entry point. Parses findings, plans operations, and applies them.""" + args = build_arg_parser().parse_args(argv) + context = EmitContext( + source_repo=args.source_repo, + issues_repo=args.issues_repo, + pr_number=str(args.pr_number or ""), + head_sha=str(args.head_sha or ""), + run_url=str(args.run_url or ""), + scan_complete=bool(args.scan_complete), + scan_scope=args.scope, + ) + + findings = parse_run_dir(Path(args.run_dir), context.source_repo) + print(f"Parsed {len(findings)} distinct Strix finding(s) from {args.run_dir}.") + if not context.scan_complete: + print("Scan not marked complete: close-on-fix is disabled for this run.") + elif context.scan_scope != SCOPE_FULL: + print( + f"Scan scope is '{context.scan_scope}' (PR-scoped): close-on-fix is disabled; " + "absent findings are treated as out-of-scope, not fixed." + ) + + token = os.environ.get(args.token_env, "").strip() + dry_run = bool(args.dry_run) or not token + if dry_run and not args.dry_run: + print( + f"::notice::{args.token_env} is not set; running Strix issue emitter in DRY-RUN " + "(no issues will be created). Provision the GitHub App to enable emission." + ) + + repo_short = context.source_repo.split("/")[-1] + client: GitHubIssueClient | None = None + existing_issues: list[dict[str, Any]] = [] + if not dry_run: + client = GitHubIssueClient(context.issues_repo, token) + try: + existing_issues = client.list_scope_issues(repo_short) + except Exception as exc: # noqa: BLE001 - degrade to dry-run on read failure + print(f"::warning::Could not read existing appguardrail issues: {_scrub(str(exc))}; planning in DRY-RUN.") + dry_run = True + client = None + + if not findings and not (context.close_on_fix_enabled and existing_issues): + print("No findings and nothing to reconcile; exiting.") + return 0 + + operations = plan_operations(findings, existing_issues, context) + counts = execute_plan(operations, client, dry_run=dry_run) + print( + "Strix issue emit summary: " + + ", ".join(f"{key}={counts.get(key, 0)}" for key in ("create", "update", "comment", "close", "error")) + + (" (dry-run)" if dry_run else "") + ) + return 0 + + +if __name__ == "__main__": # pragma: no cover - CLI shim + sys.exit(run()) diff --git a/tests/test_noema_review_gate.py b/tests/test_noema_review_gate.py index cc68ff28..8285fceb 100644 --- a/tests/test_noema_review_gate.py +++ b/tests/test_noema_review_gate.py @@ -206,7 +206,7 @@ def test_call_llm_handles_configuration_and_verdicts(monkeypatch): monkeypatch.setenv("NOEMA_LLM_API_URL", "file:///etc/passwd") monkeypatch.setenv("NOEMA_LLM_API_KEY", "secret") - with pytest.raises(ValueError, match="must start with http:// or https://"): + with pytest.raises(ValueError, match="URL scheme must be http or https"): noema.call_llm("owner/repo", 1, pr, "diff", False) monkeypatch.setenv("NOEMA_LLM_API_URL", "https://llm.example.test/chat") diff --git a/tests/test_opencode_agent_contract.py b/tests/test_opencode_agent_contract.py index ece0a6bc..38a391d0 100644 --- a/tests/test_opencode_agent_contract.py +++ b/tests/test_opencode_agent_contract.py @@ -280,7 +280,7 @@ def test_workflow_provisions_sandbox_tool_and_reviewer_agent(): assert '"## Review outcome"' in workflow assert '"## Check outcome"' not in workflow assert "publish REQUEST_CHANGES when coverage-evidence blocker states" in workflow - assert re.search(r"opencode-review-target:[\s\S]{0,240}timeout-minutes: 360", workflow) + assert re.search(r"opencode-review-target:[\s\S]{0,400}timeout-minutes: 360", workflow) assert 'timeout-minutes: 75' in workflow assert re.search(r"Run OpenCode PR Review model pool[\s\S]{0,240}timeout-minutes: 350", workflow) assert 'APPROVAL_CHECK_WAIT_ATTEMPTS: "81"' in workflow @@ -412,7 +412,7 @@ def test_merge_scheduler_uses_escalating_mutation_credentials(): assert "steps.scheduler_app_token.outputs.token" in workflow assert "SCHEDULER_READ_TOKEN: ${{ github.token }}" in workflow assert "SCHEDULER_MUTATION_TOKEN_SOURCE" in workflow - assert 'default: "-1"' in workflow + assert 'default: "1"' in workflow assert 'review_dispatch_limit="-1"' in workflow diff --git a/tests/test_strix_emit_appguardrail_issues.py b/tests/test_strix_emit_appguardrail_issues.py new file mode 100644 index 00000000..43b548ef --- /dev/null +++ b/tests/test_strix_emit_appguardrail_issues.py @@ -0,0 +1,729 @@ +"""Unit tests for the source-side Strix -> appguardrail issue emitter.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from scripts.ci import strix_emit_appguardrail_issues as emit + +SOURCE_REPO = "ContextualWisdomLab/example-service" + +SQLI_REPORT = """\ +# Vulnerability Report + +Model: github_models/openai/gpt-5 +Title: SQL Injection in login handler +Severity: HIGH +CVSS Score: 8.1 +CVSS Vector: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H +Target: backend/app/auth.py +Endpoint: /api/login +Method: POST + +Description: +User input is concatenated directly into a SQL query. + +Impact: +An attacker can read or modify arbitrary rows. + +Code Locations: +backend/app/auth.py:42-45 + +Remediation: +Use parameterized queries. +""" + +# Bold-markdown variant with a /workspace-prefixed location. +XSS_REPORT = """\ +## Reflected XSS in search page + +- **Severity:** MEDIUM +- **CVSS Score:** 5.4 +- **Target:** frontend/src/search.tsx +- **Endpoint:** /search +- **Method:** GET + +**Description:** Query parameter is rendered without escaping. + +**Code Locations** +/workspace/example-service/frontend/src/search.tsx:88 + +**Remediation:** Escape user-controlled output. +""" + +# Critical finding with no code location at all. +NO_LOCATION_REPORT = """\ +Title: Missing Content Security Policy +Severity: CRITICAL +Endpoint: all frontend pages +Description: No CSP header is set on any response. +Remediation: Add a restrictive Content-Security-Policy header. +""" + +NOT_A_FINDING = """\ +# Scan Notes + +Some prose with no Title field. +""" + + +def write_run(tmp_path: Path, reports: dict[str, str], run_name: str = "run-1") -> Path: + """Create a strix_runs/<run>/vulnerabilities tree and return the run dir.""" + run_dir = tmp_path / "strix_runs" / run_name + vuln_dir = run_dir / "vulnerabilities" + vuln_dir.mkdir(parents=True) + for name, text in reports.items(): + (vuln_dir / name).write_text(text, encoding="utf-8") + return tmp_path / "strix_runs" + + +def make_context(**overrides) -> emit.EmitContext: + """Build an EmitContext with sensible test defaults. + + Defaults to a full-repo scan (``scan_scope=SCOPE_FULL``) so close-on-fix + reconciliation is exercised; override ``scan_scope`` to model a PR-scoped run. + """ + values = { + "source_repo": SOURCE_REPO, + "pr_number": "42", + "head_sha": "a" * 40, + "run_url": "https://github.com/ContextualWisdomLab/.github/actions/runs/1", + "scan_complete": True, + "scan_scope": emit.SCOPE_FULL, + } + values.update(overrides) + return emit.EmitContext(**values) + + +class FakeClient: + """In-memory stand-in for GitHubIssueClient used to assert executed ops.""" + + def __init__(self, existing: list[dict] | None = None) -> None: + self.existing = existing or [] + self.created: list[dict] = [] + self.updated: list[dict] = [] + self.comments: list[dict] = [] + self.closed: list[int] = [] + + def list_scope_issues(self, repo_short): + return self.existing + + def create_issue(self, title, body, labels): + self.created.append({"title": title, "body": body, "labels": list(labels)}) + + def update_issue(self, number, body, labels): + self.updated.append({"number": number, "body": body, "labels": list(labels)}) + + def comment_issue(self, number, comment): + self.comments.append({"number": number, "comment": comment}) + + def close_issue(self, number, comment): + self.comments.append({"number": number, "comment": comment}) + self.closed.append(number) + + +# --------------------------------------------------------------------------- # +# Parsing +# --------------------------------------------------------------------------- # + + +def test_parses_all_fields_from_plain_report(): + """A plain-text report yields a fully populated finding.""" + finding = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO, "sqli.md") + assert finding is not None + assert finding.title == "SQL Injection in login handler" + assert finding.severity == "HIGH" + assert finding.cvss == "8.1" + assert finding.cvss_vector.startswith("CVSS:3.1") + assert finding.endpoint == "/api/login" + assert finding.method == "POST" + assert finding.model == "github_models/openai/gpt-5" + assert finding.code_location == "backend/app/auth.py:42-45" + assert "parameterized" in finding.remediation + assert "arbitrary rows" in finding.impact + + +def test_parses_bold_markdown_and_normalizes_workspace_location(): + """Bold-markdown fields parse and /workspace prefixes are stripped for dedup.""" + finding = emit.parse_finding_markdown(XSS_REPORT, SOURCE_REPO, "xss.md") + assert finding is not None + assert finding.title == "Reflected XSS in search page" + assert finding.severity == "MEDIUM" + assert finding.code_location == "/workspace/example-service/frontend/src/search.tsx:88" + assert finding.normalized_location == "frontend/src/search.tsx:88" + + +def test_no_location_finding_parses_with_empty_location(): + """A finding without any code location keeps an empty location but still parses.""" + finding = emit.parse_finding_markdown(NO_LOCATION_REPORT, SOURCE_REPO, "csp.md") + assert finding is not None + assert finding.severity == "CRITICAL" + assert finding.code_location == "" + assert finding.normalized_location == "" + + +def test_non_finding_report_returns_none(): + """Reports without a Title are not findings.""" + assert emit.parse_finding_markdown(NOT_A_FINDING, SOURCE_REPO) is None + + +def test_parse_run_dir_dedups_duplicate_model_reports(tmp_path): + """Duplicate reports of the same vulnerability collapse to one finding.""" + runs = write_run( + tmp_path, + { + "a.md": SQLI_REPORT, + "b.md": SQLI_REPORT, # duplicate finding from another model + "c.md": XSS_REPORT, + "d.md": NO_LOCATION_REPORT, + "e.md": NOT_A_FINDING, + }, + ) + findings = emit.parse_run_dir(runs, SOURCE_REPO) + assert len(findings) == 3 + titles = {f.title for f in findings} + assert "SQL Injection in login handler" in titles + + +# --------------------------------------------------------------------------- # +# Hashing +# --------------------------------------------------------------------------- # + + +def test_dedup_hash_is_stable_and_whitespace_insensitive(): + """Cosmetic title whitespace does not change the dedup hash; location does.""" + base = emit.finding_dedup_hash(SOURCE_REPO, "SQL Injection", "backend/app/auth.py:42-45") + wrapped = emit.finding_dedup_hash(SOURCE_REPO, "SQL Injection\n", "backend/app/auth.py:42-45") + workspace = emit.finding_dedup_hash( + SOURCE_REPO, "SQL Injection", "/workspace/example-service/backend/app/auth.py:42-45" + ) + other = emit.finding_dedup_hash(SOURCE_REPO, "SQL Injection", "backend/app/auth.py:99") + assert base == wrapped == workspace + assert base != other + assert len(base) == 64 + + +def test_short_hash_length(): + """Short hash is the configured prefix length.""" + finding = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + assert len(finding.short_hash) == emit.SHORT_HASH_LENGTH + assert finding.finding_hash.startswith(finding.short_hash) + + +# --------------------------------------------------------------------------- # +# Issue content +# --------------------------------------------------------------------------- # + + +def test_issue_title_and_labels(): + """Issue title and labels follow the documented format.""" + finding = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + title = emit.build_issue_title(finding) + assert title == "[strix] example-service HIGH: SQL Injection in login handler (backend/app/auth.py:42-45)" + labels = emit.build_issue_labels(finding) + assert "strix" in labels + assert "security" in labels + assert "repo:example-service" in labels + assert "severity:high" in labels + assert f"strix-finding:{finding.short_hash}" in labels + + +def test_issue_body_carries_markers(): + """Issue body embeds the finding/severity/location reconciliation markers.""" + finding = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + body = emit.build_issue_body(finding, make_context()) + assert emit.marker_value(body, emit.FINDING_MARKER_PREFIX) == finding.finding_hash + assert emit.marker_value(body, emit.SEVERITY_MARKER_PREFIX) == "HIGH" + assert emit.marker_value(body, emit.LOCATION_MARKER_PREFIX) == "backend/app/auth.py:42-45" + assert "pull/42" in body + assert ("commit/" + "a" * 40) in body + + +def test_issue_finding_hash_prefers_marker_then_label(): + """Existing-issue hash extraction reads the marker, falling back to the label.""" + finding = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + body = emit.build_issue_body(finding, make_context()) + assert emit.issue_finding_hash({"body": body}) == finding.finding_hash + label_only = {"body": "", "labels": [{"name": f"strix-finding:{finding.short_hash}"}]} + assert emit.issue_finding_hash(label_only) == finding.short_hash + assert emit.issue_finding_hash({"body": "", "labels": ["unrelated"]}) == "" + + +# --------------------------------------------------------------------------- # +# Planning +# --------------------------------------------------------------------------- # + + +def existing_issue_for(report: str, *, number: int, state: str = "open", context=None): + """Build an existing-issue dict mirroring what the tracker would return.""" + finding = emit.parse_finding_markdown(report, SOURCE_REPO) + body = emit.build_issue_body(finding, context or make_context()) + return { + "number": number, + "state": state, + "title": emit.build_issue_title(finding), + "body": body, + "labels": [{"name": name} for name in emit.build_issue_labels(finding)], + } + + +def test_plan_creates_issue_for_new_finding(): + """A finding with no existing issue plans a create.""" + findings = [emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO)] + ops = emit.plan_operations(findings, [], make_context(scan_complete=False)) + assert len(ops) == 1 + assert ops[0].action == "create" + assert "strix" in ops[0].labels + + +def test_plan_updates_without_comment_when_unchanged(): + """An unchanged finding refreshes the body but does not comment.""" + finding = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + existing = existing_issue_for(SQLI_REPORT, number=7) + ops = emit.plan_operations([finding], [existing], make_context()) + actions = [op.action for op in ops] + assert actions == ["update"] + assert ops[0].issue_number == 7 + + +def test_plan_comments_when_severity_changes(): + """A severity change on an existing finding plans an update plus a comment.""" + # Existing issue recorded HIGH; new run reports the same finding as CRITICAL. + existing = existing_issue_for(SQLI_REPORT, number=7) + escalated = SQLI_REPORT.replace("Severity: HIGH", "Severity: CRITICAL") + finding = emit.parse_finding_markdown(escalated, SOURCE_REPO) + ops = emit.plan_operations([finding], [existing], make_context()) + actions = [op.action for op in ops] + assert actions == ["update", "comment"] + assert "severity" in ops[1].comment.lower() + + +def test_plan_reopens_closed_issue_when_finding_returns(): + """A closed issue whose finding reappears is updated back to open.""" + finding = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + existing = existing_issue_for(SQLI_REPORT, number=7, state="closed") + ops = emit.plan_operations([finding], [existing], make_context()) + assert ops[0].action == "update" + assert "reopen" in ops[0].reason + + +def test_close_on_fix_closes_missing_open_issue_only_for_full_scope(): + """An open issue absent from a complete FULL-repo scan is closed. + + Close-on-fix is safe only for a whole-repo scan, which sees every finding; + this test pins that a clean full scan closes the stale issue. + """ + current = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + stale = existing_issue_for(XSS_REPORT, number=9) # not in current run + live = existing_issue_for(SQLI_REPORT, number=7) + context = make_context(scan_complete=True, scan_scope=emit.SCOPE_FULL) + ops = emit.plan_operations([current], [live, stale], context) + close_ops = [op for op in ops if op.action == "close"] + assert len(close_ops) == 1 + assert close_ops[0].issue_number == 9 + assert ("a" * 40) in close_ops[0].comment + + +def test_pr_scoped_complete_scan_closes_nothing(): + """The scope guard: a completed PR-scoped scan never closes issues. + + A PR-scoped scan only inspects the PR's changed files, so a finding's + absence means "outside this PR", not "fixed". Neither a zero-finding clean + PR nor a subset scan may close still-valid open issues in untouched files. + """ + stale = existing_issue_for(XSS_REPORT, number=9) + live = existing_issue_for(SQLI_REPORT, number=7) + # Zero findings (a clean PR): must not close every open Strix issue. + zero_ops = emit.plan_operations( + [], [live, stale], make_context(scan_complete=True, scan_scope=emit.SCOPE_PR) + ) + assert all(op.action != "close" for op in zero_ops) + # A subset scan that only re-reports one finding must not close the other. + current = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + subset_ops = emit.plan_operations( + [current], [live, stale], make_context(scan_complete=True, scan_scope=emit.SCOPE_PR) + ) + assert all(op.action != "close" for op in subset_ops) + + +def test_unknown_scope_complete_scan_closes_nothing(): + """An unknown/unset scope is treated as unsafe: close-on-fix stays off.""" + current = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + stale = existing_issue_for(XSS_REPORT, number=9) + context = make_context(scan_complete=True, scan_scope="unknown") + assert context.close_on_fix_enabled is False + ops = emit.plan_operations([current], [stale], context) + assert all(op.action != "close" for op in ops) + + +def test_incomplete_scan_never_closes(): + """The close-on-fix guard: an incomplete scan closes nothing even at full scope.""" + current = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + stale = existing_issue_for(XSS_REPORT, number=9) + context = make_context(scan_complete=False, scan_scope=emit.SCOPE_FULL) + assert context.close_on_fix_enabled is False + ops = emit.plan_operations([current], [stale], context) + assert all(op.action != "close" for op in ops) + + +def test_close_on_fix_ignores_already_closed_issues(): + """Already-closed stale issues are not re-closed.""" + current = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + stale_closed = existing_issue_for(XSS_REPORT, number=9, state="closed") + ops = emit.plan_operations([current], [stale_closed], make_context(scan_complete=True)) + assert all(op.action != "close" for op in ops) + + +# --------------------------------------------------------------------------- # +# Execution +# --------------------------------------------------------------------------- # + + +def test_execute_plan_dry_run_logs_without_mutation(): + """Dry-run logs each op and counts it, calling no client methods.""" + findings = [emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO)] + ops = emit.plan_operations(findings, [], make_context(scan_complete=False)) + messages: list[str] = [] + counts = emit.execute_plan(ops, None, dry_run=True, log=messages.append) + assert counts["create"] == 1 + assert any("DRY-RUN" in m and "CREATE" in m for m in messages) + + +def test_execute_plan_applies_operations_via_client(): + """A live plan drives create/comment/close through the client.""" + context = make_context(scan_complete=True) + current = emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO) + stale = existing_issue_for(XSS_REPORT, number=9, context=context) + ops = emit.plan_operations([current], [stale], context) + client = FakeClient() + messages: list[str] = [] + counts = emit.execute_plan(ops, client, dry_run=False, log=messages.append) + assert counts["create"] == 1 + assert counts["close"] == 1 + assert len(client.created) == 1 + assert client.closed == [9] + + +def test_execute_plan_applies_update_and_comment(): + """A severity escalation drives update + comment through the client.""" + existing = existing_issue_for(SQLI_REPORT, number=7) + escalated = SQLI_REPORT.replace("Severity: HIGH", "Severity: CRITICAL") + finding = emit.parse_finding_markdown(escalated, SOURCE_REPO) + ops = emit.plan_operations([finding], [existing], make_context(scan_complete=False)) + client = FakeClient() + counts = emit.execute_plan(ops, client, dry_run=False, log=lambda *_: None) + assert counts["update"] == 1 + assert counts["comment"] == 1 + assert client.updated[0]["number"] == 7 + assert client.comments[0]["number"] == 7 + + +def test_execute_plan_swallows_client_errors(): + """A failing operation is logged as a warning and counted, not raised.""" + + class BoomClient(FakeClient): + def create_issue(self, title, body, labels): + raise RuntimeError("token gho_deadbeefdeadbeefdeadbeef expired") + + findings = [emit.parse_finding_markdown(SQLI_REPORT, SOURCE_REPO)] + ops = emit.plan_operations(findings, [], make_context(scan_complete=False)) + messages: list[str] = [] + counts = emit.execute_plan(ops, BoomClient(), dry_run=False, log=messages.append) + assert counts["error"] == 1 + # Token must be scrubbed from the warning. + assert all("gho_deadbeef" not in m for m in messages) + + +# --------------------------------------------------------------------------- # +# CLI +# --------------------------------------------------------------------------- # + + +def test_run_dry_run_without_token(tmp_path, monkeypatch, capsys): + """Without a token the CLI plans in dry-run and exits 0 without mutating.""" + runs = write_run(tmp_path, {"a.md": SQLI_REPORT, "b.md": XSS_REPORT}) + monkeypatch.delenv(emit.DEFAULT_TOKEN_ENV, raising=False) + code = emit.run( + [ + "--run-dir", + str(runs), + "--source-repo", + SOURCE_REPO, + "--pr-number", + "42", + "--head-sha", + "a" * 40, + ] + ) + out = capsys.readouterr().out + assert code == 0 + assert "DRY-RUN" in out + assert "Parsed 2 distinct" in out + assert "close-on-fix is disabled" in out + + +def test_run_forced_dry_run_flag(tmp_path, monkeypatch, capsys): + """--dry-run forces planning even when a token is present.""" + runs = write_run(tmp_path, {"a.md": SQLI_REPORT}) + monkeypatch.setenv(emit.DEFAULT_TOKEN_ENV, "gho_" + "x" * 30) + code = emit.run( + ["--run-dir", str(runs), "--source-repo", SOURCE_REPO, "--scan-complete", "--dry-run"] + ) + out = capsys.readouterr().out + assert code == 0 + assert "DRY-RUN" in out + + +def test_iter_vulnerability_files_skips_symlinked_dir(tmp_path): + """Symlinked vulnerabilities directories are ignored.""" + runs = write_run(tmp_path, {"a.md": SQLI_REPORT}) + evil = tmp_path / "strix_runs" / "run-evil" + evil.mkdir() + (evil / "vulnerabilities").symlink_to(runs / "run-1" / "vulnerabilities") + files = emit.iter_vulnerability_files(runs) + # Only the real directory's report is returned. + assert len(files) == 1 + + +def test_severity_rank_orders_and_handles_unknown(): + """Severity ranking orders known levels and returns -1 for unknown.""" + assert emit.severity_rank("CRITICAL") > emit.severity_rank("LOW") + assert emit.severity_rank("bogus") == -1 + + +def test_scrub_redacts_tokens(): + """Token scrubbing removes GitHub token shapes.""" + assert "gho_" not in emit._scrub("leak gho_" + "a" * 30) + assert "github_pat_" not in emit._scrub("pat github_pat_" + "a" * 30) + + +# --------------------------------------------------------------------------- # +# Location edge cases +# --------------------------------------------------------------------------- # + + +PROSE_LOCATION_REPORT = """\ +Title: Insecure Deserialization +Severity: HIGH + +Description: +The bug is triggered at backend/loader.py:73 inside the request handler. +""" + +KEYWORD_LOCATION_REPORT = """\ +Title: Directory Traversal +Severity: HIGH +Affected file backend/files.py:12 handles the path unsafely. +""" + + +def test_location_falls_back_to_prose_reference(): + """A location mentioned only in prose is recovered by the end-of-text fallback.""" + finding = emit.parse_finding_markdown(PROSE_LOCATION_REPORT, SOURCE_REPO) + assert finding.code_location == "backend/loader.py:73" + + +def test_location_line_with_file_keyword_is_used(): + """A location line containing a file/path keyword is treated as a location.""" + finding = emit.parse_finding_markdown(KEYWORD_LOCATION_REPORT, SOURCE_REPO) + assert finding.code_location == "backend/files.py:12" + + +def test_iter_vulnerability_files_on_missing_dir(tmp_path): + """A non-existent run directory yields no report files.""" + assert emit.iter_vulnerability_files(tmp_path / "nope") == [] + + +def test_parse_run_dir_skips_unreadable_file(tmp_path, monkeypatch): + """An unreadable report file is skipped rather than aborting the run.""" + runs = write_run(tmp_path, {"a.md": SQLI_REPORT, "b.md": XSS_REPORT}) + original = Path.read_text + + def flaky_read_text(self, *args, **kwargs): + if self.name == "a.md": + raise OSError("permission denied") + return original(self, *args, **kwargs) + + monkeypatch.setattr(Path, "read_text", flaky_read_text) + findings = emit.parse_run_dir(runs, SOURCE_REPO) + assert len(findings) == 1 + assert findings[0].title == "Reflected XSS in search page" + + +def test_relocated_finding_is_new_identity_and_closes_old(): + """A moved finding forks a new hash: create the new issue, close the stale one.""" + existing = existing_issue_for(SQLI_REPORT, number=7) # old location + moved = SQLI_REPORT.replace("backend/app/auth.py:42-45", "backend/app/auth.py:200-210") + finding = emit.parse_finding_markdown(moved, SOURCE_REPO) + ops = emit.plan_operations([finding], [existing], make_context(scan_complete=True)) + actions = [op.action for op in ops] + assert "create" in actions + close_ops = [op for op in ops if op.action == "close"] + assert [op.issue_number for op in close_ops] == [7] + + +def test_issue_number_handles_bad_values(): + """Issue-number parsing tolerates missing/invalid numbers.""" + assert emit._issue_number({"number": "5"}) == 5 + assert emit._issue_number({}) is None + assert emit._issue_number({"number": "x"}) is None + + +# --------------------------------------------------------------------------- # +# GitHubIssueClient (gh subprocess wrapper) +# --------------------------------------------------------------------------- # + + +class FakeCompleted: + """Stand-in for subprocess.CompletedProcess.""" + + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +def test_client_list_scope_issues_filters_pull_requests(monkeypatch): + """list_scope_issues slurps pages and drops pull requests.""" + calls = {} + + def fake_run(args, input=None, capture_output=None, text=None, env=None, check=None): + calls["args"] = args + calls["token"] = env["GH_TOKEN"] + page = [{"number": 1, "title": "issue"}, {"number": 2, "pull_request": {}}] + return FakeCompleted(stdout=json.dumps([page])) + + monkeypatch.setattr(emit.subprocess, "run", fake_run) + client = emit.GitHubIssueClient("ContextualWisdomLab/appguardrail", "gho_" + "t" * 30) + issues = client.list_scope_issues("example-service") + assert [i["number"] for i in issues] == [1] + assert calls["token"].startswith("gho_") + assert "labels=strix,repo:example-service" in calls["args"] + + +def test_client_write_methods_invoke_gh(monkeypatch): + """create/update/comment/close send the expected gh payloads.""" + recorded = [] + + def fake_run(args, input=None, **kwargs): + recorded.append((args, input)) + return FakeCompleted(stdout="{}") + + monkeypatch.setattr(emit.subprocess, "run", fake_run) + client = emit.GitHubIssueClient("owner/repo", "gho_" + "t" * 30) + client.create_issue("t", "b", ["strix"]) + client.update_issue(3, "b2", ["strix"]) + client.comment_issue(3, "hi") + client.close_issue(4, "bye") + joined = " ".join(" ".join(a) for a, _ in recorded) + assert "repos/owner/repo/issues" in joined + assert "repos/owner/repo/issues/3" in joined + assert "PATCH" in joined + # close_issue comments then patches state closed. + assert any("closed" in (payload or "") for _, payload in recorded) + + +def test_client_raises_scrubbed_error_on_failure(monkeypatch): + """A non-zero gh exit raises a scrubbed RuntimeError.""" + + def fake_run(args, input=None, **kwargs): + return FakeCompleted(returncode=1, stderr="bad token gho_" + "z" * 30) + + monkeypatch.setattr(emit.subprocess, "run", fake_run) + client = emit.GitHubIssueClient("owner/repo", "gho_" + "t" * 30) + with pytest.raises(RuntimeError) as excinfo: + client.create_issue("t", "b", []) + assert "gho_" not in str(excinfo.value) + + +def test_client_error_defaults_message_when_stderr_empty(monkeypatch): + """A silent gh failure still raises a generic error message.""" + + def fake_run(args, input=None, **kwargs): + return FakeCompleted(returncode=1, stderr="") + + monkeypatch.setattr(emit.subprocess, "run", fake_run) + client = emit.GitHubIssueClient("owner/repo", "gho_" + "t" * 30) + with pytest.raises(RuntimeError, match="gh command failed"): + client.comment_issue(1, "x") + + +# --------------------------------------------------------------------------- # +# CLI live + degradation paths +# --------------------------------------------------------------------------- # + + +def test_run_live_applies_plan(tmp_path, monkeypatch, capsys): + """With a token and a working client the CLI creates issues live.""" + runs = write_run(tmp_path, {"a.md": SQLI_REPORT}) + monkeypatch.setenv(emit.DEFAULT_TOKEN_ENV, "gho_" + "t" * 30) + client = FakeClient(existing=[]) + monkeypatch.setattr(emit, "GitHubIssueClient", lambda repo, token: client) + code = emit.run(["--run-dir", str(runs), "--source-repo", SOURCE_REPO, "--scan-complete"]) + out = capsys.readouterr().out + assert code == 0 + assert len(client.created) == 1 + assert "create=1" in out + assert "dry-run" not in out + + +def test_run_full_scope_closes_stale_issue(tmp_path, monkeypatch, capsys): + """A full-repo scan (--scope full) reconciles and closes stale issues via the CLI.""" + runs = write_run(tmp_path, {"a.md": SQLI_REPORT}) + monkeypatch.setenv(emit.DEFAULT_TOKEN_ENV, "gho_" + "t" * 30) + stale = existing_issue_for(XSS_REPORT, number=9) + client = FakeClient(existing=[stale]) + monkeypatch.setattr(emit, "GitHubIssueClient", lambda repo, token: client) + code = emit.run( + ["--run-dir", str(runs), "--source-repo", SOURCE_REPO, "--scan-complete", "--scope", "full"] + ) + out = capsys.readouterr().out + assert code == 0 + assert client.closed == [9] + assert "close=1" in out + + +def test_run_pr_scope_never_closes_stale_issue(tmp_path, monkeypatch, capsys): + """A completed PR-scoped scan (default --scope pr) closes nothing via the CLI.""" + runs = write_run(tmp_path, {"a.md": SQLI_REPORT}) + monkeypatch.setenv(emit.DEFAULT_TOKEN_ENV, "gho_" + "t" * 30) + stale = existing_issue_for(XSS_REPORT, number=9) + client = FakeClient(existing=[stale]) + monkeypatch.setattr(emit, "GitHubIssueClient", lambda repo, token: client) + code = emit.run(["--run-dir", str(runs), "--source-repo", SOURCE_REPO, "--scan-complete"]) + out = capsys.readouterr().out + assert code == 0 + assert client.closed == [] + assert "close=0" in out + assert "PR-scoped" in out + + +def test_run_degrades_to_dry_run_when_read_fails(tmp_path, monkeypatch, capsys): + """A failure reading existing issues degrades to dry-run instead of failing.""" + runs = write_run(tmp_path, {"a.md": SQLI_REPORT}) + monkeypatch.setenv(emit.DEFAULT_TOKEN_ENV, "gho_" + "t" * 30) + + class BadReadClient(FakeClient): + def list_scope_issues(self, repo_short): + raise RuntimeError("gho_" + "z" * 30 + " unauthorized") + + monkeypatch.setattr(emit, "GitHubIssueClient", lambda repo, token: BadReadClient()) + code = emit.run(["--run-dir", str(runs), "--source-repo", SOURCE_REPO]) + out = capsys.readouterr().out + assert code == 0 + assert "DRY-RUN" in out + assert "gho_" not in out + + +def test_run_exits_early_when_nothing_to_do(tmp_path, monkeypatch, capsys): + """No findings and nothing to reconcile exits cleanly without planning.""" + empty = tmp_path / "strix_runs" / "run-1" / "vulnerabilities" + empty.mkdir(parents=True) + monkeypatch.delenv(emit.DEFAULT_TOKEN_ENV, raising=False) + code = emit.run(["--run-dir", str(tmp_path / "strix_runs"), "--source-repo", SOURCE_REPO]) + out = capsys.readouterr().out + assert code == 0 + assert "nothing to reconcile" in out