diff --git a/Mirror Flowers.png b/Mirror Flowers.png new file mode 100644 index 0000000..f9c82b2 Binary files /dev/null and b/Mirror Flowers.png differ diff --git a/README.MD b/README.MD new file mode 100644 index 0000000..0368a7f --- /dev/null +++ b/README.MD @@ -0,0 +1,160 @@ +# Mirror Flowers (镜花) + +![image-20250205181045094](https://raw.githubusercontent.com/Ky0toFu/Mirror-Flowers/refs/heads/main/Mirror%20Flowers/Mirror%20Flowers.png) + +MirrorFlower(镜花)是一款基于 AI 的代码安全审计工具，支持多种编程语言的代码分析，可以帮助开发者快速发现代码中的潜在安全漏洞。支持DeepSeek-R1，ChatGPT-4o等多种大模型。 + +## 支持的API接口 + +FREEGPTAPI：https://github.com/popjane/free_chatgpt_api +SiliconFlow(硅基流动)：https://cloud.siliconflow.cn/i/JzMCyiJ3 + +如需要使用GPT大模型则使用FREEGPTAPI，使用DeepSeek-R1大模型则使用SiliconFlow API。 + +SiliconFlow(硅基流动)注册可免费领取14元使用额度，可通过SMS接码平台注册账号，理论可无限免费使用API KEY。 + +## 核心功能 + +### 多语言支持 +- PHP +- Java +- Python +- JavaScript + +### 依赖分析 +- Maven (pom.xml) +- NPM (package.json) +- Python (requirements.txt) +- Composer (composer.json) + +### 框架安全分析 +- Spring Framework + - Spring Security配置审计 + - 权限控制检查 + - CSRF/XSS防护 + - 跨域配置检查 + +- Django + - 中间件配置检查 + - CSRF Token验证 + - XSS防护配置 + - 调试模式检查 + +- Express.js + - 安全中间件配置 + - 认证机制检查 + - CORS配置审计 + - 输入验证检查 + +- Hibernate + - HQL注入检测 + - 缓存配置审计 + - 实体验证检查 + - 会话管理检查 + +### 安全检查特性 + +#### PHP安全检查 +- 文件包含漏洞（include/require） +- SQL注入（mysql_*函数） +- 命令注入（system, exec, shell_exec） +- 文件上传漏洞 +- 反序列化漏洞（unserialize） +- XSS（echo, print） +- SSRF漏洞 +- 目录遍历 +- 会话管理问题 +- 配置文件泄露 + +#### Java安全检查 +- SQL注入（PreparedStatement相关） +- 命令注入（Runtime.exec, ProcessBuilder） +- XXE漏洞（XML解析器配置） +- 反序列化漏洞（readObject） +- 不安全的文件操作 +- CSRF/XSS防护 +- 权限控制缺陷 +- 线程安全问题 +- 密码学实现缺陷 +- 日志信息泄露 + +#### Python安全检查 +- 不安全的反序列化（pickle.loads, yaml.load） +- 命令注入（os.system, eval, exec, subprocess） +- 不安全的模块导入（__import__） +- SQL注入（字符串格式化, execute） +- 路径遍历（open, os.path） +- 模板注入（render_template_string） +- 密码学实现问题（weak random） +- 环境变量泄露 +- 调试配置泄露 +- 不安全的依赖加载 + +#### JavaScript安全检查 +- XSS（DOM型和反射型） +- 原型污染 +- 不安全的第三方依赖 +- 客户端存储安全 +- 不安全的随机数生成 +- CSRF防护缺失 +- 跨域配置问题 +- 敏感信息泄露 +- 不安全的正则表达式 +- 事件监听器泄露 + +### 分析器组件 +- TaintAnalyzer: 污点分析和数据流追踪 +- SecurityAnalyzer: 通用安全问题检测 +- DependencyAnalyzer: 依赖组件安全分析 +- FrameworkAnalyzer: 框架特定安全检查 +- ConfigAnalyzer: 配置文件安全分析 +- ContextAnalyzer: 上下文感知分析 + +## 安装和配置 + +### 环境要求 +- Python 3.8+ +- FastAPI +- Node.js (前端开发) + +### 快速开始 + +1. 克隆项目 +```bash +git clone https://github.com/Ky0toFu/Mirror-Flowers.git +cd Mirror-Flowers +``` + +2. 安装依赖 +```bash +pip install -r requirements.txt +``` + +3. 配置环境变量 +```env +OPENAI_API_KEY=your_api_key_here +OPENAI_API_BASE=your_api_base_url +OPENAI_MODEL=your_preferred_model +``` + +4. 启动服务 +```bash +uvicorn backend.app:app --reload +``` + +## 注意事项 + +1. 文件大小限制：10MB +2. 支持的文件类型：.php, .java, .py, .js +3. API密钥安全：请妥善保管API密钥 +4. 分析时间：大型项目可能需要较长时间 +5. 结果验证：建议结合人工审查 +6. API配置问题：如果提示"API配置更新错误"或"不存在某个模型"，请检查API Base URL格式，尝试在URL后添加/v1或删除/v1（例如：https://api.example.com 或 https://api.example.com/v1） + +## 许可证 + +MIT License + +## 联系方式 + +如有问题或建议，请通过 Issue 与我联系。 diff --git a/backend/__pycache__/app.cpython-313.pyc b/backend/__pycache__/app.cpython-313.pyc new file mode 100644 index 0000000..7cecd47 Binary files /dev/null and b/backend/__pycache__/app.cpython-313.pyc differ diff --git a/backend/app.py b/backend/app.py new file mode 100644 index 0000000..9833dff --- /dev/null +++ b/backend/app.py @@ -0,0 +1,1119 @@ +from fastapi import FastAPI, UploadFile, File, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +from pydantic import BaseModel, validator +from openai import AsyncOpenAI +import os +from typing import List, Optional, Dict, Set, Tuple +import aiohttp +import logging +from pydantic_settings import BaseSettings +from fastapi.staticfiles import StaticFiles +from fastapi.responses import FileResponse +from enum import Enum +from urllib.parse import urlparse +import asyncio +import zipfile +from pathlib import Path +import tempfile +import json + +# 配置类 +class ModelType(str, Enum): + GPT35 = "gpt-3.5-turbo" + GPT4 = "gpt-4" + CLAUDE = "claude-2" + CLAUDE3 = "claude-3" + +class Settings(BaseSettings): + OPENAI_API_KEY: str = "sk-LshpT3NTpaT3HDw317634b552f22499c9aE9E75e003b9fA3" + OPENAI_API_BASE: str = "https://free.v36.cm/v1" + OPENAI_MODEL: str = "gpt-4o-mini-2024-07-18" + + # 更新可用模型配置，按域名和类型分组 + AVAILABLE_MODELS: Dict[str, Dict[str, List[str]]] = { + "guji.ai": { + "Chat": [ + "gpt-3.5-turbo", + "gpt-3.5-turbo-1106", + "gpt-3.5-turbo-0125", + "gpt-3.5-turbo-16k", + "gpt-4", + "gpt-4-1106-preview", + "gpt-4-0125-preview", + "gpt-4-32k", + "claude-2", + "claude-3", + "01-ai/Yi-1.5-6B-Chat", + "01-ai/Yi-1.5-9B-Chat-16K", + "01-ai/Yi-1.5-34B-Chat-16K", + "THUDM/chatglm3-6b", + "THUDM/glm-4-9b-chat", + "Qwen/Qwen2-7B-Instruct", + "Qwen/Qwen2-1.5B-Instruct", + "internlm/internlm2_5-7b-chat" + ], + "Embedding": [ + "BAAI/bge-large-en-v1.5", + "BAAI/bge-large-zh-v1.5", + "BAAI/bge-m3", + "netease-youdao/bce-embedding-base_v1" + ], + "Image": [ + "stabilityai/stable-diffusion-xl-base-1.0", + "stabilityai/stable-diffusion-2-1", + "stabilityai/stable-diffusion-3-medium", + "stabilityai/stable-diffusion-3-5-large", + "stabilityai/stable-diffusion-3-5-large-turbo" + ], + "Audio": [ + "FunAudioLLM/SenseVoiceSmall", + "fishaudio/fish-speech-1.4", + "fishaudio/fish-speech-1.5", + "FunAudioLLM/CosyVoice2-0.5B" + ], + "Pro": [ + "Pro/Qwen/Qwen2-7B-Instruct", + "Pro/Qwen/Qwen2-1.5B-Instruct", + "Pro/THUDM/glm-4-9b-chat", + "Pro/BAAI/bge-m3", + "Pro/OpenGVLab/InternVL2-8B" + ] + }, + "360.com": { + "Chat": [ + "360GPT_S2_V9", + "360GPT_S2_V9.4", + "360GPT_S2_V9.4-4K", + "360GPT_S2_V9.4-8K" + ] + }, + "v36.cm": { + "Chat": [ + "360GPT_S2_V9", + "360GPT_S2_V9.4", + "360GPT_S2_V9.4-4K", + "360GPT_S2_V9.4-8K" + ] + }, + "api.siliconflow.cn": { + "Chat": [ + "gpt-3.5-turbo", + "gpt-3.5-turbo-1106", + "gpt-3.5-turbo-0125", + "gpt-3.5-turbo-16k", + "gpt-4", + "gpt-4-1106-preview", + "gpt-4-0125-preview", + "gpt-4-32k", + "claude-2", + "claude-3", + "360GPT_S2_V9", + "360GPT_S2_V9.4", + "360GPT_S2_V9.4-4K", + "360GPT_S2_V9.4-8K", + "01-ai/Yi-1.5-6B-Chat", + "01-ai/Yi-1.5-9B-Chat-16K", + "01-ai/Yi-1.5-34B-Chat-16K", + "THUDM/chatglm3-6b", + "THUDM/glm-4-9b-chat", + "Qwen/Qwen2-7B-Instruct", + "Qwen/Qwen2-1.5B-Instruct", + "internlm/internlm2_5-7b-chat" + ] + } + } + + @validator('OPENAI_API_BASE') + def validate_api_base(cls, v): + v = v.rstrip('/') + if not v.endswith('/v1'): + v = v + '/v1' + return v + + async def fetch_models_from_api(self, api_base: str, api_key: str) -> Dict[str, List[str]]: + """从API获取可用模型列表""" + try: + async with aiohttp.ClientSession() as session: + headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json" + } + async with session.get(f"{api_base}/models", headers=headers) as response: + if response.status == 200: + data = await response.json() + models = data.get("data", []) + + # 按类型分组模型 + grouped_models = { + "Chat": [], + "Embedding": [], + "Image": [], + "Audio": [], + "Pro": [] + } + + for model in models: + model_id = model.get("id") + if not model_id: + continue + + # 根据模型ID分类 + if model_id.startswith("360GPT"): + grouped_models["Chat"].append(model_id) + elif "embedding" in model_id.lower(): + grouped_models["Embedding"].append(model_id) + elif any(img_key in model_id.lower() for img_key in ["diffusion", "dall-e"]): + grouped_models["Image"].append(model_id) + elif any(audio_key in model_id.lower() for audio_key in ["speech", "voice", "audio"]): + grouped_models["Audio"].append(model_id) + elif model_id.startswith("Pro/"): + grouped_models["Pro"].append(model_id) + else: + grouped_models["Chat"].append(model_id) + + # 移除空类别 + return {k: v for k, v in grouped_models.items() if v} + else: + error_data = await response.text() + logger.error(f"获取模型列表失败: {error_data}") + return self.get_default_models(api_base) + + except Exception as e: + logger.error(f"获取模型列表时出错: {str(e)}") + return self.get_default_models(api_base) + + def get_default_models(self, api_base: str) -> Dict[str, List[str]]: + """根据API地址获取默认模型列表""" + try: + parsed_url = urlparse(api_base) + domain = parsed_url.netloc.lower() + + # 提取基本域名 + base_domain = '.'.join(domain.split('.')[-2:]) + + # 尝试直接匹配域名 + if domain in self.AVAILABLE_MODELS: + logger.info(f"直接匹配到域名: {domain}") + return self.AVAILABLE_MODELS[domain] + + # 尝试匹配基本域名 + if base_domain in self.AVAILABLE_MODELS: + logger.info(f"匹配到基本域名: {base_domain}") + return self.AVAILABLE_MODELS[base_domain] + + # 尝试部分匹配 + for configured_domain in self.AVAILABLE_MODELS.keys(): + if (configured_domain in domain or + domain in configured_domain or + configured_domain in base_domain or + base_domain in configured_domain): + logger.info(f"部分匹配到域名: {configured_domain}") + return self.AVAILABLE_MODELS[configured_domain] + + # 如果是特定域名，返回对应配置 + if "v36.cm" in domain: + logger.info("匹配到 v36.cm 域名") + return self.AVAILABLE_MODELS["v36.cm"] + if "360.com" in domain: + logger.info("匹配到 360.com 域名") + return self.AVAILABLE_MODELS["360.com"] + if "siliconflow.cn" in domain: + logger.info("匹配到 siliconflow.cn 域名") + return self.AVAILABLE_MODELS["api.siliconflow.cn"] + + logger.warning(f"未找到匹配的域名配置: {domain}") + return {"Chat": ["gpt-3.5-turbo"]} + + except Exception as e: + logger.error(f"获取默认模型列表时出错: {str(e)}") + return {"Chat": ["gpt-3.5-turbo"]} + + async def get_models_for_api(self, api_base: str, api_key: str = None) -> Dict[str, List[str]]: + """获取指定API地址支持的模型列表""" + try: + # 尝试从API获取模型列表 + if api_key: + try: + models = await self.fetch_models_from_api(api_base, api_key) + if models: + logger.info(f"从API获取到模型列表: {models}") + return models + except Exception as e: + logger.warning(f"从API获取模型列表失败: {str(e)}") + + # 如果API获取失败，使用预配置的模型列表 + return self.get_default_models(api_base) + + except Exception as e: + logger.error(f"获取模型列表时出错: {str(e)}") + return {"Chat": ["gpt-3.5-turbo"]} + + class Config: + env_file = ".env" + case_sensitive = True + +settings = Settings() + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +app = FastAPI(title="代码审计工具API") + +# 配置CORS +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# 挂载静态文件目录（如果前端文件在static目录下） +static_dir = os.path.join(os.path.dirname(__file__), "static") +if os.path.exists(static_dir): + app.mount("/static", StaticFiles(directory=static_dir), name="static") + +# 添加新的数据模型 +class ProjectAuditResult(BaseModel): + file_path: str + language: str + vulnerabilities: List[dict] + related_files: List[str] + context_analysis: str + +class ProjectAnalysis: + def __init__(self): + self.file_dependencies: Dict[str, Set[str]] = {} # 文件依赖关系 + self.shared_variables: Dict[str, Set[str]] = {} # 共享变量 + self.function_calls: Dict[str, Set[str]] = {} # 函数调用关系 + self.vulnerability_context: Dict[str, List[dict]] = {} # 漏洞上下文 + + def add_dependency(self, file: str, depends_on: str): + if file not in self.file_dependencies: + self.file_dependencies[file] = set() + self.file_dependencies[file].add(depends_on) + + def add_shared_variable(self, file: str, variable: str): + if file not in self.shared_variables: + self.shared_variables[file] = set() + self.shared_variables[file].add(variable) + + def add_function_call(self, source_file: str, target_file: str): + if source_file not in self.function_calls: + self.function_calls[source_file] = set() + self.function_calls[source_file].add(target_file) + + def get_related_files(self, file: str) -> Set[str]: + """获取与指定文件相关的所有文件""" + related = set() + if file in self.file_dependencies: + related.update(self.file_dependencies[file]) + if file in self.function_calls: + related.update(self.function_calls[file]) + return related + +class CodeAuditService: + def __init__(self): + """初始化服务""" + self.client = None + self.openai_api_key = None + self.api_base = None + self.model = None + # 初始化时设置基本配置，但不进行异步操作 + self._init_config(settings.OPENAI_API_KEY, settings.OPENAI_API_BASE) + self.project_analysis = ProjectAnalysis() + self.supported_languages = { + '.php': 'php', + '.java': 'java', + '.py': 'python', + '.js': 'javascript' + } + + def _init_config(self, api_key: str, api_base: str): + """初始化基本配置""" + self.openai_api_key = api_key + self.api_base = api_base.rstrip('/') + '/v1' if not api_base.endswith('/v1') else api_base + self.model = "gpt-3.5-turbo" # 设置默认模型 + + async def ensure_initialized(self): + """确保服务完全初始化""" + if not self.client: + await self.configure_openai(self.openai_api_key, self.api_base) + + async def configure_openai(self, api_key: str = None, api_base: str = None, model: str = None): + """配置OpenAI API设置""" + if not api_key and not self.openai_api_key: + raise ValueError("未设置OPENAI_API_KEY") + + # 更新配置 + if api_key: + self.openai_api_key = api_key + if api_base: + api_base = api_base.rstrip('/') + if not api_base.endswith('/v1'): + api_base = api_base + '/v1' + self.api_base = api_base + elif not self.api_base: + self.api_base = settings.OPENAI_API_BASE + + # 获取当前API地址支持的模型列表 + available_models = await settings.get_models_for_api(self.api_base, self.openai_api_key) + + # 设置模型 + if model: + model_found = False + for category_models in available_models.values(): + if model in category_models: + self.model = model + model_found = True + break + if not model_found: + raise ValueError(f"该API地址不支持模型: {model}") + elif not self.model: + if "Chat" in available_models and available_models["Chat"]: + self.model = available_models["Chat"][0] + else: + first_category = next(iter(available_models)) + if available_models[first_category]: + self.model = available_models[first_category][0] + else: + self.model = "gpt-3.5-turbo" + + try: + self.client = AsyncOpenAI( + api_key=self.openai_api_key, + base_url=self.api_base, + timeout=120.0, # 增加超时时间到120秒 + max_retries=5 # 增加重试次数 + ) + logger.info(f"OpenAI API已配置: {self.api_base}, 使用模型: {self.model}") + except Exception as e: + logger.error(f"OpenAI客户端配置失败: {str(e)}") + raise ValueError(f"API配置失败: {str(e)}") + + async def analyze_code(self, code: str, language: str, api_key: str = None, api_base: str = None) -> dict: + """分析代码，支持自定义API设置""" + try: + # 如果提供了新的API设置，重新配置 + if api_key or api_base: + self.configure_openai(api_key, api_base) + + # 验证语言支持 + if language not in self.supported_languages.values(): + raise HTTPException( + status_code=400, + detail=f"不支持的语言类型。支持的语言包括: {', '.join(self.supported_languages.values())}" + ) + + # 第一轮AI分析 + logger.info(f"开始第一轮{language}代码分析") + first_response = await self._get_openai_response( + self._generate_first_prompt(code, language) + ) + + # 第二轮AI验证 + logger.info("开始第二轮验证分析") + second_response = await self._get_openai_response( + self._generate_second_prompt(code, first_response) + ) + + return { + "first_analysis": first_response, + "second_analysis": second_response + } + except Exception as e: + logger.error(f"代码分析过程中发生错误: {str(e)}") + raise HTTPException(status_code=500, detail=str(e)) + + def _generate_first_prompt(self, code: str, language: str) -> str: + """生成第一轮分析的提示""" + language_specific_prompts = { + 'php': """ + 请特别注意以下PHP特有的安全问题： + 1. 文件包含漏洞（include/require） + 2. SQL注入（mysql_*函数） + 3. 命令注入（system, exec, shell_exec） + 4. 文件上传漏洞 + 5. 反序列化漏洞（unserialize） + 6. XSS（echo, print） + 7. SSRF漏洞 + 8. 目录遍历 + 9. 会话管理问题 + 10. 配置文件泄露 + """, + 'java': """ + 请特别注意以下Java特有的安全问题： + 1. SQL注入（PreparedStatement相关） + 2. 命令注入（Runtime.exec, ProcessBuilder） + 3. XXE漏洞（XML解析器配置） + 4. 反序列化漏洞（readObject） + 5. 不安全的文件操作 + 6. CSRF/XSS防护 + 7. 权限控制缺陷 + 8. 线程安全问题 + 9. 密码学实现缺陷 + 10. 日志信息泄露 + """, + 'python': """ + 请特别注意以下Python特有的安全问题： + 1. 不安全的反序列化（如pickle.loads） + 2. 命令注入（如os.system, eval） + 3. 不安全的模块导入 + 4. SQL注入（特别是字符串格式化） + 5. 路径遍历 + """, + 'javascript': """ + 请特别注意以下JavaScript特有的安全问题： + 1. XSS（DOM型和反射型） + 2. 原型污染 + 3. 不安全的第三方依赖 + 4. 客户端存储安全 + 5. 不安全的随机数生成 + """ + } + + base_prompt = f"""请分析以下{language}代码中的安全漏洞： + {code} + + 请详细说明每个潜在的安全问题，包括： + 1. 漏洞类型 + 2. 漏洞位置 + 3. 可能的影响 + 4. 修复建议 + """ + + if language in language_specific_prompts: + base_prompt += f"\n{language_specific_prompts[language]}" + + return base_prompt + + def _check_file_type(self, filename: str) -> str: + """检查文件类型是否支持""" + ext = os.path.splitext(filename)[1].lower() + if ext not in self.supported_languages: + supported_exts = ', '.join(self.supported_languages.keys()) + raise HTTPException( + status_code=400, + detail=f"不支持的文件类型。支持的文件类型包括: {supported_exts}" + ) + return self.supported_languages[ext] + + def _generate_second_prompt(self, code: str, first_response: str) -> str: + return f"""请验证以下代码审计结果的准确性，并提供可能的payload： + {first_response} + 代码： + {code}""" + + async def _get_openai_response(self, prompt: str) -> str: + if not self.client: + raise ValueError("OpenAI客户端未初始化") + + try: + logger.info(f"正在发送请求到: {self.api_base}, 使用模型: {self.model}") + + # 添加重试逻辑 + max_retries = 3 + retry_count = 0 + retry_delay = 1 # 初始延迟1秒 + + while retry_count < max_retries: + try: + response = await self.client.chat.completions.create( + model=self.model, + messages=[ + {"role": "system", "content": "你是一个专业的代码安全审计专家。"}, + {"role": "user", "content": prompt} + ], + temperature=0.7, + max_tokens=2000 + ) + + if hasattr(response, 'choices') and len(response.choices) > 0: + content = response.choices[0].message.content + logger.debug(f"收到响应: {content[:100]}...") + return content + else: + raise ValueError("API响应格式错误") + + except Exception as e: + retry_count += 1 + if retry_count >= max_retries: + raise + + logger.warning(f"请求失败，正在进行第{retry_count}次重试: {str(e)}") + await asyncio.sleep(retry_delay) + retry_delay *= 2 # 指数退避 + + except Exception as e: + error_msg = str(e) + logger.error(f"OpenAI API调用失败: {error_msg}") + + if "401" in error_msg: + raise HTTPException(status_code=401, detail="API密钥无效或未授权") + elif "timeout" in error_msg.lower(): + raise HTTPException(status_code=504, detail="API请求超时，请稍后重试") + elif "404" in error_msg: + raise HTTPException(status_code=404, detail="API端点不存在，请检查API基础URL") + else: + raise HTTPException(status_code=500, detail=f"AI分析服务错误: {error_msg}") + + async def analyze_project(self, zip_file: UploadFile) -> Dict[str, ProjectAuditResult]: + """分析整个项目代码""" + results = {} + + # 创建临时目录 + with tempfile.TemporaryDirectory() as temp_dir: + # 保存并解压ZIP文件 + zip_path = Path(temp_dir) / "project.zip" + with open(zip_path, "wb") as f: + content = await zip_file.read() + f.write(content) + + # 解压文件 + with zipfile.ZipFile(zip_path, 'r') as zip_ref: + zip_ref.extractall(temp_dir) + + # 首先进行项目结构分析 + await self._analyze_project_structure(temp_dir) + + # 分析每个文件 + for file_path in Path(temp_dir).rglob('*'): + if file_path.suffix in self.supported_languages: + rel_path = str(file_path.relative_to(temp_dir)) + try: + result = await self._analyze_file_with_context( + file_path, + self.project_analysis.get_related_files(rel_path) + ) + results[rel_path] = result + except Exception as e: + logger.error(f"分析文件 {rel_path} 时出错: {str(e)}") + + # 进行交叉验证和上下文关联分析 + await self._cross_validate_results(results) + + return results + + async def _analyze_project_structure(self, project_dir: str): + """分析项目结构，建立依赖关系""" + for file_path in Path(project_dir).rglob('*'): + if file_path.suffix not in self.supported_languages: + continue + + rel_path = str(file_path.relative_to(project_dir)) + content = file_path.read_text(errors='ignore') + + # 分析文件依赖 + await self._analyze_dependencies(rel_path, content) + + # 分析共享变量 + await self._analyze_shared_variables(rel_path, content) + + # 分析函数调用 + await self._analyze_function_calls(rel_path, content) + + async def _analyze_file_with_context( + self, + file_path: Path, + related_files: Set[str] + ) -> ProjectAuditResult: + """分析单个文件，考虑上下文""" + content = file_path.read_text(errors='ignore') + language = file_path.suffix[1:] # 移除点号 + + # 构建包含上下文的提示 + context_prompt = f"""请分析以下{language}代码，重点关注安全漏洞。请提供详细的JSON格式分析结果： + +代码内容： +{content} + +相关文件： +{', '.join(related_files)} + +请提供以下格式的分析结果： +{{ + "vulnerabilities": [ + {{ + "type": "漏洞类型（如SQL注入、XSS等）", + "location": "具体代码行号和代码片段", + "severity": "严重程度（高/中/低）", + "description": "详细的漏洞描述", + "impact": "潜在影响", + "fix": "修复建议", + "related_context": "相关的上下文信息" + }} + ], + "context_analysis": "整体代码安全性分析", + "related_files": {{ + "dependencies": ["相关的依赖文件"], + "includes": ["包含的文件"], + "functions": ["调用的函数"], + "affected_by": ["受影响的文件"], + "affects": ["可能影响的文件"] + }} +}} + +请特别注意： +1. 详细分析每个可能的漏洞 +2. 提供具体的代码位置 +3. 给出可行的修复建议 +4. 分析代码与其他文件的关联 +5. 考虑整体的安全影响 +""" + + # 获取AI分析结果 + analysis_result = await self._get_openai_response(context_prompt) + + # 解析结果 + try: + result_dict = json.loads(analysis_result) + + # 确保结果包含所有必要字段 + if 'vulnerabilities' not in result_dict: + result_dict['vulnerabilities'] = [] + if 'context_analysis' not in result_dict: + result_dict['context_analysis'] = "未提供分析结果" + if 'related_files' not in result_dict: + result_dict['related_files'] = { + "dependencies": [], + "includes": [], + "functions": [], + "affected_by": [], + "affects": [] + } + + # 转换相关文件格式 + related_files_list = [] + for category, files in result_dict['related_files'].items(): + if files: + related_files_list.extend([f"{category}: {file}" for file in files]) + + return ProjectAuditResult( + file_path=str(file_path), + language=language, + vulnerabilities=result_dict['vulnerabilities'], + related_files=related_files_list, + context_analysis=result_dict['context_analysis'] + ) + except json.JSONDecodeError: + # 如果结果不是JSON格式，返回基本结构 + return ProjectAuditResult( + file_path=str(file_path), + language=language, + vulnerabilities=[], + related_files=[], + context_analysis=analysis_result + ) + + async def _cross_validate_results(self, results: Dict[str, ProjectAuditResult]): + """交叉验证分析结果""" + # 收集所有漏洞 + all_vulnerabilities = [] + for result in results.values(): + all_vulnerabilities.extend(result.vulnerabilities) + + # 生成交叉验证提示 + validation_prompt = f"""请验证以下项目漏洞分析结果的准确性和完整性： + +发现的漏洞： +{json.dumps(all_vulnerabilities, indent=2, ensure_ascii=False)} + +请考虑： +1. 漏洞之间的关联性 +2. 漏洞的优先级 +3. 误报可能性 +4. 修复建议的可行性 +""" + + # 获取验证结果 + validation_result = await self._get_openai_response(validation_prompt) + + # 更新结果 + for result in results.values(): + result.context_analysis += f"\n\n交叉验证结果：\n{validation_result}" + + async def _analyze_dependencies(self, file_path: str, content: str): + """分析文件依赖关系""" + try: + language = Path(file_path).suffix[1:] + + # 根据不同语言分析依赖 + if language == 'php': + await self._analyze_php_dependencies(file_path, content) + elif language == 'java': + await self._analyze_java_dependencies(file_path, content) + elif language == 'py': + await self._analyze_python_dependencies(file_path, content) + elif language == 'js': + await self._analyze_js_dependencies(file_path, content) + + except Exception as e: + logger.error(f"分析文件依赖时出错 {file_path}: {str(e)}") + + async def _analyze_php_dependencies(self, file_path: str, content: str): + """分析PHP文件依赖""" + import re + patterns = [ + r'(?:include|require|include_once|require_once)\s*[\'"]([^\'"]+)[\'"]', + r'use\s+([^;]+)', + r'namespace\s+([^;{\s]+)' + ] + + for pattern in patterns: + matches = re.finditer(pattern, content) + for match in matches: + dependency = match.group(1) + self.project_analysis.add_dependency(file_path, dependency) + + async def _analyze_java_dependencies(self, file_path: str, content: str): + """分析Java文件依赖""" + import re + patterns = [ + r'import\s+([^;]+)', + r'extends\s+([^\s{]+)', + r'implements\s+([^{]+)' + ] + + for pattern in patterns: + matches = re.finditer(pattern, content) + for match in matches: + dependency = match.group(1) + self.project_analysis.add_dependency(file_path, dependency) + + async def _analyze_python_dependencies(self, file_path: str, content: str): + """分析Python文件依赖""" + import re + patterns = [ + r'(?:from|import)\s+([^\s]+)', + r'__import__\([\'"]([^\'"]+)[\'"]\)' + ] + + for pattern in patterns: + matches = re.finditer(pattern, content) + for match in matches: + dependency = match.group(1) + self.project_analysis.add_dependency(file_path, dependency) + + async def _analyze_js_dependencies(self, file_path: str, content: str): + """分析JavaScript文件依赖""" + import re + patterns = [ + r'(?:import|require)\s*\([\'"]([^\'"]+)[\'"]\)', + r'import\s+.*\s+from\s+[\'"]([^\'"]+)[\'"]', + r'import\s+[\'"]([^\'"]+)[\'"]' + ] + + for pattern in patterns: + matches = re.finditer(pattern, content) + for match in matches: + dependency = match.group(1) + self.project_analysis.add_dependency(file_path, dependency) + + async def _analyze_shared_variables(self, file_path: str, content: str): + """分析共享变量""" + try: + language = Path(file_path).suffix[1:] + + # 根据不同语言分析共享变量 + if language == 'php': + await self._analyze_php_shared_vars(file_path, content) + elif language == 'java': + await self._analyze_java_shared_vars(file_path, content) + elif language == 'py': + await self._analyze_python_shared_vars(file_path, content) + elif language == 'js': + await self._analyze_js_shared_vars(file_path, content) + + except Exception as e: + logger.error(f"分析共享变量时出错 {file_path}: {str(e)}") + + async def _analyze_php_shared_vars(self, file_path: str, content: str): + """分析PHP共享变量""" + import re + patterns = [ + r'\$GLOBALS\[[\'"](\w+)[\'"]\]', + r'\$_(?:GET|POST|REQUEST|SESSION|COOKIE)\[[\'"](\w+)[\'"]\]', + r'global\s+\$(\w+)' + ] + + for pattern in patterns: + matches = re.finditer(pattern, content) + for match in matches: + var_name = match.group(1) + self.project_analysis.add_shared_variable(file_path, var_name) + + async def _analyze_java_shared_vars(self, file_path: str, content: str): + """分析Java共享变量""" + import re + patterns = [ + r'static\s+(?:final\s+)?(?:\w+)\s+(\w+)', + r'public\s+(?:static\s+)?(?:\w+)\s+(\w+)' + ] + + for pattern in patterns: + matches = re.finditer(pattern, content) + for match in matches: + var_name = match.group(1) + self.project_analysis.add_shared_variable(file_path, var_name) + + async def _analyze_python_shared_vars(self, file_path: str, content: str): + """分析Python共享变量""" + import re + patterns = [ + r'global\s+(\w+)', + r'(\w+)\s*=\s*[^=]' # 简单的全局变量定义 + ] + + for pattern in patterns: + matches = re.finditer(pattern, content) + for match in matches: + var_name = match.group(1) + self.project_analysis.add_shared_variable(file_path, var_name) + + async def _analyze_js_shared_vars(self, file_path: str, content: str): + """分析JavaScript共享变量""" + import re + patterns = [ + r'(?:var|let|const)\s+(\w+)\s*=', + r'window\.(\w+)\s*=', + r'global\.(\w+)\s*=' + ] + + for pattern in patterns: + matches = re.finditer(pattern, content) + for match in matches: + var_name = match.group(1) + self.project_analysis.add_shared_variable(file_path, var_name) + + async def _analyze_function_calls(self, file_path: str, content: str): + """分析函数调用关系""" + try: + language = Path(file_path).suffix[1:] + + # 根据不同语言分析函数调用 + if language == 'php': + await self._analyze_php_function_calls(file_path, content) + elif language == 'java': + await self._analyze_java_function_calls(file_path, content) + elif language == 'py': + await self._analyze_python_function_calls(file_path, content) + elif language == 'js': + await self._analyze_js_function_calls(file_path, content) + + except Exception as e: + logger.error(f"分析函数调用时出错 {file_path}: {str(e)}") + + async def _analyze_php_function_calls(self, file_path: str, content: str): + """分析PHP函数调用""" + import re + pattern = r'(?:function\s+(\w+)|(\w+)\s*\()' + + matches = re.finditer(pattern, content) + for match in matches: + func_name = match.group(1) or match.group(2) + # 在项目中查找调用此函数的文件 + await self._find_function_callers(file_path, func_name) + + async def _analyze_java_function_calls(self, file_path: str, content: str): + """分析Java函数调用""" + import re + pattern = r'(?:public|private|protected)?\s*(?:static)?\s*\w+\s+(\w+)\s*\([^)]*\)' + + matches = re.finditer(pattern, content) + for match in matches: + func_name = match.group(1) + await self._find_function_callers(file_path, func_name) + + async def _analyze_python_function_calls(self, file_path: str, content: str): + """分析Python函数调用""" + import re + pattern = r'def\s+(\w+)\s*\(' + + matches = re.finditer(pattern, content) + for match in matches: + func_name = match.group(1) + await self._find_function_callers(file_path, func_name) + + async def _analyze_js_function_calls(self, file_path: str, content: str): + """分析JavaScript函数调用""" + import re + pattern = r'(?:function\s+(\w+)|(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s+)?function)' + + matches = re.finditer(pattern, content) + for match in matches: + func_name = match.group(1) or match.group(2) + await self._find_function_callers(file_path, func_name) + + async def _find_function_callers(self, source_file: str, function_name: str): + """查找调用指定函数的文件""" + # 这里可以实现更复杂的函数调用分析 + # 当前简单记录函数定义所在文件 + self.project_analysis.add_function_call(source_file, function_name) + +code_audit_service = CodeAuditService() + +# 添加请求体模型 +class ConfigureRequest(BaseModel): + api_key: str + api_base: Optional[str] = None + model: Optional[str] = None # 添加模型选择 + +# 添加获取可用模型的API +@app.get("/api/models") +async def get_available_models(): + """获取当前API地址支持的模型列表""" + try: + # 确保服务已初始化 + await code_audit_service.ensure_initialized() + + api_base = code_audit_service.api_base or settings.OPENAI_API_BASE + available_models = await settings.get_models_for_api( + api_base, + code_audit_service.openai_api_key or settings.OPENAI_API_KEY + ) + current_model = code_audit_service.model or settings.OPENAI_MODEL + + logger.info(f"当前API地址: {api_base}") + logger.info(f"可用模型: {available_models}") + logger.info(f"当前使用的模型: {current_model}") + + return { + "models": available_models, + "current_model": current_model + } + except Exception as e: + logger.error(f"获取模型列表失败: {str(e)}") + raise HTTPException(status_code=500, detail=str(e)) + +# 修改配置路由 +@app.post("/api/configure") +async def configure_api(config: ConfigureRequest): + """配置OpenAI API设置""" + try: + await code_audit_service.configure_openai( + config.api_key, + config.api_base, + config.model + ) + return { + "status": "success", + "message": "API配置已更新", + "model": code_audit_service.model + } + except Exception as e: + logger.error(f"配置更新失败: {str(e)}") + raise HTTPException(status_code=400, detail=str(e)) + +@app.post("/api/audit") +async def audit_code( + file: UploadFile = File(...), + api_key: str = None, + api_base: str = None +): + """审计代码，支持自定义API设置""" + try: + # 确保服务已初始化 + await code_audit_service.ensure_initialized() + + content = await file.read() + code = content.decode() + + # 获取文件扩展名并检查支持的类型 + file_extension = os.path.splitext(file.filename)[1].lower() + language = code_audit_service._check_file_type(file.filename) + + logger.info(f"开始分析{file.filename}") + + result = await code_audit_service.analyze_code(code, language, api_key, api_base) + return result + except UnicodeDecodeError: + raise HTTPException(status_code=400, detail="文件编码错误") + except Exception as e: + logger.error(f"处理文件时发生错误: {str(e)}") + if isinstance(e, HTTPException): + raise e + raise HTTPException(status_code=500, detail=str(e)) + +@app.post("/api/audit/project") +async def audit_project( + project: UploadFile = File(...), + api_key: str = None, + api_base: str = None +): + """审计整个项目代码""" + try: + # 确保服务已初始化 + await code_audit_service.ensure_initialized() + + # 验证文件类型 + if not project.filename.endswith('.zip'): + raise HTTPException(status_code=400, detail="请上传ZIP格式的项目文件") + + # 分析项目 + results = await code_audit_service.analyze_project(project) + + return { + "status": "success", + "results": results + } + except Exception as e: + logger.error(f"项目审计过程中发生错误: {str(e)}") + raise HTTPException(status_code=500, detail=str(e)) + +@app.get("/health") +async def health_check(): + return {"status": "healthy"} + +@app.get("/") +async def root(): + """ + 根路径处理程序，返回API基本信息 + """ + return { + "name": "代码审计工具API", + "version": "1.0.0", + "status": "running", + "endpoints": { + "audit": "/api/audit", + "configure": "/api/configure", + "health": "/health", + "docs": "/docs" + } + } + +@app.exception_handler(Exception) +async def global_exception_handler(request, exc): + """ + 全局异常处理 + """ + error_msg = str(exc) + logger.error(f"发生错误: {error_msg}") + return JSONResponse( + status_code=500, + content={ + "error": "内部服务器错误", + "detail": error_msg + } + ) + +@app.middleware("http") +async def log_requests(request, call_next): + """ + 请求日志中间件 + """ + logger.info(f"收到请求: {request.method} {request.url}") + response = await call_next(request) + logger.info(f"响应状态码: {response.status_code}") + return response + +@app.get("/ui") +async def serve_spa(): + """ + 服务前端单页应用 + """ + return FileResponse(os.path.join(static_dir, "index.html")) \ No newline at end of file diff --git a/backend/static/index.html b/backend/static/index.html new file mode 100644 index 0000000..dbcd7ec --- /dev/null +++ b/backend/static/index.html @@ -0,0 +1,924 @@ + + + + + + Mirror Flowers + + + + + + + + + +
+
+

Mirror Flowers

+
镜花 · 代码安全审计工具
+
+ + +
+
+
API配置
+
+
+
+
+ +
+
+ +
+
+ +
+
+ +
+
+
+
+ + +
+
+
代码审计
+
+
+
+ +
+ +
+ +
+
+ + +
+
+
+ + + +
+ + +
+
+ + + + + + +
+ + + + + + + \ No newline at end of file diff --git a/core/analyzers/config_analyzer.py b/core/analyzers/config_analyzer.py new file mode 100644 index 0000000..2b19951 --- /dev/null +++ b/core/analyzers/config_analyzer.py @@ -0,0 +1,55 @@ +class ConfigAnalyzer: + def __init__(self): + self.dangerous_configs = { + 'php': { + 'allow_url_fopen': 'On', + 'allow_url_include': 'On', + 'display_errors': 'On', + 'expose_php': 'On' + }, + 'java': { + 'debug': 'true', + 'trace': 'true', + 'security.basic.enabled': 'false' + } + } + + # 添加更多配置检查规则 + self.security_rules = { + 'authentication': { + 'required_settings': ['session.cookie_secure', 'session.cookie_httponly'], + 'forbidden_settings': ['session.use_only_cookies=0'] + }, + 'file_upload': { + 'check_settings': ['upload_max_filesize', 'max_file_uploads'], + 'risk_values': ['unlimited', '-1'] + }, + 'error_reporting': { + 'production_settings': { + 'display_errors': 'Off', + 'log_errors': 'On', + 'error_reporting': 'E_ALL & ~E_DEPRECATED & ~E_STRICT' + } + } + } + + def analyze(self, config_files): + """增强的配置分析""" + issues = [] + + for file_path in config_files: + config_type = self._detect_config_type(file_path) + + # 基本配置检查 + basic_issues = self._check_dangerous_settings(file_path, config_type) + issues.extend(basic_issues) + + # 环境特定检查 + env_issues = self._check_environment_specific(file_path, config_type) + issues.extend(env_issues) + + # 安全规则检查 + security_issues = self._check_security_rules(file_path, config_type) + issues.extend(security_issues) + + return self._prioritize_issues(issues) \ No newline at end of file diff --git a/core/analyzers/context_analyzer.py b/core/analyzers/context_analyzer.py new file mode 100644 index 0000000..2660b7d --- /dev/null +++ b/core/analyzers/context_analyzer.py @@ -0,0 +1,47 @@ +class ContextAnalyzer: + def __init__(self): + self.file_dependencies = {} + self.global_variables = {} + self.function_calls = {} + self.class_hierarchy = {} + + def analyze_project_context(self, files): + """分析项目整体上下文""" + for file_path in files: + self._analyze_file_context(file_path) + + def _analyze_file_context(self, file_path): + """分析单个文件的上下文""" + with open(file_path, 'r') as f: + content = f.read() + + # 分析文件依赖 + self._analyze_dependencies(content, file_path) + + # 分析全局变量 + self._analyze_globals(content, file_path) + + # 分析函数调用关系 + self._analyze_function_calls(content, file_path) + + # 分析类继承关系 + self._analyze_class_hierarchy(content, file_path) + + def get_call_graph(self, function_name): + """获取函数调用图""" + call_graph = { + 'name': function_name, + 'calls': self.function_calls.get(function_name, []), + 'called_by': self._find_callers(function_name) + } + return call_graph + + def get_variable_scope(self, variable_name): + """获取变量作用域""" + if variable_name in self.global_variables: + return { + 'type': 'global', + 'defined_in': self.global_variables[variable_name], + 'used_in': self._find_variable_usage(variable_name) + } + return None \ No newline at end of file diff --git a/core/analyzers/dependency_analyzer.py b/core/analyzers/dependency_analyzer.py new file mode 100644 index 0000000..52930ba --- /dev/null +++ b/core/analyzers/dependency_analyzer.py @@ -0,0 +1,98 @@ +import json + +class DependencyAnalyzer: + def __init__(self): + self.vulnerability_database = self._load_vulnerability_database() + self.dependency_graph = {} + + def analyze_dependencies(self, project_files): + """完整的项目依赖分析""" + issues = [] + for file in project_files: + if file.endswith('pom.xml'): + issues.extend(self._analyze_maven_dependencies(file)) + elif file.endswith('package.json'): + issues.extend(self._analyze_npm_dependencies(file)) + elif file.endswith('requirements.txt'): + issues.extend(self._analyze_python_dependencies(file)) + elif file.endswith('composer.json'): + issues.extend(self._analyze_composer_dependencies(file)) + return issues + + def _analyze_maven_dependencies(self, pom_file): + """分析Maven依赖""" + try: + with open(pom_file, 'r') as f: + content = f.read() + dependencies = self._parse_pom_xml(content) + return self._check_dependencies(dependencies, 'maven') + except Exception as e: + return [{'error': f'Maven依赖分析失败: {str(e)}'}] + + def _analyze_npm_dependencies(self, package_file): + """分析NPM依赖""" + try: + with open(package_file, 'r') as f: + data = json.load(f) + dependencies = {**data.get('dependencies', {}), **data.get('devDependencies', {})} + return self._check_dependencies(dependencies, 'npm') + except Exception as e: + return [{'error': f'NPM依赖分析失败: {str(e)}'}] + + def _analyze_python_dependencies(self, requirements_file): + """分析Python依赖""" + try: + with open(requirements_file, 'r') as f: + dependencies = {} + for line in f: + if '==' in line: + name, version = line.strip().split('==') + dependencies[name] = version + return self._check_dependencies(dependencies, 'python') + except Exception as e: + return [{'error': f'Python依赖分析失败: {str(e)}'}] + + def _check_dependencies(self, dependencies, ecosystem): + """检查依赖的安全问题""" + issues = [] + for name, version in dependencies.items(): + # 检查已知漏洞 + vulns = self._check_known_vulnerabilities(name, version, ecosystem) + if vulns: + issues.extend(vulns) + + # 检查版本过时 + if self._is_outdated_version(name, version, ecosystem): + issues.append({ + 'type': 'outdated_dependency', + 'name': name, + 'current_version': version, + 'latest_version': self._get_latest_version(name, ecosystem), + 'ecosystem': ecosystem, + 'severity': 'medium', + 'recommendation': '建议更新到最新的稳定版本' + }) + + # 检查许可证 + license_issue = self._check_license_compatibility(name, ecosystem) + if license_issue: + issues.append(license_issue) + + return issues + + def analyze(self, dependencies): + """分析项目依赖中的安全问题""" + vulnerabilities = [] + + for dep in dependencies: + if dep in self.vulnerability_database: + vulnerabilities.append({ + 'type': 'vulnerable_dependency', + 'name': dep, + 'version': dependencies[dep], + 'known_vulnerabilities': self.vulnerability_database[dep], + 'severity': 'high', + 'recommendation': '更新到最新的安全版本' + }) + + return vulnerabilities \ No newline at end of file diff --git a/core/analyzers/framework_analyzer.py b/core/analyzers/framework_analyzer.py new file mode 100644 index 0000000..c039c83 --- /dev/null +++ b/core/analyzers/framework_analyzer.py @@ -0,0 +1,189 @@ +class FrameworkAnalyzer: + def __init__(self): + self.framework_rules = { + 'spring': { + 'security_checks': { + 'csrf': [ + r'@EnableWebSecurity(?!.*csrf\(\)\.disable\(\))', + r'csrf\(\)\.disable\(\)' + ], + 'auth': [ + r'@PreAuthorize', + r'@Secured', + r'SecurityContextHolder' + ], + 'cors': [ + r'@CrossOrigin\(.*allowCredentials\s*=\s*true.*\)', + r'addCorsMappings\(' + ] + }, + 'dangerous_configs': [ + r'security\.basic\.enabled\s*=\s*false', + r'management\.security\.enabled\s*=\s*false' + ] + }, + 'django': { + 'security_checks': { + 'csrf': [ + r'@csrf_exempt', + r'CSRF_COOKIE_SECURE\s*=\s*False' + ], + 'auth': [ + r'@login_required', + r'@permission_required' + ], + 'xss': [ + r'mark_safe\(', + r'safe\s+filter' + ] + }, + 'dangerous_configs': [ + r'DEBUG\s*=\s*True', + r'ALLOWED_HOSTS\s*=\s*\[\s*\'\*\'\s*\]' + ] + }, + 'express': { + 'security_checks': { + 'helmet': [ + r'app\.use\(helmet\(\)\)', + r'app\.use\(cors\(\)\)' + ], + 'auth': [ + r'passport\.authenticate', + r'jwt\.verify' + ], + 'input': [ + r'body-parser', + r'express-validator' + ] + }, + 'dangerous_configs': [ + r'app\.disable\(.*trust\s*proxy.*\)', + r'app\.use\(bodyParser\.raw\(\)\)' + ] + }, + 'hibernate': { + 'security_checks': { + 'sql': [ + r'createQuery\(.*\+.*\)', + r'createSQLQuery\(.*\+.*\)' + ], + 'cache': [ + r'@Cache\(', + r'setCacheable\(' + ], + 'validation': [ + r'@Valid', + r'@Validated' + ] + }, + 'dangerous_configs': [ + r'show_sql\s*=\s*true', + r'hibernate\.format_sql\s*=\s*true' + ] + } + } + + def analyze_framework(self, code: str, framework: str) -> dict: + """分析框架特定的安全问题""" + if framework not in self.framework_rules: + return {"error": f"不支持的框架: {framework}"} + + issues = [] + rules = self.framework_rules[framework] + + # 检查安全配置 + for check_type, patterns in rules['security_checks'].items(): + for pattern in patterns: + matches = re.finditer(pattern, code) + for match in matches: + issues.append({ + 'type': f'{framework}_{check_type}_issue', + 'pattern': pattern, + 'location': self._get_location(code, match.start()), + 'description': self._get_issue_description(framework, check_type, pattern), + 'severity': self._calculate_severity(framework, check_type) + }) + + # 检查危险配置 + for pattern in rules['dangerous_configs']: + matches = re.finditer(pattern, code) + for match in matches: + issues.append({ + 'type': f'{framework}_dangerous_config', + 'pattern': pattern, + 'location': self._get_location(code, match.start()), + 'description': f'发现危险的{framework}配置', + 'severity': 'high' + }) + + return { + 'framework': framework, + 'issues': issues, + 'analysis_summary': self._generate_summary(issues) + } + + def _get_location(self, code: str, pos: int) -> dict: + """获取代码位置信息""" + lines = code[:pos].splitlines() + return { + 'line': len(lines), + 'column': len(lines[-1]) if lines else 0 + } + + def _calculate_severity(self, framework: str, check_type: str) -> str: + """计算问题严重程度""" + high_severity = { + 'spring': ['csrf', 'auth'], + 'django': ['csrf', 'auth'], + 'express': ['helmet', 'auth'], + 'hibernate': ['sql'] + } + + if framework in high_severity and check_type in high_severity[framework]: + return 'high' + return 'medium' + + def _get_issue_description(self, framework: str, check_type: str, pattern: str) -> str: + """获取问题描述""" + descriptions = { + 'spring': { + 'csrf': 'Spring Security CSRF 保护配置问题', + 'auth': 'Spring Security 认证授权配置问题', + 'cors': 'Spring CORS 配置可能存在安全风险' + }, + 'django': { + 'csrf': 'Django CSRF 保护被禁用', + 'auth': 'Django 认证装饰器使用不当', + 'xss': 'Django XSS 防护被绕过' + }, + 'express': { + 'helmet': 'Express 安全中间件配置问题', + 'auth': 'Express 认证机制实现问题', + 'input': 'Express 输入验证配置问题' + }, + 'hibernate': { + 'sql': 'Hibernate SQL 注入风险', + 'cache': 'Hibernate 缓存配置问题', + 'validation': 'Hibernate 验证配置问题' + } + } + + return descriptions.get(framework, {}).get(check_type, '未知问题') + + def _generate_summary(self, issues: list) -> str: + """生成分析总结""" + if not issues: + return "未发现框架相关的安全问题" + + summary = "框架安全分析总结:\n" + severity_count = {'high': 0, 'medium': 0, 'low': 0} + + for issue in issues: + severity_count[issue['severity']] += 1 + + summary += f"- 高危问题: {severity_count['high']} 个\n" + summary += f"- 中危问题: {severity_count['medium']} 个\n" + summary += f"- 低危问题: {severity_count['low']} 个\n" + + return summary \ No newline at end of file diff --git a/core/analyzers/security_analyzer.py b/core/analyzers/security_analyzer.py new file mode 100644 index 0000000..7310130 --- /dev/null +++ b/core/analyzers/security_analyzer.py @@ -0,0 +1,36 @@ +class SecurityAnalyzer: + def __init__(self): + self.patterns = { + 'weak_crypto': [ + r'md5\(', + r'sha1\(', + r'crypt\(', + r'des_encrypt\(' + ], + 'hardcoded_secrets': [ + r'password\s*=\s*[\'"][^\'"]+[\'"]', + r'secret\s*=\s*[\'"][^\'"]+[\'"]', + r'api_key\s*=\s*[\'"][^\'"]+[\'"]' + ], + 'insecure_config': [ + r'display_errors\s*=\s*On', + r'allow_url_include\s*=\s*On', + r'register_globals\s*=\s*On' + ], + 'csrf_vulnerability': [ + r'form.*method=[\'"]post[\'"].*(?!.*csrf)', + r'ajax\.post\(.*(?!.*token)' + ] + } + + def analyze(self, code, file_type): + """执行安全分析""" + vulnerabilities = [] + + # 根据文件类型选择不同的分析策略 + if file_type == 'php': + vulnerabilities.extend(self._analyze_php(code)) + elif file_type == 'java': + vulnerabilities.extend(self._analyze_java(code)) + + return vulnerabilities \ No newline at end of file diff --git a/core/analyzers/taint_analyzer.py b/core/analyzers/taint_analyzer.py new file mode 100644 index 0000000..12a63f9 --- /dev/null +++ b/core/analyzers/taint_analyzer.py @@ -0,0 +1,573 @@ +import re + +class TaintAnalyzer: + def __init__(self): + self.sources = set([ + 'GET', 'POST', 'REQUEST', 'FILES', 'COOKIE', + 'file_get_contents', 'fgets', 'fread', + 'stdin', '$_SERVER', '$_ENV', 'getenv', + 'mysqli_query', 'mysql_query', 'PDO->query', + 'curl_exec', 'file', 'readfile', 'unserialize' + ]) + self.sinks = set([ + 'eval', 'exec', 'system', 'shell_exec', + 'passthru', 'popen', 'proc_open', + 'include', 'include_once', 'require', 'require_once', + 'mysqli_query', 'mysql_query', 'PDO->query', + 'echo', 'print', 'printf', + 'header', + 'file_put_contents', 'fwrite', + 'unserialize', + 'mail' + ]) + self.sanitizers = set([ + 'htmlspecialchars', 'htmlentities', 'strip_tags', + 'addslashes', 'escapeshellarg', 'escapeshellcmd' + ]) + self.vulnerability_types = { + 'rce': ['eval', 'exec', 'system', 'shell_exec'], + 'sqli': ['mysqli_query', 'mysql_query', 'PDO->query'], + 'xss': ['echo', 'print', 'printf'], + 'file_inclusion': ['include', 'include_once', 'require'], + 'file_operation': ['file_put_contents', 'fwrite'], + 'deserialization': ['unserialize'], + 'header_injection': ['header', 'mail'] + } + + # 添加变量追踪映射 + self.variable_mapping = {} + + # 添加函数调用栈 + self.call_stack = [] + + # 添加更多的漏洞模式 + self.vulnerability_patterns = { + 'sql_injection': { + 'risk_functions': ['query', 'execute'], + 'safe_patterns': [r'\?|:[a-zA-Z_][a-zA-Z0-9_]*'], # 参数化查询模式 + 'risk_patterns': [r".*\+.*'.*|.*'.*\+.*"] # 字符串拼接模式 + }, + 'xss': { + 'risk_functions': ['echo', 'print'], + 'safe_patterns': [r'htmlspecialchars\(.*\)|htmlentities\(.*\)'], + 'risk_patterns': [r'<.*>|javascript:'] + }, + 'path_traversal': { + 'risk_functions': ['file_get_contents', 'fopen'], + 'risk_patterns': [r'\.\.\/|\.\.\\'] + } + } + + # 添加更细粒度的漏洞检测规则 + self.detection_rules = { + 'java': { + 'sql_injection': { + 'patterns': [ + r'.*Statement\.executeQuery\(.*\+.*\)', + r'.*Statement\.execute\(.*\+.*\)', + r'.*PreparedStatement.*\+.*\)' + ], + 'safe_patterns': [ + r'PreparedStatement.*\?.*\)', + r'.*createQuery\(.*:.*\)' + ] + }, + 'command_injection': { + 'patterns': [ + r'Runtime\.exec\(.*\+.*\)', + r'ProcessBuilder.*\+.*\)', + ], + 'safe_patterns': [ + r'Runtime\.exec\(new String\[\].*\)' + ] + }, + 'xxe': { + 'patterns': [ + r'DocumentBuilder.*parse\(', + r'SAXParser.*parse\(', + r'XMLReader.*parse\(' + ], + 'safe_patterns': [ + r'setFeature\(.*XMLConstants\.FEATURE_SECURE_PROCESSING.*true\)' + ] + } + } + } + + # 添加框架特定的漏洞模式 + self.framework_patterns = { + 'spring': { + 'unsafe_redirects': [ + r'redirect:.*\+', + r'sendRedirect\(.*\+.*\)' + ], + 'csrf_vulnerable': [ + r'@CrossOrigin\(.*allowCredentials\s*=\s*true.*\)', + r'@RequestMapping.*method\s*=\s*RequestMethod\.POST.*(?!@CrossOrigin)' + ] + }, + 'hibernate': { + 'hql_injection': [ + r'createQuery\(.*\+.*\)', + r'createSQLQuery\(.*\+.*\)' + ] + } + } + + # 添加 Python 相关的检测规则 + self.detection_rules['python'] = { + 'command_injection': { + 'patterns': [ + r'os\.system\(.*\+.*\)', + r'subprocess\.call\(.*\+.*\)', + r'subprocess\.Popen\(.*\+.*\)', + r'eval\(.*\+.*\)', + r'exec\(.*\+.*\)' + ], + 'safe_patterns': [ + r'subprocess\.run\([^,]+,\s*shell\s*=\s*False\)', + r'shlex\.quote\(.*\)' + ] + }, + 'sql_injection': { + 'patterns': [ + r'execute\(.*\+.*\)', + r'executemany\(.*\+.*\)', + r'raw\(.*\+.*\)', + r'\.format\(.*\)' + ], + 'safe_patterns': [ + r'execute\([^,]+,\s*\(.*\)\)', + r'execute\([^,]+,\s*\[.*\]\)' + ] + }, + 'path_traversal': { + 'patterns': [ + r'open\(.*\+.*\)', + r'os\.path\.join\(.*\+.*\)', + r'__import__\(.*\+.*\)' + ], + 'safe_patterns': [ + r'os\.path\.abspath\(.*\)', + r'os\.path\.realpath\(.*\)' + ] + }, + 'deserialization': { + 'patterns': [ + r'pickle\.loads\(', + r'yaml\.load\(', + r'marshal\.loads\(' + ], + 'safe_patterns': [ + r'yaml\.safe_load\(', + r'json\.loads\(' + ] + } + } + + # 添加 JavaScript 相关的检测规则 + self.detection_rules['javascript'] = { + 'xss': { + 'patterns': [ + r'innerHTML\s*=', + r'outerHTML\s*=', + r'document\.write\(', + r'eval\(', + r'\$\(.*\)\.html\(' + ], + 'safe_patterns': [ + r'textContent\s*=', + r'innerText\s*=', + r'createElement\(' + ] + }, + 'dom_xss': { + 'patterns': [ + r'location\s*=', + r'location\.href\s*=', + r'location\.search', + r'location\.hash' + ], + 'safe_patterns': [ + r'encodeURIComponent\(', + r'encodeURI\(' + ] + }, + 'prototype_pollution': { + 'patterns': [ + r'Object\.assign\(', + r'Object\.prototype', + r'\.__proto__', + r'\.constructor\.prototype' + ] + }, + 'insecure_randomness': { + 'patterns': [ + r'Math\.random\(', + ], + 'safe_patterns': [ + r'crypto\.getRandomValues\(', + r'window\.crypto\.subtle' + ] + } + } + + # 添加框架特定的检测规则 + self.framework_patterns['django'] = { + 'csrf_vulnerable': [ + r'@csrf_exempt', + r'CSRF_COOKIE_SECURE\s*=\s*False' + ], + 'sql_injection_risk': [ + r'raw\(', + r'extra\(', + r'RawSQL\(' + ] + } + + self.framework_patterns['express'] = { + 'nosql_injection': [ + r'findOne\(.*\+.*\)', + r'find\(.*\+.*\)', + r'update\(.*\+.*\)' + ], + 'security_misconfiguration': [ + r'app\.disable\(.*trust\s*proxy.*\)', + r'app\.use\(bodyParser\.raw\(\)\)' + ] + } + + def analyze(self, ast_tree): + """ + 执行污点分析 + """ + vulnerabilities = [] + + # 遍历AST寻找污点传播路径 + for node in ast_tree.traverse(): + if self._is_source(node): + taint = self._track_taint(node) + if taint: + vulnerabilities.append(taint) + + return vulnerabilities + + def _is_source(self, node): + """检查节点是否为污点源""" + # 实现基本的污点源检查 + if hasattr(node, 'name'): + return str(node.name) in self.sources + return False + + def _track_taint(self, node): + """增强的污点追踪""" + if not node: + return None + + vulnerabilities = [] + visited = set() + + def track_recursive(current_node, taint_chain=None, context=None): + if not current_node or id(current_node) in visited: + return + + visited.add(id(current_node)) + taint_chain = taint_chain or [] + context = context or {} + + # 记录变量赋值 + if self._is_assignment(current_node): + self._track_variable_assignment(current_node) + + # 函数调用分析 + if self._is_function_call(current_node): + self._analyze_function_call(current_node, context) + + # 条件语句分析 + if self._is_condition(current_node): + self._analyze_condition_branch(current_node, context) + + # 检查是否经过安全的过滤 + if self._is_sanitized(current_node, context): + context['sanitized'] = True + return + + # 检查漏洞模式 + vuln = self._check_vulnerability_patterns(current_node, context) + if vuln: + vulnerabilities.append(vuln) + + # 递归分析 + for child in self._get_node_children(current_node): + track_recursive(child, taint_chain + [current_node], context.copy()) + + track_recursive(node) + return vulnerabilities + + def _calculate_severity(self, vuln_type): + """计算漏洞严重程度""" + severity_map = { + 'rce': 'critical', + 'sqli': 'high', + 'xss': 'medium', + 'file_inclusion': 'high', + 'file_operation': 'medium', + 'deserialization': 'high', + 'header_injection': 'medium' + } + return severity_map.get(vuln_type, 'low') + + def _get_vulnerability_type(self, node): + """确定漏洞类型""" + if hasattr(node, 'name'): + node_name = str(node.name) + for vuln_type, sinks in self.vulnerability_types.items(): + if node_name in sinks: + return vuln_type + return 'unknown' + + def _extract_context(self, node): + """提取漏洞上下文""" + context = { + 'code_snippet': self._get_code_snippet(node), + 'variables': self._get_related_variables(node), + 'function_scope': self._get_function_scope(node) + } + return context + + def _is_sink(self, node): + """检查节点是否为危险函数""" + if hasattr(node, 'name'): + return str(node.name) in self.sinks + return False + + def _analyze_function_call(self, node, context): + """分析函数调用的安全性""" + if not hasattr(node, 'name'): + return + + func_name = str(node.name) + + # 检查是否是高风险函数 + for vuln_type, patterns in self.vulnerability_patterns.items(): + if func_name in patterns['risk_functions']: + # 分析函数参数 + args = self._get_function_args(node) + for arg in args: + if self._is_tainted(arg, context): + # 检查是否使用了安全的编码/过滤方式 + if not self._has_safe_encoding(arg, patterns['safe_patterns']): + context['risks'].append({ + 'type': vuln_type, + 'function': func_name, + 'argument': str(arg) + }) + + def _track_variable_assignment(self, node): + """追踪变量赋值""" + if hasattr(node, 'target') and hasattr(node, 'value'): + var_name = str(node.target) + self.variable_mapping[var_name] = { + 'value': str(node.value), + 'tainted': self._is_tainted(node.value), + 'sanitized': self._is_sanitized(node.value), + 'location': self._get_node_location(node) + } + + def _analyze_condition_branch(self, node, context): + """分析条件分支中的安全检查""" + if hasattr(node, 'test'): + # 检查是否包含安全验证 + if self._has_security_check(node.test): + context['security_checked'] = True + + # 检查是否有风险的条件判断 + if self._has_risky_condition(node.test): + context['risks'].append({ + 'type': 'unsafe_condition', + 'condition': str(node.test), + 'location': self._get_node_location(node) + }) + + def _has_security_check(self, node): + """检查是否包含安全验证""" + security_patterns = [ + r'validate|verify|check|auth|permission', + r'is[A-Z]|has[A-Z]', + r'sanitize|escape|encode' + ] + node_str = str(node) + return any(re.search(pattern, node_str, re.I) for pattern in security_patterns) + + def _get_data_flow_path(self, node): + """获取数据流路径""" + path = [] + current = node + while current and hasattr(current, 'parent'): + path.append({ + 'type': type(current).__name__, + 'value': str(current), + 'location': self._get_node_location(current) + }) + current = current.parent + return path[::-1] + + def _analyze_framework_specific(self, node, context): + """分析框架特定的安全问题""" + framework = self._detect_framework(context) + if framework and framework in self.framework_patterns: + patterns = self.framework_patterns[framework] + for vuln_type, rules in patterns.items(): + if self._match_patterns(str(node), rules): + return { + 'type': vuln_type, + 'framework': framework, + 'location': self._get_node_location(node), + 'severity': 'high', + 'description': f'发现{framework}框架相关的{vuln_type}漏洞' + } + return None + + def _analyze_data_validation(self, node): + """分析数据验证逻辑""" + validation_info = { + 'has_validation': False, + 'validation_type': None, + 'validation_coverage': 0.0 + } + + # 检查是否使用了验证注解 + if self._has_validation_annotations(node): + validation_info['has_validation'] = True + validation_info['validation_type'] = 'annotation' + + # 检查是否有手动验证代码 + elif self._has_manual_validation(node): + validation_info['has_validation'] = True + validation_info['validation_type'] = 'manual' + + # 计算验证覆盖率 + validation_info['validation_coverage'] = self._calculate_validation_coverage(node) + + return validation_info + + def _analyze_authentication(self, node, context): + """分析认证相关的安全问题""" + auth_issues = [] + + # 检查认证绕过 + if self._check_auth_bypass(node): + auth_issues.append({ + 'type': 'auth_bypass', + 'severity': 'critical', + 'location': self._get_node_location(node) + }) + + # 检查权限检查 + if not self._has_permission_check(node): + auth_issues.append({ + 'type': 'missing_permission_check', + 'severity': 'high', + 'location': self._get_node_location(node) + }) + + return auth_issues + + def _analyze_secure_configuration(self, node): + """分析安全配置""" + config_issues = [] + + # 检查安全标头配置 + if not self._has_security_headers(node): + config_issues.append({ + 'type': 'missing_security_headers', + 'severity': 'medium' + }) + + # 检查安全cookie配置 + if not self._has_secure_cookie_config(node): + config_issues.append({ + 'type': 'insecure_cookie_config', + 'severity': 'medium' + }) + + return config_issues + + def _analyze_language_specific(self, node, language): + """基于语言特性的分析""" + if language not in self.detection_rules: + return None + + rules = self.detection_rules[language] + node_str = str(node) + + for vuln_type, patterns in rules.items(): + # 检查危险模式 + if 'patterns' in patterns: + for pattern in patterns['patterns']: + if re.search(pattern, node_str): + # 检查是否有安全模式 + if 'safe_patterns' in patterns: + if any(re.search(safe_pattern, node_str) + for safe_pattern in patterns['safe_patterns']): + continue + + return { + 'type': vuln_type, + 'language': language, + 'location': self._get_node_location(node), + 'code': node_str, + 'severity': self._calculate_severity(vuln_type), + 'description': f'发现{language}代码中的{vuln_type}漏洞' + } + return None + + def _check_js_specific_issues(self, node): + """检查JavaScript特有的安全问题""" + issues = [] + + # 检查不安全的第三方脚本引用 + if self._is_script_tag(node): + if not self._has_integrity_check(node): + issues.append({ + 'type': 'insecure_script_include', + 'severity': 'medium', + 'location': self._get_node_location(node), + 'recommendation': '添加 SRI (Subresource Integrity) 校验' + }) + + # 检查敏感信息泄露 + if self._contains_sensitive_data(node): + issues.append({ + 'type': 'sensitive_data_exposure', + 'severity': 'high', + 'location': self._get_node_location(node), + 'recommendation': '避免在前端代码中硬编码敏感信息' + }) + + return issues + + def _check_python_specific_issues(self, node): + """检查Python特有的安全问题""" + issues = [] + + # 检查不安全的模块导入 + if self._is_import(node): + if self._is_dangerous_import(node): + issues.append({ + 'type': 'dangerous_import', + 'severity': 'medium', + 'location': self._get_node_location(node), + 'recommendation': '谨慎使用潜在危险的模块' + }) + + # 检查调试配置 + if self._is_debug_config(node): + issues.append({ + 'type': 'debug_enabled', + 'severity': 'medium', + 'location': self._get_node_location(node), + 'recommendation': '在生产环境中禁用调试模式' + }) + + return issues \ No newline at end of file diff --git a/core/parsers/java_parser.py b/core/parsers/java_parser.py new file mode 100644 index 0000000..7147b56 --- /dev/null +++ b/core/parsers/java_parser.py @@ -0,0 +1,12 @@ +import javalang + +class JavaParser: + def parse(self, code: str): + """ + 解析Java代码生成AST + """ + try: + ast = javalang.parse.parse(code) + return ast + except Exception as e: + raise Exception(f"Java解析错误: {str(e)}") \ No newline at end of file diff --git a/core/parsers/php_parser.py b/core/parsers/php_parser.py new file mode 100644 index 0000000..2a7bd0a --- /dev/null +++ b/core/parsers/php_parser.py @@ -0,0 +1,40 @@ +import ast +import subprocess +import tempfile +import os + +class PHPParser: + def parse(self, code: str): + """ + 解析PHP代码生成AST + 使用 php -l 进行语法检查 + """ + try: + # 创建临时文件存储PHP代码 + with tempfile.NamedTemporaryFile(suffix='.php', mode='w', delete=False) as tmp: + tmp.write(code) + tmp_path = tmp.name + + # 使用 PHP 命令行进行语法检查 + result = subprocess.run(['php', '-l', tmp_path], + capture_output=True, + text=True) + + # 清理临时文件 + os.unlink(tmp_path) + + if "No syntax errors detected" not in result.stdout: + raise Exception(result.stderr) + + # 这里可以添加更详细的AST分析 + # 目前先返回简单的语法检查结果 + return { + "type": "php_file", + "syntax_valid": True, + "content": code + } + + except subprocess.CalledProcessError as e: + raise Exception(f"PHP解析错误: {str(e)}") + except Exception as e: + raise Exception(f"PHP解析错误: {str(e)}") \ No newline at end of file diff --git a/docker/Dockerfile b/docker/Dockerfile new file mode 100644 index 0000000..29b5198 --- /dev/null +++ b/docker/Dockerfile @@ -0,0 +1,10 @@ +FROM python:3.9 + +WORKDIR /app + +COPY requirements.txt . +RUN pip install -r requirements.txt + +COPY . . + +CMD ["uvicorn", "backend.app:app", "--host", "0.0.0.0", "--port", "8000"] \ No newline at end of file diff --git a/frontend/src/App.vue b/frontend/src/App.vue new file mode 100644 index 0000000..05ea60f --- /dev/null +++ b/frontend/src/App.vue @@ -0,0 +1,188 @@ + + + + + \ No newline at end of file diff --git a/project_structure b/project_structure new file mode 100644 index 0000000..a339f0b --- /dev/null +++ b/project_structure @@ -0,0 +1,8 @@ +code-audit-tool/ +├── frontend/ # Vue.js前端 +├── backend/ # Python FastAPI后端 +├── core/ # 核心审计逻辑 +│ ├── analyzers/ # 各种分析器 +│ ├── parsers/ # 代码解析器 +│ └── ai/ # AI分析模块 +└── docker/ # Docker配置文件 \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c5f293b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,10 @@ +fastapi>=0.68.0 +uvicorn>=0.15.0 +python-multipart>=0.0.5 +openai>=1.0.0 +javalang>=0.13.0 +aiohttp>=3.8.1 +python-dotenv>=0.19.0 +pydantic>=2.0.0 +pydantic-settings>=2.0.0 +php-ast>=1.1.0 \ No newline at end of file