v0.3: phone check (iPhone, Android) as the Space front page, catalog 0.3 (74 checks), shared vulnerability rules, real-CVE training data
6ccf48e verified | #!/usr/bin/env python3 | |
| """Validate audit evidence (collector or synthetic) against the schema rules and the check catalog. | |
| Usage: | |
| python3 tools/validate_evidence.py audit-macos-host-20261007.json | |
| python3 tools/validate_evidence.py synthetic/out/evidence.jsonl | |
| Accepts .json (one record) or .jsonl (one record per line). Exit code 1 if any record is invalid. | |
| Standard library only. | |
| """ | |
| import json | |
| import os | |
| import sys | |
| ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| REQUIRED = {"schema_version", "catalog_version", "source", "audit", "host", "baseline", "results", "summary"} | |
| RESULT_KEYS = {"id", "section", "name", "status", "actual", "expected"} | |
| OPTIONAL_RESULT_KEYS = {"detail", "evidence"} | |
| def load_catalog(): | |
| with open(os.path.join(ROOT, "catalog", "checks.json"), encoding="utf-8") as handle: | |
| catalog = json.load(handle) | |
| return catalog, {c["id"]: c for c in catalog["checks"]} | |
| def validate(record, catalog, checks): | |
| errors = [] | |
| missing = REQUIRED - record.keys() | |
| extra = record.keys() - REQUIRED | |
| if missing: | |
| errors.append(f"missing top-level keys: {sorted(missing)}") | |
| if extra: | |
| errors.append(f"unexpected top-level keys: {sorted(extra)}") | |
| if errors: | |
| return errors | |
| if record["schema_version"] != "1.0": | |
| errors.append(f"schema_version {record['schema_version']!r} != '1.0'") | |
| if record["catalog_version"] != catalog["catalog_version"]: | |
| errors.append(f"catalog_version {record['catalog_version']} != {catalog['catalog_version']}") | |
| if record["source"].get("kind") not in ("collector", "synthetic", "web"): | |
| errors.append("source.kind must be collector, synthetic or web") | |
| os_family = record["host"].get("os_family") | |
| if os_family not in ("windows", "linux", "macos", "ios", "android"): | |
| errors.append(f"host.os_family invalid: {os_family!r}") | |
| return errors | |
| if record["source"].get("collector") != os_family: | |
| errors.append("source.collector does not match host.os_family") | |
| if record["audit"].get("mode") != "READ-ONLY": | |
| errors.append("audit.mode must be READ-ONLY") | |
| if not isinstance(record["audit"].get("is_admin"), bool): | |
| errors.append("audit.is_admin must be true/false") | |
| for key in ("hostname", "os_name", "os_version"): | |
| if not record["host"].get(key): | |
| errors.append(f"host.{key} is empty") | |
| if not all(isinstance(v, str) for v in record["baseline"].values()): | |
| errors.append("baseline values must all be strings") | |
| seen = set() | |
| for index, result in enumerate(record["results"]): | |
| where = f"results[{index}]" | |
| keys = set(result.keys()) | |
| if not RESULT_KEYS <= keys: | |
| errors.append(f"{where}: missing {sorted(RESULT_KEYS - keys)}") | |
| continue | |
| if keys - RESULT_KEYS - OPTIONAL_RESULT_KEYS: | |
| errors.append(f"{where}: unexpected keys {sorted(keys - RESULT_KEYS - OPTIONAL_RESULT_KEYS)}") | |
| check_id = result["id"] | |
| if check_id in seen: | |
| errors.append(f"{where}: duplicate id {check_id}") | |
| seen.add(check_id) | |
| check = checks.get(check_id) | |
| if not check: | |
| errors.append(f"{where}: unknown check id {check_id}") | |
| continue | |
| if os_family not in check["os"]: | |
| errors.append(f"{where}: {check_id} does not apply to {os_family}") | |
| if result["section"] != check["section"]: | |
| errors.append(f"{where}: {check_id} section {result['section']} != {check['section']}") | |
| expected_name = check.get("names", {}).get(os_family, check["name"]) | |
| if result["name"] != expected_name: | |
| errors.append(f"{where}: {check_id} name {result['name']!r} != catalog {expected_name!r}") | |
| if result["status"] not in catalog["statuses"]: | |
| errors.append(f"{where}: {check_id} invalid status {result['status']!r}") | |
| if "evidence" in result and result["evidence"] not in ("detected", "self-reported"): | |
| errors.append(f"{where}: {check_id} evidence must be detected or self-reported") | |
| if os_family in ("ios", "android") and check["section"] != "org" and "evidence" not in result: | |
| errors.append(f"{where}: {check_id} needs an evidence source on phones") | |
| if os_family not in ("ios", "android") and "evidence" in result: | |
| errors.append(f"{where}: {check_id} evidence source is only used on phones") | |
| for key in ("actual", "expected") + (("detail",) if "detail" in result else ()): | |
| if not isinstance(result[key], str): | |
| errors.append(f"{where}: {check_id}.{key} must be a string") | |
| expected_ids = {c["id"] for c in catalog["checks"] if os_family in c["os"]} | |
| if seen != expected_ids: | |
| if expected_ids - seen: | |
| errors.append(f"checks missing for {os_family}: {sorted(expected_ids - seen)}") | |
| if seen - expected_ids: | |
| errors.append(f"checks not expected for {os_family}: {sorted(seen - expected_ids)}") | |
| counts = {s: sum(1 for r in record["results"] if r.get("status") == s) for s in catalog["statuses"]} | |
| if record["summary"] != counts: | |
| errors.append(f"summary {record['summary']} != counted {counts}") | |
| return errors | |
| def read_records(path): | |
| with open(path, encoding="utf-8-sig") as handle: | |
| if path.endswith(".jsonl"): | |
| return [json.loads(line) for line in handle if line.strip()] | |
| return [json.load(handle)] | |
| def main(): | |
| if len(sys.argv) < 2: | |
| print(__doc__) | |
| return 2 | |
| catalog, checks = load_catalog() | |
| total = bad = 0 | |
| for path in sys.argv[1:]: | |
| for number, record in enumerate(read_records(path), 1): | |
| total += 1 | |
| errors = validate(record, catalog, checks) | |
| if errors: | |
| bad += 1 | |
| print(f"INVALID {path} record {number}:") | |
| for error in errors[:20]: | |
| print(f" - {error}") | |
| print(f"{total - bad}/{total} records valid") | |
| return 1 if bad else 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |