diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index cc8bb9b60..d8e59da4c 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -25,10 +25,6 @@ from application.utils import db_backend from application.utils import gap_analysis from application.utils import cres_csv_export -from application.utils.external_project_parsers.parsers import ( - owasp_kubernetes_top10_2022, - owasp_kubernetes_top10_2025, -) if TYPE_CHECKING: from application.prompt_client import prompt_client as prompt_client diff --git a/application/tests/owasp_mapping_fixtures_test.py b/application/tests/owasp_mapping_fixtures_test.py index ee9a38238..3f58eb88f 100644 --- a/application/tests/owasp_mapping_fixtures_test.py +++ b/application/tests/owasp_mapping_fixtures_test.py @@ -1,10 +1,9 @@ -import json import re import unittest -from pathlib import Path + +from application.utils import mapping_fixtures -FIXTURE_DIR = Path(__file__).parent / "fixtures" / "owasp_mappings" EXPECTED_FIXTURES = { "owasp_aisvs_1_0.json", "owasp_api_top10_2023.json", @@ -19,13 +18,55 @@ class TestOwaspMappingFixtures(unittest.TestCase): def test_fixture_set_is_complete(self) -> None: - actual = {path.name for path in FIXTURE_DIR.glob("*.json")} - self.assertEqual(actual, EXPECTED_FIXTURES) + self.assertEqual( + set(mapping_fixtures.list_owasp_mapping_fixtures()), EXPECTED_FIXTURES + ) + + def test_load_all_returns_every_named_fixture(self) -> None: + loaded = mapping_fixtures.load_all_owasp_mapping_fixtures() + self.assertEqual(set(loaded), EXPECTED_FIXTURES) + for filename, payload in loaded.items(): + with self.subTest(fixture=filename): + self.assertIsInstance(payload, list) + self.assertGreater(len(payload), 0) + + def test_load_accepts_stem_without_json_suffix(self) -> None: + by_stem = mapping_fixtures.load_owasp_mapping_fixture("owasp_top10_2025") + by_name = mapping_fixtures.load_owasp_mapping_fixture("owasp_top10_2025.json") + self.assertEqual(by_stem, by_name) + + def test_load_unknown_fixture_raises(self) -> None: + with self.assertRaises(FileNotFoundError) as ctx: + mapping_fixtures.load_owasp_mapping_fixture("not_a_real_mapping") + self.assertIn("not_a_real_mapping.json", str(ctx.exception)) + + def test_k8s_2025_uses_per_item_owasp_hyperlinks(self) -> None: + # Data from #953 (Bornunique911): section pages, not the family homepage. + entries = mapping_fixtures.load_owasp_mapping_fixture( + "owasp_kubernetes_top10_2025" + ) + self.assertEqual(10, len(entries)) + self.assertEqual("K01", entries[0]["section_id"]) + self.assertIn("/2025/en/src/K01-", entries[0]["hyperlink"]) + self.assertEqual(["233-748", "486-813"], entries[0]["cre_ids"]) + + def test_resolved_cre_ids_follow_fallback_when_own_ids_empty(self) -> None: + by_id = { + "K05": {"cre_ids": ["148-420"]}, + "K10": { + "cre_ids": [], + "fallback_section_ids": ["K05"], + }, + } + self.assertEqual( + ["148-420"], + mapping_fixtures.resolved_cre_ids(by_id["K10"], by_id), + ) def test_fixtures_have_expected_mapping_shape(self) -> None: - for path in sorted(FIXTURE_DIR.glob("*.json")): - with self.subTest(fixture=path.name): - payload = json.loads(path.read_text(encoding="utf-8")) + loaded = mapping_fixtures.load_all_owasp_mapping_fixtures() + for filename, payload in loaded.items(): + with self.subTest(fixture=filename): self.assertIsInstance(payload, list) self.assertGreater(len(payload), 0) @@ -53,7 +94,7 @@ def test_fixtures_have_expected_mapping_shape(self) -> None: self.assertNotIn( entry["section_id"], seen_section_ids, - msg=f"Duplicate section_id {entry['section_id']} in {path.name}", + msg=f"Duplicate section_id {entry['section_id']} in {filename}", ) seen_section_ids.add(entry["section_id"]) @@ -68,7 +109,7 @@ def test_fixtures_have_expected_mapping_shape(self) -> None: known_section_ids, msg=( f"Fallback section id {fallback_section_id} " - f"in {path.name} is not a known section_id" + f"in {filename} is not a known section_id" ), ) diff --git a/application/utils/external_project_parsers/parsers/cheatsheets_parser.py b/application/utils/external_project_parsers/parsers/cheatsheets_parser.py index 15fdab76d..a6600f68e 100644 --- a/application/utils/external_project_parsers/parsers/cheatsheets_parser.py +++ b/application/utils/external_project_parsers/parsers/cheatsheets_parser.py @@ -8,8 +8,8 @@ import re from application.utils.external_project_parsers import base_parser_defs import json -from pathlib import Path import logging +from application.utils.mapping_fixtures import OWASP_MAPPING_FIXTURE_DIR from application.utils.external_project_parsers.base_parser_defs import ( ParserInterface, ParseResult, @@ -21,11 +21,7 @@ class Cheatsheets(ParserInterface): name = "OWASP Cheat Sheets" cheatsheetseries_base_url = "https://cheatsheetseries.owasp.org/cheatsheets" supplement_data_file = ( - Path(__file__).resolve().parents[3] - / "tests" - / "fixtures" - / "owasp_mappings" - / "owasp_cheatsheets_supplement.json" + OWASP_MAPPING_FIXTURE_DIR / "owasp_cheatsheets_supplement.json" ) logger = logging.getLogger(__name__) diff --git a/application/utils/external_project_parsers/parsers/owasp_aisvs.py b/application/utils/external_project_parsers/parsers/owasp_aisvs.py index d32595666..cb2269948 100644 --- a/application/utils/external_project_parsers/parsers/owasp_aisvs.py +++ b/application/utils/external_project_parsers/parsers/owasp_aisvs.py @@ -1,51 +1,8 @@ -import json -from pathlib import Path - -from application.database import db -from application.defs import cre_defs as defs -from application.prompt_client import prompt_client -from application.utils.external_project_parsers.base_parser_defs import ( - ParseResult, - ParserInterface, +from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import ( + OwaspMappingFixtureParser, ) -class OwaspAisvs(ParserInterface): +class OwaspAisvs(OwaspMappingFixtureParser): name = "OWASP AI Security Verification Standard (AISVS)" - data_file = ( - Path(__file__).resolve().parents[3] - / "tests" - / "fixtures" - / "owasp_mappings" - / "owasp_aisvs_1_0.json" - ) - - def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler): - with self.data_file.open("r", encoding="utf-8") as handle: - raw_entries = json.load(handle) - - entries = [] - for entry in raw_entries: - standard = defs.Standard( - name=self.name, - sectionID=entry["section_id"], - section=entry["section"], - hyperlink=entry["hyperlink"], - ) - for cre_id in entry.get("cre_ids", []): - cres = cache.get_CREs(external_id=cre_id) - if not cres: - continue - standard.add_link( - defs.Link( - ltype=defs.LinkTypes.LinkedTo, - document=cres[0].shallow_copy(), - ) - ) - entries.append(standard) - - return ParseResult( - results={self.name: entries}, - calculate_gap_analysis=False, - calculate_embeddings=False, - ) + fixture_name = "owasp_aisvs_1_0" diff --git a/application/utils/external_project_parsers/parsers/owasp_api_top10_2023.py b/application/utils/external_project_parsers/parsers/owasp_api_top10_2023.py index 18c8310b2..224dc01c9 100644 --- a/application/utils/external_project_parsers/parsers/owasp_api_top10_2023.py +++ b/application/utils/external_project_parsers/parsers/owasp_api_top10_2023.py @@ -1,51 +1,8 @@ -import json -from pathlib import Path - -from application.database import db -from application.defs import cre_defs as defs -from application.prompt_client import prompt_client -from application.utils.external_project_parsers.base_parser_defs import ( - ParseResult, - ParserInterface, +from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import ( + OwaspMappingFixtureParser, ) -class OwaspApiTop10_2023(ParserInterface): +class OwaspApiTop10_2023(OwaspMappingFixtureParser): name = "OWASP API Security Top 10 2023" - data_file = ( - Path(__file__).resolve().parents[3] - / "tests" - / "fixtures" - / "owasp_mappings" - / "owasp_api_top10_2023.json" - ) - - def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler): - with self.data_file.open("r", encoding="utf-8") as handle: - raw_entries = json.load(handle) - - entries = [] - for entry in raw_entries: - standard = defs.Standard( - name=self.name, - sectionID=entry["section_id"], - section=entry["section"], - hyperlink=entry["hyperlink"], - ) - for cre_id in entry.get("cre_ids", []): - cres = cache.get_CREs(external_id=cre_id) - if not cres: - continue - standard.add_link( - defs.Link( - ltype=defs.LinkTypes.LinkedTo, - document=cres[0].shallow_copy(), - ) - ) - entries.append(standard) - - return ParseResult( - results={self.name: entries}, - calculate_gap_analysis=False, - calculate_embeddings=False, - ) + fixture_name = "owasp_api_top10_2023" diff --git a/application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2022.py b/application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2022.py index cfd3f3894..239f44a0e 100644 --- a/application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2022.py +++ b/application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2022.py @@ -1,51 +1,8 @@ -import json -from pathlib import Path - -from application.database import db -from application.defs import cre_defs as defs -from application.prompt_client import prompt_client -from application.utils.external_project_parsers.base_parser_defs import ( - ParseResult, - ParserInterface, +from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import ( + OwaspMappingFixtureParser, ) -class OwaspKubernetesTop10_2022(ParserInterface): +class OwaspKubernetesTop10_2022(OwaspMappingFixtureParser): name = "OWASP Kubernetes Top Ten 2022" - data_file = ( - Path(__file__).resolve().parents[3] - / "tests" - / "fixtures" - / "owasp_mappings" - / "owasp_kubernetes_top10_2022.json" - ) - - def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler): - with self.data_file.open("r", encoding="utf-8") as handle: - raw_entries = json.load(handle) - - entries = [] - for entry in raw_entries: - standard = defs.Standard( - name=self.name, - sectionID=entry["section_id"], - section=entry["section"], - hyperlink=entry["hyperlink"], - ) - for cre_id in entry.get("cre_ids", []): - cres = cache.get_CREs(external_id=cre_id) - if not cres: - continue - standard.add_link( - defs.Link( - ltype=defs.LinkTypes.LinkedTo, - document=cres[0].shallow_copy(), - ) - ) - entries.append(standard) - - return ParseResult( - results={self.name: entries}, - calculate_gap_analysis=False, - calculate_embeddings=False, - ) + fixture_name = "owasp_kubernetes_top10_2022" diff --git a/application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2025.py b/application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2025.py index cd3a09336..ec64b3853 100644 --- a/application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2025.py +++ b/application/utils/external_project_parsers/parsers/owasp_kubernetes_top10_2025.py @@ -1,82 +1,9 @@ -import json -from pathlib import Path - -from application.database import db -from application.defs import cre_defs as defs -from application.prompt_client import prompt_client -from application.utils.external_project_parsers.base_parser_defs import ( - ParseResult, - ParserInterface, +from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import ( + OwaspMappingFixtureParser, ) -class OwaspKubernetesTop10_2025(ParserInterface): +class OwaspKubernetesTop10_2025(OwaspMappingFixtureParser): name = "OWASP Kubernetes Top Ten 2025 (Draft)" - data_file = ( - Path(__file__).resolve().parents[3] - / "tests" - / "fixtures" - / "owasp_mappings" - / "owasp_kubernetes_top10_2025.json" - ) - fallback_data_file = ( - Path(__file__).resolve().parents[3] - / "tests" - / "fixtures" - / "owasp_mappings" - / "owasp_kubernetes_top10_2022.json" - ) - - def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler): - with self.data_file.open("r", encoding="utf-8") as handle: - raw_entries = json.load(handle) - with self.fallback_data_file.open("r", encoding="utf-8") as handle: - fallback_entries = { - entry["section_id"]: entry for entry in json.load(handle) - } - - entries = [] - for entry in raw_entries: - standard = defs.Standard( - name=self.name, - sectionID=entry["section_id"], - section=entry["section"], - hyperlink=entry["hyperlink"], - ) - linked_cre_ids = [] - for cre_id in entry.get("cre_ids", []): - cres = cache.get_CREs(external_id=cre_id) - if not cres: - continue - linked_cre_ids.append(cre_id) - standard.add_link( - defs.Link( - ltype=defs.LinkTypes.LinkedTo, - document=cres[0].shallow_copy(), - ) - ) - if not linked_cre_ids: - for section_id in entry.get("fallback_section_ids", []): - fallback_entry = fallback_entries.get(section_id) - if not fallback_entry: - continue - for cre_id in fallback_entry.get("cre_ids", []): - if cre_id in linked_cre_ids: - continue - cres = cache.get_CREs(external_id=cre_id) - if not cres: - continue - linked_cre_ids.append(cre_id) - standard.add_link( - defs.Link( - ltype=defs.LinkTypes.LinkedTo, - document=cres[0].shallow_copy(), - ) - ) - entries.append(standard) - - return ParseResult( - results={self.name: entries}, - calculate_gap_analysis=False, - calculate_embeddings=False, - ) + fixture_name = "owasp_kubernetes_top10_2025" + fallback_fixture_name = "owasp_kubernetes_top10_2022" diff --git a/application/utils/external_project_parsers/parsers/owasp_llm_top10_2025.py b/application/utils/external_project_parsers/parsers/owasp_llm_top10_2025.py index 233596cd6..3c2426213 100644 --- a/application/utils/external_project_parsers/parsers/owasp_llm_top10_2025.py +++ b/application/utils/external_project_parsers/parsers/owasp_llm_top10_2025.py @@ -1,51 +1,8 @@ -import json -from pathlib import Path - -from application.database import db -from application.defs import cre_defs as defs -from application.prompt_client import prompt_client -from application.utils.external_project_parsers.base_parser_defs import ( - ParseResult, - ParserInterface, +from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import ( + OwaspMappingFixtureParser, ) -class OwaspLlmTop10_2025(ParserInterface): +class OwaspLlmTop10_2025(OwaspMappingFixtureParser): name = "OWASP Top 10 for LLM and Gen AI Apps 2025" - data_file = ( - Path(__file__).resolve().parents[3] - / "tests" - / "fixtures" - / "owasp_mappings" - / "owasp_llm_top10_2025.json" - ) - - def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler): - with self.data_file.open("r", encoding="utf-8") as handle: - raw_entries = json.load(handle) - - entries = [] - for entry in raw_entries: - standard = defs.Standard( - name=self.name, - sectionID=entry["section_id"], - section=entry["section"], - hyperlink=entry["hyperlink"], - ) - for cre_id in entry.get("cre_ids", []): - cres = cache.get_CREs(external_id=cre_id) - if not cres: - continue - standard.add_link( - defs.Link( - ltype=defs.LinkTypes.LinkedTo, - document=cres[0].shallow_copy(), - ) - ) - entries.append(standard) - - return ParseResult( - results={self.name: entries}, - calculate_gap_analysis=False, - calculate_embeddings=False, - ) + fixture_name = "owasp_llm_top10_2025" diff --git a/application/utils/external_project_parsers/parsers/owasp_mapping_fixture_parser.py b/application/utils/external_project_parsers/parsers/owasp_mapping_fixture_parser.py new file mode 100644 index 000000000..9f62f34a1 --- /dev/null +++ b/application/utils/external_project_parsers/parsers/owasp_mapping_fixture_parser.py @@ -0,0 +1,54 @@ +"""Opt-in parsers that hydrate Standards from OWASP mapping *fixtures*. + +These are eval/test gold, not import-all catalog sources. Other projects should +link to OpenCRE; we parse at the source. The exception is GSoC/OIE validation. +""" + +from __future__ import annotations + +from pathlib import Path + +from typing import Optional + +from application.database import db +from application.prompt_client import prompt_client +from application.utils.external_project_parsers.base_parser_defs import ( + ParseResult, + ParserInterface, +) +from application.utils.mapping_fixtures import ( + linked_standards_from_mapping, + load_mapping_entries, +) + + +class OwaspMappingFixtureParser(ParserInterface): + fixture_name: str + fallback_fixture_name: str | None = None + data_file: Path | None = None + fallback_data_file: Path | None = None + + def parse( + self, + database: db.Node_collection, + prompt_client: Optional[prompt_client.PromptHandler], + ) -> ParseResult: + del prompt_client + entries = load_mapping_entries(self.fixture_name, self.data_file) + fallback_entries = None + if self.fallback_fixture_name or self.fallback_data_file: + fallback_entries = load_mapping_entries( + self.fallback_fixture_name or "", + self.fallback_data_file, + ) + documents = linked_standards_from_mapping( + database, + self.name, + entries, + fallback_entries=fallback_entries, + ) + return ParseResult( + results={self.name: documents}, + calculate_gap_analysis=False, + calculate_embeddings=False, + ) diff --git a/application/utils/external_project_parsers/parsers/owasp_top10_2025.py b/application/utils/external_project_parsers/parsers/owasp_top10_2025.py index 51b59262f..797f4ab54 100644 --- a/application/utils/external_project_parsers/parsers/owasp_top10_2025.py +++ b/application/utils/external_project_parsers/parsers/owasp_top10_2025.py @@ -1,51 +1,8 @@ -import json -from pathlib import Path - -from application.database import db -from application.defs import cre_defs as defs -from application.prompt_client import prompt_client -from application.utils.external_project_parsers.base_parser_defs import ( - ParseResult, - ParserInterface, +from application.utils.external_project_parsers.parsers.owasp_mapping_fixture_parser import ( + OwaspMappingFixtureParser, ) -class OwaspTop10_2025(ParserInterface): +class OwaspTop10_2025(OwaspMappingFixtureParser): name = "OWASP Top 10 2025" - data_file = ( - Path(__file__).resolve().parents[3] - / "tests" - / "fixtures" - / "owasp_mappings" - / "owasp_top10_2025.json" - ) - - def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler): - with self.data_file.open("r", encoding="utf-8") as handle: - raw_entries = json.load(handle) - - entries = [] - for entry in raw_entries: - standard = defs.Standard( - name=self.name, - sectionID=entry["section_id"], - section=entry["section"], - hyperlink=entry["hyperlink"], - ) - for cre_id in entry.get("cre_ids", []): - cres = cache.get_CREs(external_id=cre_id) - if not cres: - continue - standard.add_link( - defs.Link( - ltype=defs.LinkTypes.LinkedTo, - document=cres[0].shallow_copy(), - ) - ) - entries.append(standard) - - return ParseResult( - results={self.name: entries}, - calculate_gap_analysis=False, - calculate_embeddings=False, - ) + fixture_name = "owasp_top10_2025" diff --git a/application/utils/mapping_fixtures.py b/application/utils/mapping_fixtures.py new file mode 100644 index 000000000..e9a676b4d --- /dev/null +++ b/application/utils/mapping_fixtures.py @@ -0,0 +1,170 @@ +"""Load OWASP mapping JSON from ``application/tests/fixtures/owasp_mappings``. + +These files are **eval / unit-test gold**, not a production catalog. Other +projects should link to OpenCRE so we can parse at the source; shipping +hand-maintained CRE mappings as importers would make OpenCRE another +spreadsheet. The exception is GSoC/OIE pipeline validation. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List, Mapping, Sequence + +OWASP_MAPPING_FIXTURE_DIR = ( + Path(__file__).resolve().parents[1] / "tests" / "fixtures" / "owasp_mappings" +) + +MappingEntry = Dict[str, Any] + + +def list_owasp_mapping_fixtures() -> List[str]: + """Return sorted JSON filenames in the mapping fixture directory.""" + return sorted(path.name for path in OWASP_MAPPING_FIXTURE_DIR.glob("*.json")) + + +def load_owasp_mapping_fixture(name: str) -> List[MappingEntry]: + """Load one mapping fixture by filename or stem. + + ``name`` may be ``owasp_aisvs_1_0`` or ``owasp_aisvs_1_0.json``. + """ + filename = name if name.endswith(".json") else f"{name}.json" + path = OWASP_MAPPING_FIXTURE_DIR / filename + if not path.is_file(): + known = ", ".join(list_owasp_mapping_fixtures()) or "(none)" + raise FileNotFoundError( + f"Unknown OWASP mapping fixture {filename!r}. Known: {known}" + ) + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, list): + raise ValueError(f"Fixture {filename} must be a JSON list, got {type(payload)}") + return payload + + +def load_all_owasp_mapping_fixtures() -> Dict[str, List[MappingEntry]]: + """Load every ``*.json`` mapping fixture, keyed by filename.""" + return { + filename: load_owasp_mapping_fixture(filename) + for filename in list_owasp_mapping_fixtures() + } + + +def load_mapping_entries( + name: str, override_path: Path | None = None +) -> List[MappingEntry]: + """Load mapping rows from the named fixture, or from ``override_path`` (tests).""" + if override_path is not None: + payload = json.loads(Path(override_path).read_text(encoding="utf-8")) + if not isinstance(payload, list): + raise ValueError( + f"{override_path} must be a JSON list, got {type(payload)}" + ) + return payload + return load_owasp_mapping_fixture(name) + + +def index_by_section_id( + entries: Sequence[Mapping[str, Any]], +) -> Dict[str, Mapping[str, Any]]: + return { + entry["section_id"]: entry + for entry in entries + if isinstance(entry.get("section_id"), str) + } + + +def cre_ids_in_cache( + cache: Any, + entry: Mapping[str, Any], + fallback_by_section_id: Mapping[str, Mapping[str, Any]] | None = None, +) -> List[str]: + """CRE ids that exist in ``cache``, then ``fallback_section_ids`` if none match.""" + found: List[str] = [] + for cre_id in entry.get("cre_ids") or []: + if isinstance(cre_id, str) and cache.get_CREs(external_id=cre_id): + found.append(cre_id) + if found: + return found + if not fallback_by_section_id: + return [] + seen: set[str] = set() + resolved: List[str] = [] + for section_id in entry.get("fallback_section_ids") or []: + if not isinstance(section_id, str): + continue + fallback_entry = fallback_by_section_id.get(section_id) + if not fallback_entry: + continue + for cre_id in fallback_entry.get("cre_ids") or []: + if ( + isinstance(cre_id, str) + and cre_id not in seen + and cache.get_CREs(external_id=cre_id) + ): + seen.add(cre_id) + resolved.append(cre_id) + return resolved + + +def linked_standards_from_mapping( + cache: Any, + standard_name: str, + entries: Sequence[Mapping[str, Any]], + *, + fallback_entries: Sequence[Mapping[str, Any]] | None = None, +) -> List[Any]: + """Build ``Standard`` documents with LinkedTo CREs from fixture rows.""" + from application.defs import cre_defs as defs + + fallback_index = index_by_section_id(fallback_entries or []) + documents: List[Any] = [] + for entry in entries: + standard = defs.Standard( + name=standard_name, + sectionID=str(entry.get("section_id") or ""), + section=str(entry.get("section") or ""), + hyperlink=str(entry.get("hyperlink") or ""), + ) + for cre_id in cre_ids_in_cache(cache, entry, fallback_index): + cres = cache.get_CREs(external_id=cre_id) + if not cres: + continue + standard.add_link( + defs.Link( + ltype=defs.LinkTypes.LinkedTo, + document=cres[0].shallow_copy(), + ) + ) + documents.append(standard) + return documents + + +def resolved_cre_ids( + entry: Mapping[str, Any], + by_section_id: Mapping[str, Mapping[str, Any]] | None = None, +) -> List[str]: + """Return ``cre_ids`` for an entry, optionally following ``fallback_section_ids``. + + Fallback is used when the entry's own ``cre_ids`` are empty. Callers that + also need "CRE missing from cache" behavior should filter after lookup. + """ + own = [cre_id for cre_id in entry.get("cre_ids") or [] if isinstance(cre_id, str)] + if own: + return own + if not by_section_id: + return [] + fallbacks: Sequence[Any] = entry.get("fallback_section_ids") or [] + resolved: List[str] = [] + seen: set[str] = set() + for section_id in fallbacks: + if not isinstance(section_id, str): + continue + fallback_entry = by_section_id.get(section_id) + if not fallback_entry: + continue + for cre_id in fallback_entry.get("cre_ids") or []: + if isinstance(cre_id, str) and cre_id not in seen: + seen.add(cre_id) + resolved.append(cre_id) + return resolved diff --git a/cre.py b/cre.py index b6a89a57f..63d8d7cce 100644 --- a/cre.py +++ b/cre.py @@ -171,32 +171,32 @@ def main() -> None: parser.add_argument( "--owasp_kubernetes_top10_2022_in", action="store_true", - help="import OWASP Kubernetes Top Ten 2022", + help="load OWASP Kubernetes Top Ten 2022 mapping fixture (eval gold, not import-all)", ) parser.add_argument( "--owasp_kubernetes_top10_2025_in", action="store_true", - help="import OWASP Kubernetes Top Ten 2025 draft", + help="load OWASP Kubernetes Top Ten 2025 mapping fixture (eval gold, not import-all)", ) parser.add_argument( "--owasp_top10_2025_in", action="store_true", - help="import OWASP Top 10 2025", + help="load OWASP Top 10 2025 mapping fixture (eval gold, not import-all)", ) parser.add_argument( "--owasp_api_top10_2023_in", action="store_true", - help="import OWASP API Security Top 10 2023", + help="load OWASP API Security Top 10 2023 mapping fixture (eval gold, not import-all)", ) parser.add_argument( "--owasp_llm_top10_2025_in", action="store_true", - help="import OWASP Top 10 for LLM and Gen AI Apps 2025", + help="load OWASP LLM Top 10 2025 mapping fixture (eval gold, not import-all)", ) parser.add_argument( "--owasp_aisvs_in", action="store_true", - help="import OWASP AI Security Verification Standard (AISVS)", + help="load OWASP AISVS mapping fixture (eval gold, not import-all)", ) parser.add_argument( "--pci_dss_3_2_in",