安服 & 实施实习生双修实战：本地脱敏+大模型API，让排障、验收、报告不再加班

# 下载Ollama安装包（约200MB）
# 官网：https://ollama.com/download

# 安装完成后，在PowerShell中验证
ollama --version
# 输出：ollama version 0.5.0

# 启动Ollama服务（默认端口11434）
ollama serve

# 使用curl安装
curl -fsSL https://ollama.com/install.sh | sh

# 验证安装
ollama --version

# 启动服务
ollama serve &

# 拉取Qwen-7B模型（约4GB）
ollama pull qwen:7b

# 拉取Codellama-7B模型（约4GB）
ollama pull codellama:7b

# 查看已安装模型
ollama list

你是一个专业的数据脱敏工具。请严格按照以下规则处理文本：

【必须替换的内容】
- IP地址：替换为 <IP_0>、<IP_1>...（同一IP始终使用相同占位符）
- 手机号：替换为 <PHONE_0>、<PHONE_1>...（同一手机号始终使用相同占位符）
- 身份证号：替换为 <ID_0>、<ID_1>...
- 邮箱：替换为 <EMAIL_0>、<EMAIL_1>...
- 银行卡号：替换为 <CARD_0>、<CARD_1>...
- 域名/URL中的客户标识：替换为 <DOMAIN_0>、<DOMAIN_1>...

【必须保留的内容】
- 文件路径（即使包含客户名称，保留原样）
- 命令和命令参数
- 错误码和状态码
- 端口号
- 技术术语和专业词汇
- JSON/XML结构
- 代码片段

【输出格式】
仅输出脱敏后的文本，不要添加任何解释、注释或前缀。

【示例】
输入：用户从 192.168.1.100 访问 api客户A.com，触发错误
输出：用户从 <IP_0> 访问 api<DOMAIN_0>，触发错误

import requests
import re
import json
from typing import Dict, List, Optional
from pathlib import Path
import hashlib

class LocalSanitizer:
    def __init__(
        self,
        model: str = "qwen:7b",
        base_url: str = "http://localhost:11434",
        cache_dir: Optional[str] = None
    ):
        self.model = model
        self.base_url = base_url
        self.cache: Dict[str, str] = {}
        self.cache_dir = Path(cache_dir) if cache_dir else None
        self._load_cache()

    def _load_cache(self):
        if self.cache_dir and (self.cache_dir / "sanitize_cache.json").exists():
            with open(self.cache_dir / "sanitize_cache.json", "r", encoding="utf-8") as f:
                self.cache = json.load(f)

    def _save_cache(self):
        if self.cache_dir:
            self.cache_dir.mkdir(parents=True, exist_ok=True)
            with open(self.cache_dir / "sanitize_cache.json", "w", encoding="utf-8") as f:
                json.dump(self.cache, f, ensure_ascii=False, indent=2)

    def _compute_hash(self, text: str) -> str:
        return hashlib.sha256(text.encode("utf-8")).hexdigest()

    def _call_ollama(self, prompt: str) -> str:
        response = requests.post(
            f"{self.base_url}/api/generate",
            json={
                "model": self.model,
                "prompt": prompt,
                "stream": False,
                "options": {"temperature": 0.1, "num_predict": 2048}
            },
            timeout=60
        )
        response.raise_for_status()
        return response.json()["response"]

    def sanitize(self, text: str) -> str:
        text_hash = self._compute_hash(text)
        if text_hash in self.cache:
            return self.cache[text_hash]

        full_prompt = f"""你是一个专业的数据脱敏工具。请严格按照以下规则处理文本：

【必须替换的内容】
- IP地址：替换为 <IP_0>、<IP_1>...（同一IP始终使用相同占位符）
- 手机号：替换为 <PHONE_0>、<PHONE_1>...（同一手机号始终使用相同占位符）
- 身份证号：替换为 <ID_0>、<ID_1>...
- 邮箱：替换为 <EMAIL_0>、<EMAIL_1>...
- 银行卡号：替换为 <CARD_0>、<CARD_1>...
- 域名/URL中的客户标识：替换为 <DOMAIN_0>、<DOMAIN_1>...

【必须保留的内容】
- 文件路径（即使包含客户名称，保留原样）
- 命令和命令参数
- 错误码和状态码
- 端口号
- 技术术语和专业词汇
- JSON/XML结构
- 代码片段

【输出格式】
仅输出脱敏后的文本，不要添加任何解释、注释或前缀。

待处理文本：
{text}"""

        result = self._call_ollama(full_prompt)
        self.cache[text_hash] = result
        self._save_cache()
        return result

    def batch_sanitize(self, texts: List[str], progress_callback=None) -> List[str]:
        results = []
        total = len(texts)
        for idx, text in enumerate(texts):
            try:
                results.append(self.sanitize(text))
            except Exception as e:
                print(f"脱敏第{idx+1}条时出错: {e}，保留原文")
                results.append(text)
            if progress_callback:
                progress_callback(idx + 1, total)
        return results

import csv
import xml.etree.ElementTree as ET
from typing import List, Dict
import re

class BurpLogParser:
    @staticmethod
    def parse_csv(csv_path: str) -> List[Dict[str, str]]:
        results = []
        with open(csv_path, "r", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                entry = {
                    "method": row.get("method", ""),
                    "url": row.get("url", ""),
                    "param": row.get("param", ""),
                    "status": row.get("status", ""),
                    "response_body": row.get("response", "")[:2000],
                    "extension": row.get("extension", ""),
                    "host": row.get("host", "")
                }
                if entry["extension"] in ["jpg", "png", "gif", "css", "js"]:
                    continue
                results.append(entry)
        return results

    @staticmethod
    def parse_xml(xml_path: str) -> List[Dict[str, str]]:
        results = []
        tree = ET.parse(xml_path)
        root = tree.getroot()
        for item in root.findall(".//item"):
            entry = {
                "method": item.findtext("method", ""),
                "url": item.findtext("url", ""),
                "param": item.findtext("param", ""),
                "status": item.findtext("status", ""),
                "response_body": (item.findtext("response") or "")[:2000],
                "host": item.findtext("host", "")
            }
            extension = item.findtext("extension", "")
            if extension in ["jpg", "png", "gif", "css", "js"]:
                continue
            results.append(entry)
        return results

    @staticmethod
    def extract_features(entries: List[Dict[str, str]]) -> List[Dict[str, str]]:
        suspicious_patterns = [
            (r"'|and|or|union|select", "potential_sql_injection"),
            (r"<script|alert|onerror", "potential_xss"),
            (r"\.\./|\.\.\\", "potential_path_traversal"),
            (r"<[^>]+>", "potential_html_injection"),
            (r"document\.cookie|location\.href", "potential_csrf")
        ]
        results = []
        for entry in entries:
            combined = f"{entry['url']} {entry['param']} {entry['response_body']}"
            for pattern, vuln_type in suspicious_patterns:
                if re.search(pattern, combined, re.IGNORECASE):
                    entry["suspected_type"] = vuln_type
                    entry["matched_pattern"] = pattern
                    results.append(entry)
                    break
        return results

import requests
from typing import Literal, Optional
from dataclasses import dataclass
import json

@dataclass
class VulnerabilityResult:
    risk_level: Literal["高", "中", "低", "无"]
    vuln_type: str
    evidence: str
    fix_suggestion: str
    raw_response: str

class LLMAnalyzer:
    def __init__(
        self,
        provider: Literal["deepseek", "openai", "anthropic"] = "deepseek",
        api_key: Optional[str] = None,
        base_url: Optional[str] = None
    ):
        self.provider = provider
        self.api_key = api_key
        self.base_url = base_url or self._get_default_base_url(provider)

    def _get_default_base_url(self, provider: str) -> str:
        urls = {
            "deepseek": "https://api.deepseek.com",
            "openai": "https://api.openai.com/v1",
            "anthropic": "https://api.anthropic.com"
        }
        return urls.get(provider, "")

    def _build_headers(self) -> dict:
        headers = {"Content-Type": "application/json"}
        if self.provider == "deepseek":
            headers["Authorization"] = f"Bearer {self.api_key}"
        elif self.provider == "openai":
            headers["Authorization"] = f"Bearer {self.api_key}"
        elif self.provider == "anthropic":
            headers["x-api-key"] = self.api_key
            headers["anthropic-version"] = "2023-06-01"
        return headers

    def _build_payload(self, prompt: str, model: str) -> dict:
        if self.provider == "anthropic":
            return {"model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": 1024}
        else:
            return {"model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, "max_tokens": 1024}

    def analyze_vulnerability(self, method: str, url: str, param: str, response_body: str, model: str = "deepseek-chat") -> VulnerabilityResult:
        prompt = f"""你是安服工程师。以下是一条脱敏后的HTTP请求/响应：

[请求方法] {method}
[URL] {url}
[参数] {param}
[响应] {response_body[:1000]}

请判断是否存在漏洞类型，按以下严格JSON格式输出（不要输出任何其他内容）：
{{"risk_level": "高/中/低/无", "vuln_type": "漏洞类型", "evidence": "证据简述", "fix_suggestion": "修复建议"}}"""

        try:
            response = requests.post(f"{self.base_url}/chat/completions", headers=self._build_headers(), json=self._build_payload(prompt, model), timeout=30)
            response.raise_for_status()
            result_text = response.json()["choices"][0]["message"]["content"]
            result_json = json.loads(result_text)
            return VulnerabilityResult(risk_level=result_json["risk_level"], vuln_type=result_json["vuln_type"], evidence=result_json["evidence"], fix_suggestion=result_json["fix_suggestion"], raw_response=result_text)
        except Exception as e:
            return VulnerabilityResult(risk_level="无", vuln_type="分析失败", evidence=f"API调用异常: {str(e)}", fix_suggestion="请人工复核", raw_response="")

    def batch_analyze(self, entries: List[dict], model: str = "deepseek-chat", progress_callback=None) -> List[VulnerabilityResult]:
        results = []
        total = len(entries)
        for idx, entry in enumerate(entries):
            result = self.analyze_vulnerability(method=entry.get("method", "GET"), url=entry.get("url", ""), param=entry.get("param", ""), response_body=entry.get("response_body", ""), model=model)
            results.append(result)
            if progress_callback:
                progress_callback(idx + 1, total)
        return results

class DebugErrorAssistant:
    SYSTEM_PROMPT = """你是一位经验丰富的DevOps工程师，专长于Linux系统、网络协议、中间件的故障排查。
你的分析风格是：
1. 先判断"是什么"（What）- 识别错误类型
2. 再分析"为什么"（Why）- 分析可能原因
3. 最后解决"怎么办"（How）- 给出具体命令和步骤

回答格式：
1. 最可能的根本原因（2-3句话）
2. 3条具体的排查命令（直接可运行）
3. 修复步骤（按顺序编号）"""

    def __init__(self, analyzer: LLMAnalyzer):
        self.analyzer = analyzer

    def debug(self, error_log: str, os_info: str = "CentOS 7.9", software: str = "Nginx 1.20", context: str = "") -> str:
        prompt = f"""{self.SYSTEM_PROMPT}

## 错误日志（已脱敏）

## 环境信息
- 操作系统：{os_info}
- 相关软件：{software}

## 额外上下文
{context if context else "无"}"""
        try:
            return self.analyzer._call_llm_raw(prompt)
        except Exception as e:
            return f"分析失败: {str(e)}\n\n建议：请人工检查错误日志，或联系技术支持。"

class POCScriptGenerator:
    def __init__(self, analyzer: LLMAnalyzer):
        self.analyzer = analyzer

    def generate(self, requirement: str, language: str = "python", framework: str = "requests") -> str:
        prompt = f"""你是一位安全研究员，负责生成POC脚本。

请生成一个完整的、可直接运行的{language}脚本，实现以下功能：

{requirement}

## 要求
1. 使用{framework}库进行HTTP请求
2. 代码必须完整、可运行（包含所有import）
3. 添加必要的注释说明关键逻辑
4. 输出结果必须清晰（成功/失败统计）
5. 包含异常处理"""
        return self.analyzer._call_llm_raw(prompt)

class AcceptanceReportGenerator:
    def __init__(self, analyzer: LLMAnalyzer):
        self.analyzer = analyzer

    def generate(self, test_records: str, project_name: str = "XX项目") -> str:
        prompt = f"""以下是我们对{project_name}的测试执行记录（已脱敏）：

{test_records}

请将上述内容整理成一份正式的"系统验收报告"初版，包含：
- 测试环境概述
- 各功能项测试结论（表格形式）
- 遗留问题清单（重点标注"磁盘满"等问题）
- 建议

输出格式为Markdown。"""
        return self.analyzer._call_llm_raw(prompt)

#!/bin/bash
# environment_check.sh - 服务器环境自动检测脚本
# 作者：HOS(安全风信子)
# 日期：2026-05-09

set -e

echo "=========================================="
echo "     服务器环境自动检测脚本 v1.0"
echo "=========================================="
echo ""

# 初始化结果
CHECK_PASS=0
CHECK_FAIL=0

check_item() {
    local name="$1"
    local command="$2"
    local expected="$3"

    echo -n "[检测] $name ... "

    result=$(eval "$command" 2>/dev/null || echo "ERROR")

    if [[ "$result" == "$expected" ]] || [[ "$result" =~ $expected ]]; then
        echo "✓ 通过 (实际值: $result)"
        ((CHECK_PASS++))
    else
        echo "✗ 失败 (期望: $expected, 实际: $result)"
        ((CHECK_FAIL++))
    fi
}

# 1. 内存检测
echo "【1. 硬件资源检测】"
MEM_KB=$(grep MemTotal /proc/meminfo | awk '{print $2}')
MEM_GB=$((MEM_KB / 1024 / 1024))
if [ $MEM_GB -ge 4 ]; then
    echo "✓ 内存检测通过：${MEM_GB}GB (要求≥4GB)"
    ((CHECK_PASS++))
else
    echo "✗ 内存检测失败：${MEM_GB}GB (要求≥4GB)"
    ((CHECK_FAIL++))
fi

# 2. 硬盘检测
DISK_KB=$(df / | tail -1 | awk '{print $2}')
DISK_GB=$((DISK_KB / 1024 / 1024))
if [ $DISK_GB -ge 50 ]; then
    echo "✓ 硬盘检测通过：${DISK_GB}GB (要求≥50GB)"
    ((CHECK_PASS++))
else
    echo "✗ 硬盘检测失败：${DISK_GB}GB (要求≥50GB)"
    ((CHECK_FAIL++))
fi

# 3. 端口检测
echo ""
echo "【2. 端口检测】"
for port in 80 443 3306; do
    if netstat -tuln 2>/dev/null | grep -q ":$port "; then
        echo "✓ 端口 $port 已开放"
        ((CHECK_PASS++))
    else
        echo "✗ 端口 $port 未开放"
        ((CHECK_FAIL++))
    fi
done

# 4. Docker检测
echo ""
echo "【3. Docker检测】"
if command -v docker &> /dev/null; then
    DOCKER_VERSION=$(docker --version | awk '{print $3}' | sed 's/,//')
    echo "✓ Docker已安装，版本：$DOCKER_VERSION"
    ((CHECK_PASS++))

    if docker ps &> /dev/null; then
        echo "✓ Docker服务运行正常"
        ((CHECK_PASS++))
    else
        echo "✗ Docker服务未运行"
        ((CHECK_FAIL++))
    fi
else
    echo "✗ Docker未安装"
    ((CHECK_FAIL++))
fi

# 5. 操作系统检测
echo ""
echo "【4. 操作系统信息】"
if [ -f /etc/os-release ]; then
    . /etc/os-release
    echo "  系统：$NAME $VERSION"
else
    echo "  无法获取系统信息"
fi
echo "  内核：$(uname -r)"
echo "  架构：$(uname -m)"

# 输出汇总
echo ""
echo "=========================================="
echo "            检测结果汇总"
echo "=========================================="
echo "  通过：$CHECK_PASS 项"
echo "  失败：$CHECK_FAIL 项"
echo ""

if [ $CHECK_FAIL -eq 0 ]; then
    echo "✅ 环境检测通过，可以进行安装部署！"
    exit 0
else
    echo "❌ 环境检测未通过，请先解决上述问题！"
    exit 1
fi

llm_helper/
├── llm_helper.py          # API调用封装
├── debug_tool.py          # 排障工具
├── sanitizer.py           # 脱敏工具
├── poc_generator.py      # POC生成器
└── prompt_library.md      # 提示词模板库

"""
llm_helper.py - 大模型API调用封装工具箱
作者：HOS(安全风信子)
日期：2026-05-09
"""

import requests
from typing import Literal, Optional, List, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime
import json
import hashlib
from pathlib import Path

@dataclass
class LLMResponse:
    """大模型响应数据结构"""
    content: str
    model: str
    usage: Dict[str, int]
    raw_response: Any
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class LLMHelper:
    """
    大模型API调用封装类
    支持DeepSeek、OpenAI GPT、Claude等多种后端
    """

    SUPPORTED_PROVIDERS = ["deepseek", "openai", "anthropic"]

    def __init__(
        self,
        provider: Literal["deepseek", "openai", "anthropic"] = "deepseek",
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        default_model: Optional[str] = None
    ):
        if provider not in self.SUPPORTED_PROVIDERS:
            raise ValueError(f"不支持的provider: {provider}，支持的列表: {self.SUPPORTED_PROVIDERS}")

        self.provider = provider
        self.api_key = api_key or self._get_env_key(provider)
        self.base_url = base_url or self._get_default_base_url(provider)
        self.default_model = default_model or self._get_default_model(provider)
        self.request_count = 0
        self.total_tokens = 0

    def _get_env_key(self, provider: str) -> str:
        """从环境变量获取API Key"""
        env_vars = {
            "deepseek": "DEEPSEEK_API_KEY",
            "openai": "OPENAI_API_KEY",
            "anthropic": "ANTHROPIC_API_KEY"
        }
        import os
        return os.getenv(env_vars.get(provider, ""), "")

    def _get_default_base_url(self, provider: str) -> str:
        """获取默认API地址"""
        urls = {
            "deepseek": "https://api.deepseek.com",
            "openai": "https://api.openai.com/v1",
            "anthropic": "https://api.anthropic.com"
        }
        return urls.get(provider, "")

    def _get_default_model(self, provider: str) -> str:
        """获取默认模型"""
        models = {
            "deepseek": "deepseek-chat",
            "openai": "gpt-4o-mini",
            "anthropic": "claude-3-haiku-20240307"
        }
        return models.get(provider, "")

    def _build_headers(self) -> dict:
        """构建请求头"""
        headers = {"Content-Type": "application/json"}

        if self.provider == "deepseek":
            headers["Authorization"] = f"Bearer {self.api_key}"
        elif self.provider == "openai":
            headers["Authorization"] = f"Bearer {self.api_key}"
        elif self.provider == "anthropic":
            headers["x-api-key"] = self.api_key
            headers["anthropic-version"] = "2023-06-01"

        return headers

    def _build_payload(self, messages: List[Dict], model: str, **kwargs) -> dict:
        """构建请求Payload"""
        if self.provider == "anthropic":
            system_msg = ""
            user_msgs = []
            for msg in messages:
                if msg["role"] == "system":
                    system_msg = msg["content"]
                else:
                    user_msgs.append(msg)

            payload = {
                "model": model,
                "messages": user_msgs,
                "max_tokens": kwargs.get("max_tokens", 1024)
            }
            if system_msg:
                payload["system"] = system_msg
            return payload
        else:
            return {
                "model": model,
                "messages": messages,
                "temperature": kwargs.get("temperature", 0.7),
                "max_tokens": kwargs.get("max_tokens", 1024)
            }

    def chat(
        self,
        prompt: str,
        system: str = "",
        model: Optional[str] = None,
        **kwargs
    ) -> LLMResponse:
        """
        发送对话请求

        Args:
            prompt: 用户输入
            system: 系统提示
            model: 使用的模型（默认为配置中的默认模型）
            **kwargs: 其他参数（temperature, max_tokens等）

        Returns:
            LLMResponse对象
        """
        model = model or self.default_model

        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})

        try:
            if self.provider == "anthropic":
                endpoint = f"{self.base_url}/v1/messages"
            else:
                endpoint = f"{self.base_url}/chat/completions"

            response = requests.post(
                endpoint,
                headers=self._build_headers(),
                json=self._build_payload(messages, model, **kwargs),
                timeout=kwargs.get("timeout", 60)
            )

            response.raise_for_status()
            self.request_count += 1

            result = response.json()

            if self.provider == "anthropic":
                content = result["content"][0]["text"]
                usage = result.get("usage", {})
            else:
                content = result["choices"][0]["message"]["content"]
                usage = result.get("usage", {})

            self.total_tokens += usage.get("total_tokens", 0)

            return LLMResponse(
                content=content,
                model=model,
                usage=usage,
                raw_response=result
            )

        except requests.exceptions.RequestException as e:
            raise LLMException(f"请求失败: {str(e)}")
        except (KeyError, IndexError, json.JSONDecodeError) as e:
            raise LLMException(f"响应解析失败: {str(e)}")

    def batch_chat(
        self,
        prompts: List[str],
        system: str = "",
        model: Optional[str] = None,
        progress_callback=None
    ) -> List[LLMResponse]:
        """
        批量发送对话请求

        Args:
            prompts: 用户输入列表
            system: 系统提示
            model: 使用的模型
            progress_callback: 进度回调函数

        Returns:
            LLMResponse列表
        """
        results = []
        total = len(prompts)

        for idx, prompt in enumerate(prompts):
            try:
                result = self.chat(prompt=prompt, system=system, model=model)
                results.append(result)
            except LLMException as e:
                print(f"第{idx+1}条请求失败: {e}")
                results.append(LLMResponse(content=f"错误: {str(e)}", model=model or "", usage={}, raw_response=None))

            if progress_callback:
                progress_callback(idx + 1, total)

        return results

    def get_stats(self) -> Dict[str, int]:
        """获取使用统计"""
        return {
            "request_count": self.request_count,
            "total_tokens": self.total_tokens
        }

class LLMException(Exception):
    """大模型API异常"""
    pass

# 便捷函数
def create_helper(provider: str = "deepseek", **kwargs) -> LLMHelper:
    """创建LLMHelper实例的便捷函数"""
    return LLMHelper(provider=provider, **kwargs)

# 常用Prompt模板库

## 1. 数据脱敏模板

### 通用脱敏

### 日志脱敏

## 2. 漏洞分析模板

## 3. POC脚本生成模板

## 4. 验收报告生成模板

### 附录B：环境配置参考

```yaml
# docker-compose.yml 示例
version: '3.8'
services:
  ollama:
    image: ollama/ollama:latest
    container_name: local-llm
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: all
              capabilities: [gpu]

volumes:
  ollama_data: