diff --git a/.github/workflows/datasets-validate-all.yml b/.github/workflows/datasets-validate-all.yml index 3359d8d..f1f87a8 100644 --- a/.github/workflows/datasets-validate-all.yml +++ b/.github/workflows/datasets-validate-all.yml @@ -39,7 +39,7 @@ jobs: run: python -m pip install --quiet -r data_validation/requirements.txt - name: Run every dataset validator and the shared checks - run: python data_validation/run_all_checks.py --verbose + run: python data_validation/run_all_checks.py --verbose --github - name: Cross-dataset consistency run: python data_validation/qc_tools/consistency_check.py diff --git a/data_validation/README.md b/data_validation/README.md index b6bb24f..bc1cf27 100644 --- a/data_validation/README.md +++ b/data_validation/README.md @@ -22,7 +22,7 @@ The tooling has two jobs: keep the research data behind OWASP deliverables (the `run_all_checks.py` runs two layers. -**1. Dataset validators.** Each dataset that has a `validate.py` (at its root, or in a sub-collection folder) is run in its own directory. These are the authoritative checks for their data: schema conformance against the dataset's own `schema.json`, plus rules a schema cannot express. Datasets without one are listed as `NO VALIDATOR` so the gap is visible. +**1. Dataset validators.** Each dataset that has a `validate.py` (at its root, or in a sub-collection folder) is run in its own directory. Most are thin wrappers around a shared core, [`datasets/_shared/validate_lib.py`](../datasets/_shared/validate_lib.py) (schema with formats, DSGAI IDs, ID matches file name, duplicate IDs), plus the dataset's own extra rules. These are the authoritative checks for their data: schema conformance against the dataset's own `schema.json`, plus rules a schema cannot express. Datasets without one are listed as `NO VALIDATOR` so the gap is visible. | Dataset | Own validator | |---|---| @@ -92,6 +92,7 @@ python data_validation/run_all_checks.py # everything python data_validation/run_all_checks.py --dataset datasets/exploit_dataset python data_validation/run_all_checks.py --verbose # every validator's output and every warning python data_validation/run_all_checks.py --strict # warnings fail too +python data_validation/run_all_checks.py --github # also emit annotations (CI uses this) # One check, on chosen files or datasets (each script's docstring has details) python data_validation/validators/schema_validator.py --file datasets/rag_dataset/entries/RAG-0001.json @@ -107,7 +108,7 @@ GITHUB_TOKEN=... python data_validation/qc_tools/online_check.py --summary onlin python -m pytest data_validation/tests ``` -`run_all_checks.py` exits 0 when everything passed, 1 on any failure, and 2 when there was nothing to check. +`run_all_checks.py` exits 0 when everything passed, 1 on any failure, and 2 when there was nothing to check. In CI it runs with `--github`, so every finding also appears as an annotation on the exact line of the file in the pull request diff. --- diff --git a/data_validation/run_all_checks.py b/data_validation/run_all_checks.py index b444077..7167165 100644 --- a/data_validation/run_all_checks.py +++ b/data_validation/run_all_checks.py @@ -28,6 +28,7 @@ from __future__ import annotations import argparse +import re import subprocess import sys from pathlib import Path @@ -41,6 +42,7 @@ from validators._common import ( # noqa: E402 ERROR, SKIP_DIRS, Finding, ensure_utf8_stdout, iter_data_files, load_json, load_taxonomy_ids, ) +from validators.json_lines import line_for # noqa: E402 DATASETS_ROOT = Path(__file__).resolve().parent.parent / "datasets" DEFAULT_TIMEOUT = 300 @@ -88,6 +90,44 @@ def run_validator(dataset: Path, timeout: float = DEFAULT_TIMEOUT) -> tuple[str, return "FAIL", output +VALIDATOR_LINE_RE = re.compile(r"^(?P[^:\s][^:]*\.json): (?P[^:]+): (?P.+)$") + + +def _escape(text: str, prop: bool = False) -> str: + text = text.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A") + return text.replace(":", "%3A").replace(",", "%2C") if prop else text + + +def annotation(level: str, path: Path, pointer: str, title: str, message: str) -> str: + """A GitHub Actions workflow command that shows the finding on the PR diff.""" + command = "error" if level == ERROR else "warning" + try: + shown = path.resolve().relative_to(DATASETS_ROOT.resolve().parent).as_posix() + except ValueError: + shown = path.as_posix() + props = f"file={_escape(shown, True)}" + line = line_for(path, pointer) + if line: + props += f",line={line}" + return f"::{command} {props},title={_escape(title, True)}::{_escape(message)}" + + +def validator_annotations(script: Path, output: str) -> list[str]: + """Annotations for a dataset validator's "file.json: location: message" lines.""" + out = [] + for raw in output.splitlines(): + m = VALIDATOR_LINE_RE.match(raw.strip()) + if not m: + continue + name = m.group("file") + path = next((p for p in (script.parent / "entries" / name, script.parent / name) if p.is_file()), + script.parent / name) + loc = m.group("loc").strip() + pointer = "" if loc == "" else "/" + loc.lstrip("/") + out.append(annotation(ERROR, path, pointer, f"{script.parent.name}/validate.py", m.group("msg"))) + return out + + def shared_checks(dataset: Path, datasets_root: Path) -> tuple[int, list[Finding]]: """(files checked, findings) for the shared checks on one dataset.""" files = iter_data_files(dataset) @@ -129,6 +169,8 @@ def main(argv: list[str] | None = None) -> int: help="Treat shared-check warnings as failures") parser.add_argument("--no-shared", action="store_true", help="Run only the dataset validators") + parser.add_argument("--github", action="store_true", + help="Also print GitHub Actions annotations, so findings show on the PR diff") parser.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, help=f"Seconds each dataset validator may run (default {DEFAULT_TIMEOUT})") args = parser.parse_args(argv) @@ -149,6 +191,7 @@ def main(argv: list[str] | None = None) -> int: return 2 results = [] + annotations: list[str] = [] for dataset in datasets: validators = find_validators(dataset) if not validators: @@ -165,6 +208,8 @@ def main(argv: list[str] | None = None) -> int: if output and (args.verbose or status in ("FAIL", "ERROR")): for line in output.splitlines(): print(f" {line}") + if status == "FAIL": + annotations += validator_annotations(script, output) checked_files = errors = warnings = 0 if not args.no_shared: @@ -181,6 +226,11 @@ def main(argv: list[str] | None = None) -> int: for f in findings: if f.level == ERROR or args.verbose or args.strict: print(f" {f.format(datasets_root.parent)}") + annotations.append(annotation(f.level, f.path, f.location, f.check, f.message)) + + if args.github: + for line in annotations: + print(line) ran = [s for s in results if s != "NO VALIDATOR"] failed = [s for s in ran if s != "PASS"] diff --git a/data_validation/tests/test_json_lines.py b/data_validation/tests/test_json_lines.py new file mode 100644 index 0000000..8b3793a --- /dev/null +++ b/data_validation/tests/test_json_lines.py @@ -0,0 +1,45 @@ +"""Tests for json_lines.py and the runner's GitHub annotations.""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import run_all_checks # noqa: E402 +from validators.json_lines import pointer_line # noqa: E402 + +TEXT = """{ + "id": "X-1", + "tricky": "a \\" quote, a } and a ] inside", + "list": [ + "first", + {"inner": ["q", + "r"]} + ], + "a/b": 1 +}""" + + +def test_pointer_lines(): + assert pointer_line(TEXT, "/id") == 2 + assert pointer_line(TEXT, "/list/0") == 5 + assert pointer_line(TEXT, "/list/1/inner/1") == 7 + assert pointer_line(TEXT, "/a~1b") == 9 + assert pointer_line(TEXT, "") == 1 + assert pointer_line(TEXT, "/missing") is None + + +def test_annotation_format(tmp_path): + f = tmp_path / "e.json" + f.write_text(TEXT) + line = run_all_checks.annotation("ERROR", f, "/list/1/inner/1", "dsgai", "bad: value, 100%") + assert line.startswith("::error file=") and ",line=7,title=dsgai::bad: value, 100%25" in line + + +def test_validator_output_is_annotated(tmp_path): + (tmp_path / "entries").mkdir() + (tmp_path / "entries" / "e.json").write_text(TEXT) + out = "e.json: list/1/inner/1: 'r' is not valid\nunrelated line\n\nFAIL: 1 issue(s) across 1 entries." + [line] = run_all_checks.validator_annotations(tmp_path / "validate.py", out) + assert "entries/e.json,line=7" in line.replace("\\", "/") and line.endswith("::'r' is not valid") diff --git a/data_validation/tests/test_standards.py b/data_validation/tests/test_standards.py index 83da9f2..85a54d1 100644 --- a/data_validation/tests/test_standards.py +++ b/data_validation/tests/test_standards.py @@ -66,9 +66,16 @@ def test_owasp_table_is_complete_and_crosswalked(): @pytest.mark.parametrize("validate", sorted(DATASETS_ROOT.glob("*/validate.py")), ids=lambda p: p.parent.name) def test_dataset_validators_enforce_format(validate): - """Without a format checker, "format": "date" and "uri" are never checked.""" + """Without a format checker, "format": "date" and "uri" are never checked. + + A validator either builds its own (agentdataflow) or uses the shared core, + which does.""" source = validate.read_text(encoding="utf-8") - assert re.search(r"format_checker\s*=", source), f"{validate} builds its validator without a format checker" + uses_shared = "from validate_lib import" in source + assert uses_shared or re.search(r"format_checker\s*=", source), f"{validate} has no format checker" + if uses_shared: + lib = (DATASETS_ROOT / "_shared" / "validate_lib.py").read_text(encoding="utf-8") + assert re.search(r"format_checker\s*=", lib) def test_nist_table_matches_the_publication(): diff --git a/data_validation/tests/test_validate_lib.py b/data_validation/tests/test_validate_lib.py new file mode 100644 index 0000000..c8fbbc8 --- /dev/null +++ b/data_validation/tests/test_validate_lib.py @@ -0,0 +1,90 @@ +"""Tests for datasets/_shared/validate_lib.py, the core of the dataset validators.""" +from __future__ import annotations + +import json +import shutil +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "datasets" / "_shared")) + +import validate_lib # noqa: E402 + +SCHEMA = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "required": ["item_id", "date_added"], + "properties": {"item_id": {"type": "string"}, "date_added": {"type": "string", "format": "date"}, + "dsgai_mapping": {"type": "array"}}, +} + + +def make(tmp_path: Path, entries: dict[str, dict]) -> Path: + (tmp_path / "schema.json").write_text(json.dumps(SCHEMA)) + (tmp_path / "entries").mkdir() + for name, entry in entries.items(): + (tmp_path / "entries" / f"{name}.json").write_text(json.dumps(entry)) + return tmp_path + + +def run(tmp_path, capsys, **kwargs) -> tuple[int, str]: + code = validate_lib.run(tmp_path, id_field="item_id", **kwargs) + return code, capsys.readouterr().out + + +def test_valid_entries_pass(tmp_path, capsys): + make(tmp_path, {"A-1": {"item_id": "A-1", "date_added": "2026-01-01", "dsgai_mapping": ["DSGAI01"]}}) + assert run(tmp_path, capsys) == (0, "OK: 1 entries validated against schema.\n") + + +@pytest.mark.parametrize("entry,expected", [ + ({"item_id": "A-1"}, "'date_added' is a required property"), + ({"item_id": "A-1", "date_added": "2026-13-01"}, "is not a 'date'"), + ({"item_id": "A-1", "date_added": "2026-01-01", "dsgai_mapping": ["DSGAI99"]}, "dsgai_mapping 'DSGAI99' not in taxonomy"), + ({"item_id": "A-1", "date_added": "2026-01-01", "risks": [{"dsgai_id": "DSGAI00"}]}, "risks/0/dsgai_id: dsgai_id 'DSGAI00'"), + ({"item_id": "B-2", "date_added": "2026-01-01"}, "does not match the file name"), +]) +def test_each_rule_fails(tmp_path, capsys, entry, expected): + make(tmp_path, {"A-1": entry}) + code, out = run(tmp_path, capsys) + assert code == 1 and expected in out + + +def test_duplicate_ids(tmp_path, capsys): + make(tmp_path, {"A-1": {"item_id": "A-1", "date_added": "2026-01-01"}, + "A-2": {"item_id": "A-1", "date_added": "2026-01-01"}}) + code, out = run(tmp_path, capsys) + assert code == 1 and "is also used by A-1.json" in out + + +def test_extra_rule(tmp_path, capsys): + make(tmp_path, {"A-1": {"item_id": "A-1", "date_added": "2026-01-01"}}) + code, out = run(tmp_path, capsys, extra=lambda entry, path: ["custom problem"]) + assert code == 1 and "A-1.json: custom problem" in out + + +def test_empty_entries(tmp_path, capsys): + (tmp_path / "schema.json").write_text(json.dumps(SCHEMA)) + assert run(tmp_path, capsys, allow_empty=True)[0] == 0 + assert run(tmp_path, capsys)[0] == 1 + + +@pytest.mark.parametrize("dataset,mutate,expected", [ + ("exploit_dataset", lambda e: e.update(mitre_atlas_mapping=[]), "mitre_atlas_mapping does not include it"), + ("vulnerability_dataset", lambda e: e.update(cve_id="CVE-2000-0001"), "does not match cve_id"), +]) +def test_dataset_specific_rules_still_fire(tmp_path, dataset, mutate, expected): + """Run the real validate.py on a copy with one entry broken.""" + import subprocess + copy = tmp_path / "datasets" + shutil.copytree(ROOT / "datasets" / "_shared", copy / "_shared") + shutil.copytree(ROOT / "datasets" / dataset, copy / dataset) + entry_path = sorted((copy / dataset / "entries").glob("CVE-*.json" if dataset == "vulnerability_dataset" else "AML.*.json"))[0] + entry = json.loads(entry_path.read_text(encoding="utf-8")) + mutate(entry) + entry_path.write_text(json.dumps(entry), encoding="utf-8") + proc = subprocess.run([sys.executable, "validate.py"], cwd=copy / dataset, capture_output=True, text=True) + assert proc.returncode == 1 and expected in proc.stdout diff --git a/data_validation/validators/json_lines.py b/data_validation/validators/json_lines.py new file mode 100644 index 0000000..467d741 --- /dev/null +++ b/data_validation/validators/json_lines.py @@ -0,0 +1,111 @@ +""" +Line numbers for JSON pointers, so findings can be annotated on the exact +line of a pull request diff. The standard json module does not report +positions, so this walks the text with a small scanner that understands +just enough JSON (strings, nesting) to find where each value starts. +""" +from __future__ import annotations + +import json +from functools import lru_cache +from pathlib import Path + +_WS = " \t\r\n" + + +def _skip_ws(text: str, i: int) -> int: + while i < len(text) and text[i] in _WS: + i += 1 + return i + + +def _string_end(text: str, i: int) -> int: + """Index just past the string that starts at text[i] == '"'.""" + i += 1 + while i < len(text): + c = text[i] + if c == "\\": + i += 2 + continue + if c == '"': + return i + 1 + i += 1 + return i + + +def _value_end(text: str, i: int) -> int: + """Index just past the JSON value that starts at i.""" + c = text[i] + if c == '"': + return _string_end(text, i) + if c in "{[": + depth = 0 + while i < len(text): + c = text[i] + if c == '"': + i = _string_end(text, i) + continue + if c in "{[": + depth += 1 + elif c in "}]": + depth -= 1 + if depth == 0: + return i + 1 + i += 1 + return i + while i < len(text) and text[i] not in ",}]" + _WS: + i += 1 + return i + + +def _child_start(text: str, i: int, token: str) -> int | None: + """Start of the child `token` (object key or array index) of the container at i.""" + i = _skip_ws(text, i) + if i >= len(text) or text[i] not in "{[": + return None + is_object = text[i] == "{" + i += 1 + index = 0 + while True: + i = _skip_ws(text, i) + if i >= len(text) or text[i] in "}]": + return None + if is_object: + key_end = _string_end(text, i) + key = json.loads(text[i:key_end]) + i = _skip_ws(text, key_end) + 1 # past ':' + i = _skip_ws(text, i) + if key == token: + return i + else: + if str(index) == token: + return i + index += 1 + i = _skip_ws(text, _value_end(text, i)) + if i < len(text) and text[i] == ",": + i += 1 + + +def pointer_line(text: str, pointer: str) -> int | None: + """1-based line where the value at a JSON pointer ("/a/0/b") starts.""" + if pointer in ("", ""): + return 1 + i: int | None = _skip_ws(text, 0) + for raw in pointer.lstrip("/").split("/"): + token = raw.replace("~1", "/").replace("~0", "~") + i = _child_start(text, i, token) + if i is None: + return None + return text.count("\n", 0, i) + 1 + + +@lru_cache(maxsize=256) +def _text(path: str) -> str: + return Path(path).read_text(encoding="utf-8") + + +def line_for(path: Path, pointer: str) -> int | None: + try: + return pointer_line(_text(str(path)), pointer) + except (OSError, ValueError, IndexError): + return None diff --git a/datasets/_shared/validate_lib.py b/datasets/_shared/validate_lib.py new file mode 100644 index 0000000..c8a6453 --- /dev/null +++ b/datasets/_shared/validate_lib.py @@ -0,0 +1,116 @@ +"""Shared core of the dataset validators (datasets/*/validate.py). + +Each dataset's validate.py calls run() with what is particular to it: where +its schema is, which field is the entry ID, and any extra rules a schema +cannot express. run() does the rest the same way for every dataset: + + * every entries/*.json validates against the schema, with "format" + (dates, URIs) enforced + * every DSGAI ID (dsgai_mapping, dsgai_id, addresses_dsgai, at any depth) + is in datasets/_shared/dsgai_taxonomy.json + * the entry ID matches the file name, and no two entries share an ID + * output: one line per problem, then "OK: ..." or "FAIL: ..."; exit code + 0 on success, 1 on any problem, 2 when jsonschema is not installed + +Standard library plus jsonschema only, so a dataset can still be checked +with nothing but `pip install jsonschema`. +""" +from __future__ import annotations + +import json +import sys +from collections.abc import Callable, Iterator +from pathlib import Path +from typing import Any + +SHARED = Path(__file__).resolve().parent +TAXONOMY_PATH = SHARED / "dsgai_taxonomy.json" +DSGAI_KEYS = {"dsgai_mapping", "dsgai_id", "addresses_dsgai"} + +Extra = Callable[[dict, Path], list[str]] + + +def load_taxonomy_ids() -> set[str]: + data = json.loads(TAXONOMY_PATH.read_text(encoding="utf-8")) + return {e["id"] for e in data.get("entries", [])} + + +def dsgai_references(obj: Any, pointer: str = "") -> Iterator[tuple[str, str, Any]]: + """(pointer, key, value) for every value under a DSGAI key, at any depth.""" + if isinstance(obj, dict): + for key, value in obj.items(): + here = f"{pointer}/{key}" + if key in DSGAI_KEYS: + for i, item in enumerate(value if isinstance(value, list) else [value]): + yield (f"{here}/{i}" if isinstance(value, list) else here), key, item + else: + yield from dsgai_references(value, here) + elif isinstance(obj, list): + for i, item in enumerate(obj): + yield from dsgai_references(item, f"{pointer}/{i}") + + +def run(root: Path, *, id_field: str, schema_path: Path | None = None, + allow_empty: bool = False, extra: Extra | None = None) -> int: + """Validate root/entries/*.json. Returns the process exit code.""" + try: + from jsonschema import validators + except ImportError: + sys.stderr.write("jsonschema not installed. Run: pip install jsonschema\n") + return 2 + + schema = json.loads((schema_path or root / "schema.json").read_text(encoding="utf-8")) + cls = validators.validator_for(schema) + validator = cls(schema, format_checker=cls.FORMAT_CHECKER) + taxonomy_ids = load_taxonomy_ids() + + entries_dir = root / "entries" + entry_files = sorted(entries_dir.glob("*.json")) if entries_dir.is_dir() else [] + if not entry_files: + if allow_empty: + print(f"OK: no entries yet in {entries_dir.name}/.") + return 0 + sys.stderr.write(f"No entries found in {entries_dir}\n") + return 1 + + errors: list[str] = [] + seen: dict[str, str] = {} + for path in entry_files: + try: + entry = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as e: + errors.append(f"{path.name}: invalid JSON — {e}") + continue + + for err in validator.iter_errors(entry): + loc = "/".join(str(p) for p in err.absolute_path) or "" + errors.append(f"{path.name}: {loc}: {err.message}") + + for pointer, key, value in dsgai_references(entry): + if value not in taxonomy_ids: + errors.append(f"{path.name}: {pointer.lstrip('/')}: {key} {value!r} not in taxonomy") + + entry_id = entry.get(id_field) if isinstance(entry, dict) else None + if isinstance(entry_id, str): + if entry_id != path.stem: + errors.append(f"{path.name}: {id_field} {entry_id!r} does not match the file name") + if entry_id in seen: + errors.append(f"{path.name}: {id_field} {entry_id!r} is also used by {seen[entry_id]}") + seen.setdefault(entry_id, path.name) + + if extra is not None and isinstance(entry, dict): + errors += [f"{path.name}: {message}" for message in extra(entry, path)] + + if errors: + for line in errors: + print(line) + print(f"\nFAIL: {len(errors)} issue(s) across {len(entry_files)} entries.") + return 1 + + print(f"OK: {len(entry_files)} entries validated against schema.") + return 0 + + +def main_for(file: str, **kwargs) -> None: + """Entry point for a dataset's validate.py: run() on its folder and exit.""" + sys.exit(run(Path(file).resolve().parent, **kwargs)) diff --git a/datasets/exploit_dataset/validate.py b/datasets/exploit_dataset/validate.py index 599edb4..562da8e 100644 --- a/datasets/exploit_dataset/validate.py +++ b/datasets/exploit_dataset/validate.py @@ -1,85 +1,32 @@ """Validate every entry in ./entries/ against ./schema.json. Usage: python validate.py -Exit code 0 on success, 1 on any validation failure. +Exit code 0 on success, 1 on any validation failure, 2 without jsonschema. Requires: jsonschema (pip install jsonschema) + +The shared checks (schema with formats, DSGAI IDs, ID matches file name, +duplicate IDs) are in ../_shared/validate_lib.py. Specific to this dataset: +an entry whose exploit_id is an ATLAS ID must list it in mitre_atlas_mapping. """ from __future__ import annotations -import json -import re import sys from pathlib import Path -try: - from jsonschema import Draft202012Validator -except ImportError: - sys.stderr.write("jsonschema not installed. Run: pip install jsonschema\n") - sys.exit(2) +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "_shared")) +from validate_lib import main_for # noqa: E402 -ROOT = Path(__file__).resolve().parent -SCHEMA_PATH = ROOT / "schema.json" -ENTRIES_DIR = ROOT / "entries" -TAXONOMY_PATH = ROOT.parent / "_shared" / "dsgai_taxonomy.json" +import re # noqa: E402 ATLAS_ID_RE = re.compile(r"^AML\.T\d{4}(\.\d{3})?$") -def load_taxonomy_ids() -> set[str]: - if not TAXONOMY_PATH.exists(): - return set() - data = json.loads(TAXONOMY_PATH.read_text(encoding="utf-8")) - return {e["id"] for e in data.get("entries", [])} - - -def main() -> int: - schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) - validator = Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER) - taxonomy_ids = load_taxonomy_ids() - - errors: list[str] = [] - entry_files = sorted(ENTRIES_DIR.glob("*.json")) - if not entry_files: - sys.stderr.write(f"No entries found in {ENTRIES_DIR}\n") - return 1 - - for path in entry_files: - try: - entry = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as e: - errors.append(f"{path.name}: invalid JSON — {e}") - continue - - for err in validator.iter_errors(entry): - loc = "/".join(str(p) for p in err.absolute_path) or "" - errors.append(f"{path.name}: {loc}: {err.message}") - - if taxonomy_ids: - for dsgai_id in entry.get("dsgai_mapping", []): - if dsgai_id not in taxonomy_ids: - errors.append( - f"{path.name}: dsgai_mapping {dsgai_id!r} not in taxonomy" - ) - - # If the exploit_id is an ATLAS ID, mitre_atlas_mapping should include it. - eid = entry.get("exploit_id", "") - if ATLAS_ID_RE.match(eid): - atlas_map = entry.get("mitre_atlas_mapping") or [] - if eid not in atlas_map: - errors.append( - f"{path.name}: exploit_id is an ATLAS ID ({eid!r}) " - f"but mitre_atlas_mapping does not include it" - ) - - if errors: - for line in errors: - print(line) - print(f"\nFAIL: {len(errors)} issue(s) across {len(entry_files)} entries.") - return 1 - - print(f"OK: {len(entry_files)} entries validated against schema.") - return 0 +def atlas_id_is_mapped(entry: dict, path: Path) -> list[str]: + eid = entry.get("exploit_id", "") + if ATLAS_ID_RE.match(eid) and eid not in (entry.get("mitre_atlas_mapping") or []): + return [f"exploit_id is an ATLAS ID ({eid!r}) but mitre_atlas_mapping does not include it"] + return [] if __name__ == "__main__": - sys.exit(main()) + main_for(__file__, id_field="exploit_id", extra=atlas_id_is_mapped) diff --git a/datasets/incident_dataset/validate.py b/datasets/incident_dataset/validate.py index e2a4d7b..7f5adae 100644 --- a/datasets/incident_dataset/validate.py +++ b/datasets/incident_dataset/validate.py @@ -5,79 +5,19 @@ validation failure, 2 when jsonschema is missing. Requires: jsonschema (pip install jsonschema) -The schema is data_validation/schemas/incident.schema.json, the only schema for -this dataset. Beyond the schema this checks that incident_id matches the -file name, that no two entries share an incident_id, and that every -dsgai_mapping value is in datasets/_shared/dsgai_taxonomy.json. +The schema is data_validation/schemas/incident.schema.json, the only schema +for this dataset. The checks (schema with formats, DSGAI IDs, incident_id +matches the file name, duplicate IDs) are in ../_shared/validate_lib.py. """ from __future__ import annotations -import json import sys from pathlib import Path -try: - from jsonschema import Draft202012Validator -except ImportError: - sys.stderr.write("jsonschema not installed. Run: pip install jsonschema\n") - sys.exit(2) - -ROOT = Path(__file__).resolve().parent -SCHEMA_PATH = ROOT.parent.parent / "data_validation" / "schemas" / "incident.schema.json" -ENTRIES_DIR = ROOT / "entries" -TAXONOMY_PATH = ROOT.parent / "_shared" / "dsgai_taxonomy.json" -ID_FIELD = "incident_id" - - -def load_taxonomy_ids() -> set[str]: - data = json.loads(TAXONOMY_PATH.read_text(encoding="utf-8")) - return {e["id"] for e in data.get("entries", [])} - - -def main() -> int: - schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) - validator = Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER) - taxonomy_ids = load_taxonomy_ids() - - entry_files = sorted(ENTRIES_DIR.glob("*.json")) if ENTRIES_DIR.is_dir() else [] - if not entry_files: - print(f"OK: no entries yet in {ENTRIES_DIR.name}/.") - return 0 - - errors: list[str] = [] - seen: dict[str, str] = {} - for path in entry_files: - try: - entry = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as e: - errors.append(f"{path.name}: invalid JSON — {e}") - continue - - for err in validator.iter_errors(entry): - loc = "/".join(str(p) for p in err.absolute_path) or "" - errors.append(f"{path.name}: {loc}: {err.message}") - - for dsgai_id in entry.get("dsgai_mapping", []): - if dsgai_id not in taxonomy_ids: - errors.append(f"{path.name}: dsgai_mapping {dsgai_id!r} not in taxonomy") - - entry_id = entry.get(ID_FIELD) - if isinstance(entry_id, str): - if entry_id != path.stem: - errors.append(f"{path.name}: {ID_FIELD} {entry_id!r} does not match the file name") - if entry_id in seen: - errors.append(f"{path.name}: {ID_FIELD} {entry_id!r} is also used by {seen[entry_id]}") - seen.setdefault(entry_id, path.name) - - if errors: - for line in errors: - print(line) - print(f"\nFAIL: {len(errors)} issue(s) across {len(entry_files)} entries.") - return 1 - - print(f"OK: {len(entry_files)} entries validated against schema.") - return 0 +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "_shared")) +from validate_lib import main_for # noqa: E402 +SCHEMA = Path(__file__).resolve().parents[2] / "data_validation" / "schemas" / "incident.schema.json" if __name__ == "__main__": - sys.exit(main()) + main_for(__file__, id_field="incident_id", schema_path=SCHEMA, allow_empty=True) diff --git a/datasets/rag_dataset/validate.py b/datasets/rag_dataset/validate.py index 2c5ed9a..7253089 100644 --- a/datasets/rag_dataset/validate.py +++ b/datasets/rag_dataset/validate.py @@ -6,78 +6,18 @@ Requires: jsonschema (pip install jsonschema) The schema is data_validation/schemas/rag.schema.json, the only schema for -this dataset. Beyond the schema this checks that document_id matches the -file name, that no two entries share a document_id, and that every -dsgai_mapping value is in datasets/_shared/dsgai_taxonomy.json. +this dataset. The checks (schema with formats, DSGAI IDs, document_id +matches the file name, duplicate IDs) are in ../_shared/validate_lib.py. """ from __future__ import annotations -import json import sys from pathlib import Path -try: - from jsonschema import Draft202012Validator -except ImportError: - sys.stderr.write("jsonschema not installed. Run: pip install jsonschema\n") - sys.exit(2) - -ROOT = Path(__file__).resolve().parent -SCHEMA_PATH = ROOT.parent.parent / "data_validation" / "schemas" / "rag.schema.json" -ENTRIES_DIR = ROOT / "entries" -TAXONOMY_PATH = ROOT.parent / "_shared" / "dsgai_taxonomy.json" -ID_FIELD = "document_id" - - -def load_taxonomy_ids() -> set[str]: - data = json.loads(TAXONOMY_PATH.read_text(encoding="utf-8")) - return {e["id"] for e in data.get("entries", [])} - - -def main() -> int: - schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) - validator = Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER) - taxonomy_ids = load_taxonomy_ids() - - entry_files = sorted(ENTRIES_DIR.glob("*.json")) if ENTRIES_DIR.is_dir() else [] - if not entry_files: - print(f"OK: no entries yet in {ENTRIES_DIR.name}/.") - return 0 - - errors: list[str] = [] - seen: dict[str, str] = {} - for path in entry_files: - try: - entry = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as e: - errors.append(f"{path.name}: invalid JSON — {e}") - continue - - for err in validator.iter_errors(entry): - loc = "/".join(str(p) for p in err.absolute_path) or "" - errors.append(f"{path.name}: {loc}: {err.message}") - - for dsgai_id in entry.get("dsgai_mapping", []): - if dsgai_id not in taxonomy_ids: - errors.append(f"{path.name}: dsgai_mapping {dsgai_id!r} not in taxonomy") - - entry_id = entry.get(ID_FIELD) - if isinstance(entry_id, str): - if entry_id != path.stem: - errors.append(f"{path.name}: {ID_FIELD} {entry_id!r} does not match the file name") - if entry_id in seen: - errors.append(f"{path.name}: {ID_FIELD} {entry_id!r} is also used by {seen[entry_id]}") - seen.setdefault(entry_id, path.name) - - if errors: - for line in errors: - print(line) - print(f"\nFAIL: {len(errors)} issue(s) across {len(entry_files)} entries.") - return 1 - - print(f"OK: {len(entry_files)} entries validated against schema.") - return 0 +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "_shared")) +from validate_lib import main_for # noqa: E402 +SCHEMA = Path(__file__).resolve().parents[2] / "data_validation" / "schemas" / "rag.schema.json" if __name__ == "__main__": - sys.exit(main()) + main_for(__file__, id_field="document_id", schema_path=SCHEMA, allow_empty=True) diff --git a/datasets/riskassessment_dataset/validate.py b/datasets/riskassessment_dataset/validate.py index b574132..50b7479 100644 --- a/datasets/riskassessment_dataset/validate.py +++ b/datasets/riskassessment_dataset/validate.py @@ -1,90 +1,21 @@ """Validate every entry in ./entries/ against ./schema.json. Usage: python validate.py -Exit code 0 on success, 1 on any validation failure. +Exit code 0 on success, 1 on any validation failure, 2 without jsonschema. Requires: jsonschema (pip install jsonschema) + +The shared checks (schema with formats, DSGAI IDs, including +risks_identified[].dsgai_id and existing_controls/control_gaps +addresses_dsgai, ID matches file name, duplicate IDs) are in +../_shared/validate_lib.py. """ from __future__ import annotations -import json import sys from pathlib import Path -try: - from jsonschema import Draft202012Validator -except ImportError: - sys.stderr.write("jsonschema not installed. Run: pip install jsonschema\n") - sys.exit(2) - -ROOT = Path(__file__).resolve().parent -SCHEMA_PATH = ROOT / "schema.json" -ENTRIES_DIR = ROOT / "entries" -TAXONOMY_PATH = ROOT.parent / "_shared" / "dsgai_taxonomy.json" - - -def load_taxonomy_ids() -> set[str]: - if not TAXONOMY_PATH.exists(): - return set() - data = json.loads(TAXONOMY_PATH.read_text(encoding="utf-8")) - return {e["id"] for e in data.get("entries", [])} - - -def main() -> int: - schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) - validator = Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER) - taxonomy_ids = load_taxonomy_ids() - - errors: list[str] = [] - entry_files = sorted(ENTRIES_DIR.glob("*.json")) - if not entry_files: - sys.stderr.write(f"No entries found in {ENTRIES_DIR}\n") - return 1 - - for path in entry_files: - try: - entry = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as e: - errors.append(f"{path.name}: invalid JSON — {e}") - continue - - for err in validator.iter_errors(entry): - loc = "/".join(str(p) for p in err.absolute_path) or "" - errors.append(f"{path.name}: {loc}: {err.message}") - - if taxonomy_ids: - seen_dsgai: set[str] = set() - for risk in entry.get("risks_identified", []): - seen_dsgai.add(risk.get("dsgai_id", "")) - if risk.get("dsgai_id") not in taxonomy_ids: - errors.append( - f"{path.name}: risks_identified.dsgai_id " - f"{risk.get('dsgai_id')!r} not in taxonomy" - ) - - for control in entry.get("existing_controls", []): - for did in control.get("addresses_dsgai", []) or []: - if did not in taxonomy_ids: - errors.append( - f"{path.name}: existing_controls.addresses_dsgai " - f"{did!r} not in taxonomy" - ) - for gap in entry.get("control_gaps", []): - for did in gap.get("addresses_dsgai", []) or []: - if did not in taxonomy_ids: - errors.append( - f"{path.name}: control_gaps.addresses_dsgai " - f"{did!r} not in taxonomy" - ) - - if errors: - for line in errors: - print(line) - print(f"\nFAIL: {len(errors)} issue(s) across {len(entry_files)} entries.") - return 1 - - print(f"OK: {len(entry_files)} entries validated against schema.") - return 0 - +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "_shared")) +from validate_lib import main_for # noqa: E402 if __name__ == "__main__": - sys.exit(main()) + main_for(__file__, id_field="assessment_id") diff --git a/datasets/vulnerability_dataset/validate.py b/datasets/vulnerability_dataset/validate.py index a7d74b1..d8545f6 100644 --- a/datasets/vulnerability_dataset/validate.py +++ b/datasets/vulnerability_dataset/validate.py @@ -1,79 +1,28 @@ """Validate every entry in ./entries/ against ./schema.json. Usage: python validate.py -Exit code 0 on success, 1 on any validation failure. +Exit code 0 on success, 1 on any validation failure, 2 without jsonschema. Requires: jsonschema (pip install jsonschema) + +The shared checks (schema with formats, DSGAI IDs, ID matches file name, +duplicate IDs) are in ../_shared/validate_lib.py. Specific to this dataset: +a CVE vulnerability_id must equal cve_id. """ from __future__ import annotations -import json import sys from pathlib import Path -try: - from jsonschema import Draft202012Validator -except ImportError: - sys.stderr.write("jsonschema not installed. Run: pip install jsonschema\n") - sys.exit(2) - -ROOT = Path(__file__).resolve().parent -SCHEMA_PATH = ROOT / "schema.json" -ENTRIES_DIR = ROOT / "entries" -TAXONOMY_PATH = ROOT.parent / "_shared" / "dsgai_taxonomy.json" - - -def load_taxonomy_ids() -> set[str]: - if not TAXONOMY_PATH.exists(): - return set() - data = json.loads(TAXONOMY_PATH.read_text(encoding="utf-8")) - return {e["id"] for e in data.get("entries", [])} - - -def main() -> int: - schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) - validator = Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER) - taxonomy_ids = load_taxonomy_ids() - - errors: list[str] = [] - entry_files = sorted(ENTRIES_DIR.glob("*.json")) - if not entry_files: - sys.stderr.write(f"No entries found in {ENTRIES_DIR}\n") - return 1 - - for path in entry_files: - try: - entry = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as e: - errors.append(f"{path.name}: invalid JSON — {e}") - continue - - for err in validator.iter_errors(entry): - loc = "/".join(str(p) for p in err.absolute_path) or "" - errors.append(f"{path.name}: {loc}: {err.message}") - - if taxonomy_ids: - for dsgai_id in entry.get("dsgai_mapping", []): - if dsgai_id not in taxonomy_ids: - errors.append( - f"{path.name}: dsgai_mapping {dsgai_id!r} not in taxonomy" - ) - - if entry.get("cve_id") and entry.get("vulnerability_id", "").startswith("CVE-"): - if entry["cve_id"] != entry["vulnerability_id"]: - errors.append( - f"{path.name}: vulnerability_id ({entry['vulnerability_id']!r}) " - f"does not match cve_id ({entry['cve_id']!r})" - ) +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "_shared")) +from validate_lib import main_for # noqa: E402 - if errors: - for line in errors: - print(line) - print(f"\nFAIL: {len(errors)} issue(s) across {len(entry_files)} entries.") - return 1 - print(f"OK: {len(entry_files)} entries validated against schema.") - return 0 +def cve_id_matches(entry: dict, path: Path) -> list[str]: + vid, cve = entry.get("vulnerability_id", ""), entry.get("cve_id") + if cve and vid.startswith("CVE-") and cve != vid: + return [f"vulnerability_id ({vid!r}) does not match cve_id ({cve!r})"] + return [] if __name__ == "__main__": - sys.exit(main()) + main_for(__file__, id_field="vulnerability_id", extra=cve_id_matches)