7장: 보안 취약점 분석과 자동 수정

# 취약한 코드: SQL Injection
def get_user(username: str) -> dict | None:
    query = f"SELECT * FROM users WHERE username = '{username}'"
    cursor.execute(query)
    return cursor.fetchone()

# 수정된 코드: 매개변수화된 쿼리
def get_user(username: str) -> dict | None:
    query = "SELECT * FROM users WHERE username = %s"
    cursor.execute(query, (username,))
    return cursor.fetchone()

// 취약한 코드: Stored XSS
function CommentDisplay({ comment }: { comment: string }) {
  return <div dangerouslySetInnerHTML={{ __html: comment }} />;
}

// 수정된 코드: 안전한 렌더링
import DOMPurify from "dompurify";
 
function CommentDisplay({ comment }: { comment: string }) {
  const sanitized = DOMPurify.sanitize(comment);
  return <div dangerouslySetInnerHTML={{ __html: sanitized }} />;
}

from dataclasses import dataclass
from enum import Enum
 
 
class VulnerabilityType(Enum):
    SQL_INJECTION = "A03:2021-Injection"
    XSS = "A03:2021-Injection"
    BROKEN_AUTH = "A07:2021-Authentication"
    SENSITIVE_DATA = "A02:2021-Cryptographic Failures"
    BROKEN_ACCESS = "A01:2021-Broken Access Control"
    SSRF = "A10:2021-SSRF"
    INSECURE_DESERIALIZATION = "A08:2021-Software Integrity"
 
 
class Severity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"
 
 
@dataclass
class Vulnerability:
    vuln_type: VulnerabilityType
    severity: Severity
    file_path: str
    line_number: int
    description: str
    fix_suggestion: str
    fix_code: str | None
    confidence: float
    is_false_positive: bool
 
 
SECURITY_ANALYSIS_PROMPT = """보안 전문가로서 다음 코드를 분석하세요.
 
파일: {filepath}
코드:
---
{code}
---
 
SAST 도구가 탐지한 잠재적 취약점:
{sast_findings}
 
호출 컨텍스트:
{call_context}
 
각 SAST 발견에 대해 다음을 판단하세요:
1. 실제 취약점인지 오탐인지 (이유 포함)
2. 실제 취약점이라면 심각도 (critical/high/medium/low)
3. OWASP 분류
4. 구체적인 수정 코드
5. 확신도 (0.0-1.0)
 
추가로 SAST가 놓친 취약점이 있다면 보고하세요:
- 비즈니스 로직 취약점
- 인증/인가 결함
- 경쟁 조건(race condition)
- 안전하지 않은 기본 설정"""
 
 
class SecurityAnalyzer:
    """SAST + LLM 하이브리드 보안 분석기"""
 
    def __init__(self, sast_scanner, llm_client):
        self.sast_scanner = sast_scanner
        self.llm_client = llm_client
 
    async def analyze(
        self, filepath: str, source: str, call_context: str = ""
    ) -> list[Vulnerability]:
        # 1단계: SAST 스캔
        sast_findings = self.sast_scanner.scan(filepath, source)
 
        # 2단계: LLM 컨텍스트 분석
        prompt = SECURITY_ANALYSIS_PROMPT.format(
            filepath=filepath,
            code=source,
            sast_findings=self._format_findings(sast_findings),
            call_context=call_context,
        )
 
        response = await self.llm_client.generate(prompt)
        vulnerabilities = self._parse_response(response)
 
        # 3단계: 오탐 필터링
        real_vulns = [
            v for v in vulnerabilities
            if not v.is_false_positive and v.confidence > 0.7
        ]
 
        return sorted(
            real_vulns,
            key=lambda v: v.severity.value,
        )
 
    def _format_findings(self, findings: list[dict]) -> str:
        return "\n".join(
            f"- [{f['rule_id']}] {f['message']} (줄 {f['line']})"
            for f in findings
        )
 
    def _parse_response(self, response: str) -> list[Vulnerability]:
        # LLM 응답 파싱 (간략화)
        return []

@dataclass
class SecurityFix:
    vulnerability: Vulnerability
    original_code: str
    fixed_code: str
    explanation: str
    breaking_changes: list[str]
    test_cases: list[str]
 
 
FIX_GENERATION_PROMPT = """다음 보안 취약점을 수정하는 코드를 생성하세요.
 
취약점: {vuln_description}
심각도: {severity}
OWASP 분류: {owasp_category}
 
원본 코드:
---
{original_code}
---
 
규칙:
1. 최소한의 변경으로 취약점을 수정합니다
2. 기존 기능을 유지합니다
3. 새로운 취약점을 도입하지 않습니다
4. 수정 이유를 설명합니다
5. 변경으로 인한 영향(breaking changes)을 나열합니다
6. 수정을 검증하는 테스트 케이스를 제안합니다"""
 
 
class SecurityFixer:
    """보안 취약점 자동 수정기"""
 
    def __init__(self, llm_client):
        self.llm_client = llm_client
 
    async def generate_fix(
        self, vulnerability: Vulnerability, source: str
    ) -> SecurityFix:
        prompt = FIX_GENERATION_PROMPT.format(
            vuln_description=vulnerability.description,
            severity=vulnerability.severity.value,
            owasp_category=vulnerability.vuln_type.value,
            original_code=source,
        )
 
        response = await self.llm_client.generate(prompt)
        parsed = self._parse_fix(response)
 
        # 수정된 코드가 새로운 취약점을 도입하지 않는지 검증
        await self._verify_no_new_vulnerabilities(
            parsed["fixed_code"],
            vulnerability.file_path,
        )
 
        return SecurityFix(
            vulnerability=vulnerability,
            original_code=source,
            fixed_code=parsed["fixed_code"],
            explanation=parsed["explanation"],
            breaking_changes=parsed["breaking_changes"],
            test_cases=parsed["test_cases"],
        )
 
    async def _verify_no_new_vulnerabilities(
        self, fixed_code: str, filepath: str
    ):
        """수정된 코드에 새로운 취약점이 없는지 검증"""
        # SAST 재스캔
        new_findings = self.sast_scanner.scan(filepath, fixed_code)
        if new_findings:
            raise SecurityError(
                "수정된 코드에서 새로운 취약점이 발견되었습니다: "
                + str(new_findings)
            )
 
    def _parse_fix(self, response: str) -> dict:
        return {
            "fixed_code": "",
            "explanation": "",
            "breaking_changes": [],
            "test_cases": [],
        }

@dataclass
class SecurityReviewComment:
    file_path: str
    line_number: int
    severity: str
    comment: str
    suggestion: str | None
 
 
class SecurityReviewer:
    """PR 보안 자동 리뷰"""
 
    def __init__(self, security_analyzer, github_client):
        self.analyzer = security_analyzer
        self.github = github_client
 
    async def review_pr(self, pr_number: int) -> list[SecurityReviewComment]:
        # PR의 변경된 파일 목록 가져오기
        changed_files = await self.github.get_pr_files(pr_number)
        comments = []
 
        for file_info in changed_files:
            if not self._is_code_file(file_info["filename"]):
                continue
 
            # 변경된 코드만 분석
            patch_content = file_info.get("patch", "")
            added_lines = self._extract_added_lines(patch_content)
 
            if not added_lines:
                continue
 
            # 보안 분석 실행
            vulns = await self.analyzer.analyze(
                filepath=file_info["filename"],
                source=added_lines,
            )
 
            for vuln in vulns:
                comments.append(SecurityReviewComment(
                    file_path=vuln.file_path,
                    line_number=vuln.line_number,
                    severity=vuln.severity.value,
                    comment=(
                        f"[{vuln.severity.value.upper()}] "
                        f"{vuln.description}"
                    ),
                    suggestion=vuln.fix_code,
                ))
 
        # GitHub PR에 코멘트 작성
        if comments:
            await self._post_review(pr_number, comments)
 
        return comments
 
    def _is_code_file(self, filename: str) -> bool:
        code_extensions = {
            ".py", ".ts", ".tsx", ".js", ".jsx",
            ".java", ".go", ".rs",
        }
        return any(filename.endswith(ext) for ext in code_extensions)
 
    def _extract_added_lines(self, patch: str) -> str:
        lines = []
        for line in patch.splitlines():
            if line.startswith("+") and not line.startswith("+++"):
                lines.append(line[1:])
        return "\n".join(lines)
 
    async def _post_review(
        self,
        pr_number: int,
        comments: list[SecurityReviewComment],
    ):
        review_body = self._build_review_summary(comments)
        await self.github.create_review(
            pr_number=pr_number,
            body=review_body,
            comments=[
                {
                    "path": c.file_path,
                    "line": c.line_number,
                    "body": f"{c.comment}\n\n제안: {c.suggestion}"
                    if c.suggestion else c.comment,
                }
                for c in comments
            ],
        )
 
    def _build_review_summary(
        self, comments: list[SecurityReviewComment]
    ) -> str:
        critical = sum(1 for c in comments if c.severity == "critical")
        high = sum(1 for c in comments if c.severity == "high")
        medium = sum(1 for c in comments if c.severity == "medium")
 
        return (
            f"보안 분석 결과: "
            f"Critical {critical}건, High {high}건, Medium {medium}건"
        )

# GitHub Actions 보안 게이트 예시
name: Security Gate
 
on:
  pull_request:
    branches: [main]
 
jobs:
  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: SAST 스캔
        uses: checkmarx/ast-github-action@v2
        with:
          project_name: my-project
          cx_tenant: ${{ secrets.CX_TENANT }}
 
      - name: 의존성 취약점 스캔
        uses: snyk/actions/node@master
        with:
          args: --severity-threshold=high
 
      - name: LLM 보안 분석
        run: |
          python scripts/llm_security_review.py \
            --pr-number=${{ github.event.pull_request.number }} \
            --threshold=medium
 
      - name: 보안 게이트 판정
        run: |
          python scripts/security_gate.py \
            --block-on=critical,high \
            --warn-on=medium

from dataclasses import dataclass
 
 
@dataclass
class GatePolicy:
    block_severities: list[str]
    warn_severities: list[str]
    max_open_vulns: int
    require_fix_plan: bool
 
 
class SecurityGate:
    """CI/CD 보안 게이트"""
 
    def __init__(self, policy: GatePolicy):
        self.policy = policy
 
    def evaluate(self, vulnerabilities: list[dict]) -> dict:
        blocked = [
            v for v in vulnerabilities
            if v["severity"] in self.policy.block_severities
        ]
        warnings = [
            v for v in vulnerabilities
            if v["severity"] in self.policy.warn_severities
        ]
 
        passed = len(blocked) == 0
 
        return {
            "passed": passed,
            "blocked_count": len(blocked),
            "warning_count": len(warnings),
            "message": self._build_message(passed, blocked, warnings),
        }
 
    def _build_message(
        self,
        passed: bool,
        blocked: list[dict],
        warnings: list[dict],
    ) -> str:
        if passed and not warnings:
            return "보안 게이트 통과: 취약점이 발견되지 않았습니다."
 
        parts = []
        if blocked:
            parts.append(
                f"차단: {len(blocked)}건의 심각한 취약점이 발견되었습니다."
            )
        if warnings:
            parts.append(
                f"경고: {len(warnings)}건의 중간 수준 취약점이 있습니다."
            )
 
        return " ".join(parts)