diff --git a/.github/scripts/test_validate_mappings.py b/.github/scripts/test_validate_mappings.py new file mode 100644 index 0000000..8c1e045 --- /dev/null +++ b/.github/scripts/test_validate_mappings.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +"""Offline tests for the cross-framework mapping validator.""" + +from __future__ import annotations + +import copy +import io +import json +import sys +import tempfile +import unittest +import zipfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import validate_mappings as validator # noqa: E402 + + +def cwe_source(version: str = "4.20") -> bytes: + xml = f""" + + + + + +""".encode() + result = io.BytesIO() + with zipfile.ZipFile(result, "w") as archive: + archive.writestr(f"cwec_v{version}.xml", xml) + return result.getvalue() + + +def atlas_source( + content_version: str = "2026.06", format_version: str = "6.0.0" +) -> bytes: + return f"""format-version: {format_version} +collection: + name: ATLAS + version: '{content_version}' +tactics: + AML.TA0000: + name: AI Model Access + id: AML.TA0000 + object-type: tactic +techniques: {{}} +""".encode() + + +class MappingValidatorTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + self.root = Path(self.temporary_directory.name) + self.mapping_dir = self.root / "mappings" + self.source_dir = self.root / "sources" + self.mapping_dir.mkdir() + self.source_dir.mkdir() + + cwe_bytes = cwe_source() + atlas_bytes = atlas_source() + (self.source_dir / "cwec_v4.20.xml.zip").write_bytes(cwe_bytes) + (self.source_dir / "ATLAS-2026.06.yaml").write_bytes(atlas_bytes) + + self.cwe_mapping = { + "framework": "MITRE CWE (Common Weakness Enumeration)", + "version": "4.20", + "source_url": "https://cwe.mitre.org/data/xml/cwec_v4.20.xml.zip", + "source_sha": validator.sha256(cwe_bytes), + "elements": [{"id": "CWE-79", "title": "Cross-site Scripting (Base)"}], + } + self.atlas_mapping = { + "framework": "MITRE ATLAS", + "version": "content v2026.06 (format-version 6.0.0)", + "source_url": ( + "https://raw.githubusercontent.com/mitre-atlas/atlas-data/" + "main/dist/v6/ATLAS-2026.06.yaml" + ), + "source_sha": validator.sha256(atlas_bytes), + "elements": [{"id": "AML.TA0000", "title": "AI Model Access"}], + } + self.write_catalogs() + + def tearDown(self) -> None: + self.temporary_directory.cleanup() + + def write_catalogs(self) -> None: + (self.mapping_dir / "cwe-4.20.json").write_text( + json.dumps(self.cwe_mapping), encoding="utf-8" + ) + (self.mapping_dir / "mitre-atlas-2026.06.json").write_text( + json.dumps(self.atlas_mapping), encoding="utf-8" + ) + + def errors_for(self, filename: str) -> list[str]: + return validator.validate_catalog( + self.mapping_dir / filename, self.source_dir, download=False + ) + + def test_valid_catalogs_pass_offline(self) -> None: + self.assertEqual([], self.errors_for("cwe-4.20.json")) + self.assertEqual([], self.errors_for("mitre-atlas-2026.06.json")) + + def test_nonexistent_identifier_fails(self) -> None: + self.cwe_mapping["elements"][0]["id"] = "CWE-999999" + self.write_catalogs() + self.assertTrue( + any("identifier not found" in error for error in self.errors_for("cwe-4.20.json")) + ) + + def test_duplicate_identifier_fails(self) -> None: + self.cwe_mapping["elements"].append( + copy.deepcopy(self.cwe_mapping["elements"][0]) + ) + self.write_catalogs() + self.assertTrue( + any("duplicate element id" in error for error in self.errors_for("cwe-4.20.json")) + ) + + def test_title_mismatch_fails(self) -> None: + self.atlas_mapping["elements"][0]["title"] = "ML Model Access" + self.write_catalogs() + self.assertTrue( + any("title mismatch" in error for error in self.errors_for("mitre-atlas-2026.06.json")) + ) + + def test_cwe_abstraction_mismatch_fails(self) -> None: + self.cwe_mapping["elements"][0]["title"] = "Cross-site Scripting (Class)" + self.write_catalogs() + self.assertTrue( + any("title mismatch" in error for error in self.errors_for("cwe-4.20.json")) + ) + + def test_source_version_mismatch_fails(self) -> None: + mismatched = atlas_source(content_version="2025.10") + (self.source_dir / "ATLAS-2026.06.yaml").write_bytes(mismatched) + self.atlas_mapping["source_sha"] = validator.sha256(mismatched) + self.write_catalogs() + self.assertTrue( + any( + "content version mismatch" in error + for error in self.errors_for("mitre-atlas-2026.06.json") + ) + ) + + def test_source_sha_mismatch_fails(self) -> None: + self.cwe_mapping["source_sha"] = "sha256:" + ("0" * 64) + self.write_catalogs() + self.assertTrue( + any("SHA-256 mismatch" in error for error in self.errors_for("cwe-4.20.json")) + ) + + def test_deprecated_atlas_path_fails(self) -> None: + self.atlas_mapping["source_url"] = ( + "https://raw.githubusercontent.com/mitre-atlas/atlas-data/main/dist/ATLAS.yaml" + ) + self.write_catalogs() + self.assertTrue( + any( + "pinned v6 source" in error + for error in self.errors_for("mitre-atlas-2026.06.json") + ) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/scripts/validate_mappings.py b/.github/scripts/validate_mappings.py new file mode 100644 index 0000000..9ed370f --- /dev/null +++ b/.github/scripts/validate_mappings.py @@ -0,0 +1,315 @@ +#!/usr/bin/env python3 +"""Validate mapping catalogs against version-pinned primary datasets.""" + +from __future__ import annotations + +import argparse +import hashlib +import io +import json +import re +import sys +import urllib.request +import xml.etree.ElementTree as ET +import zipfile +from pathlib import Path +from urllib.parse import urlparse + +ROOT = Path(__file__).resolve().parents[2] +DEFAULT_MAPPING_DIR = ROOT / "2026" / "final" / "mappings" +DEFAULT_SOURCE_DIR = ROOT / ".cache" / "mapping-sources" +SUPPORTED_CATALOGS = { + "cwe-4.20.json": "cwe", + "mitre-atlas-2026.06.json": "atlas", +} +MAX_SOURCE_BYTES = 25 * 1024 * 1024 + + +def sha256(data: bytes) -> str: + return f"sha256:{hashlib.sha256(data).hexdigest()}" + + +def parse_expected_sha(value: object) -> str: + if not isinstance(value, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", value): + raise ValueError("source_sha must use the form sha256:<64 lowercase hex digits>") + return value + + +def read_source(mapping: dict, source_dir: Path, download: bool) -> bytes: + source_url = mapping.get("source_url") + if not isinstance(source_url, str) or urlparse(source_url).scheme != "https": + raise ValueError("source_url must be an HTTPS URL") + + filename = Path(urlparse(source_url).path).name + if not filename: + raise ValueError("source_url must identify a source file") + + source_path = source_dir / filename + if not source_path.exists(): + if not download: + raise ValueError( + f"source file not found: {source_path} " + "(use --download to populate the cache)" + ) + source_dir.mkdir(parents=True, exist_ok=True) + try: + with urllib.request.urlopen(source_url, timeout=60) as response: + data = response.read(MAX_SOURCE_BYTES + 1) + except Exception as exc: + raise ValueError(f"could not download {source_url}: {exc}") from exc + if len(data) > MAX_SOURCE_BYTES: + raise ValueError( + f"source exceeds the {MAX_SOURCE_BYTES // (1024 * 1024)} MiB limit" + ) + source_path.write_bytes(data) + + return source_path.read_bytes() + + +def validate_source_location(catalog: str, mapping: dict) -> list[str]: + source_url = mapping.get("source_url") + if not isinstance(source_url, str): + return ["source_url must be an HTTPS URL"] + + if catalog == "cwe": + expected = ( + "https://cwe.mitre.org/data/xml/" + f"cwec_v{mapping.get('version', '')}.xml.zip" + ) + if source_url != expected: + return [f"CWE source_url must be the pinned source {expected}"] + return [] + + version_match = re.fullmatch( + r"content v(\d{4}\.\d{2}) \(format-version ([^)]+)\)", + str(mapping.get("version", "")), + ) + if not version_match: + return ["ATLAS version must be 'content vYYYY.MM (format-version X.Y.Z)'"] + content_version = version_match.group(1) + expected = ( + "https://raw.githubusercontent.com/mitre-atlas/atlas-data/" + f"main/dist/v6/ATLAS-{content_version}.yaml" + ) + if source_url != expected: + return [ + "ATLAS source_url must use the pinned v6 source " + f"{expected}" + ] + return [] + + +def validate_common(mapping: dict) -> list[str]: + errors: list[str] = [] + elements = mapping.get("elements") + if not isinstance(elements, list): + return ["elements must be an array"] + + seen: set[str] = set() + for index, element in enumerate(elements): + location = f"elements[{index}]" + if not isinstance(element, dict): + errors.append(f"{location} must be an object") + continue + element_id = element.get("id") + title = element.get("title") + if not isinstance(element_id, str) or not element_id: + errors.append(f"{location}.id must be a non-empty string") + elif element_id in seen: + errors.append(f"duplicate element id: {element_id}") + else: + seen.add(element_id) + if not isinstance(title, str) or not title: + errors.append(f"{location}.title must be a non-empty string") + return errors + + +def validate_cwe(mapping: dict, source: bytes) -> list[str]: + errors: list[str] = [] + expected_version = str(mapping.get("version", "")) + + try: + with zipfile.ZipFile(io.BytesIO(source)) as archive: + expected_member = f"cwec_v{expected_version}.xml" + if archive.namelist() != [expected_member]: + errors.append( + f"CWE archive must contain only {expected_member}; " + f"found {archive.namelist()}" + ) + return errors + root = ET.fromstring(archive.read(expected_member)) + except (zipfile.BadZipFile, ET.ParseError, KeyError) as exc: + return errors + [f"invalid CWE source archive: {exc}"] + + source_version = root.attrib.get("Version") + if source_version != expected_version: + errors.append( + f"CWE source version mismatch: expected {expected_version}, " + f"found {source_version or 'missing'}" + ) + + namespace = root.tag.partition("}")[0].lstrip("{") + weaknesses = {} + for weakness in root.findall(f".//{{{namespace}}}Weakness"): + weakness_id = f"CWE-{weakness.attrib.get('ID', '')}" + title = weakness.attrib.get("Name", "") + abstraction = weakness.attrib.get("Abstraction", "") + weaknesses[weakness_id] = f"{title} ({abstraction})" + + for element in mapping.get("elements", []): + if not isinstance(element, dict): + continue + element_id = element.get("id") + if element_id not in weaknesses: + errors.append(f"{element_id}: identifier not found in CWE {expected_version}") + elif element.get("title") != weaknesses[element_id]: + errors.append( + f"{element_id}: title mismatch; expected " + f"{weaknesses[element_id]!r}, found {element.get('title')!r}" + ) + return errors + + +def atlas_metadata(source: bytes) -> tuple[str | None, str | None, dict[str, str]]: + try: + text = source.decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError(f"ATLAS source is not UTF-8: {exc}") from exc + + format_version = None + content_version = None + tactics: dict[str, str] = {} + in_collection = False + in_tactics = False + current_tactic = None + + for line in text.splitlines(): + if line.startswith("format-version:"): + format_version = line.split(":", 1)[1].strip().strip("'\"") + elif line == "collection:": + in_collection = True + in_tactics = False + elif line == "tactics:": + in_collection = False + in_tactics = True + elif line and not line.startswith(" "): + in_collection = False + in_tactics = False + elif in_collection and line.startswith(" version:"): + content_version = line.split(":", 1)[1].strip().strip("'\"") + elif in_tactics: + tactic_match = re.fullmatch(r" (AML\.TA\d{4}):", line) + if tactic_match: + current_tactic = tactic_match.group(1) + elif current_tactic and line.startswith(" name:"): + tactics[current_tactic] = line.split(":", 1)[1].strip().strip("'\"") + current_tactic = None + + return format_version, content_version, tactics + + +def validate_atlas(mapping: dict, source: bytes) -> list[str]: + errors: list[str] = [] + version_match = re.fullmatch( + r"content v(\d{4}\.\d{2}) \(format-version ([^)]+)\)", + str(mapping.get("version", "")), + ) + if not version_match: + return ["ATLAS version must be 'content vYYYY.MM (format-version X.Y.Z)'"] + + expected_content, expected_format = version_match.groups() + + try: + source_format, source_content, tactics = atlas_metadata(source) + except ValueError as exc: + return errors + [str(exc)] + + if source_format != expected_format: + errors.append( + f"ATLAS format-version mismatch: expected {expected_format}, " + f"found {source_format or 'missing'}" + ) + if source_content != expected_content: + errors.append( + f"ATLAS content version mismatch: expected {expected_content}, " + f"found {source_content or 'missing'}" + ) + + for element in mapping.get("elements", []): + if not isinstance(element, dict): + continue + element_id = element.get("id") + if element_id not in tactics: + errors.append( + f"{element_id}: identifier not found in ATLAS {expected_content} tactics" + ) + elif element.get("title") != tactics[element_id]: + errors.append( + f"{element_id}: title mismatch; expected " + f"{tactics[element_id]!r}, found {element.get('title')!r}" + ) + return errors + + +def validate_catalog(path: Path, source_dir: Path, download: bool) -> list[str]: + try: + mapping = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + return [f"could not read mapping: {exc}"] + if not isinstance(mapping, dict): + return ["mapping root must be an object"] + + errors = validate_common(mapping) + validator = SUPPORTED_CATALOGS[path.name] + location_errors = validate_source_location(validator, mapping) + if location_errors: + return errors + location_errors + try: + expected_sha = parse_expected_sha(mapping.get("source_sha")) + source = read_source(mapping, source_dir, download) + except ValueError as exc: + return errors + [str(exc)] + + actual_sha = sha256(source) + if actual_sha != expected_sha: + return errors + [ + f"source SHA-256 mismatch: expected {expected_sha}, found {actual_sha}" + ] + + if validator == "cwe": + errors.extend(validate_cwe(mapping, source)) + else: + errors.extend(validate_atlas(mapping, source)) + return errors + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--mapping-dir", type=Path, default=DEFAULT_MAPPING_DIR) + parser.add_argument("--source-dir", type=Path, default=DEFAULT_SOURCE_DIR) + parser.add_argument( + "--download", + action="store_true", + help="download a pinned source when it is absent from --source-dir", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + failed = False + for filename in SUPPORTED_CATALOGS: + path = args.mapping_dir / filename + errors = validate_catalog(path, args.source_dir, args.download) + if errors: + failed = True + print(f"FAIL {path}") + for error in errors: + print(f" - {error}") + else: + print(f"PASS {path}") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/workflows/mapping-integrity.yml b/.github/workflows/mapping-integrity.yml new file mode 100644 index 0000000..3149693 --- /dev/null +++ b/.github/workflows/mapping-integrity.yml @@ -0,0 +1,35 @@ +name: Mapping integrity + +on: + pull_request: + paths: + - "2026/final/mappings/**" + - ".github/scripts/validate_mappings.py" + - ".github/scripts/test_validate_mappings.py" + - ".github/workflows/mapping-integrity.yml" + push: + branches: + - main + paths: + - "2026/final/mappings/**" + - ".github/scripts/validate_mappings.py" + - ".github/scripts/test_validate_mappings.py" + - ".github/workflows/mapping-integrity.yml" + workflow_dispatch: + +permissions: + contents: read + +jobs: + validate: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Run offline negative tests + run: python .github/scripts/test_validate_mappings.py + + - name: Validate mapping catalogs + run: python .github/scripts/validate_mappings.py --download diff --git a/.gitignore b/.gitignore index ee1c095..aa094e6 100644 --- a/.gitignore +++ b/.gitignore @@ -5,5 +5,6 @@ 2026/working/polling/forms/create_forms_execution_log.txt __pycache__/ *.pyc +.cache/ .DS_Store dist/ diff --git a/2026/final/mappings/README.md b/2026/final/mappings/README.md new file mode 100644 index 0000000..5e46379 --- /dev/null +++ b/2026/final/mappings/README.md @@ -0,0 +1,38 @@ +# Mapping Integrity Validation + +The mapping integrity validator currently supports the CWE 4.20 and MITRE +ATLAS 2026.06 catalogs. It checks: + +* Duplicate identifiers in a mapping catalog. +* Identifier existence in the corresponding primary dataset. +* Exact title and abstraction metadata for CWE entries. +* Exact tactic names for MITRE ATLAS entries. +* The self-reported source version and format version. +* The SHA-256 digest of every downloaded or cached primary dataset. +* Use of the pinned MITRE ATLAS v6 source instead of the deprecated default + `dist/ATLAS.yaml` file. + +Run the validator locally: + +```bash +python .github/scripts/validate_mappings.py --download +``` + +The first run downloads the version-specific primary datasets into +`.cache/mapping-sources/`. Later runs can validate entirely offline by omitting +`--download`: + +```bash +python .github/scripts/validate_mappings.py +``` + +Run the offline negative test cases separately: + +```bash +python .github/scripts/test_validate_mappings.py +``` + +Each supported mapping records `source_url` and `source_sha`. When a framework +source is intentionally updated, review its contents and version metadata, +update the mapping, and replace `source_sha` with the SHA-256 digest of the +exact source bytes. diff --git a/2026/final/mappings/cwe-4.20.json b/2026/final/mappings/cwe-4.20.json index e5680a2..13d8e7a 100644 --- a/2026/final/mappings/cwe-4.20.json +++ b/2026/final/mappings/cwe-4.20.json @@ -1,7 +1,8 @@ { "framework": "MITRE CWE (Common Weakness Enumeration)", "version": "4.20", - "source_url": "https://cwe.mitre.org/", + "source_url": "https://cwe.mitre.org/data/xml/cwec_v4.20.xml.zip", + "source_sha": "sha256:3976f599e5e5200219a3108bb896d06e2a88fbb293369e1883cb423a5e9d7d50", "accessed": "2026-07-11", "element_type": "weakness", "coarse_level": "pillars+base-weaknesses", diff --git a/2026/final/mappings/mitre-atlas-2026.06.json b/2026/final/mappings/mitre-atlas-2026.06.json index 2bd28c7..c7fd0ca 100644 --- a/2026/final/mappings/mitre-atlas-2026.06.json +++ b/2026/final/mappings/mitre-atlas-2026.06.json @@ -2,6 +2,7 @@ "framework": "MITRE ATLAS", "version": "content v2026.06 (format-version 6.0.0)", "source_url": "https://raw.githubusercontent.com/mitre-atlas/atlas-data/main/dist/v6/ATLAS-2026.06.yaml", + "source_sha": "sha256:b771de8b1489564b2838a709c7429849a9575dbd94073928817fe1a21661e70a", "accessed": "2026-07-11", "element_type": "attack-technique", "coarse_level": "tactics",