diff --git a/gitgalaxy/core/detector.py b/gitgalaxy/core/detector.py index 685f782f..2366c42c 100644 --- a/gitgalaxy/core/detector.py +++ b/gitgalaxy/core/detector.py @@ -422,7 +422,7 @@ def splice( # --- EXISTING STRUCTURAL PIPELINE --- segments = self._partition_segments(code_stream, self.primary_lang_id) - equations, mitigation_telemetry, segment_spatial_maps, extracted_parents = self.coding_analysis( + equations, mitigation_telemetry, segment_spatial_maps, extracted_parents, threat_locations = self.coding_analysis( segments, regex_telemetry if profile_regex else None ) @@ -552,6 +552,7 @@ def splice( "financial_read_cost": ( round((file_token_mass / 1000000) * 3.00, 5) if file_token_mass is not None else None ), + "threat_locations": threat_locations, } if profile_regex: result_payload["regex_telemetry"] = regex_telemetry @@ -840,7 +841,7 @@ def _correlate_signals(self, targets: List[int], dampeners: List[int], max_dista def coding_analysis( self, segments: List[Tuple[str, str, int]], regex_telemetry: dict = None - ) -> Tuple[Dict[str, int], Dict[str, int], List[Dict[str, List[int]]], List[str]]: + ) -> Tuple[Dict[str, int], Dict[str, int], List[Dict[str, List[int]]], List[str], Dict[str, List[int]]]: counts: Dict[str, int] = {key: 0 for key in self.UNIVERSAL_METRICS_SCHEMA} # --- THE FIX: INJECT APPSEC SENSORS --- @@ -858,8 +859,9 @@ def coding_analysis( } segment_spatial_maps = [] extracted_parents = [] + threat_locations: Dict[str, List[int]] = {} - for seg_lang, seg_code, _ in segments: + for seg_lang, seg_code, current_line_offset in segments: # 1. Grab the language-specific rules rules = self.languages.get(seg_lang, {}).get("rules", {}).copy() @@ -895,6 +897,11 @@ def coding_analysis( if hasattr(pattern, "finditer"): matches = list(pattern.finditer(seg_code)) hit_indices = [m.start() for m in matches] + + # ---> NEW: Offset to LOC Conversion <--- + for m in matches: + line_number = current_line_offset + seg_code.count("\n", 0, m.start()) + 1 + threat_locations.setdefault(mapped_key, []).append(line_number) # ---> THE LINEAGE EXTRACTOR <--- # If the regex has 2+ capture groups, group 2 contains the inheritance mapping @@ -903,7 +910,13 @@ def coding_analysis( if m.group(2): extracted_parents.append(m.group(2).strip()) else: - hit_indices = [m.start() for m in re.finditer(str(pattern), seg_code)] + matches = list(re.finditer(str(pattern), seg_code)) + hit_indices = [m.start() for m in matches] + + # ---> NEW: Offset to LOC Conversion <--- + for m in matches: + line_number = current_line_offset + seg_code.count("\n", 0, m.start()) + 1 + threat_locations.setdefault(mapped_key, []).append(line_number) c = len(hit_indices) @@ -1026,7 +1039,7 @@ def coding_analysis( counts["indent_spaces"] += len(re.findall(r"^[ ]{2,}(?=\S)", seg_code, flags=re.MULTILINE)) segment_spatial_maps.append(spatial_map) - return counts, mitigations, segment_spatial_maps, extracted_parents + return counts, mitigations, segment_spatial_maps, extracted_parents, threat_locations def comment_analysis(self, comment_stream: str, lang_id: str, counts: Dict[str, int]) -> Dict[str, int]: """ diff --git a/gitgalaxy/metrics/signal_processor.py b/gitgalaxy/metrics/signal_processor.py index fc193f4b..0b57ea4a 100644 --- a/gitgalaxy/metrics/signal_processor.py +++ b/gitgalaxy/metrics/signal_processor.py @@ -333,19 +333,20 @@ def calculate_risk_vector( return { "risk_vector": blanket_risk_vector, "hit_vector": [0] * len(self.SIGNAL_SCHEMA), - "file_impact": 150.0, # Massive structural footprint for the topological map + "file_impact": 150.0, # <-- FIX: Restored the 150.0 mass spike for critical leaks "telemetry": { "archetype": getattr(config, "STATIC_ARCHETYPES", {}).get( - "data", "Static: Declarative Data & Configurations" + "quarantine", "Static: Critical Contraband Leak" ), "control_flow_ratio": 0.0, - "ownership_entropy": self._calc_ownership_entropy(authors_map), - "author_distribution": self._calculate_silo_risk(authors_map), + "ownership_entropy": 0.0, + "author_distribution": 0.0, "ownership": dominant_author, "domain_context": { "alert": "CRITICAL LEAK BYPASS", **ghost_meta, }, + "threat_locations": meta.get("threat_locations", {}), }, } @@ -389,6 +390,7 @@ def calculate_risk_vector( "alert": "MINIFIED VENDOR BYPASS", **ghost_meta, }, + "threat_locations": meta.get("threat_locations", {}), }, } @@ -431,6 +433,7 @@ def calculate_risk_vector( "author_distribution": 0.0, # <-- FIX: Plaintext changelogs don't have Authorship Centralization risk "ownership": dominant_author, "domain_context": ghost_meta, + "threat_locations": meta.get("threat_locations", {}), }, } @@ -830,6 +833,7 @@ def calculate_risk_vector( "ownership": dominant_author, "domain_context": ghost_meta, "mitigation_telemetry": meta.get("mitigations", []), + "threat_locations": meta.get("threat_locations", {}), } if mp_map: diff --git a/gitgalaxy/recorders/sarif_recorder.py b/gitgalaxy/recorders/sarif_recorder.py index 956d94c2..239e67af 100644 --- a/gitgalaxy/recorders/sarif_recorder.py +++ b/gitgalaxy/recorders/sarif_recorder.py @@ -86,18 +86,26 @@ def generate_report( # B. Extract Passive Security Lens Snippets (SAST) threat_snippets = telemetry.get("threat_snippets", {}) + threat_locations = telemetry.get("threat_locations", {}) + for threat_type, snippets in threat_snippets.items(): - for snippet in snippets: + # Attempt to retrieve exact line locations, matching by raw or sec_ prefixed key + locs = threat_locations.get(threat_type) or threat_locations.get(f"sec_{threat_type}", []) + + for idx, snippet in enumerate(snippets): # Safely determine severity severity = "error" if any(x in threat_type for x in ["secret", "injection", "execution", "corruption"]) else "warning" + # Fallback to file's start_line if specific LOC isn't captured + exact_line = locs[idx] if idx < len(locs) else start_line + results_list.append({ "ruleId": f"GG-SAST-{threat_type.upper()}", "level": severity, "message": { "text": f"Vulnerability signature triggered: {snippet}" }, - "locations": [self._build_location(rel_path, start_line)], + "locations": [self._build_location(rel_path, exact_line)], "properties": { "category": "Structural SAST" } diff --git a/tests/core_engine/test_detector.py b/tests/core_engine/test_detector.py index d95406f1..fc81fcc0 100644 --- a/tests/core_engine/test_detector.py +++ b/tests/core_engine/test_detector.py @@ -1307,7 +1307,7 @@ def test_detector_empty_pattern_continuations(): opt.languages["python"]["rules"]["none_rule"] = None # This should run without accumulating any hits and without crashing - counts, _, _, _ = opt.coding_analysis([("python", "code", 0)]) + counts, _, _, _, _ = opt.coding_analysis([("python", "code", 0)]) assert counts.get("empty_rule_1", 0) == 0, "Empty rule falsely triggered a hit!" @@ -1508,4 +1508,34 @@ def test_detector_zero_branch_massive_state(): ) assert result["mitigation_telemetry"].get("amplified_cascading_flux", 0) == 0, ( "Detector falsely amplified an OOM Bomb in a file with no algorithmic loops!" - ) \ No newline at end of file + ) + +# ============================================================================== +# TEST 46: EXACT LOC MAPPING (Offset to LOC) +# ============================================================================== +def test_detector_exact_loc_mapping(): + """Proves the coding_analysis phase accurately converts regex offsets to line numbers.""" + from gitgalaxy.core.detector import StructuralExtractor + opt_detector = StructuralExtractor("python", MOCK_LANG_DEFS) + + # Inject rules for testing + opt_detector.primary_rules["sec_hardcoded_secrets"] = re.compile(r"password") + opt_detector.primary_rules["high_risk_execution"] = re.compile(r"eval") + + code = ( + "def safe_func():\n" # Line 1 + " pass\n" # Line 2 + "\n" # Line 3 + "def bad_func():\n" # Line 4 + " x = 'password'\n" # Line 5 (sec_hardcoded_secrets) + " eval(x)\n" # Line 6 (high_risk_execution) + ) + + # Manually run coding analysis + segments = [("python", code, 0)] + counts, mitigations, spatial_maps, parents, threat_locations = opt_detector.coding_analysis(segments) + + # Verify the exact line numbers were captured + assert "sec_hardcoded_secrets" in threat_locations, "Failed to map threat location!" + assert threat_locations["sec_hardcoded_secrets"][0] == 5, f"Expected line 5, got {threat_locations['sec_hardcoded_secrets'][0]}" + assert threat_locations["high_risk_execution"][0] == 6, "Failed to map subsequent line threat!" \ No newline at end of file diff --git a/tests/core_engine/test_signal_processor.py b/tests/core_engine/test_signal_processor.py index 54601fa1..326ed031 100644 --- a/tests/core_engine/test_signal_processor.py +++ b/tests/core_engine/test_signal_processor.py @@ -1557,3 +1557,52 @@ def test_signal_processor_inline_suppressions(processor): assert "made_up_phantom_123" in res["telemetry"]["mitigation_telemetry"], ( "Phantom risk was not passed to the UI telemetry payload!" ) + + +# ============================================================================== +# TEST 51: SARIF EXACT LOC INJECTION +# ============================================================================== +def test_sarif_exact_loc_injection(): + """ + COVERAGE TARGET: sarif_recorder.py (_build_location). + Ensures the SARIF exporter consumes the threat_locations array and outputs + the exact line number instead of falling back to line 1. + """ + from gitgalaxy.recorders.sarif_recorder import SarifRecorder + import json + import tempfile + import os + + recorder = SarifRecorder() + + mock_file = { + "path": "src/vulnerable.py", + "start_line": 1, + "telemetry": { + "threat_snippets": { + "hardcoded_secrets": ["password='123'"] + }, + "threat_locations": { + "sec_hardcoded_secrets": [42] # The exact line number + } + } + } + + fd, temp_path = tempfile.mkstemp(suffix=".json") + os.close(fd) + + try: + recorder.generate_report([mock_file], {}, {}, temp_path) + + with open(temp_path, "r") as f: + sarif_output = json.load(f) + + results = sarif_output["runs"][0]["results"] + assert len(results) == 1, "Failed to generate SARIF result!" + + # Extract the exact line number from the payload + exact_line = results[0]["locations"][0]["physicalLocation"]["region"]["startLine"] + assert exact_line == 42, f"SARIF fell back to start_line! Expected 42, got {exact_line}." + + finally: + os.remove(temp_path)