diff --git a/app/jobs/analysis_job.py b/app/jobs/analysis_job.py index 3fdc5f8..17c8001 100644 --- a/app/jobs/analysis_job.py +++ b/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( diff --git a/app/services/normalizer/finding_normalizer.py b/app/services/normalizer/finding_normalizer.py index 9ce5be8..ec64ff2 100644 --- a/app/services/normalizer/finding_normalizer.py +++ b/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으로 변환 diff --git a/app/services/normalizer/repository_path.py b/app/services/normalizer/repository_path.py new file mode 100644 index 0000000..5b2f0fc --- /dev/null +++ b/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 diff --git a/app/services/scanner/codeql_sarif_parser.py b/app/services/scanner/codeql_sarif_parser.py index 0c6412f..978898d 100644 --- a/app/services/scanner/codeql_sarif_parser.py +++ b/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 텍스트를 추출 diff --git a/tests/jobs/test_analysis_job.py b/tests/jobs/test_analysis_job.py index 5870e00..29349cf 100644 --- a/tests/jobs/test_analysis_job.py +++ b/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, diff --git a/tests/services/normalizer/test_repository_path.py b/tests/services/normalizer/test_repository_path.py new file mode 100644 index 0000000..a966741 --- /dev/null +++ b/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) diff --git a/tests/services/scanner/test_codeql_sarif_parser.py b/tests/services/scanner/test_codeql_sarif_parser.py index 47ef6f6..6fe2b14 100644 --- a/tests/services/scanner/test_codeql_sarif_parser.py +++ b/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", diff --git a/tests/services/scanner/test_semgrep_path_integration.py b/tests/services/scanner/test_semgrep_path_integration.py new file mode 100644 index 0000000..06c5e60 --- /dev/null +++ b/tests/services/scanner/test_semgrep_path_integration.py @@ -0,0 +1,39 @@ +"""Offline CLI check; requires semgrep on PATH, never downloads registry rules.""" + +import os +import shutil +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from app.services.scanner.semgrep_runner import build_semgrep_command, parse_semgrep_output +from app.services.normalizer.finding_normalizer import normalize_findings + + +@unittest.skipUnless(shutil.which("semgrep"), "Semgrep CLI is not installed") +class SemgrepPathIntegrationTest(unittest.TestCase): + def test_actual_absolute_and_repository_relative_output(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp).resolve() / "repository" + source = root / "src" / "한글 file.py" + source.parent.mkdir(parents=True) + source.write_text("eval(input())\n", encoding="utf-8") + config = Path(tmp) / "rule.yaml" + config.write_text("rules:\n- id: local-eval\n languages: [python]\n severity: ERROR\n message: unsafe eval\n pattern: eval(...)\n", encoding="utf-8") + env = {**os.environ, "SEMGREP_ENABLE_VERSION_CHECK": "0", + "SEMGREP_SEND_METRICS": "off", "SEMGREP_USER_DATA_FOLDER": str(Path(tmp) / "data")} + for target in [root, Path(".")]: + with self.subTest(target=target), patch( + "app.services.scanner.semgrep_runner.settings.SEMGREP_CONFIG", str(config), + ): + command = build_semgrep_command(target) + command[0] = shutil.which("semgrep") + result = subprocess.run(command, cwd=root, env=env, capture_output=True, + text=True, timeout=60) + self.assertEqual(result.returncode, 0, result.stderr + result.stdout) + raw = parse_semgrep_output(result.stdout) + self.assertEqual(len(raw), 1, result.stderr) + self.assertEqual(raw[0].file_path, str(source) if target.is_absolute() else "src/한글 file.py") + self.assertEqual(normalize_findings(raw, root)[0].file_path, "src/한글 file.py")