diff --git a/app/services/normalizer/deduplicator.py b/app/services/normalizer/deduplicator.py index 6e31fe3..e638e2e 100644 --- a/app/services/normalizer/deduplicator.py +++ b/app/services/normalizer/deduplicator.py @@ -1,6 +1,6 @@ -from app.schemas.finding import Finding, FindingSeverity +from app.schemas.finding import Finding, FindingSeverity, FindingTool -DeduplicationKey = tuple[str | None, str, str, int | None] +DeduplicationKey = tuple[str, str, int | None] SEVERITY_RANK = { FindingSeverity.INFO: 0, @@ -10,6 +10,12 @@ FindingSeverity.CRITICAL: 4, } +TOOL_RANK = { + FindingTool.INFRA: 0, + FindingTool.SEMGREP: 1, + FindingTool.CODEQL: 2, +} + # 공통 Finding 목록에서 동일 key를 가진 중복 finding을 제거 def deduplicate_findings(findings: list[Finding]) -> list[Finding]: @@ -24,12 +30,11 @@ def deduplicate_findings(findings: list[Finding]) -> list[Finding]: return list(deduplicated.values()) -# CWE, type, file path, 시작 라인을 기준으로 중복 판단 key 생성 +# 취약점 식별값, file path, 시작 라인을 기준으로 중복 판단 key 생성 def build_deduplication_key(finding: Finding) -> DeduplicationKey: return ( - finding.cwe_id, - finding.type, - finding.file_path, + vulnerability_identity(finding), + normalize_file_path(finding.file_path), finding.line_start, ) @@ -42,12 +47,43 @@ def should_replace_finding(current: Finding, candidate: Finding) -> bool: if candidate_rank != current_rank: return candidate_rank > current_rank - return evidence_length(candidate) > evidence_length(current) + current_evidence_length = evidence_length(current) + candidate_evidence_length = evidence_length(candidate) + if candidate_evidence_length != current_evidence_length: + return candidate_evidence_length > current_evidence_length + + return tool_rank(candidate.tool) > tool_rank(current.tool) + + +# CWE가 있으면 CWE를, 없으면 type을 취약점 식별값으로 사용 +def vulnerability_identity(finding: Finding) -> str: + return normalize_key_part(finding.cwe_id) or normalize_key_part(finding.type) + + +# file path 중복 비교용 문자열을 생성 +def normalize_file_path(file_path: str) -> str: + return normalize_key_part(file_path) + + +# 중복 비교용 문자열을 trim/lowercase 형태로 정규화 +def normalize_key_part(value: str | None) -> str: + return str(value or "").strip().lower() # severity enum/string 값을 비교 가능한 우선순위 숫자로 변환 def severity_rank(severity: FindingSeverity | str) -> int: - return SEVERITY_RANK[FindingSeverity(severity)] + try: + return SEVERITY_RANK[FindingSeverity(severity)] + except ValueError: + return 0 + + +# analyzer tool enum/string 값을 비교 가능한 우선순위 숫자로 변환 +def tool_rank(tool: FindingTool | str) -> int: + try: + return TOOL_RANK[FindingTool(tool)] + except ValueError: + return 0 # evidence가 풍부한 finding을 고르기 위해 evidence 길이 계산 diff --git a/app/services/normalizer/finding_normalizer.py b/app/services/normalizer/finding_normalizer.py index 19a26fa..6d38691 100644 --- a/app/services/normalizer/finding_normalizer.py +++ b/app/services/normalizer/finding_normalizer.py @@ -1,8 +1,11 @@ from collections.abc import Iterable -from app.schemas.finding import Finding, FindingTool +from app.schemas.finding import Finding, FindingSeverity, FindingTool from app.services.scanner.base import RawFinding +DEFAULT_FINDING_TYPE_SUFFIX = "FINDING" +DEFAULT_MESSAGE = "Security finding detected" + class UnknownFindingToolError(ValueError): pass @@ -16,45 +19,145 @@ def normalize_findings(raw_findings: Iterable[RawFinding]) -> list[Finding]: # raw finding의 tool에 맞는 normalizer를 선택해 공통 Finding으로 변환 def normalize_finding(raw_finding: RawFinding) -> Finding: try: - normalizer = FINDING_NORMALIZERS[FindingTool(raw_finding.tool)] + tool = normalize_tool(raw_finding.tool) + normalizer = FINDING_NORMALIZERS[tool] except ValueError as exc: raise UnknownFindingToolError(f"Unsupported finding tool: {raw_finding.tool}") from exc - return normalizer(raw_finding) + return normalizer(raw_finding, tool) # Semgrep raw finding을 공통 Finding으로 변환 -def normalize_semgrep_finding(raw_finding: RawFinding) -> Finding: - return build_finding(raw_finding) +def normalize_semgrep_finding(raw_finding: RawFinding, tool: FindingTool) -> Finding: + return build_finding(raw_finding, tool) # CodeQL raw finding을 공통 Finding으로 변환 -def normalize_codeql_finding(raw_finding: RawFinding) -> Finding: - return build_finding(raw_finding) +def normalize_codeql_finding(raw_finding: RawFinding, tool: FindingTool) -> Finding: + return build_finding(raw_finding, tool) # Infra raw finding을 공통 Finding으로 변환 -def normalize_infra_finding(raw_finding: RawFinding) -> Finding: - return build_finding(raw_finding) +def normalize_infra_finding(raw_finding: RawFinding, tool: FindingTool) -> Finding: + return build_finding(raw_finding, tool) # 도구별 raw finding의 공통 필드를 Finding schema에 매핑 -def build_finding(raw_finding: RawFinding) -> Finding: +def build_finding(raw_finding: RawFinding, tool: FindingTool) -> Finding: + finding_type = normalize_type(raw_finding.type, raw_finding.rule_id, tool) + line_start, line_end = normalize_line_range( + raw_finding.line_start, + raw_finding.line_end, + ) + return Finding( - tool=FindingTool(raw_finding.tool), - type=raw_finding.type, - cwe_id=raw_finding.cwe_id, - severity=raw_finding.severity, - file_path=raw_finding.file_path, - line_start=raw_finding.line_start, - line_end=raw_finding.line_end, - message=raw_finding.message, - evidence=raw_finding.evidence, + tool=tool, + type=finding_type, + cwe_id=normalize_optional_text(raw_finding.cwe_id), + severity=normalize_severity(raw_finding.severity), + file_path=normalize_file_path(raw_finding.file_path), + line_start=line_start, + line_end=line_end, + message=normalize_message( + raw_finding.message, + raw_finding.rule_id, + finding_type, + ), + evidence=normalize_optional_text(raw_finding.evidence), recommendation=None, references=[], ) +# analyzer tool 값을 FindingTool enum으로 정규화 +def normalize_tool(tool: FindingTool | str) -> FindingTool: + return FindingTool(tool) + + +# finding type이 비어 있으면 rule id 또는 tool 기반 기본값을 생성 +def normalize_type( + finding_type: str | None, + rule_id: str | None, + tool: FindingTool, +) -> str: + normalized_type = normalize_required_text(finding_type) + if normalized_type: + return normalized_type + + normalized_rule_id = normalize_required_text(rule_id) + if normalized_rule_id: + return normalized_rule_id + + return f"{tool.value}_{DEFAULT_FINDING_TYPE_SUFFIX}" + + +# severity 값을 FindingSeverity enum으로 정규화 +def normalize_severity(severity: FindingSeverity | str | None) -> FindingSeverity: + try: + return FindingSeverity(severity) + except (TypeError, ValueError): + return FindingSeverity.INFO + + +# file path를 안전한 문자열로 정규화 +def normalize_file_path(file_path: str | None) -> str: + return normalize_required_text(file_path) or "unknown" + + +# message가 비어 있으면 rule id, finding type, 기본 메시지 순서로 대체 +def normalize_message( + message: str | None, + rule_id: str | None, + finding_type: str, +) -> str: + return ( + normalize_required_text(message) + or normalize_required_text(rule_id) + or finding_type + or DEFAULT_MESSAGE + ) + + +# line range를 양수 기반으로 보정하고 역전된 범위를 정리 +def normalize_line_range( + line_start: int | None, + line_end: int | None, +) -> tuple[int | None, int | None]: + normalized_start = normalize_line_number(line_start) + normalized_end = normalize_line_number(line_end) + + if ( + normalized_start is not None + and normalized_end is not None + and normalized_end < normalized_start + ): + return normalized_start, normalized_start + + return normalized_start, normalized_end + + +# line number가 양수 정수일 때만 유지 +def normalize_line_number(line_number: int | None) -> int | None: + if not isinstance(line_number, int) or line_number <= 0: + return None + + return line_number + + +# 빈 문자열을 None으로 정규화 +def normalize_optional_text(value: str | None) -> str | None: + normalized_value = normalize_required_text(value) + return normalized_value or None + + +# 문자열 값을 trim하고 빈 값이면 빈 문자열로 정규화 +def normalize_required_text(value: str | None) -> str: + if value is None: + return "" + + return str(value).strip() + + FINDING_NORMALIZERS = { FindingTool.SEMGREP: normalize_semgrep_finding, FindingTool.CODEQL: normalize_codeql_finding, diff --git a/tests/services/normalizer/__init__.py b/tests/services/normalizer/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/tests/services/normalizer/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/services/normalizer/test_deduplicator.py b/tests/services/normalizer/test_deduplicator.py new file mode 100644 index 0000000..b31decc --- /dev/null +++ b/tests/services/normalizer/test_deduplicator.py @@ -0,0 +1,155 @@ +import unittest + +from app.schemas.finding import Finding, FindingSeverity, FindingTool +from app.services.normalizer.deduplicator import ( + build_deduplication_key, + deduplicate_findings, +) +from app.services.normalizer.finding_normalizer import normalize_findings +from app.services.scanner.base import RawFinding + + +class DeduplicatorTest(unittest.TestCase): + def test_deduplicate_findings_keeps_unique_findings(self): + findings = [ + build_finding("CWE-89", "SQL_INJECTION", "src/db.py", 10), + build_finding("CWE-79", "XSS", "src/view.py", 20), + ] + + deduplicated = deduplicate_findings(findings) + + self.assertEqual(len(deduplicated), 2) + + def test_build_deduplication_key_prefers_cwe_over_type(self): + semgrep = build_finding("CWE-89", "SQL_INJECTION", " src/db.py ", 10) + codeql = build_finding("CWE-89", "SQL_QUERY_BUILT_FROM_USER", "src/db.py", 10) + + self.assertEqual( + build_deduplication_key(semgrep), + build_deduplication_key(codeql), + ) + + def test_build_deduplication_key_uses_type_when_cwe_is_missing(self): + finding = build_finding(None, "SQL_INJECTION", "src/db.py", 10) + + self.assertEqual( + build_deduplication_key(finding), + ("sql_injection", "src/db.py", 10), + ) + + def test_deduplicate_findings_keeps_higher_severity(self): + low = build_finding( + "CWE-89", + "SQL_INJECTION", + "src/db.py", + 10, + severity=FindingSeverity.LOW, + ) + high = build_finding( + "CWE-89", + "SQL_INJECTION", + "src/db.py", + 10, + severity=FindingSeverity.HIGH, + ) + + deduplicated = deduplicate_findings([low, high]) + + self.assertEqual(len(deduplicated), 1) + self.assertEqual(deduplicated[0].severity, FindingSeverity.HIGH) + + def test_deduplicate_findings_keeps_richer_evidence_on_severity_tie(self): + short = build_finding("CWE-89", "SQL_INJECTION", "src/db.py", 10, evidence="x") + rich = build_finding( + "CWE-89", + "SQL_INJECTION", + "src/db.py", + 10, + evidence="cursor.execute(user_input)", + ) + + deduplicated = deduplicate_findings([short, rich]) + + self.assertEqual(len(deduplicated), 1) + self.assertEqual(deduplicated[0].evidence, "cursor.execute(user_input)") + + def test_deduplicate_findings_prefers_codeql_on_full_tie(self): + semgrep = build_finding( + "CWE-89", + "SQL_INJECTION", + "src/db.py", + 10, + tool=FindingTool.SEMGREP, + evidence="same", + ) + codeql = build_finding( + "CWE-89", + "SQL_QUERY_BUILT_FROM_USER", + "src/db.py", + 10, + tool=FindingTool.CODEQL, + evidence="same", + ) + + deduplicated = deduplicate_findings([semgrep, codeql]) + + self.assertEqual(len(deduplicated), 1) + self.assertEqual(deduplicated[0].tool, FindingTool.CODEQL) + + def test_normalize_and_deduplicate_flow_merges_cross_tool_same_cwe(self): + raw_findings = [ + RawFinding( + tool=FindingTool.SEMGREP, + type="SQL_INJECTION", + severity=FindingSeverity.MEDIUM, + file_path="src/db.py", + message="semgrep message", + cwe_id="CWE-89", + line_start=10, + line_end=10, + evidence="same", + ), + RawFinding( + tool=FindingTool.CODEQL, + type="SQL_QUERY_BUILT_FROM_USER", + severity=FindingSeverity.HIGH, + file_path="src/db.py", + message="codeql message", + cwe_id="CWE-89", + line_start=10, + line_end=12, + evidence="same", + ), + ] + + deduplicated = deduplicate_findings(normalize_findings(raw_findings)) + + self.assertEqual(len(deduplicated), 1) + self.assertEqual(deduplicated[0].tool, FindingTool.CODEQL) + self.assertEqual(deduplicated[0].severity, FindingSeverity.HIGH) + + +def build_finding( + cwe_id: str | None, + finding_type: str, + file_path: str, + line_start: int | None, + severity: FindingSeverity = FindingSeverity.MEDIUM, + tool: FindingTool = FindingTool.SEMGREP, + evidence: str | None = None, +) -> Finding: + return Finding( + tool=tool, + type=finding_type, + cwe_id=cwe_id, + severity=severity, + file_path=file_path, + line_start=line_start, + line_end=line_start, + message="message", + evidence=evidence, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/services/normalizer/test_finding_normalizer.py b/tests/services/normalizer/test_finding_normalizer.py new file mode 100644 index 0000000..98716a2 --- /dev/null +++ b/tests/services/normalizer/test_finding_normalizer.py @@ -0,0 +1,122 @@ +import unittest + +from app.schemas.finding import FindingSeverity, FindingTool +from app.services.normalizer.finding_normalizer import ( + UnknownFindingToolError, + normalize_finding, + normalize_findings, +) +from app.services.scanner.base import RawFinding + + +class FindingNormalizerTest(unittest.TestCase): + def test_normalize_findings_maps_common_raw_fields(self): + raw_finding = RawFinding( + tool=FindingTool.SEMGREP, + type="SQL_INJECTION", + severity=FindingSeverity.HIGH, + file_path="src/db.py", + message="User input flows into SQL query", + rule_id="python.sql-injection", + cwe_id="CWE-89", + line_start=10, + line_end=12, + evidence="cursor.execute(query)", + ) + + findings = normalize_findings([raw_finding]) + + self.assertEqual(len(findings), 1) + finding = findings[0] + self.assertEqual(finding.tool, FindingTool.SEMGREP) + self.assertEqual(finding.type, "SQL_INJECTION") + self.assertEqual(finding.severity, FindingSeverity.HIGH) + self.assertEqual(finding.file_path, "src/db.py") + self.assertEqual(finding.message, "User input flows into SQL query") + self.assertEqual(finding.cwe_id, "CWE-89") + self.assertEqual(finding.line_start, 10) + self.assertEqual(finding.line_end, 12) + self.assertEqual(finding.evidence, "cursor.execute(query)") + + def test_normalize_finding_uses_fallbacks_for_blank_required_fields(self): + raw_finding = RawFinding( + tool=FindingTool.CODEQL, + type=" ", + severity=FindingSeverity.INFO, + file_path=" ", + message=" ", + rule_id="py/sql-injection", + cwe_id=" ", + line_start=0, + line_end=-1, + evidence=" ", + ) + + finding = normalize_finding(raw_finding) + + self.assertEqual(finding.type, "py/sql-injection") + self.assertEqual(finding.file_path, "unknown") + self.assertEqual(finding.message, "py/sql-injection") + self.assertIsNone(finding.cwe_id) + self.assertIsNone(finding.line_start) + self.assertIsNone(finding.line_end) + self.assertIsNone(finding.evidence) + + def test_normalize_finding_falls_back_to_tool_type_without_rule_id(self): + raw_finding = RawFinding( + tool=FindingTool.INFRA, + type=" ", + severity=FindingSeverity.LOW, + file_path="Dockerfile", + message=" ", + ) + + finding = normalize_finding(raw_finding) + + self.assertEqual(finding.type, "INFRA_FINDING") + self.assertEqual(finding.message, "INFRA_FINDING") + + def test_normalize_finding_coerces_invalid_severity_to_info(self): + raw_finding = RawFinding.model_construct( + tool=FindingTool.CODEQL, + type="SQL_INJECTION", + severity="INVALID", + file_path="src/db.py", + message="message", + ) + + finding = normalize_finding(raw_finding) + + self.assertEqual(finding.severity, FindingSeverity.INFO) + + def test_normalize_finding_clamps_reversed_line_range(self): + raw_finding = RawFinding( + tool=FindingTool.SEMGREP, + type="SQL_INJECTION", + severity=FindingSeverity.HIGH, + file_path="src/db.py", + message="message", + line_start=12, + line_end=10, + ) + + finding = normalize_finding(raw_finding) + + self.assertEqual(finding.line_start, 12) + self.assertEqual(finding.line_end, 12) + + def test_normalize_finding_rejects_unknown_tool(self): + raw_finding = RawFinding.model_construct( + tool="UNKNOWN", + type="SQL_INJECTION", + severity=FindingSeverity.HIGH, + file_path="src/db.py", + message="message", + ) + + with self.assertRaisesRegex(UnknownFindingToolError, "Unsupported finding tool"): + normalize_finding(raw_finding) + + +if __name__ == "__main__": + unittest.main()