#!/usr/bin/env python3 """ Code Reviewer - 코드 보안 검사 기존 red-team-auto-review.py의 기능을 모듈화하여 재구현 """ import json import re import subprocess import sys from pathlib import Path from typing import Any, Dict, List # 로깅 설정 try: from base_reviewer import ReviewerInterface from logging_config import log_error, log_execution, setup_logging from security_patterns import VULNERABILITY_PATTERNS except ImportError: sys.path.insert(0, str(Path(__file__).parent.parent)) from base_reviewer import ReviewerInterface from logging_config import log_error, log_execution, setup_logging from security_patterns import VULNERABILITY_PATTERNS class CodeReviewer(ReviewerInterface): """코드 검토자""" def __init__(self): """초기화""" super().__init__() self.logger = setup_logging("red_team.code_reviewer") log_execution(self.logger, "CodeReviewer initialized") def review(self, doc: str) -> Dict[str, Any]: """ 코드 파일 검토 Args: doc: 파일 경로 (인터페이스 통일을 위해 doc 사용) Returns: 검토 결과 """ try: # doc을 파일 경로로 처리 (인터페이스 통일) file_path = doc log_execution(self.logger, "Starting code review", {"file": file_path}) # 1. 파일 로드 content = self._load_file(file_path) if not content: error_msg = f"Failed to load file: {file_path}" log_error(self.logger, FileNotFoundError(error_msg), {"file": file_path}) return {"error": error_msg, "risk_level": "unknown"} # 2. 취약점 스캔 (security_patterns.py 사용) vulnerabilities = self._scan_vulnerabilities(content) # 3. 의존성 검사 dependencies = self._check_dependencies(file_path) # 4. 아키텍처 리스크 (추가) arch_issues = self._assess_architecture_risks(content) # 리스크 평가 all_issues = vulnerabilities + arch_issues risk_level = self._assess_risk(all_issues) # 의존성 이슈 반영 if dependencies.get("vulnerable_deps", 0) > 0: if risk_level == "low": risk_level = "medium" # passed 기준: low/medium = passed, high/critical = failed # 이유: 아키텍처 리스크(200줄 초과)는 코드 품질 문제로 경고 수준 # 보안 취약점(SQL Injection 등)은 실제 위험으로 실패 처리 passed = risk_level in ["low", "medium"] result = { "risk_level": risk_level, "vulnerabilities": vulnerabilities, "vulnerability_count": len(vulnerabilities), "dependency_issues": dependencies.get("vulnerable_deps", 0), "architecture_issues": arch_issues, "recommendation": self._get_recommendation(risk_level), "passed": passed, } log_execution( self.logger, "Code review completed", {"file": file_path, "risk_level": risk_level, "vulnerability_count": len(vulnerabilities)}, ) return result except Exception as e: log_error(self.logger, e, {"file": doc}) return {"error": str(e), "risk_level": "critical", "passed": False} def _load_file(self, file_path: str) -> str | None: """파일 로드""" try: path = Path(file_path) if not path.exists(): return None with open(path, "r", encoding="utf-8") as f: return f.read() except Exception: return None def _scan_vulnerabilities(self, content: str) -> List[Dict]: """취약점 스캔 (security_patterns.py 사용)""" vulnerabilities = [] for vuln_type, patterns in VULNERABILITY_PATTERNS.items(): if vuln_type == "Insecure Dependencies": continue # 의존성은 별도 검사 for pattern in patterns: matches = re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE) for match in matches: vulnerabilities.append( { "type": vuln_type, "line": content[: match.start()].count("\n") + 1, "severity": self._get_severity(vuln_type), "snippet": match.group(), "pattern": pattern, } ) return vulnerabilities def _check_dependencies(self, file_path: str) -> Dict: """의존성 검사""" results = {"vulnerable_deps": 0, "dependencies": [], "tool_used": None} # Python 파일인 경우 requirements.txt 확인 if file_path.endswith(".py"): req_file = Path(file_path).parent / "requirements.txt" if req_file.exists(): # pip-audit 사용 가능한지 확인 if self._check_tool_available("pip-audit"): try: result = subprocess.run( ["pip-audit", "-r", str(req_file)], capture_output=True, text=True, timeout=30 ) if result.returncode != 0: results["vulnerable_deps"] = result.stdout.count("Vuln") results["tool_used"] = "pip-audit" except Exception: pass return results def _assess_architecture_risks(self, content: str) -> List[Dict]: """아키텍처 리스크 평가""" issues = [] # 파일 크기 검사 lines = content.count("\n") + 1 if lines > 200: issues.append( { "type": "Large File", "description": f"파일이 {lines}줄로 200줄을 초과 (AI 친화적 크기 원칙 위반)", "severity": "medium", "recommendation": "모듈로 분리 권장", } ) # 클래스 개수 검사 class_count = len(re.findall(r"^class\s+\w+", content, re.MULTILINE)) if class_count > 3: issues.append( { "type": "Multiple Classes", "description": f"파일에 {class_count}개 클래스 존재 (단일 책임 원칙 위반 가능성)", "severity": "low", "recommendation": "각 클래스를 별도 파일로 분리 검토", } ) return issues def _get_severity(self, vuln_type: str) -> str: """취약점 심각도 반환""" severity_map = { "SQL Injection": "high", "XSS (Cross-Site Scripting)": "medium", "Hardcoded Secret": "high", "Code Injection": "critical", "Path Traversal": "medium", "Command Injection": "critical", "Insecure Dependencies": "medium", } return severity_map.get(vuln_type, "low") def _check_tool_available(self, tool: str) -> bool: """외부 도구 사용 가능 여부 확인""" try: result = subprocess.run(["which", tool], capture_output=True, timeout=5) return result.returncode == 0 # returncode 체크 except Exception: return False def main(): """CLI 인터페이스""" if len(sys.argv) < 2: print(json.dumps({"error": "Usage: python3 code_reviewer.py [file]", "commands": ["test", "scan"]})) sys.exit(1) command = sys.argv[1] if command == "test": # 테스트 모드 print( json.dumps( {"status": "loaded", "reviewer": "CodeReviewer", "usage": "python3 code_reviewer.py scan "}, indent=2, ) ) elif command == "scan": # 실제 검사 if len(sys.argv) < 3: print( json.dumps( {"error": "scan command requires file path", "usage": "python3 code_reviewer.py scan "} ) ) sys.exit(1) file_path = sys.argv[2] reviewer = CodeReviewer() result = reviewer.review(file_path) print(json.dumps(result, indent=2, ensure_ascii=False)) else: print(json.dumps({"error": f"Unknown command: {command}", "commands": ["test", "scan"]})) sys.exit(1) if __name__ == "__main__": main()