feat(sarif): inject exact LOC precision into security dashboard alert…

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]:
         """

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:

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"
                         }

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!"
-    )
+    )
+
+# ==============================================================================
+# 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!"

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)