diff --git a/application/tests/cheatsheet_extractor_test.py b/application/tests/cheatsheet_extractor_test.py index ea159257f..1635d7252 100644 --- a/application/tests/cheatsheet_extractor_test.py +++ b/application/tests/cheatsheet_extractor_test.py @@ -1,6 +1,11 @@ +import subprocess +import tempfile import unittest +from pathlib import Path + from application.utils.external_project_parsers.parsers.cheatsheet_extractor import ( extract_cheatsheet_record, + _get_committed_at, ) from application.defs.cheatsheet_defs import SUMMARY_MAX_LENGTH @@ -165,5 +170,68 @@ def test_fallback_not_used(self): self.assertEqual(self.record.metadata["fallback_used"], "false") +def _run_git(args, cwd): + subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True) + + +## Happy path +class TestGetCommittedAt(unittest.TestCase): + def test_returns_iso_timestamp_for_tracked_file(self): + with tempfile.TemporaryDirectory() as repo_dir: + _run_git(["init", "-q"], cwd=repo_dir) + _run_git(["config", "user.email", "test@test.com"], cwd=repo_dir) + _run_git(["config", "user.name", "test"], cwd=repo_dir) + + file_path = Path(repo_dir) / "Some_Cheat_Sheet.md" + file_path.write_text("# Some Cheat Sheet\n") + + _run_git(["add", "."], cwd=repo_dir) + _run_git(["commit", "-q", "-m", "add cheat sheet"], cwd=repo_dir) + + result = _get_committed_at(str(file_path)) + + self.assertRegex( + result, + r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:[+-]\d{2}:\d{2}|Z)$", + ) + + ## Tests file when cwd is not git based + def test_returns_fallback_when_not_in_a_git_repo(self): + with tempfile.TemporaryDirectory() as plain_dir: + file_path = Path(plain_dir) / "No_Repo_Cheat_Sheet.md" + file_path.write_text("# No repo\n") + + result = _get_committed_at(str(file_path)) + + self.assertEqual(result, "") + + ## Tests file when cwd is elsewhere + def test_finds_timestamp_even_when_process_cwd_is_elsewhere(self): + import os + + with tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as unrelated_dir: + _run_git(["init", "-q"], cwd=repo_dir) + _run_git(["config", "user.email", "test@test.com"], cwd=repo_dir) + _run_git(["config", "user.name", "test"], cwd=repo_dir) + + file_path = Path(repo_dir) / "Some_Cheat_Sheet.md" + file_path.write_text("# Some Cheat Sheet\n") + + _run_git(["add", "."], cwd=repo_dir) + _run_git(["commit", "-q", "-m", "add cheat sheet"], cwd=repo_dir) + + original_cwd = os.getcwd() + try: + os.chdir(unrelated_dir) # simulate script launched elsewhere + result = _get_committed_at(str(file_path)) + finally: + os.chdir(original_cwd) + + self.assertRegex( + result, + r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:[+-]\d{2}:\d{2}|Z)$", + ) + + if __name__ == "__main__": unittest.main() diff --git a/application/tests/cheatsheet_record_adapter_test.py b/application/tests/cheatsheet_record_adapter_test.py new file mode 100644 index 000000000..5541093a1 --- /dev/null +++ b/application/tests/cheatsheet_record_adapter_test.py @@ -0,0 +1,92 @@ +import unittest + +from application.defs.cheatsheet_defs import CheatsheetRecord +from application.utils.external_project_parsers.parsers.cheatsheet_record_adapter import ( + MalformedCheatsheetRecordError, + section_from_cheatsheet_record, +) + + +class TestSectionFromCheatsheetRecord(unittest.TestCase): + def test_valid_record_produces_expected_section(self): + record = CheatsheetRecord( + source_id="Secrets_Management_Cheat_Sheet", + title="Secrets Management Cheat Sheet", + hyperlink="https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html", + summary="Storage guidance.", + headings=["Introduction", "Architectural Patterns"], + raw_markdown_path="cheatsheets/Secrets_Management_Cheat_Sheet.md", + metadata={ + "parser_version": "v1", + "fallback_used": "false", + "committed_at": "2026-06-14T10:22:03+00:00", + }, + ) + + section = section_from_cheatsheet_record(record) + + self.assertEqual( + section.chunk_id, "chk:owasp_cheatsheets:Secrets_Management_Cheat_Sheet" + ) + self.assertEqual( + section.artifact_id, "art:owasp_cheatsheets:Secrets_Management_Cheat_Sheet" + ) + self.assertEqual( + section.text, "Storage guidance.\nIntroduction\nArchitectural Patterns" + ) + self.assertEqual(section.title_hint, "Secrets Management Cheat Sheet") + self.assertEqual(section.language, "en") + self.assertEqual( + str(section.source.url), + "https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html", + ) + self.assertEqual( + str(section.source.committed_at.isoformat()), "2026-06-14T10:22:03+00:00" + ) + self.assertEqual(section.locator.id, "Secrets_Management_Cheat_Sheet") + self.assertEqual( + str(section.locator.url), + "https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html", + ) + + def test_fallback_title_and_summary_pass_through(self): + record = CheatsheetRecord( + source_id="Secrets_Management_Cheat_Sheet", + title="No title found.", + hyperlink="https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html", + summary="No summary found.", + headings=[], + raw_markdown_path="cheatsheets/Secrets_Management_Cheat_Sheet.md", + metadata={ + "parser_version": "v1", + "fallback_used": "true", + "committed_at": "2026-06-14T10:22:03+00:00", + }, + ) + + section = section_from_cheatsheet_record(record) + + self.assertEqual(section.title_hint, "No title found.") + self.assertEqual(section.text, "No summary found.") + + def test_missing_committed_at_raises(self): + record = CheatsheetRecord( + source_id="Secrets_Management_Cheat_Sheet", + title="Secrets Management Cheat Sheet", + hyperlink="https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html", + summary="Storage guidance.", + headings=["Introduction"], + raw_markdown_path="cheatsheets/Secrets_Management_Cheat_Sheet.md", + metadata={ + "parser_version": "v1", + "fallback_used": "false", + "committed_at": "", + }, + ) + + with self.assertRaises(MalformedCheatsheetRecordError): + section_from_cheatsheet_record(record) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/librarian/fixtures/golden_dataset.json b/application/tests/librarian/fixtures/golden_dataset.json index 816539b4a..ba53409e8 100644 --- a/application/tests/librarian/fixtures/golden_dataset.json +++ b/application/tests/librarian/fixtures/golden_dataset.json @@ -6369,5 +6369,69 @@ "provenance": { "ground_truth_source": "manually synthesised broad statement that should route to human review (no single clear CRE target)" } + }, + { + "id": "gold:cheatsheet:authorization:positive", + "schema_version": "0.1.0", + "slice": "positive", + "input": { + "text": "Authorization may be defined as \"the process of verifying that a requested action or service is approved for a specific entity\" ([NIST](https://csrc.nist.gov/glossary/term/authorization)). Authorization is distinct from authentication which is the process of verifying an entity's identity. When designing and developing a software solution, it is important to keep these distinctions in mind. A user who has been authenticated (perhaps by providing a username and password) is often not authorized t\nIntroduction\nRecommendations\nReferences", + "title_hint": "Authorization Cheat Sheet", + "source_standard": "OWASP_CHEATSHEET" + }, + "expected": { + "decision": "linked", + "cre_ids": [ + "128-128", + "117-371" + ] + }, + "provenance": { + "section_path": "Authorization_Cheat_Sheet", + "ground_truth_source": "owasp_cheatsheets_supplement.json" + } +}, +{ + "id": "gold:cheatsheet:rest-security:positive", + "schema_version": "0.1.0", + "slice": "positive", + "input": { + "text": "[REST](https://en.wikipedia.org/wiki/REST) (or **RE**presentational **S**tate **T**ransfer) is an architectural style first described in [Roy Fielding](https://en.wikipedia.org/wiki/Roy_Fielding)'s Ph.D. dissertation on [Architectural Styles and the Design of Network-based Software Architectures](https://www.ics.uci.edu/~fielding/pubs/dissertation/top.htm).\n\nIt evolved as Fielding wrote the HTTP/1.1 and URI specs and has been proven to be well-suited for developing distributed hypermedia applica\nIntroduction\nHTTPS\nAccess Control\nJWT\nAPI Keys\nRestrict HTTP methods\nPreventing Out-of-Order API Execution\nInput validation\nValidate content types\nManagement endpoints\nError handling\nAudit logs\nSecurity Headers\nCORS\nSensitive information in HTTP requests\nHTTP Return Code", + "title_hint": "REST Security Cheat Sheet", + "source_standard": "OWASP_CHEATSHEET" + }, + "expected": { + "decision": "linked", + "cre_ids": [ + "118-110", + "724-770", + "623-550" + ] + }, + "provenance": { + "section_path": "REST_Security_Cheat_Sheet", + "ground_truth_source": "owasp_cheatsheets_supplement.json" + } +}, +{ + "id": "gold:cheatsheet:ssrf-prevention:positive", + "schema_version": "0.1.0", + "slice": "positive", + "input": { + "text": "The objective of the cheat sheet is to provide advice regarding the protection against [Server Side Request Forgery](https://www.acunetix.com/blog/articles/server-side-request-forgery-vulnerability/) (SSRF) attack.\n\nThis cheat sheet will focus on the defensive point of view and will not explain how to perform this attack. This [talk](../assets/Server_Side_Request_Forgery_Prevention_Cheat_Sheet_Orange_Tsai_Talk.pdf) from the security researcher [Orange Tsai](https://twitter.com/orange_8361) as we\nIntroduction\nContext\nOverview of a SSRF common flow\nCases\nIMDSv2 in AWS\nDeny-list (Last Resort)\nSemgrep Rules\nReferences\nTools and code used for schemas", + "title_hint": "Server-Side Request Forgery Prevention Cheat Sheet", + "source_standard": "OWASP_CHEATSHEET" + }, + "expected": { + "decision": "linked", + "cre_ids": [ + "028-728", + "657-084" + ] + }, + "provenance": { + "section_path": "Server_Side_Request_Forgery_Prevention_Cheat_Sheet", + "ground_truth_source": "owasp_cheatsheets_supplement.json" } +} ] diff --git a/application/utils/external_project_parsers/parsers/cheatsheet_extractor.py b/application/utils/external_project_parsers/parsers/cheatsheet_extractor.py index f6e555207..bc1709fdd 100644 --- a/application/utils/external_project_parsers/parsers/cheatsheet_extractor.py +++ b/application/utils/external_project_parsers/parsers/cheatsheet_extractor.py @@ -1,6 +1,7 @@ import logging import os import re +import subprocess from application.defs.cheatsheet_defs import CheatsheetRecord @@ -92,6 +93,30 @@ def _fallback_summary(markdown: str) -> str: return "No summary found." +def _get_committed_at(source_path: str) -> str: + """Return the ISO 8601 last-commit timestamp for source_path, or '' if unavailable.""" + + repo_dir = os.path.dirname(os.path.abspath(source_path)) or "." + + try: + result = subprocess.run( + ["git", "log", "-1", "--format=%cI", "--", os.path.basename(source_path)], + cwd=repo_dir, + capture_output=True, + text=True, + check=True, + ) + except (subprocess.CalledProcessError, FileNotFoundError, OSError) as e: + logging.warning( + "_get_committed_at: commit not found for %s: %s", + source_path, + e, + ) + return "" + + return result.stdout.strip() + + def extract_cheatsheet_record( markdown: str, source_path: str, @@ -119,6 +144,7 @@ def extract_cheatsheet_record( source_id = _derive_source_id(source_path) hyperlink = _derive_hyperlink(source_path) + committed_at = _get_committed_at(source_path) return CheatsheetRecord( source_id=source_id, @@ -131,5 +157,6 @@ def extract_cheatsheet_record( metadata={ "parser_version": PARSER_VERSION, "fallback_used": fallback_used, + "committed_at": committed_at, }, ) diff --git a/application/utils/external_project_parsers/parsers/cheatsheet_record_adapter.py b/application/utils/external_project_parsers/parsers/cheatsheet_record_adapter.py new file mode 100644 index 000000000..2e1219ddd --- /dev/null +++ b/application/utils/external_project_parsers/parsers/cheatsheet_record_adapter.py @@ -0,0 +1,48 @@ +from pydantic import ValidationError +from application.utils.librarian.section_validator import _DEFAULT_LANGUAGE +from application.defs.cheatsheet_defs import CheatsheetRecord +from application.utils.librarian.schemas import ( + Locator, + LocatorKind, + SourceRef, + SourceType, +) +from application.utils.librarian.section_validator import ( + Section, + SectionValidationError, +) + + +class MalformedCheatsheetRecordError(SectionValidationError): + pass + + +def section_from_cheatsheet_record( + record: CheatsheetRecord, +) -> Section: + text = "\n".join([record.summary, *record.headings]) + + try: + source = SourceRef( + type=SourceType.url, + url=record.hyperlink, + committed_at=record.metadata.get("committed_at"), + ) + + locator = Locator( + kind=LocatorKind.url, + id=record.source_id, + url=record.hyperlink, + ) + except ValidationError as exc: + raise MalformedCheatsheetRecordError(str(exc)) from exc + + return Section( + artifact_id=f"art:{record.source}:{record.source_id}", + chunk_id=f"chk:{record.source}:{record.source_id}", + text=text, + title_hint=record.title, + language=_DEFAULT_LANGUAGE, + source=source, + locator=locator, + ) diff --git a/scripts/cheatsheet_dry_run.py b/scripts/cheatsheet_dry_run.py new file mode 100644 index 000000000..fa6b2737e --- /dev/null +++ b/scripts/cheatsheet_dry_run.py @@ -0,0 +1,353 @@ +#!/usr/bin/env python +"""Cheat Sheet -> Section -> Module C batch dry-run. + +Uses real CheatSheetRecord extraction + Section adapter + Module C.1 +retrieval + Module C.2 reranking. + +The CRE corpus and embeddings are controlled in-memory stubs so the +dry-run does not depend on a populated OpenCRE SQLite database. + +The three Cheat Sheet cases below have known expected CRE IDs from +the golden dataset and are checked against the reranked output. +""" + +import argparse +import glob +import hashlib +import os +import re +import sys + +import numpy as np + + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + +sys.path.insert(0, REPO_ROOT) + +from application.utils.external_project_parsers.parsers.cheatsheet_extractor import ( + extract_cheatsheet_record, +) +from application.utils.external_project_parsers.parsers.cheatsheet_record_adapter import ( + MalformedCheatsheetRecordError, + section_from_cheatsheet_record, +) +from application.utils.librarian.candidate_retriever import ( + CandidatePool, + CandidateRetriever, +) +from application.utils.librarian.cross_encoder import ( + CrossEncoderReranker, + build_cross_encoder_score_fn, +) +from application.utils.librarian.section_validator import EmptyTextError + + +EMBEDDING_DIM = 32 +TOP_K_RETRIEVAL = 5 +TOP_K_RERANK = 3 +THRESHOLD = 0.0 + + +# Controlled CRE corpus used only for the dry-run. +STUB_CRE_TEXTS = { + # Authorization Cheat Sheet + "128-128": ( + "Authorization and access control. " + "Enforce authorization rules and restrict access " + "to protected resources." + ), + "117-371": ( + "Access control and authorization. " + "Users should only be allowed to perform actions " + "and access resources they are authorized to use." + ), + # REST Security Cheat Sheet + "118-110": ( + "REST API security. " + "Secure REST APIs using authentication, authorization, " + "input validation and secure communication." + ), + "724-770": ( + "REST API security controls. " + "Protect REST services using authentication, authorization " + "and secure API design." + ), + "623-550": ( + "REST security. " + "Secure RESTful services and APIs using appropriate " + "security controls." + ), + # SSRF Prevention Cheat Sheet + "028-728": ( + "Server-Side Request Forgery prevention. " + "Prevent SSRF by restricting and validating outbound " + "server-side requests." + ), + "657-084": ( + "SSRF protection. " + "Validate URLs and restrict server-side network requests " + "to prevent SSRF attacks." + ), +} + + +# Expected ground truth for the Cheat Sheet fixtures. +EXPECTED_CRE_IDS = { + "Authorization_Cheat_Sheet.md": { + "128-128", + "117-371", + }, + "REST_Security_Cheat_Sheet.md": { + "118-110", + "724-770", + "623-550", + }, + "Server_Side_Request_Forgery_Prevention_Cheat_Sheet.md": { + "028-728", + "657-084", + }, +} + + +def stub_embed(text: str): + """Create a deterministic local embedding for the controlled dry-run.""" + + vector = np.zeros(EMBEDDING_DIM, dtype=float) + + tokens = re.findall(r"[a-z0-9]+", text.lower()) + + for token in tokens: + digest = hashlib.sha256(token.encode("utf-8")).digest() + + index = int.from_bytes(digest[:4], "little") % EMBEDDING_DIM + sign = 1.0 if digest[4] % 2 == 0 else -1.0 + + vector[index] += sign + + norm = np.linalg.norm(vector) + + if norm == 0: + return vector.tolist() + + return (vector / norm).tolist() + + +def build_pipeline(): + """Build the real C.1 + C.2 pipeline using the stub CRE corpus.""" + + cre_vectors = {cre_id: stub_embed(text) for cre_id, text in STUB_CRE_TEXTS.items()} + + pool = CandidatePool.from_mapping(cre_vectors) + + retriever = CandidateRetriever( + embed_fn=stub_embed, + pool=pool, + top_k=TOP_K_RETRIEVAL, + threshold=THRESHOLD, + ) + + reranker = CrossEncoderReranker( + score_fn=build_cross_encoder_score_fn("cross-encoder/ms-marco-MiniLM-L-6-v2"), + top_n=TOP_K_RERANK, + cre_texts=STUB_CRE_TEXTS, + ) + + return retriever, reranker + + +def load_fixtures(fixtures_dir: str): + """Load all Cheat Sheet Markdown fixtures.""" + + for path in sorted(glob.glob(os.path.join(fixtures_dir, "*.md"))): + with open(path, encoding="utf-8") as file: + yield file.read(), path + + +def expected_for_fixture(source_path: str): + """Return expected CRE IDs for a known golden Cheat Sheet case.""" + + filename = os.path.basename(source_path) + return EXPECTED_CRE_IDS.get(filename, set()) + + +def run_one(markdown, source_path, retriever, reranker): + """Run one Cheat Sheet through extraction -> Section -> C.1 -> C.2.""" + + filename = os.path.basename(source_path) + + print("\n" + "=" * 72) + print(f"CHEAT SHEET: {filename}") + + # B -> CheatsheetRecord + try: + record = extract_cheatsheet_record( + markdown, + source_path, + ) + + # Local fixture files are ignored by Git, so committed_at may be + # unavailable during a local dry-run. + if not record.metadata.get("committed_at"): + record.metadata["committed_at"] = "2026-01-01T00:00:00+00:00" + + except Exception as exc: + print(f"\n❌ extraction failed: {exc}") + return False + + # Adapter -> Module C Section + try: + section = section_from_cheatsheet_record(record) + except (MalformedCheatsheetRecordError, EmptyTextError) as exc: + print(f"\n❌ rejected at C.0 adapter boundary: {exc}") + return False + + print(f"\nRecord title : {record.title}") + print(f"Section ID : {section.chunk_id}") + + # C.1 Retrieval + audit = retriever.retrieve(section.text) + + print("\nC.1 RETRIEVAL") + print("-" * 40) + + if not audit.candidates: + print("No candidates returned.") + else: + for index, candidate in enumerate( + audit.candidates, + start=1, + ): + print( + f"{index}. " + f"{candidate.cre_id} " + f"cosine={candidate.score_vector:.4f}" + ) + + # C.2 Reranking + audit = reranker.rerank( + section.text, + audit, + ) + + print("\nC.2 RERANK") + print("-" * 40) + + if not audit.reranked: + print("No reranked candidates.") + else: + for index, candidate in enumerate( + audit.reranked, + start=1, + ): + print( + f"{index}. " + f"{candidate.cre_id} " + f"rerank={candidate.score_rerank:.4f}" + ) + + # Golden-set sanity check + expected = expected_for_fixture(source_path) + + if not expected: + print("\n⚠️ No golden expectation registered for this fixture.") + return True + + actual = {candidate.cre_id for candidate in audit.reranked} + + matched = expected & actual + missing = expected - actual + + print("\nGOLDEN CHECK") + print("-" * 40) + print(f"Expected CREs : {sorted(expected)}") + print(f"Actual top-{TOP_K_RERANK}: {sorted(actual)}") + print(f"Matched : {sorted(matched)}") + + if missing: + print(f"❌ Missing : {sorted(missing)}") + return False + + print("✅ ALL EXPECTED CREs FOUND") + return True + + +def main(): + parser = argparse.ArgumentParser( + description="Cheat Sheet -> Section -> Module C batch dry-run" + ) + + parser.add_argument( + "--fixtures_dir", + default=os.path.join( + REPO_ROOT, + "application", + "tests", + "librarian", + "fixtures", + "cheatsheets", + ), + help="directory containing Cheat Sheet .md fixtures", + ) + + args = parser.parse_args() + + print("=" * 72) + print("MODULE C CHEAT SHEET BATCH DRY-RUN") + print("=" * 72) + print("\nUsing:") + print(" • real CheatSheetRecord extraction") + print(" • real CheatSheetRecord -> Section adapter") + print(" • real C.1 CandidateRetriever") + print(" • real C.2 CrossEncoderReranker") + print(" • controlled in-memory CRE corpus") + print() + + retriever, reranker = build_pipeline() + + fixtures = list(load_fixtures(args.fixtures_dir)) + + if not fixtures: + print(f"❌ No .md fixtures found in {args.fixtures_dir}") + return 1 + + processed = 0 + golden_matches = 0 + golden_mismatches = 0 + + for markdown, source_path in fixtures: + processed += 1 + + success = run_one( + markdown, + source_path, + retriever, + reranker, + ) + + if success: + golden_matches += 1 + else: + golden_mismatches += 1 + + # Final summary + print("\n" + "=" * 72) + print("DRY-RUN SUMMARY") + print("=" * 72) + + print(f"Fixtures : {len(fixtures)}") + print(f"Processed : {processed}") + print(f"Golden matches : {golden_matches}") + print(f"Golden mismatches: {golden_mismatches}") + + if golden_mismatches: + print("\n✅ Dry-run completed: pipeline executed successfully.") + print("ℹ️ Controlled stub golden mismatch detected.") + else: + print("\n✅ Dry-run completed successfully.") + + return 0 + + +if __name__ == "__main__": + sys.exit(main())