data_validation: check mapping rows that cite published test records

data_validation/requirements-dev.txt

@@ -1,6 +1,10 @@
 # Hash-pinned lock generated from requirements-dev.in by data_validation/update_locks.py
 # (uv pip compile --universal --generate-hashes --python-version 3.11).
 # Do not edit by hand: change requirements-dev.in and rerun the script.
+agent-evidence-vectors==0.24.0 \
+    --hash=sha256:bc8c5785dc9768676361151a4ca2fb4dc1d9bd5a22a67381f6db7adda014539d \
+    --hash=sha256:fa1c4b804f700a0f71f1a1776978546d9df571b4e5a451b58220478f75e3b6d3
+    # via -r requirements.in
 arrow==1.4.0 \
     --hash=sha256:749f0769958ebdc79c173ff0b0670d59051a535fa26e8eba02953dc19eb43205 \
     --hash=sha256:ed0cc050e98001b8779e84d461b0098c4ac597e88704a655582b21d116e526d7

data_validation/requirements.in

@@ -5,3 +5,7 @@ jsonschema[format-nongpl]>=4.18.0
 pytest>=7.0.0
 # Only for reference_data/update_reference_data.py (refreshing the ATLAS table)
 pyyaml>=6.0
+# Signed conformance records and their reference verifier, replayed by
+# validators/evidence_mapping_check.py for mapping rows that cite test_records.
+# Exact pin: a new corpus release can change a verdict, so Dependabot proposes each bump.
+agent-evidence-vectors==0.24.0

data_validation/requirements.txt

@@ -1,6 +1,10 @@
 # Hash-pinned lock generated from requirements.in by data_validation/update_locks.py
 # (uv pip compile --universal --generate-hashes --python-version 3.11).
 # Do not edit by hand: change requirements.in and rerun the script.
+agent-evidence-vectors==0.24.0 \
+    --hash=sha256:bc8c5785dc9768676361151a4ca2fb4dc1d9bd5a22a67381f6db7adda014539d \
+    --hash=sha256:fa1c4b804f700a0f71f1a1776978546d9df571b4e5a451b58220478f75e3b6d3
+    # via -r requirements.in
 arrow==1.4.0 \
     --hash=sha256:749f0769958ebdc79c173ff0b0670d59051a535fa26e8eba02953dc19eb43205 \
     --hash=sha256:ed0cc050e98001b8779e84d461b0098c4ac597e88704a655582b21d116e526d7

data_validation/run_all_checks.py

@@ -10,8 +10,8 @@
 2. Shared checks (data_validation/validators/) on every data file:
    schema (for files no dataset validator covers), DSGAI IDs, CVE/CWE/ATLAS
    OWASP Top 10 and NIST AI 100-2 IDs, CVSS scores and ratings, dates,
-   SPDX licences and BCP 47 language tags, anonymization patterns and
-   duplicates. Errors fail the run;
+   SPDX licences and BCP 47 language tags, mapping rows that cite
+   published test records, anonymization patterns and duplicates. Errors fail the run;
    heuristic warnings (anonymization, near-duplicates, retired IDs) are
    counted, and printed with --verbose.
 
@@ -41,6 +41,7 @@
     date_check,
     dedup_checker,
     dsgai_mapping_check,
+    evidence_mapping_check,
     metadata_check,
     schema_validator,
     severity_check,
@@ -158,6 +159,7 @@ def shared_checks(dataset: Path, datasets_root: Path) -> tuple[int, list[Finding
         findings += severity_check.check_data(path, data)
         findings += date_check.check_data(path, data)
         findings += metadata_check.check_data(path, data)
+        findings += evidence_mapping_check.check_data(path, data)
         findings += anonymization_scanner.check_data(path, data)
     records, load_errors = dedup_checker.load_records(files)
     findings += load_errors + dedup_checker.check_records(records)
@@ -225,7 +227,10 @@ def main(argv: list[str] | None = None) -> int:
 
     checked_files = errors = warnings = 0
     if not args.no_shared:
-        print("\nShared checks (schema, DSGAI, CVE/CWE/ATLAS/OWASP/NIST IDs, CVSS, dates, licences, languages, anonymization, duplicates):")
+        print(
+            "\nShared checks (schema, DSGAI, CVE/CWE/ATLAS/OWASP/NIST IDs, CVSS, dates, licences, languages,"
+            " test records, anonymization, duplicates):"
+        )
         for dataset in datasets:
             n, findings = shared_checks(dataset, datasets_root)
             if not n:

data_validation/tests/test_evidence_mapping_check.py

@@ -0,0 +1,106 @@
+"""Tests for evidence_mapping_check.py, against the pinned agent-evidence-vectors corpora."""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from validators import evidence_mapping_check as em  # noqa: E402
+from validators._common import ERROR, WARN  # noqa: E402
+
+P = Path("entry.json")
+
+
+def row(control_id="MCP03", status="satisfied", **records):
+    cited = {"package": "agent-evidence-vectors", "corpus": "vectors-mcp-owasp", **records}
+    return {"framework_alignments": [
+        {"framework": "OWASP MCP Top 10", "control_id": control_id, "alignment_status": status, "test_records": cited},
+    ]}
+
+
+def test_rows_without_test_records_are_not_checked():
+    assert em.check_data(P, {"framework_alignments": [{"control_id": "PR.DS-05", "alignment_status": "satisfied"}]}) == []
+
+
+def test_cited_requirement_that_tests_the_control_passes():
+    assert em.check_data(P, row(requirement="MCPVS-1")) == []
+
+
+def test_requirement_that_tests_another_control_fails():
+    [f] = em.check_data(P, row(control_id="MCP01", requirement="MCPVS-1"))
+    assert f.level == ERROR and "MCPVS-1 tests MCP03, not MCP01" in f.message
+    assert f.location == "/framework_alignments/0/test_records"
+
+
+def test_unknown_requirement_fails():
+    [f] = em.check_data(P, row(requirement="MCPVS-999"))
+    assert f.level == ERROR and "not in the vectors-mcp-owasp registry" in f.message
+
+
+def test_withdrawn_requirement_fails(monkeypatch):
+    monkeypatch.setattr(em, "registry", lambda corpus: {"MCPVS-1": {"id": "MCPVS-1", "status": "withdrawn", "top10": ["MCP03"]}})
+    [f] = em.check_data(P, row(requirement="MCPVS-1"))
+    assert f.level == ERROR and "withdrawn" in f.message
+
+
+def test_corpus_not_shipped_fails():
+    data = row()
+    data["framework_alignments"][0]["test_records"]["corpus"] = "vectors-does-not-exist"
+    [f] = em.check_data(P, data)
+    assert f.level == ERROR and "ships no corpus 'vectors-does-not-exist'" in f.message
+
+
+def test_corpus_without_a_reference_reader_fails():
+    # Shipped, but the package's reference verifier has no reader for it, so nothing is judged.
+    data = row(control_id="DE.AE-02")
+    data["framework_alignments"][0]["test_records"]["corpus"] = "vectors-delegation-lifecycle"
+    [f] = em.check_data(P, data)
+    assert f.level == ERROR and "does not replay clean" in f.message
+
+
+def test_failing_replay_fails(monkeypatch):
+    monkeypatch.setattr(em, "replay", lambda corpus: (False, "FAIL v0485d4e3a7045da5 reject"))
+    [f] = em.check_data(P, row(requirement="MCPVS-1"))
+    assert f.level == ERROR and "FAIL v0485d4e3a7045da5" in f.message
+
+
+@pytest.mark.parametrize("status", ["missing", "not_applicable"])
+def test_records_on_an_unsupported_row_warn(status):
+    [f] = em.check_data(P, row(status=status, requirement="MCPVS-1"))
+    assert f.level == WARN and status in f.message
+
+
+@pytest.mark.parametrize("cited", [
+    "vectors-mcp-owasp",
+    {"corpus": "vectors-mcp-owasp"},
+    {"package": "other-package", "corpus": "vectors-mcp-owasp"},
+    {"package": "agent-evidence-vectors"},
+])
+def test_malformed_citation_fails(cited):
+    [f] = em.check_data(P, {"framework_alignments": [{"control_id": "MCP03", "test_records": cited}]})
+    assert f.level == ERROR and "expected" in f.message
+
+
+def test_package_not_installed_fails(monkeypatch):
+    monkeypatch.setattr(em, "package_version", lambda: None)
+    [f] = em.check_data(P, row(requirement="MCPVS-1"))
+    assert f.level == ERROR and "is not installed" in f.message
+
+
+def test_each_corpus_replays_once():
+    em.replay.cache_clear()
+    em.check_data(P, {"a": row(requirement="MCPVS-1"), "b": row(requirement="MCPVS-2")})
+    info = em.replay.cache_info()
+    assert info.misses == 1 and info.hits == 1
+
+
+def test_cli_on_a_file(tmp_path, capsys):
+    import json
+
+    path = tmp_path / "entry.json"
+    path.write_text(json.dumps(row(control_id="MCP01", requirement="MCPVS-1")), encoding="utf-8")
+    assert em.main(["--file", str(path)]) == 1
+    assert "MCPVS-1 tests MCP03, not MCP01" in capsys.readouterr().out

data_validation/validators/evidence_mapping_check.py

@@ -0,0 +1,187 @@
+"""
+Checks that a framework mapping row which cites published test records is
+backed by records that exist and pass their reference verifier.
+
+A row in framework_alignments (or any object with a "control_id") may carry
+
+    "test_records": {
+        "package": "agent-evidence-vectors",
+        "corpus": "vectors-mcp-owasp",
+        "requirement": "MCPVS-1"
+    }
+
+naming a conformance corpus of signed accept and reject records shipped in
+the agent-evidence-vectors package (pinned in requirements.in), and
+optionally one requirement in that corpus's REGISTRY.json. The row's
+alignment_status then rests on something a third party can re-run instead
+of on the author's word.
+
+ERROR: the package is not installed; the corpus is not shipped in the
+       installed version; the requirement is not in the corpus registry or
+       is withdrawn; the requirement does not list the row's control_id
+       when the control is one the registry indexes (MCP01-MCP10); or the
+       corpus does not replay clean under the package's reference verifier
+       (including a corpus the reference verifier has no reader for).
+WARN:  test_records on a row whose alignment_status is "missing" or
+       "not_applicable", where the records cannot support the row.
+
+Each corpus is replayed once per run, however many rows cite it.
+
+Usage:
+    python evidence_mapping_check.py --file ../../datasets/riskassessment_dataset/entries/X.json
+    python evidence_mapping_check.py                # every dataset
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import sys
+import tempfile
+from functools import cache
+from importlib import metadata
+from pathlib import Path
+from typing import Any
+
+if __package__ in (None, ""):
+    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from validators._common import (  # noqa: E402
+    DATASETS_ROOT,
+    ERROR,
+    WARN,
+    Finding,
+    add_target_args,
+    ensure_utf8_stdout,
+    load_json,
+    print_findings,
+    resolve_targets,
+)
+
+CHECK = "evidence"
+PACKAGE = "agent-evidence-vectors"
+MODULE = "agent_evidence_vectors"
+KEY = "test_records"
+REPLAY_TIMEOUT = 300
+# Control IDs a corpus registry indexes in its "top10" lists.
+INDEXED_CONTROL_RE = re.compile(r"^MCP(0[1-9]|10)$")
+NO_SUPPORT = {"missing", "not_applicable"}
+
+
+@cache
+def package_version() -> str | None:
+    try:
+        return metadata.version(PACKAGE)
+    except metadata.PackageNotFoundError:
+        return None
+
+
+@cache
+def corpora_dir() -> Path | None:
+    if package_version() is None:
+        return None
+    from agent_evidence_vectors.run_vectors import corpora_root
+
+    return Path(corpora_root())
+
+
+@cache
+def registry(corpus: str) -> dict[str, dict] | None:
+    """Requirement and threat entries of a corpus registry, by id, or None if it has none."""
+    root = corpora_dir()
+    path = root / corpus / "REGISTRY.json" if root else None
+    if path is None or not path.is_file():
+        return None
+    data = json.loads(path.read_text(encoding="utf-8"))
+    return {e["id"]: e for section in ("requirements", "threats") for e in data.get(section, []) if "id" in e}
+
+
+@cache
+def replay(corpus: str) -> tuple[bool, str]:
+    """(passed, last output line) for one replay of a shipped corpus by the reference verifier."""
+    with tempfile.TemporaryDirectory() as tmp:
+        try:
+            proc = subprocess.run(
+                [sys.executable, "-m", f"{MODULE}.run_vectors", "--corpus", corpus, "--report", str(Path(tmp) / "report.json")],
+                cwd=tmp, capture_output=True, text=True, encoding="utf-8", errors="replace",
+                timeout=REPLAY_TIMEOUT, check=False,
+            )
+        except subprocess.TimeoutExpired:
+            return False, f"replay timed out after {REPLAY_TIMEOUT}s"
+    lines = (proc.stdout + proc.stderr).strip().splitlines()
+    return proc.returncode == 0, (lines[-1] if lines else f"exit {proc.returncode}")
+
+
+def iter_cited_rows(obj: Any, pointer: str = ""):
+    """(pointer, row) for every object that carries test_records."""
+    if isinstance(obj, dict):
+        if KEY in obj:
+            yield pointer, obj
+        for k, v in obj.items():
+            yield from iter_cited_rows(v, f"{pointer}/{k}")
+    elif isinstance(obj, list):
+        for i, v in enumerate(obj):
+            yield from iter_cited_rows(v, f"{pointer}/{i}")
+
+
+def check_row(path: Path, pointer: str, row: dict) -> list[Finding]:
+    where = f"{pointer}/{KEY}"
+    cited = row[KEY]
+    if not isinstance(cited, dict) or cited.get("package") != PACKAGE or not isinstance(cited.get("corpus"), str):
+        return [Finding(ERROR, CHECK, path, where, f'expected {{"package": "{PACKAGE}", "corpus": "<name>"}}')]
+    if package_version() is None:
+        return [Finding(ERROR, CHECK, path, where, f"{PACKAGE} is not installed (pip install -r requirements.txt)")]
+    findings: list[Finding] = []
+    status = row.get("alignment_status")
+    if status in NO_SUPPORT:
+        findings.append(Finding(WARN, CHECK, path, where, f'alignment_status is "{status}"; test records cannot support it'))
+    corpus = cited["corpus"]
+    root = corpora_dir()
+    if root is None or not (root / corpus / "MANIFEST.json").is_file():
+        return findings + [Finding(ERROR, CHECK, path, where, f"{PACKAGE} {package_version()} ships no corpus {corpus!r}")]
+    requirement = cited.get("requirement")
+    if requirement is not None:
+        entries = registry(corpus)
+        entry = entries.get(requirement) if entries else None
+        if entry is None:
+            findings.append(Finding(ERROR, CHECK, path, where, f"{requirement!r} is not in the {corpus} registry"))
+        elif entry.get("status") == "withdrawn":
+            findings.append(Finding(ERROR, CHECK, path, where, f"{requirement} is withdrawn in the {corpus} registry"))
+        else:
+            control = str(row.get("control_id", "")).strip()
+            if INDEXED_CONTROL_RE.match(control) and control not in entry.get("top10", []):
+                findings.append(Finding(ERROR, CHECK, path, where,
+                                        f"{requirement} tests {', '.join(entry.get('top10', [])) or 'no control'}, not {control}"))
+    passed, last = replay(corpus)
+    if not passed:
+        findings.append(Finding(ERROR, CHECK, path, where, f"{corpus} does not replay clean under the reference verifier: {last}"))
+    return findings
+
+
+def check_data(path: Path, data: Any) -> list[Finding]:
+    return [f for pointer, row in iter_cited_rows(data) for f in check_row(path, pointer, row)]
+
+
+def check_file(path: Path) -> list[Finding]:
+    data, err = load_json(path, CHECK)
+    if err:
+        return [err]
+    return check_data(path, data)
+
+
+def main(argv: list[str] | None = None) -> int:
+    ensure_utf8_stdout()
+    parser = argparse.ArgumentParser(description="Check mapping rows that cite published test records")
+    add_target_args(parser)
+    args = parser.parse_args(argv)
+    files = resolve_targets(args)
+    if files is None:
+        return 2
+    findings = [f for p in files for f in check_file(p)]
+    print(f"Checked {len(files)} file(s)")
+    return print_findings(findings, DATASETS_ROOT.parent)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

datasets/riskassessment_dataset/README.md

@@ -47,7 +47,7 @@ Canonical schema: [`schema.json`](./schema.json). Reference entry: [`example.jso
 | `risks_identified` | yes (≥5) | Per-DSGAI risk entries: `dsgai_id`, `inherent_rating`, `rationale`, optional `residual_rating` + `residual_rationale` |
 | `existing_controls` | no | Mitigations in place; each with optional tier and `addresses_dsgai[]` |
 | `control_gaps` | no | Missing/weak mitigations; each with `urgency` (Immediate/Near-term/Strategic) and optional `suggested_remediation` |
-| `framework_alignments` | yes (≥4) | Mappings to NIST CSF 2.0, NIST AI RMF, ISO/IEC 42001/27001, OWASP LLM/Agentic Top 10, MITRE ATLAS/ATT&CK, EU AI Act, GDPR, HIPAA, FERPA, PCI DSS 4.0, SOC 2, FedRAMP, Other |
+| `framework_alignments` | yes (≥4) | Mappings to NIST CSF 2.0, NIST AI RMF, ISO/IEC 42001/27001, OWASP LLM/Agentic/MCP Top 10, MITRE ATLAS/ATT&CK, EU AI Act, GDPR, HIPAA, FERPA, PCI DSS 4.0, SOC 2, FedRAMP, Other. A row may add `test_records` naming a published corpus of signed test records (see `data_validation/validators/evidence_mapping_check.py`) |
 | `compliance_obligations` | no | Regimes the deployment likely operates under |
 | `maturity_overall` | no | CMMI-like: Initial/Developing/Defined/Managed/Optimizing |
 | `source_urls` | no | Optional inspirational references |

datasets/riskassessment_dataset/croissant.json

@@ -89,7 +89,7 @@
       "description": "JSON Schema for the records",
       "contentUrl": "schema.json",
       "encodingFormat": "application/json",
-      "sha256": "75ae9981bc08d69ad0916706acf710946f1a892724d78ba8e0ed9f596305e79d"
+      "sha256": "c4ef74e00d8008b01cdf7f9297225562326c83db5aaaa3343928040181e88b89"
     },
     {
       "@type": "cr:FileSet",