[Fix] 취약점 파일 경로를 저장소 상대경로로 정규화

app/jobs/analysis_job.py

@@ -60,7 +60,7 @@ def run_analysis_job(payload: dict[str, Any]) -> dict[str, Any]:
         #추후 runner 내부 조정 필요
         raw_findings = run_analyzers(context)
         context.raw_findings = [finding.model_dump() for finding in raw_findings] #원본
-        context.normalized_findings = normalize_findings(raw_findings) #정규화 후
+        context.normalized_findings = normalize_findings(raw_findings, repo_root=context.repo_path) #정규화 후
         context.normalized_findings = deduplicate_findings(context.normalized_findings) #중복
 
         evidence_by_finding = search_evidence_for_findings(

app/services/normalizer/finding_normalizer.py

@@ -1,7 +1,12 @@
 from collections.abc import Iterable
+import logging
+from pathlib import Path
 
 from app.schemas.finding import Finding, FindingSeverity, FindingTool
 from app.services.scanner.base import RawFinding
+from app.services.normalizer.repository_path import InvalidFindingPath, repository_relative_path
+
+logger = logging.getLogger(__name__)
 
 DEFAULT_FINDING_TYPE_SUFFIX = "FINDING"
 DEFAULT_MESSAGE = "Security finding detected"
@@ -12,8 +17,27 @@ class UnknownFindingToolError(ValueError):
 
 
 # 여러 analyzer raw finding을 공통 Finding 목록으로 변환
-def normalize_findings(raw_findings: Iterable[RawFinding]) -> list[Finding]:
-    return [normalize_finding(raw_finding) for raw_finding in raw_findings]
+def normalize_findings(
+    raw_findings: Iterable[RawFinding], repo_root: str | Path | None = None,
+) -> list[Finding]:
+    findings = []
+    for raw_finding in raw_findings:
+        if repo_root is not None:
+            try:
+                path = repository_relative_path(raw_finding.file_path, repo_root)
+            except InvalidFindingPath as exc:
+                logger.warning(
+                    "Skipping finding with invalid repository path. tool=%s rule_id=%s path=%r reason=%s",
+                    raw_finding.tool, raw_finding.rule_id, raw_finding.file_path, exc,
+                )
+                continue
+            raw_finding = raw_finding.model_copy(update={"file_path": path})
+        finding = normalize_finding(raw_finding)
+        if repo_root is not None:
+            # Keep filename whitespace; generic text normalization must not alter paths.
+            finding.file_path = path
+        findings.append(finding)
+    return findings
 
 
 # raw finding의 tool에 맞는 normalizer를 선택해 공통 Finding으로 변환

app/services/normalizer/repository_path.py

@@ -0,0 +1,94 @@
+"""Validate scanner paths against the clone, preserving repository structure."""
+
+import re
+from pathlib import Path
+from urllib.parse import unquote, urljoin, urlsplit
+
+
+class InvalidFindingPath(ValueError):
+    pass
+
+
+def repository_relative_path(value: str, repo_root: str | Path) -> str:
+    if not value or "\x00" in value:
+        raise InvalidFindingPath("missing or invalid file path")
+    value = value.replace("\\", "/")
+    if re.match(r"^[A-Za-z][A-Za-z0-9+.-]*:", value) or value.startswith("//"):
+        raise InvalidFindingPath("unsupported URI, drive or network path")
+    root = Path(repo_root).absolute()
+    candidate = Path(value)
+    if not candidate.is_absolute():
+        candidate = root / candidate
+    try:
+        # Resolve before collapsing '..' so symlink traversal cannot hide an escape.
+        resolved_relative = candidate.resolve().relative_to(root.resolve())
+        try:
+            relative = candidate.relative_to(root)
+        except ValueError:
+            # Accept the canonical clone root spelling, never an unrelated symlink
+            # outside the clone merely because its target points back inside.
+            relative = candidate.relative_to(root.resolve())
+    except (ValueError, OSError, RuntimeError) as exc:
+        raise InvalidFindingPath("path or symlink escapes repository") from exc
+    parts: list[str] = []
+    traversed = root
+    through_symlink = False
+    for part in relative.parts:
+        traversed = traversed / part
+        try:
+            if traversed.is_symlink():
+                through_symlink = True
+                traversed.resolve(strict=True).relative_to(root.resolve())
+            traversed.resolve().relative_to(root.resolve())
+        except (ValueError, OSError, RuntimeError) as exc:
+            raise InvalidFindingPath("path traverses outside repository") from exc
+        if part == "..":
+            if not parts:
+                raise InvalidFindingPath("relative traversal escapes repository")
+            parts.pop()
+        elif part != ".":
+            parts.append(part)
+    if not parts:
+        raise InvalidFindingPath("path refers to repository root")
+    if through_symlink and ".." in relative.parts:
+        if not resolved_relative.parts:
+            raise InvalidFindingPath("path refers to repository root")
+        return resolved_relative.as_posix()
+    return "/".join(parts)
+
+
+def sarif_artifact_path(artifact: dict, bases: dict) -> str:
+    def resolve(location: dict, seen: frozenset[str]) -> str:
+        uri = location.get("uri")
+        if not isinstance(uri, str) or not uri:
+            raise InvalidFindingPath("missing artifact URI (index-only locations unsupported)")
+        if any(ord(char) < 32 for char in uri):
+            raise InvalidFindingPath("URI contains control characters")
+        base_id = location.get("uriBaseId")
+        if base_id is not None:
+            if not isinstance(base_id, str) or base_id in seen or len(seen) >= 64:
+                raise InvalidFindingPath("invalid or cyclic URI base")
+            base = bases.get(base_id)
+            if not isinstance(base, dict):
+                raise InvalidFindingPath("unknown URI base")
+            base_uri = resolve(base, seen | {base_id})
+            if not urlsplit(base_uri).scheme or not base_uri.endswith("/"):
+                raise InvalidFindingPath("URI base requires an absolute directory URI")
+            uri = urljoin(base_uri, uri)
+        return uri
+
+    try:
+        uri = resolve(artifact, frozenset())
+        parsed = urlsplit(uri)
+    except ValueError as exc:
+        raise InvalidFindingPath(str(exc)) from exc
+    if parsed.scheme not in ("", "file") or parsed.netloc not in ("", "localhost"):
+        raise InvalidFindingPath("unsupported URI scheme or authority")
+    if parsed.query or parsed.fragment or (parsed.scheme == "file" and not parsed.path.startswith("/")):
+        raise InvalidFindingPath("unsupported URI query, fragment or relative file URI")
+    if re.search(r"%(?![0-9A-Fa-f]{2})", parsed.path):
+        raise InvalidFindingPath("malformed URI percent encoding")
+    try:
+        return unquote(parsed.path, encoding="utf-8", errors="strict")
+    except UnicodeError as exc:
+        raise InvalidFindingPath("invalid URI UTF-8 encoding") from exc

app/services/scanner/codeql_sarif_parser.py

@@ -1,12 +1,16 @@
 from __future__ import annotations
 
 import json
+import logging
 import re
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from app.schemas.finding import FindingSeverity, FindingTool
 from app.services.scanner.base import AnalyzerError, RawFinding
+from app.services.normalizer.repository_path import InvalidFindingPath, sarif_artifact_path
+
+logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
     from app.services.scanner.codeql_runner import CodeQLLanguage
@@ -45,10 +49,16 @@ def parse_sarif_file(
         if not isinstance(results, list):
             raise AnalyzerError("Invalid CodeQL SARIF output: results must be a list")
 
-        findings.extend(
-            convert_sarif_result(result, rules, language)
-            for result in results
-        )
+        for result_index, result in enumerate(results):
+            try:
+                findings.append(convert_sarif_result(
+                    result, rules, language, run.get("originalUriBaseIds", {}),
+                ))
+            except InvalidFindingPath as exc:
+                logger.warning(
+                    "Skipping CodeQL finding with invalid artifact URI. result_index=%s rule_id=%s reason=%s",
+                    result_index, result.get("ruleId"), exc,
+                )
 
     return findings
 
@@ -79,6 +89,7 @@ def convert_sarif_result(
     result: dict[str, Any],
     rules: dict[str, dict[str, Any]],
     language: CodeQLLanguage,
+    uri_bases: dict[str, Any] | None = None,
 ) -> RawFinding:
     if not isinstance(result, dict):
         raise AnalyzerError("Invalid CodeQL SARIF output: result must be an object")
@@ -94,7 +105,7 @@ def convert_sarif_result(
         tool=FindingTool.CODEQL,
         type=normalize_rule_type(rule_id),
         severity=map_codeql_severity(properties, rule_properties, result.get("level")),
-        file_path=get_artifact_uri(location),
+        file_path=get_artifact_uri(location, uri_bases),
         message=get_sarif_message(result, rule_id),
         rule_id=rule_id,
         cwe_id=extract_cwe_id(result, rule),
@@ -125,12 +136,12 @@ def get_primary_location(result: dict[str, Any]) -> dict[str, Any]:
 
 
 # SARIF artifact uri를 추출
-def get_artifact_uri(location: dict[str, Any]) -> str:
+def get_artifact_uri(location: dict[str, Any], uri_bases: dict | None = None) -> str:
     artifact_location = location.get("artifactLocation", {})
     if not isinstance(artifact_location, dict):
-        return ""
+        raise InvalidFindingPath("invalid artifact location")
 
-    return str(artifact_location.get("uri") or "")
+    return sarif_artifact_path(artifact_location, uri_bases if isinstance(uri_bases, dict) else {})
 
 
 # SARIF message 텍스트를 추출

tests/jobs/test_analysis_job.py

@@ -12,6 +12,24 @@
 
 
 class AnalysisJobTest(unittest.TestCase):
+    def test_callback_contains_relative_deduplicated_paths_only(self):
+        runner = Mock(tool=FindingTool.SEMGREP)
+        relative = build_raw_finding("FIRST", 10)
+        absolute = relative.model_copy(update={"file_path": "/tmp/repository/src/example.py"})
+        invalid = relative.model_copy(update={"file_path": "/tmp/outside.py"})
+        runner.run.return_value = [absolute, relative, invalid]
+        with self.pipeline_patches([runner]) as callback, patch(
+            "app.jobs.analysis_job.search_evidence_for_findings", return_value=[[]],
+        ) as search, patch(
+            "app.jobs.analysis_job.enrich_finding_with_explanation", side_effect=lambda finding, _: finding,
+        ), self.assertLogs("app.services.normalizer.finding_normalizer", level="WARNING"):
+            result = run_analysis_job(self.payload)
+        self.assertEqual(result["finding_count"], 1)
+        self.assertEqual(len(search.call_args.args[0]), 1)
+        payload = callback.send_success.call_args.args[0].model_dump(by_alias=True)
+        self.assertEqual(payload["findings"][0]["filePath"], "src/example.py")
+        self.assertNotIn("/tmp/repository", str(payload))
+
     def setUp(self):
         self.payload = {
             "analysis_id": 11,

tests/services/normalizer/test_repository_path.py

@@ -0,0 +1,95 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from app.services.normalizer.repository_path import (
+    InvalidFindingPath, repository_relative_path, sarif_artifact_path,
+)
+from app.services.normalizer.finding_normalizer import normalize_findings
+from app.services.normalizer.deduplicator import deduplicate_findings
+from app.services.scanner.base import RawFinding
+
+
+class RepositoryPathTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name) / "repository"
+        self.root.mkdir()
+        (self.root / "src").mkdir()
+        (self.root / "src" / "한글 file.py").touch()
+        (self.root / "internal").symlink_to(self.root / "src", target_is_directory=True)
+        (self.root / "external").symlink_to(self.root.parent, target_is_directory=True)
+
+    def test_paths_preserve_structure_and_names(self):
+        for path in ["src/한글 file.py", str(self.root / "src/한글 file.py"),
+                     "./src/../src/한글 file.py", "src\\한글 file.py", "src//한글 file.py"]:
+            with self.subTest(path=path):
+                self.assertEqual(repository_relative_path(path, self.root), "src/한글 file.py")
+
+    def test_internal_symlink_keeps_repository_name(self):
+        self.assertEqual(repository_relative_path("internal/한글 file.py", self.root), "internal/한글 file.py")
+
+    def test_parent_after_internal_symlink_uses_actual_target(self):
+        (self.root / "src/nested").mkdir()
+        (self.root / "nested-link").symlink_to(self.root / "src/nested", target_is_directory=True)
+        self.assertEqual(repository_relative_path("nested-link/../한글 file.py", self.root), "src/한글 file.py")
+
+    def test_rejects_external_traversal_and_symlinks(self):
+        for path in [str(self.root.parent / "outside.py"), "../outside.py", "src/../../x.py",
+                     "external/x.py", "external/../repository/src/한글 file.py",
+                     "../repository/src/한글 file.py", "", ".", "C:\\src\\x.py", "https://host/x.py"]:
+            with self.subTest(path=path), self.assertRaises(InvalidFindingPath):
+                repository_relative_path(path, self.root)
+
+    def test_looping_symlink_is_invalid(self):
+        (self.root / "loop").symlink_to("loop")
+        with self.assertRaises(InvalidFindingPath):
+            repository_relative_path("loop/x.py", self.root)
+
+    def test_external_absolute_symlink_pointing_inside_is_invalid(self):
+        alias = self.root.parent / "outside-link"
+        alias.symlink_to(self.root, target_is_directory=True)
+        with self.assertRaises(InvalidFindingPath):
+            repository_relative_path(str(alias / "src/한글 file.py"), self.root)
+
+    def test_normalization_before_dedup_isolates_invalid_finding(self):
+        def raw(path):
+            return RawFinding(tool="SEMGREP", type="SQL", severity="HIGH", file_path=path,
+                              message="message /tmp/unchanged", evidence="/tmp/evidence", line_start=1)
+        inputs = [raw(str(self.root / "src/한글 file.py")), raw("src/한글 file.py"), raw("../bad")]
+        with self.assertLogs("app.services.normalizer.finding_normalizer", level="WARNING"):
+            findings = deduplicate_findings(normalize_findings(inputs, self.root))
+        self.assertEqual(len(findings), 1)
+        self.assertEqual(findings[0].file_path, "src/한글 file.py")
+        self.assertEqual(findings[0].message, "message /tmp/unchanged")
+        self.assertEqual(findings[0].evidence, "/tmp/evidence")
+        self.assertEqual(inputs[0].file_path, str(self.root / "src/한글 file.py"))
+
+    def test_uri_formats_and_recursive_base(self):
+        uri = (self.root / "src/한글 file.py").as_uri()
+        bases = {"ROOT": {"uri": self.root.as_uri() + "/"},
+                 "SRC": {"uri": "src/", "uriBaseId": "ROOT"}}
+        for artifact in [{"uri": uri}, {"uri": uri.replace("file:///", "file://localhost/")},
+                         {"uri": "src/%ED%95%9C%EA%B8%80%20file.py"},
+                         {"uri": "%ED%95%9C%EA%B8%80%20file.py", "uriBaseId": "SRC"}]:
+            with self.subTest(artifact=artifact):
+                self.assertEqual(repository_relative_path(sarif_artifact_path(artifact, bases), self.root),
+                                 "src/한글 file.py")
+
+    def test_uri_rejections(self):
+        bases = {"A": {"uri": "a/", "uriBaseId": "B"},
+                 "B": {"uri": "b/", "uriBaseId": "A"}, "REL": {"uri": "src/"}}
+        for artifact in [{"uri": "https://host/src/x.py"}, {"uri": "file://remote/src/x.py"},
+                         {"uri": "file:src/x.py"}, {"uri": "file://[bad/x.py"},
+                         {"uri": "\nx.py"}, {"uri": "x.py?query"}, {"uri": "x.py#fragment"},
+                         {"uri": "%ZZ.py"}, {"uri": "%FF.py"}, {"index": 0},
+                         {"uri": "x.py", "uriBaseId": "MISSING"},
+                         {"uri": "x.py", "uriBaseId": "A"}, {"uri": "x.py", "uriBaseId": "REL"}]:
+            with self.subTest(artifact=artifact), self.assertRaises(InvalidFindingPath):
+                sarif_artifact_path(artifact, bases)
+
+    def test_uri_encoded_traversal_and_external_file_are_rejected_at_boundary(self):
+        for uri in ["%2e%2e/out.py", self.root.parent.as_uri() + "/out.py"]:
+            with self.subTest(uri=uri), self.assertRaises(InvalidFindingPath):
+                repository_relative_path(sarif_artifact_path({"uri": uri}, {}), self.root)

tests/services/scanner/test_codeql_sarif_parser.py

@@ -14,6 +14,21 @@
 
 
 class CodeQLSarifParserTest(unittest.TestCase):
+    def test_invalid_uri_does_not_discard_other_results_and_base_is_applied(self):
+        import copy
+        payload = build_sarif_payload()
+        run = payload["runs"][0]
+        valid = run["results"][0]
+        artifact = valid["locations"][0]["physicalLocation"]["artifactLocation"]
+        artifact.update(uri="src/a%20b.py", uriBaseId="ROOT")
+        run["originalUriBaseIds"] = {"ROOT": {"uri": "file:///tmp/repository/"}}
+        invalid = copy.deepcopy(valid)
+        invalid["locations"][0]["physicalLocation"]["artifactLocation"]["uriBaseId"] = "UNKNOWN"
+        run["results"] = [invalid, valid]
+        with self.assertLogs("app.services.scanner.codeql_sarif_parser", level="WARNING"):
+            findings = self.parse_payload(payload)
+        self.assertEqual([finding.file_path for finding in findings], ["/tmp/repository/src/a b.py"])
+
     def test_parse_sarif_file_maps_result_to_raw_finding(self):
         language = CodeQLLanguage(
             name="python",