From 52da585f37da5ab6d4962a48c08c966b8bc8a4e3 Mon Sep 17 00:00:00 2001 From: skypank Date: Wed, 19 Aug 2026 02:04:28 +0530 Subject: [PATCH 1/2] feat: add suggestions_to_parse_result import adapter (Workstream F, F3) --- .../tests/cheatsheets_workstream_f_test.py | 153 +++++++++++++++++- .../parsers/cheatsheets_workstream_f.py | 114 +++++++++++-- 2 files changed, 254 insertions(+), 13 deletions(-) diff --git a/application/tests/cheatsheets_workstream_f_test.py b/application/tests/cheatsheets_workstream_f_test.py index d7f678a66..b1e6012e4 100644 --- a/application/tests/cheatsheets_workstream_f_test.py +++ b/application/tests/cheatsheets_workstream_f_test.py @@ -1,12 +1,14 @@ -"""Tests for Cheat Sheet -> CRE mapping, Workstream F, checkpoints F1+F2. +"""Tests for Cheat Sheet -> CRE mapping, Workstream F, checkpoints F1+F2+F3. F1: the suggestions.json data contract -- SUGGESTIONS_SCHEMA plus the CandidateCRE / MappingSuggestion dataclasses. F2: the read/write adapters -- write_suggestions_json and load_approved_suggestions (schema-validate -> parse -> filter approved). +F3: the import adapter -- suggestions_to_parse_result (approved suggestions -> +ParseResult of defs.Standard with AutomaticallyLinkedTo links). F3's tests use a +lightweight Node_collection stub (get_CREs only) so no Postgres/Neo4j is needed. -F3 (suggestions_to_parse_result) and F4/F5 (CLI) are deliberately out of scope -here; these tests never touch defs.Standard / ParseResult / Node_collection. +F4/F5 (CLI) are deliberately out of scope here. """ import json @@ -16,6 +18,8 @@ import jsonschema +from application.defs import cre_defs as defs +from application.utils.external_project_parsers import base_parser_defs from application.utils.external_project_parsers.parsers import ( cheatsheets_workstream_f as wf, ) @@ -122,5 +126,148 @@ def test_written_file_validates_against_schema(self) -> None: jsonschema.validate(instance=doc, schema=wf.SUGGESTIONS_SCHEMA) +class _StubCache: + """Minimal ``db.Node_collection`` stand-in exposing only ``get_CREs``. + + Known ids resolve to a ``defs.CRE`` (shaped like a real ``external_id`` -- + e.g. ``"764-507"`` -- exactly what the live cheatsheets parser passes); + unknown ids return ``[]`` just like ``db.Node_collection.get_CREs``. No + Postgres/Neo4j is touched. + """ + + def __init__(self, known): + # known: dict of cre_id -> defs.CRE + self._known = known + + def get_CREs(self, external_id=None, **kwargs): + cre = self._known.get(external_id) + # Return a fresh copy per call, mirroring the DB (which hydrates anew). + return [cre.shallow_copy()] if cre is not None else [] + + +def _cre(external_id): + return defs.CRE(id=external_id, name=f"CRE {external_id}") + + +def _cand(cre_id): + return wf.CandidateCRE(cre_id=cre_id, score=0.9, confidence="high", reason="r") + + +def _sugg(title, candidate_ids, category="authentication", status="approved"): + slug = title.replace(" ", "_") + return wf.MappingSuggestion( + source="owasp_cheatsheets", + cheatsheet_id=slug, + title=title, + hyperlink=f"https://cheatsheetseries.owasp.org/cheatsheets/{slug}.html", + category=category, + status=status, + candidate_cres=[_cand(c) for c in candidate_ids], + ) + + +class TestSuggestionsToParseResult(unittest.TestCase): + def test_builds_standards_with_links(self) -> None: + cache = _StubCache({"764-507": _cre("764-507")}) + approved = [_sugg("Authentication Cheat Sheet", ["764-507"])] + + result = wf.suggestions_to_parse_result(approved, cache) + + self.assertIsInstance(result, base_parser_defs.ParseResult) + standards = result.results["OWASP Cheat Sheets"] + self.assertEqual(len(standards), 1) + std = standards[0] + self.assertIsInstance(std, defs.Standard) + self.assertEqual(std.name, "OWASP Cheat Sheets") + self.assertEqual(std.section, "Authentication Cheat Sheet") + self.assertEqual( + std.hyperlink, + "https://cheatsheetseries.owasp.org/cheatsheets/" + "Authentication_Cheat_Sheet.html", + ) + self.assertEqual(len(std.links), 1) + link = std.links[0] + self.assertEqual(link.ltype, defs.LinkTypes.AutomaticallyLinkedTo) + self.assertEqual(link.document.id, "764-507") + + def test_classification_tags_present_and_valid(self) -> None: + cache = _StubCache({"764-507": _cre("764-507")}) + approved = [_sugg("Authentication Cheat Sheet", ["764-507"])] + + result = wf.suggestions_to_parse_result(approved, cache) + + # The import pipeline requires the full classification tag set. + base_parser_defs.validate_classification_tags(result.results) + std = result.results["OWASP Cheat Sheets"][0] + self.assertIn(base_parser_defs.Family.GUIDANCE.value, std.tags) + self.assertIn(base_parser_defs.Subtype.CHEATSHEET.value, std.tags) + self.assertIn(base_parser_defs.Audience.DEVELOPER.value, std.tags) + self.assertIn(base_parser_defs.Maturity.STABLE.value, std.tags) + self.assertIn("source:owasp_cheatsheets", std.tags) + # category is carried as an extra tag. + self.assertIn("authentication", std.tags) + + def test_unknown_cre_id_skipped_sibling_still_linked(self) -> None: + cache = _StubCache({"764-507": _cre("764-507")}) + # One known sibling, one unknown id on the same suggestion. + approved = [_sugg("Authentication Cheat Sheet", ["764-507", "999-999"])] + + with self.assertLogs(wf.logger, level="WARNING") as cm: + result = wf.suggestions_to_parse_result(approved, cache) + + standards = result.results["OWASP Cheat Sheets"] + self.assertEqual(len(standards), 1) + self.assertEqual([l.document.id for l in standards[0].links], ["764-507"]) + # The skipped id is reported for the reviewer. + self.assertIn("999-999", "\n".join(cm.output)) + + def test_all_unknown_candidates_drops_standard(self) -> None: + cache = _StubCache({"764-507": _cre("764-507")}) + approved = [_sugg("Ghost Cheat Sheet", ["111-111", "222-222"])] + + result = wf.suggestions_to_parse_result(approved, cache) + + # A Standard with zero resolved links is omitted entirely. + self.assertEqual(result.results["OWASP Cheat Sheets"], []) + + def test_deterministic_output(self) -> None: + cache = _StubCache({"764-507": _cre("764-507"), "581-525": _cre("581-525")}) + approved = [ + _sugg("Authentication Cheat Sheet", ["764-507"]), + _sugg("Cryptographic Storage Cheat Sheet", ["581-525", "764-507"]), + ] + + first = wf.suggestions_to_parse_result(approved, cache) + second = wf.suggestions_to_parse_result(approved, cache) + + self.assertEqual(first.results, second.results) + + def test_duplicate_candidate_ids_yield_single_link(self) -> None: + # Two candidates on one suggestion resolving to the SAME cre_id must not + # raise DuplicateLinkException -- the has_link guard collapses them. + cache = _StubCache({"764-507": _cre("764-507")}) + approved = [_sugg("Authentication Cheat Sheet", ["764-507", "764-507"])] + + result = wf.suggestions_to_parse_result(approved, cache) + + std = result.results["OWASP Cheat Sheets"][0] + self.assertEqual(len(std.links), 1) + self.assertEqual(std.links[0].document.id, "764-507") + + def test_empty_category_produces_no_blank_tag(self) -> None: + cache = _StubCache({"764-507": _cre("764-507")}) + approved = [_sugg("Authentication Cheat Sheet", ["764-507"], category="")] + + result = wf.suggestions_to_parse_result(approved, cache) + + # Still a valid, classifiable Standard... + base_parser_defs.validate_classification_tags(result.results) + std = result.results["OWASP Cheat Sheets"][0] + # ...with no empty-string tag leaked in from the blank category. + self.assertNotIn("", std.tags) + # Only the five required classification tags remain (no extra). + self.assertEqual(len(std.tags), 5) + + if __name__ == "__main__": unittest.main() diff --git a/application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py b/application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py index b898b4ee0..2304fd0b5 100644 --- a/application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py +++ b/application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py @@ -6,28 +6,29 @@ in a later checkpoint, the conversion of approved suggestions into the import pipeline's ``ParseResult``. -Checkpoints implemented here (F1 + F2): +Checkpoints implemented here (F1 + F2 + F3): * F1 -- the data contract: :data:`SUGGESTIONS_SCHEMA` (JSON Schema) plus the :class:`CandidateCRE` / :class:`MappingSuggestion` dataclasses. * F2 -- the read/write adapters: :func:`write_suggestions_json` and :func:`load_approved_suggestions` (schema-validate -> parse -> filter to the reviewer-approved entries). - -Deliberately NOT in this module yet: - -* F3 -- ``suggestions_to_parse_result(approved, cache)``. When it lands it must - reconcile the review artifact with the *real* import types (real code wins): +* F3 -- the import adapter: :func:`suggestions_to_parse_result`, which reconciles + the review artifact with the *real* import types (real code wins): - fixed ``defs.Standard(name="OWASP Cheat Sheets")``; the suggestion ``title`` maps to ``Standard.section`` (matching the existing cheatsheets_parser); - the import pipeline REQUIRES ``family:/subtype:/audience:/maturity:/source:`` classification tags (``base_parser_defs.validate_classification_tags`` raises otherwise) -- built via ``build_tags`` -- even though the RFC's - §4 contract omits them; + §4 contract omits them; ``category`` rides along as an extra tag when set; - candidate CREs are resolved via ``cache.get_CREs(external_id=...)`` and - unknown ids are skipped; a Standard with zero resolved links is dropped; - - the advisory review fields dropped below (see the schema) have no home on + unknown ids are skipped (collected + logged); a Standard with zero resolved + links is dropped; + - the advisory review fields (see the schema) have no home on ``defs.Standard`` and are intentionally not persisted. + +Deliberately NOT in this module yet: + * F4/F5 -- the ``generate/validate/convert`` CLI. The advisory fields ``score``, ``confidence``, ``reason`` (per candidate) and @@ -45,11 +46,21 @@ import dataclasses import json +import logging from dataclasses import dataclass, field -from typing import Any, Dict, List +from typing import Any, Dict, List, TYPE_CHECKING import jsonschema +from application.defs import cre_defs as defs +from application.utils.external_project_parsers import base_parser_defs +from application.utils.external_project_parsers.base_parser_defs import ParseResult + +if TYPE_CHECKING: + from application.database import db + +logger = logging.getLogger(__name__) + # --- valid values ----------------------------------------------------------- #: Reviewer lifecycle for a suggestion. Only ``approved`` items are converted. @@ -232,3 +243,86 @@ def load_approved_suggestions(path: str) -> List[MappingSuggestion]: doc = json.load(fh) _validate(doc) return [_parse_suggestion(raw) for raw in doc if raw["status"] == "approved"] + + +# --- F3: import adapter ----------------------------------------------------- + +#: Fixed Standard name for every imported cheat sheet, matching the live +#: ``cheatsheets_parser`` so entries dedupe/merge against the same resource. +STANDARD_NAME = "OWASP Cheat Sheets" + + +def suggestions_to_parse_result( + approved: List[MappingSuggestion], + cache: "db.Node_collection", +) -> ParseResult: + """Convert approved suggestions into a ``ParseResult`` the import flow accepts. + + Per approved suggestion, builds a ``defs.Standard`` (fixed name + ``"OWASP Cheat Sheets"``; ``title`` -> ``section``; ``hyperlink``; the + required classification tags via ``build_tags`` with ``category`` as an extra + tag when non-empty). Each candidate ``cre_id`` is resolved via + ``cache.get_CREs(external_id=...)``; unknown ids are collected, logged, and + skipped. Resolved CREs are attached as ``AutomaticallyLinkedTo`` links (on a + ``shallow_copy`` so the CRE's own links don't leak in, and guarded by + ``has_link`` so a repeated id can't raise ``DuplicateLinkException``). A + Standard with zero resolved links is dropped. + + The advisory review fields (``score`` / ``confidence`` / ``reason`` / + ``cheatsheet_id``) have no home on ``defs.Standard`` and are not persisted. + """ + standards: List[defs.Document] = [] + unknown_cre_ids: List[str] = [] + + for suggestion in approved: + extra = [suggestion.category] if suggestion.category else [] + standard = defs.Standard( + name=STANDARD_NAME, + section=suggestion.title, + hyperlink=suggestion.hyperlink, + tags=base_parser_defs.build_tags( + family=base_parser_defs.Family.GUIDANCE, + subtype=base_parser_defs.Subtype.CHEATSHEET, + audience=base_parser_defs.Audience.DEVELOPER, + maturity=base_parser_defs.Maturity.STABLE, + source="owasp_cheatsheets", + extra=extra, + ), + ) + + for candidate in suggestion.candidate_cres: + cres = cache.get_CREs(external_id=candidate.cre_id) + if not cres: + unknown_cre_ids.append(candidate.cre_id) + logger.warning( + "Skipping unknown CRE id %s for cheat sheet %r; not in cache.", + candidate.cre_id, + suggestion.title, + ) + continue + for cre in cres: + link = defs.Link( + document=cre.shallow_copy(), + ltype=defs.LinkTypes.AutomaticallyLinkedTo, + ) + # Guard against a repeated cre_id (add_link would otherwise raise + # DuplicateLinkException on the second occurrence). + if not standard.has_link(link): + standard.add_link(link) + + if standard.links: + standards.append(standard) + else: + logger.info( + "Dropping cheat sheet %r: no candidate CREs resolved.", + suggestion.title, + ) + + if unknown_cre_ids: + logger.warning( + "suggestions_to_parse_result skipped %d unknown CRE id(s): %s", + len(unknown_cre_ids), + ", ".join(unknown_cre_ids), + ) + + return ParseResult(results={STANDARD_NAME: standards}) From d7f26744c24597f02ad15d9065771a73faf4a936 Mon Sep 17 00:00:00 2001 From: skypank Date: Wed, 19 Aug 2026 02:17:07 +0530 Subject: [PATCH 2/2] F3: suggestions_to_parse_result import adapter Convert approved MappingSuggestions into a ParseResult of defs.Standard (name='OWASP Cheat Sheets', title->section, hyperlink, classification tags via build_tags with category as an extra tag). Resolve candidate cre_ids via cache.get_CREs; skip+log unknown ids; link resolved CREs as AutomaticallyLinkedTo (shallow_copy + has_link guard); drop zero-link Standards. --- .../tests/cheatsheets_workstream_f_test.py | 37 ++++++++++++++----- .../parsers/cheatsheets_workstream_f.py | 3 +- 2 files changed, 30 insertions(+), 10 deletions(-) diff --git a/application/tests/cheatsheets_workstream_f_test.py b/application/tests/cheatsheets_workstream_f_test.py index b1e6012e4..d4af868f8 100644 --- a/application/tests/cheatsheets_workstream_f_test.py +++ b/application/tests/cheatsheets_workstream_f_test.py @@ -217,7 +217,7 @@ def test_unknown_cre_id_skipped_sibling_still_linked(self) -> None: standards = result.results["OWASP Cheat Sheets"] self.assertEqual(len(standards), 1) - self.assertEqual([l.document.id for l in standards[0].links], ["764-507"]) + self.assertEqual([link.document.id for link in standards[0].links], ["764-507"]) # The skipped id is reported for the reviewer. self.assertIn("999-999", "\n".join(cm.output)) @@ -254,19 +254,38 @@ def test_duplicate_candidate_ids_yield_single_link(self) -> None: self.assertEqual(len(std.links), 1) self.assertEqual(std.links[0].document.id, "764-507") - def test_empty_category_produces_no_blank_tag(self) -> None: + def test_blank_category_produces_no_extra_tag(self) -> None: + # Empty AND whitespace-only categories (both schema-valid) must not leak + # a blank/whitespace extra tag. + for blank in ("", " "): + with self.subTest(category=repr(blank)): + cache = _StubCache({"764-507": _cre("764-507")}) + approved = [ + _sugg("Authentication Cheat Sheet", ["764-507"], category=blank) + ] + + result = wf.suggestions_to_parse_result(approved, cache) + + # Still a valid, classifiable Standard... + base_parser_defs.validate_classification_tags(result.results) + std = result.results["OWASP Cheat Sheets"][0] + # ...with no blank/whitespace tag leaked in from the category. + self.assertNotIn("", std.tags) + self.assertNotIn(blank, std.tags) + # Only the five required classification tags remain (no extra). + self.assertEqual(len(std.tags), 5) + + def test_category_tag_is_stripped(self) -> None: cache = _StubCache({"764-507": _cre("764-507")}) - approved = [_sugg("Authentication Cheat Sheet", ["764-507"], category="")] + approved = [ + _sugg("Authentication Cheat Sheet", ["764-507"], category=" auth ") + ] result = wf.suggestions_to_parse_result(approved, cache) - # Still a valid, classifiable Standard... - base_parser_defs.validate_classification_tags(result.results) std = result.results["OWASP Cheat Sheets"][0] - # ...with no empty-string tag leaked in from the blank category. - self.assertNotIn("", std.tags) - # Only the five required classification tags remain (no extra). - self.assertEqual(len(std.tags), 5) + self.assertIn("auth", std.tags) + self.assertNotIn(" auth ", std.tags) if __name__ == "__main__": diff --git a/application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py b/application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py index 2304fd0b5..34a84c48b 100644 --- a/application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py +++ b/application/utils/external_project_parsers/parsers/cheatsheets_workstream_f.py @@ -275,7 +275,8 @@ def suggestions_to_parse_result( unknown_cre_ids: List[str] = [] for suggestion in approved: - extra = [suggestion.category] if suggestion.category else [] + category = suggestion.category.strip() + extra = [category] if category else [] standard = defs.Standard( name=STANDARD_NAME, section=suggestion.title,