diff --git a/dsgai_scanner_tool/CHANGES_v0.3.md b/dsgai_scanner_tool/CHANGES_v0.3.md index dd8fe04..182cff8 100644 --- a/dsgai_scanner_tool/CHANGES_v0.3.md +++ b/dsgai_scanner_tool/CHANGES_v0.3.md @@ -9,6 +9,19 @@ dates are ISO-8601. The previous line is recorded in [`CHANGES_v0.2.md`](CHANGES ## [Unreleased] ### Added +- **CVE pipeline, suppressions, baseline, incremental scanning** (PR-12). + - CVE fetching moved into the CLI (`cli/dsgai_cve.py`, stdlib urllib): OSV + `querybatch` is the per-version source, NVD enriches CVSS by `cveId` only (no + `keywordSearch`). Cached at `~/.dsgai/cve-cache/` (24h TTL, `--refresh-cve`); + online and offline runs are byte-identical. **The LLM never transcribes CVE data.** + - Inline `# dsgai-ignore: P##.# reason="…"` suppressions — surfaced in a visible + `suppressed` section, never silently dropped; a reason is required. + - `baseline` subcommand + `--baseline` — gate only on findings not in the baseline. + - `--diff ` incremental scans (files changed vs a ref), labelled + "INCREMENTAL — not a full assessment". + - New `cve` subcommand; `--exclude`/`--diff`/`--baseline` wired through the skill and + the Action (which now fetches `dsgai_cve.py`). CVE enrichment reaches CI Job 1 with + no WebFetch. `langchain==0.1.0` yields real OSV advisories incl. EXPLOITABLE. - Contributor infrastructure: `[scanner]` GitHub issue-form templates (false-positive, false-negative, new-rule, bug), scanner `CONTRIBUTING.md`, public `ROADMAP.md`, and this changelog scaffold. (PR-01) diff --git a/dsgai_scanner_tool/cli/dsgai_cve.py b/dsgai_scanner_tool/cli/dsgai_cve.py new file mode 100644 index 0000000..c081aa2 --- /dev/null +++ b/dsgai_scanner_tool/cli/dsgai_cve.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +"""DSGAI CVE enrichment — deterministic, cached, stdlib-only. + +The CLI fetches CVE data so the LLM never transcribes it (hallucinated CVEs +become impossible by construction — the model renders what this fetched). + +Sources: + - OSV (https://osv.dev) — the only per-version source, via POST /v1/querybatch. + - NVD — used SOLELY to enrich a known CVE id with CVSS via ?cveId= (never + keywordSearch, which returns junk for names like "ai"/"instructor"). + +Cache: ~/.dsgai/cve-cache//@.json, 24h TTL. +`--refresh-cve` ignores the cache; `offline=True` uses cache only (no network). +""" +import json +import os +import re +import time +import urllib.error +import urllib.request + +OSV_BATCH = "https://api.osv.dev/v1/querybatch" +OSV_VULN = "https://api.osv.dev/v1/vulns/" +NVD_CVE = "https://services.nvd.nist.gov/rest/json/cves/2.0" +CACHE_TTL = 24 * 3600 +CACHE_DIR = os.path.join(os.path.expanduser("~"), ".dsgai", "cve-cache") + +# Minimal ecosystem detection from manifest filename → OSV ecosystem. +ECOSYSTEMS = { + "requirements.txt": "PyPI", "requirements": "PyPI", "pyproject.toml": "PyPI", + "package.json": "npm", "go.mod": "Go", "Cargo.toml": "crates.io", + "Gemfile.lock": "RubyGems", +} + +_REQ_RE = re.compile(r'^\s*([A-Za-z0-9_.\-]+)\s*==\s*([A-Za-z0-9_.\-]+)') + + +def _cache_path(eco, pkg, ver): + safe = re.sub(r'[^A-Za-z0-9_.\-]', '_', f"{pkg}@{ver}") + return os.path.join(CACHE_DIR, re.sub(r'[^A-Za-z0-9_.\-]', '_', eco), safe + ".json") + + +def _cache_get(eco, pkg, ver, refresh): + if refresh: + return None + p = _cache_path(eco, pkg, ver) + try: + if time.time() - os.path.getmtime(p) <= CACHE_TTL: + return json.load(open(p, encoding="utf-8")) + except (OSError, ValueError): + return None + return None + + +def _cache_put(eco, pkg, ver, data): + p = _cache_path(eco, pkg, ver) + os.makedirs(os.path.dirname(p), exist_ok=True) + with open(p, "w", encoding="utf-8", newline="\n") as fh: + json.dump(data, fh, sort_keys=True) + + +def parse_dependencies(discovered): + """Return [{ecosystem, package, version}] from pinned manifest entries. + + `discovered` is a list of (abs_path, rel_path). Only exact-pinned entries are + queried (OSV needs a concrete version). + """ + deps, seen = [], set() + for ap, rel in discovered: + base = os.path.basename(rel) + eco = ECOSYSTEMS.get(base) + if eco == "PyPI" and base.startswith("requirements"): + try: + for line in open(ap, encoding="utf-8", errors="replace"): + if line.lstrip().startswith("#"): + continue + m = _REQ_RE.match(line) + if m: + key = (eco, m.group(1).lower(), m.group(2)) + if key not in seen: + seen.add(key) + deps.append({"ecosystem": eco, "package": m.group(1), + "version": m.group(2)}) + except OSError: + continue + return deps + + +def _http_json(url, data=None, timeout=20): + headers = {"Content-Type": "application/json", "User-Agent": "dsgai-scan"} + req = urllib.request.Request( + url, data=json.dumps(data).encode() if data is not None else None, + headers=headers, method="POST" if data is not None else "GET") + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode()) + + +def _severity_of(detail): + """Extract a coarse severity + CVSS score from an OSV vuln detail.""" + score = None + for sev in detail.get("severity", []) or []: + if sev.get("type", "").startswith("CVSS"): + # OSV gives a vector; we keep the label from DB-specific fields below. + score = sev.get("score") + # database_specific severity label (GHSA gives HIGH/CRITICAL/…) + label = (detail.get("database_specific", {}) or {}).get("severity") + return label, score + + +def classify(label, cvss): + """EXPLOITABLE for high/critical (CVSS >= 7 or CRITICAL/HIGH label); + VULNERABLE for anything else with a severity; INFO if wholly unknown.""" + if (cvss is not None and cvss >= 7.0) or label in ("CRITICAL", "HIGH"): + return "EXPLOITABLE" + if (cvss is not None) or label in ("MODERATE", "MEDIUM", "LOW"): + return "VULNERABLE" + return "INFO" + + +def enrich_cvss_nvd(cve_id, offline, timeout=20): + """Fetch CVSS base score for a CVE id from NVD (cveId only). None on failure.""" + if offline or not cve_id.startswith("CVE-"): + return None + try: + data = _http_json(f"{NVD_CVE}?cveId={cve_id}", timeout=timeout) + for v in data.get("vulnerabilities", []): + metrics = v.get("cve", {}).get("metrics", {}) + for key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"): + if metrics.get(key): + return metrics[key][0]["cvssData"].get("baseScore") + except (urllib.error.URLError, ValueError, KeyError, TimeoutError): + return None + return None + + +def enrich(discovered, offline=False, refresh=False): + """Return a list of CVE finding dicts (deterministic order). + + Each: {package, version, ecosystem, id, aliases, summary, status, cvss}. + Cached per {ecosystem, package, version}; second run needs no network. + """ + deps = parse_dependencies(discovered) + results = [] + uncached = [] + cache_map = {} + for d in deps: + c = _cache_get(d["ecosystem"], d["package"], d["version"], refresh) + if c is not None: + cache_map[(d["ecosystem"], d["package"].lower(), d["version"])] = c + else: + uncached.append(d) + + fetched = {} + if uncached and not offline: + try: + queries = [{"package": {"name": d["package"], "ecosystem": d["ecosystem"]}, + "version": d["version"]} for d in uncached] + batch = _http_json(OSV_BATCH, {"queries": queries}) + for d, res in zip(uncached, batch.get("results", [])): + vulns = [] + for v in res.get("vulns", []) or []: + try: + detail = _http_json(OSV_VULN + v["id"]) + except (urllib.error.URLError, ValueError, TimeoutError): + detail = {"id": v["id"]} + aliases = sorted(detail.get("aliases", []) or []) + label, _ = _severity_of(detail) + # NVD enriches CVSS for known CVE ids; stored in cache so + # offline re-runs are byte-identical to the online run. + cvss = None + for aid in [detail.get("id", v["id"])] + aliases: + if aid.startswith("CVE-"): + cvss = enrich_cvss_nvd(aid, offline=False) + if cvss is not None: + break + vulns.append({ + "id": detail.get("id", v["id"]), + "aliases": aliases, + "summary": (detail.get("summary") or "")[:300], + "cvss": cvss, + "status": classify(label, cvss), + }) + data = {"vulns": vulns} + _cache_put(d["ecosystem"], d["package"], d["version"], data) + fetched[(d["ecosystem"], d["package"].lower(), d["version"])] = data + except (urllib.error.URLError, ValueError, TimeoutError): + pass # network failure → whatever is cached still renders + + merged = dict(cache_map) + merged.update(fetched) + for d in deps: + key = (d["ecosystem"], d["package"].lower(), d["version"]) + for v in merged.get(key, {}).get("vulns", []): + # cvss/status come from the cache (populated at fetch time) so an + # offline re-run is byte-identical to the online run that seeded it. + results.append({ + "package": d["package"], "version": d["version"], + "ecosystem": d["ecosystem"], "id": v["id"], + "aliases": v.get("aliases", []), "summary": v.get("summary", ""), + "status": v.get("status", "INFO"), "cvss": v.get("cvss"), + }) + results.sort(key=lambda x: (x["package"], x["version"], x["id"])) + return results diff --git a/dsgai_scanner_tool/cli/dsgai_scan.py b/dsgai_scanner_tool/cli/dsgai_scan.py index 3d07dce..ba01143 100644 --- a/dsgai_scanner_tool/cli/dsgai_scan.py +++ b/dsgai_scanner_tool/cli/dsgai_scan.py @@ -36,7 +36,7 @@ CHECKPOINT_REQUIRED = { "schema_version", "ruleset_version", "skill_version", "framework", "engine", "git_commit", "scanned_at", "scan_scope", "obfuscation", "controls", - "findings", "cves", "file_map_ref", + "findings", "suppressed", "cves", "file_map_ref", } SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", "env", "dist", "build", ".mypy_cache", ".pytest_cache", ".tox", ".idea"} @@ -349,7 +349,8 @@ def git_commit(target): return None -def build_checkpoint(ruleset, findings, controls, scope, obfuscation, target): +def build_checkpoint(ruleset, findings, controls, scope, obfuscation, target, + cves=None, suppressed=None, diff_scope=None): return { "schema_version": SCHEMA_VERSION, "ruleset_version": ruleset["ruleset_version"], @@ -358,11 +359,12 @@ def build_checkpoint(ruleset, findings, controls, scope, obfuscation, target): "engine": "deterministic-cli", "git_commit": git_commit(target), "scanned_at": now_iso(), - "scan_scope": scope, + "scan_scope": (f"diff:{diff_scope}" if diff_scope else scope), "obfuscation": obfuscation, "controls": controls, "findings": findings, - "cves": [], + "suppressed": suppressed or [], + "cves": cves or [], "file_map_ref": None, } @@ -373,7 +375,7 @@ def self_validate_checkpoint(cp): missing = CHECKPOINT_REQUIRED - set(cp) if missing: raise ValueError(f"checkpoint missing required fields: {sorted(missing)}") - for f in cp["findings"]: + for f in cp["findings"] + (cp.get("suppressed") or []): leaked = BANNED_FINDING_FIELDS & set(f) if leaked: raise ValueError(f"finding would leak match content via {sorted(leaked)}: " @@ -461,6 +463,75 @@ def print_table(controls, findings): # --------------------------------------------------------------------------- # # Subcommands # --------------------------------------------------------------------------- # +SUPPRESS_RE = __import__("re").compile( + r'dsgai-ignore:\s*(P\d{2}\.\d+)\s+reason=["\']([^"\']+)["\']') + + +def load_suppressions(files): + """Map path -> [(line, rule_id, reason)] from inline `# dsgai-ignore` comments. + + A finding is suppressed when a matching directive sits on its line or the + line immediately above. A reason string is required (the regex enforces it). + """ + out = {} + for ap, rel in files: + try: + text = open(ap, encoding="utf-8", errors="replace").read().splitlines() + except OSError: + continue + for i, line in enumerate(text, 1): + m = SUPPRESS_RE.search(line) + if m: + out.setdefault(rel, []).append((i, m.group(1), m.group(2))) + return out + + +def split_suppressed(findings, suppressions): + """Each `# dsgai-ignore` directive suppresses at most ONE finding: the one on + the same line (inline) if present, otherwise the one on the next line (comment + directly above the code). This avoids an inline comment also swallowing an + unrelated finding on the following line.""" + removed = set() + suppressed = [] + for path, directives in suppressions.items(): + for (cline, crule, reason) in directives: + for target in (cline, cline + 1): + match = next((f for f in findings + if id(f) not in removed and f["path"] == path + and f["rule_id"] == crule and f["line"] == target), None) + if match is not None: + removed.add(id(match)) + suppressed.append({**match, "suppressed_reason": reason}) + break + active = [f for f in findings if id(f) not in removed] + return active, suppressed + + +def finding_fp(f): + """Stable fingerprint for baseline comparison (line-independent-ish).""" + return f"{f['control']}|{f['rule_id']}|{f['path']}" + + +def load_baseline(path): + try: + data = json.load(open(path, encoding="utf-8")) + return set(data.get("fingerprints", [])) + except (OSError, ValueError): + return None + + +def diff_files(target, ref): + """Return the set of rel paths changed vs `ref` (git). None if unavailable.""" + try: + r = subprocess.run(["git", "-C", os.path.abspath(target), "diff", + "--name-only", ref], capture_output=True, text=True) + if r.returncode != 0: + return None + return {p.strip().replace("/", os.sep) for p in r.stdout.splitlines() if p.strip()} + except Exception: + return None + + def cmd_scan(args): rg = find_rg() if not rg: @@ -477,6 +548,14 @@ def cmd_scan(args): scan_root = os.path.abspath(args.target) scoped_target = os.path.join(scan_root, scope) if scope != "." else scan_root files = discover_files(scoped_target, excludes) + diff_scope = None + if getattr(args, "diff", None): + changed = diff_files(args.target, args.diff) + if changed is None: + eprint(f"error: could not compute git diff vs '{args.diff}'") + return 2 + files = [f for f in files if f[1].replace("/", os.sep) in changed] + diff_scope = args.diff detected = detect_stack(rg, files, scan_root) matches_by_rule = {} nearby_matches = {} @@ -496,9 +575,36 @@ def cmd_scan(args): eprint(f"error: {exc}") return 2 findings = resolve_findings(rules, matches_by_rule, detected, nearby_matches) + + # Suppressions: inline `# dsgai-ignore` comments move findings to a visible + # Suppressed section rather than silently dropping them. + suppressions = load_suppressions(files) + findings, suppressed = split_suppressed(findings, suppressions) + + # Baseline: mark previously-known findings so only NEW ones gate CI. + baseline = load_baseline(args.baseline) if getattr(args, "baseline", None) else None + if baseline is not None: + for f in findings: + if finding_fp(f) in baseline: + f["baselined"] = True + + # CVE enrichment (deterministic, cached). Off with --no-cve. + cves = [] + if not getattr(args, "no_cve", False): + try: + import dsgai_cve + cves = dsgai_cve.enrich(files, offline=getattr(args, "offline", False), + refresh=getattr(args, "refresh_cve", False)) + except ImportError: + eprint("note: dsgai_cve module not found alongside the CLI; skipping CVE stage") + except Exception as exc: # noqa: BLE001 — CVE is best-effort + eprint(f"note: CVE enrichment failed ({exc}); continuing") + controls = classify_controls(rules, findings, detected) checkpoint = build_checkpoint(ruleset, findings, controls, scope, - args.obfuscation, args.target) + args.obfuscation, args.target, + cves=cves, suppressed=suppressed, + diff_scope=diff_scope) try: self_validate_checkpoint(checkpoint) @@ -508,17 +614,26 @@ def cmd_scan(args): if args.format == "table": print_table(controls, findings) + if diff_scope: + print(f" scope: INCREMENTAL diff vs {diff_scope} — not a full assessment") + if suppressed: + print(f" suppressed: {len(suppressed)} finding(s) via dsgai-ignore") + if cves: + print(f" CVEs: {len(cves)} " + f"({sum(1 for c in cves if c['status'] == 'EXPLOITABLE')} exploitable)") if args.json_out: _write_json(args.json_out, checkpoint) if args.sarif: _write_json(args.sarif, build_sarif(ruleset, rules, findings)) - threshold = args.fail_on - fails = sum(1 for f in findings if f["status"] == "fail") - warns = sum(1 for f in findings if f["status"] == "warn") - if threshold == "fail" and fails: + # Only NEW (non-baselined) findings gate the build. + gating = [f for f in findings if not f.get("baselined")] + fails = sum(1 for f in gating if f["status"] == "fail") + warns = sum(1 for f in gating if f["status"] == "warn") + exploitable = sum(1 for c in cves if c["status"] == "EXPLOITABLE") + if args.fail_on == "fail" and (fails or exploitable): return 1 - if threshold == "warn" and (fails or warns): + if args.fail_on == "warn" and (fails or warns or exploitable): return 1 return 0 @@ -566,6 +681,52 @@ def cmd_doctor(args): return 0 if ok else 2 +def cmd_baseline(args): + rg = find_rg() + if not rg: + eprint("error: ripgrep (rg) not found.") + return 2 + ruleset = load_rules() + rules = ruleset["rules"] + scan_root = os.path.abspath(args.target) + scoped = os.path.join(scan_root, args.scope) if args.scope != "." else scan_root + files = discover_files(scoped, args.exclude or []) + detected = detect_stack(rg, files, scan_root) + matches_by_rule, nearby = {}, {} + for rule in rules: + cands = [f for f in files if glob_match(f[1], rule["file_globs"]) + and not glob_match(f[1], rule.get("exclude_globs") or [])] + matches_by_rule[rule["id"]] = run_rule(rg, rule, cands, scan_root) + rn = rule.get("requires_nearby") or {} + if rn.get("pattern"): + nearby[rule["id"]] = run_rule(rg, {"classification": "structural", + "pcre": rn["pattern"]}, cands, scan_root) + findings, _ = split_suppressed( + resolve_findings(rules, matches_by_rule, detected, nearby), + load_suppressions(files)) + fps = sorted({finding_fp(f) for f in findings if f["status"] in ("fail", "warn")}) + _write_json(args.out, {"ruleset_version": ruleset["ruleset_version"], + "created_at": now_iso(), "fingerprints": fps}) + print(f"wrote {args.out} ({len(fps)} baselined finding fingerprints)") + return 0 + + +def cmd_cve(args): + try: + import dsgai_cve + except ImportError: + eprint("error: dsgai_cve module not found alongside the CLI.") + return 2 + files = discover_files(os.path.abspath(args.target), args.exclude or []) + cves = dsgai_cve.enrich(files, offline=args.offline, refresh=args.refresh_cve) + for c in cves: + cvss = f" CVSS {c['cvss']}" if c.get("cvss") is not None else "" + print(f"[{c['status']}] {c['package']}=={c['version']} {c['id']}{cvss}") + print(f"\n{len(cves)} advisories " + f"({sum(1 for c in cves if c['status'] == 'EXPLOITABLE')} exploitable)") + return 1 if any(c["status"] == "EXPLOITABLE" for c in cves) else 0 + + def build_parser(): p = argparse.ArgumentParser(prog="dsgai_scan", description="DSGAI deterministic scanner") p.add_argument("--version", action="version", version=f"dsgai_scan {read_version()}") @@ -583,6 +744,14 @@ def build_parser(): const="internal", default="strict", help="internal mode (full paths in report)") sp.add_argument("--fail-on", choices=["fail", "warn"], default="fail") + sp.add_argument("--diff", default=None, metavar="REF", + help="incremental: scan only files changed vs REF") + sp.add_argument("--baseline", default=None, metavar="FILE", + help="gate only on findings NOT in this baseline") + sp.add_argument("--no-cve", action="store_true", help="skip CVE enrichment") + sp.add_argument("--refresh-cve", action="store_true", help="ignore the CVE cache") + sp.add_argument("--offline", action="store_true", + help="use the CVE cache only (no network)") sp.set_defaults(func=cmd_scan) dp = sub.add_parser("detect", help="check for GenAI signals (exit 0 if present)") @@ -592,6 +761,20 @@ def build_parser(): dop = sub.add_parser("doctor", help="check environment") dop.set_defaults(func=cmd_doctor) + + bp = sub.add_parser("baseline", help="write current findings to a baseline file") + bp.add_argument("target", nargs="?", default=".") + bp.add_argument("--scope", default=".") + bp.add_argument("--exclude", action="append", default=[]) + bp.add_argument("--out", default="dsgai-baseline.json") + bp.set_defaults(func=cmd_baseline) + + cp = sub.add_parser("cve", help="CVE enrichment only (from dependency manifests)") + cp.add_argument("target", nargs="?", default=".") + cp.add_argument("--exclude", action="append", default=[]) + cp.add_argument("--refresh-cve", action="store_true") + cp.add_argument("--offline", action="store_true") + cp.set_defaults(func=cmd_cve) return p diff --git a/dsgai_scanner_tool/dsgai_scanner_tool.md b/dsgai_scanner_tool/dsgai_scanner_tool.md index a526776..fcfbd27 100644 --- a/dsgai_scanner_tool/dsgai_scanner_tool.md +++ b/dsgai_scanner_tool/dsgai_scanner_tool.md @@ -1,6 +1,6 @@ --- description: OWASP DSGAI 2026 compliance scanner — audits GenAI/agentic codebases against all 21 data security risks, produces a self-contained HTML report designed to minimize disclosure -argument-hint: '[--internal] [--no-cve] [--scope ] [--exclude ]' +argument-hint: '[--internal] [--no-cve] [--scope ] [--exclude ] [--diff ] [--baseline ]' allowed-tools: Read, Grep, Glob, Bash, Write, Edit, WebFetch, WebSearch compatible_cli: '>=0.3,<0.5' --- @@ -583,7 +583,13 @@ Also identify: $([ -n "$EXCLUDE" ] && echo --exclude "$EXCLUDE") \ --json-out DSGAI-scan.json --sarif DSGAI-scan.sarif --format table ``` - Then consume `DSGAI-scan.json` (findings + per-control classification) as your evidence. Set the report header field **`engine: deterministic-cli`**. + Then consume `DSGAI-scan.json` (findings + per-control classification + CVEs + suppressed) as your evidence. Set the report header field **`engine: deterministic-cli`**. + + **New flags (all handled by the CLI):** + - `--diff ` — incremental scan of files changed vs ``. **Label the report "INCREMENTAL — not a full assessment"** (the checkpoint's `scan_scope` is `diff:`). + - `--baseline ` — gate only on findings not already in the baseline (write one with `dsgai_scan.py baseline --out dsgai-baseline.json`). Baselined findings carry `baselined: true`. + - `--exclude ` — exclude paths/globs (repeatable). + - **CVE enrichment now runs inside the CLI** (`cves` in the checkpoint, cached at `~/.dsgai/cve-cache/`). **You never transcribe CVE data — render exactly what the CLI fetched.** Inline `# dsgai-ignore: P##.# reason="…"` comments move findings to the checkpoint's `suppressed` list; render them in a visible **Suppressed** section, never silently. 4. **If unavailable** (user installed only the `.md`): fall back to the in-context grep flow in Step 2 below — but with the value-bearing protocol replacement (never content-mode grep on secrets). Set the report header field **`engine: llm-grep`** so consumers know the reproducibility class of the artifact. 5. **Checkpoint reuse (cache invalidation).** An existing `DSGAI-scan.json` may be reused **only if** its `git_commit` equals current `HEAD`, the working tree is clean (`git status --porcelain` empty), and its `ruleset_version` matches. Otherwise delete it and rescan. Never serve stale findings with a fresh date. (The CLI's `checkpoint_is_valid()` implements exactly this.) diff --git a/dsgai_scanner_tool/integrations/dsgai-scan.yml b/dsgai_scanner_tool/integrations/dsgai-scan.yml index fbacebc..aa33c86 100644 --- a/dsgai_scanner_tool/integrations/dsgai-scan.yml +++ b/dsgai_scanner_tool/integrations/dsgai-scan.yml @@ -74,6 +74,7 @@ jobs: mkdir -p dsgai-tool/cli dsgai-tool/rules dsgai-tool/schemas base="https://raw.githubusercontent.com/GenAI-Security-Project/GenAI-Data-Security-Initiative/${DSGAI_PIN}/dsgai_scanner_tool" curl -fsSL "$base/cli/dsgai_scan.py" -o dsgai-tool/cli/dsgai_scan.py + curl -fsSL "$base/cli/dsgai_cve.py" -o dsgai-tool/cli/dsgai_cve.py curl -fsSL "$base/rules/dsgai-rules.json" -o dsgai-tool/rules/dsgai-rules.json curl -fsSL "$base/VERSION" -o dsgai-tool/VERSION diff --git a/dsgai_scanner_tool/schemas/dsgai-scan.schema.json b/dsgai_scanner_tool/schemas/dsgai-scan.schema.json index 185e1b4..f2b7276 100644 --- a/dsgai_scanner_tool/schemas/dsgai-scan.schema.json +++ b/dsgai_scanner_tool/schemas/dsgai-scan.schema.json @@ -7,7 +7,7 @@ "required": [ "schema_version", "ruleset_version", "skill_version", "framework", "engine", "git_commit", "scanned_at", "scan_scope", "obfuscation", - "controls", "findings", "cves", "file_map_ref" + "controls", "findings", "suppressed", "cves", "file_map_ref" ], "additionalProperties": false, "properties": { @@ -28,7 +28,8 @@ } }, "findings": { "type": "array", "items": { "$ref": "#/$defs/finding" } }, - "cves": { "type": "array" }, + "suppressed": { "type": "array", "items": { "$ref": "#/$defs/suppressed" } }, + "cves": { "type": "array", "items": { "$ref": "#/$defs/cve" } }, "file_map_ref": { "type": ["string", "null"] } }, "$defs": { @@ -43,6 +44,7 @@ "line": { "type": "integer", "minimum": 1 }, "status": { "enum": ["fail", "warn", "pass_signal", "count", "info"] }, "classification": { "enum": ["structural", "value_bearing"] }, + "baselined": { "type": "boolean" }, "note": { "type": "string" }, "match_text": false, @@ -50,6 +52,40 @@ "value": false, "raw_grep_output": false } + }, + "suppressed": { + "type": "object", + "required": ["control", "rule_id", "path", "line", "status", "suppressed_reason"], + "additionalProperties": false, + "properties": { + "control": { "type": "string", "pattern": "^DSGAI[0-9]{2}$" }, + "rule_id": { "type": "string", "pattern": "^P[0-9]{2}\\.[0-9]+$" }, + "path": { "type": "string", "minLength": 1 }, + "line": { "type": "integer", "minimum": 1 }, + "status": { "enum": ["fail", "warn", "pass_signal", "count", "info"] }, + "classification": { "enum": ["structural", "value_bearing"] }, + "baselined": { "type": "boolean" }, + "suppressed_reason": { "type": "string", "minLength": 1 }, + "match_text": false, + "content": false, + "value": false, + "raw_grep_output": false + } + }, + "cve": { + "type": "object", + "required": ["package", "version", "ecosystem", "id", "status"], + "additionalProperties": false, + "properties": { + "package": { "type": "string" }, + "version": { "type": "string" }, + "ecosystem": { "type": "string" }, + "id": { "type": "string" }, + "aliases": { "type": "array", "items": { "type": "string" } }, + "summary": { "type": "string" }, + "status": { "enum": ["EXPLOITABLE", "VULNERABLE", "INFO"] }, + "cvss": { "type": ["number", "null"] } + } } } } diff --git a/dsgai_scanner_tool/tests/test_runner.py b/dsgai_scanner_tool/tests/test_runner.py index 9f10332..e19f519 100644 --- a/dsgai_scanner_tool/tests/test_runner.py +++ b/dsgai_scanner_tool/tests/test_runner.py @@ -206,6 +206,53 @@ def test_cli_self_guard_rejects_leaked_finding(): dsgai_scan.self_validate_checkpoint(cp) +def _import_cli(): + sys.path.insert(0, os.path.join(SCANNER, "cli")) + import dsgai_scan + return dsgai_scan + + +def test_suppression_suppresses_exactly_one(tmp_path): + ds = _import_cli() + f = tmp_path / "x.py" + f.write_text( + 'A = "sk-proj-FAKE00000000000000000000000000" # dsgai-ignore: P02.9 reason="known test value"\n' + 'B = "sk-proj-FAKE11111111111111111111111111"\n', encoding="utf-8") + files = [(str(f), "x.py")] + supp = ds.load_suppressions(files) + findings = [ + {"control": "DSGAI02", "rule_id": "P02.9", "path": "x.py", "line": 1, "status": "fail"}, + {"control": "DSGAI02", "rule_id": "P02.9", "path": "x.py", "line": 2, "status": "fail"}, + ] + active, suppressed = ds.split_suppressed(findings, supp) + assert len(suppressed) == 1 and len(active) == 1 + assert suppressed[0]["suppressed_reason"] == "known test value" + + +def test_baseline_gating(): + ds = _import_cli() + known = {"control": "DSGAI02", "rule_id": "P02.1", "path": "a.py", "line": 7, "status": "fail"} + fresh = {"control": "DSGAI04", "rule_id": "P04.1", "path": "b.py", "line": 1, "status": "fail"} + baseline = {ds.finding_fp(known)} + assert ds.finding_fp(known) in baseline # known finding is baselined + assert ds.finding_fp(fresh) not in baseline # a new finding gates + + +@pytest.mark.skipif(not os.environ.get("DSGAI_CVE_LIVE"), + reason="set DSGAI_CVE_LIVE=1 to run the live OSV CVE test") +def test_cve_langchain_exploitable(tmp_path): + sys.path.insert(0, os.path.join(SCANNER, "cli")) + import dsgai_cve + req = tmp_path / "requirements.txt" + req.write_text("langchain==0.1.0\n", encoding="utf-8") + disc = [(str(req), "requirements.txt")] + cves = dsgai_cve.enrich(disc, refresh=True) + if not cves: + pytest.skip("OSV unreachable") + assert any(c["status"] == "EXPLOITABLE" for c in cves) + assert dsgai_cve.enrich(disc, offline=True) == cves # cache → deterministic + + def test_rules_json_in_sync(): from_yaml = yaml.safe_load(open(RULES_YAML, encoding="utf-8")) rebuilt = json.dumps(from_yaml, indent=2, sort_keys=True, ensure_ascii=False) + "\n"