别再让AI“裸奔”了！Anthropic开源的安全技能库，让Claude化身你的专属网络安全助手

import os
from anthropic import Anthropic
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 初始化Anthropic客户端
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

# 待审查的示例代码
sample_code = '''
def get_user_data(user_id, query):
    """从数据库获取用户数据"""
    # 直接拼接SQL查询，存在SQL注入风险
    sql = f"SELECT * FROM users WHERE id = {user_id} AND {query}"
    cursor.execute(sql)
    return cursor.fetchall()

def render_user_profile(username):
    """渲染用户资料页面"""
    # 直接将用户输入嵌入HTML，存在XSS风险
    template = f"<h1>Welcome, {username}</h1>"
    return template
'''

# 构建审查请求
review_prompt = f'''请对以下Python代码进行全面的安全审查：

1. 识别所有安全漏洞
2. 对每个漏洞说明：
   - 漏洞类型和严重程度
   - 攻击场景和潜在影响
   - 修复建议和最佳实践

代码内容：
{sample_code}

请以结构化的JSON格式返回审查结果。'''

# 发送请求并获取响应
message = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=2048,
    messages=[
        {
            "role": "user",
            "content": review_prompt
        }
    ]
)

# 输出审查结果
print("=== 代码安全审查报告 ===")
print(message.content[0].text)

import json
from typing import Dict, List, Any

class PenetrationTestAssistant:
    """渗透测试辅助类"""

    def __init__(self, client):
        self.client = client
        self.test_context = {
            "target": "",
            "findings": [],
            "test_phase": "reconnaissance"
        }

    def start_engagement(self, target: str, scope: Dict[str, Any]):
        """初始化渗透测试项目"""
        self.test_context["target"] = target
        self.test_context["scope"] = scope
        self.test_context["start_time"] = "当前时间戳"

        init_prompt = f'''你是一个专业的渗透测试工程师，正在对以下目标进行安全评估：

目标信息：
- URL: {target}
- 测试范围: {json.dumps(scope, ensure_ascii=False)}

请制定一份初步的渗透测试计划，包括：
1. 推荐的信息收集方法
2. 优先测试的入口点
3. 需要注意的法律和道德边界

请用中文回答。'''

        response = self._send_message(init_prompt)
        return response

    def analyze_findings(self, raw_data: str) -> Dict[str, Any]:
        """分析发现的漏洞或异常"""
        analysis_prompt = f'''作为渗透测试专家，请分析以下发现：

发现内容：
{raw_data}

请提供：
1. 初步风险评估（高/中/低/信息级）
2. 推荐的验证步骤
3. 如可利用，提供简要的利用思路
4. 修复建议

请用中文回答，保持专业的技术语气。'''

        response = self._send_message(analysis_prompt)

        # 将分析结果添加到上下文
        self.test_context["findings"].append({
            "raw_data": raw_data,
            "analysis": response,
            "timestamp": "当前时间戳"
        })

        return json.loads(response) if response.startswith("{") else {"analysis": response}

    def generate_report(self) -> str:
        """生成测试报告"""
        findings_summary = "\n".join([
            f"- 发现{i+1}: {f['raw_data'][:100]}... -> {f['analysis'][:200]}"
            for i, f in enumerate(self.test_context["findings"])
        ])

        report_prompt = f'''请根据以下渗透测试发现，生成一份专业的安全测试报告：

目标: {self.test_context["target"]}
测试范围: {json.dumps(self.test_context.get("scope", {}), ensure_ascii=False)}

发现汇总:
{findings_summary}

报告要求：
1. 执行摘要
2. 测试范围和方法说明
3. 详细发现（按严重程度排序）
4. 风险评级和统计
5. 修复建议和优先级
6. 附录（技术细节）

请生成结构化的技术报告。'''

        return self._send_message(report_prompt)

    def _send_message(self, prompt: str) -> str:
        """内部方法：发送消息到Claude"""
        message = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}]
        )
        return message.content[0].text


# 使用示例
def main():
    """主函数演示"""
    client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
    assistant = PenetrationTestAssistant(client)

    # 开始测试项目
    plan = assistant.start_engagement(
        target="https://example.com",
        scope={
            "endpoints": ["/api/users", "/admin", "/login"],
            "testing_types": ["OWASP Top 10", "Business Logic"]
        }
    )
    print("=== 渗透测试计划 ===")
    print(plan)

    # 模拟分析一个发现的漏洞
    sample_finding = """
    在登录页面发现异常行为：
    - 多次使用错误密码后，账户未被锁定
    - 响应时间随密码长度增加而延长（可能的计时攻击）
    - 用户名枚举：通过不同用户名测试，响应时间不同
    """

    analysis = assistant.analyze_findings(sample_finding)
    print("\n=== 漏洞分析 ===")
    print(analysis)

    # 生成最终报告
    report = assistant.generate_report()
    print("\n=== 测试报告 ===")
    print(report)


if __name__ == "__main__":
    main()

from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional

class IncidentSeverity(Enum):
    """事件严重程度枚举"""
    CRITICAL = "严重"
    HIGH = "高"
    MEDIUM = "中"
    LOW = "低"
    INFO = "信息级"

class SecurityIncidentResponse:
    """安全事件响应类"""

    def __init__(self, client):
        self.client = client
        self.incident = {
            "id": self._generate_incident_id(),
            "status": "OPEN",
            "timeline": [],
            "artifacts": []
        }

    def _generate_incident_id(self) -> str:
        """生成事件ID"""
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        return f"INC-{timestamp}"

    def initialize_incident(self, description: str, severity: str) -> Dict:
        """初始化安全事件"""
        self.incident["description"] = description
        self.incident["severity"] = severity
        self.incident["detected_at"] = datetime.now().isoformat()

        response_prompt = f'''一个新安全事件已被报告，请帮助我们进行初步评估：

事件描述: {description}
报告严重程度: {severity}

请提供：
1. 需要立即采取的遏制措施
2. 初步调查方向
3. 建议的证据收集优先级
4. 需要立即通知的相关方

这是紧急响应场景，请给出清晰、直接的建议。'''

        initial_response = self._send_to_claude(response_prompt)
        self.incident["initial_assessment"] = initial_response
        self._add_timeline_event("事件初始化", initial_response)

        return self.incident

    def investigate(self, evidence: List[str]) -> Dict:
        """进行事件调查"""
        evidence_text = "\n".join([f"- {e}" for e in evidence])

        investigation_prompt = f'''基于以下收集到的证据，请进行深入分析：

证据列表：
{evidence_text}

请分析：
1. 攻击的时间线和手段
2. 受影响的系统和数据范围
3. 攻击者的意图和技术水平
4. 可能的入侵路径
5. 是否存在横向移动或持续性威胁

请提供详细的分析报告，包括你的推理过程。'''

        analysis = self._send_to_claude(investigation_prompt)
        self.incident["investigation"] = analysis
        self._add_timeline_event("深入调查", analysis)

        return {"analysis": analysis}

    def develop_remediation_plan(self) -> Dict:
        """制定修复计划"""
        remediation_prompt = f'''基于目前的调查发现，请制定详细的修复计划：

当前发现：{self.incident.get("investigation", "调查中")}

请提供：
1. 短期修复措施（立即执行）
2. 中期加固措施（一周内）
3. 长期改进措施（持续改进）
4. 每项措施的责任人和完成标准
5. 修复后的验证方法

请按优先级排序，并考虑措施之间的依赖关系。'''

        plan = self._send_to_claude(remediation_prompt)
        self.incident["remediation_plan"] = plan
        self._add_timeline_event("修复计划制定", plan)

        return {"plan": plan}

    def generate_post_incident_report(self) -> str:
        """生成事件后报告"""
        timeline_summary = "\n".join([
            f"[{e['timestamp']}] {e['phase']}: {e['summary'][:100]}"
            for e in self.incident["timeline"]
        ])

        report_prompt = f'''请生成一份完整的事件后报告：

事件概要：
- 事件ID: {self.incident["id"]}
- 严重程度: {self.incident["severity"]}
- 检测时间: {self.incident["detected_at"]}
- 当前状态: {self.incident["status"]}

事件时间线：
{timeline_summary}

调查报告：
{self.incident.get("investigation", "N/A")}

修复计划：
{self.incident.get("remediation_plan", "N/A")}

请生成包含以下部分的完整报告：
1. 执行摘要
2. 事件概述
3. 影响评估
4. 根本原因分析
5. 响应措施总结
6. 修复和加固措施
7. 经验教训和改进建议
8. 附录（详细技术信息）

报告应该既适合管理层阅读，也适合技术团队参考。'''

        report = self._send_to_claude(report_prompt)
        self.incident["final_report"] = report
        self.incident["status"] = "CLOSED"
        self.incident["closed_at"] = datetime.now().isoformat()

        return report

    def _send_to_claude(self, prompt: str) -> str:
        """内部方法：与Claude交互"""
        message = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system="你是一位经验丰富的安全事件响应专家。请基于提供的信息，提供专业、准确的分析和建议。",
            messages=[{"role": "user", "content": prompt}]
        )
        return message.content[0].text

    def _add_timeline_event(self, phase: str, summary: str):
        """添加时间线事件"""
        self.incident["timeline"].append({
            "timestamp": datetime.now().isoformat(),
            "phase": phase,
            "summary": summary[:200] if len(summary) > 200 else summary
        })

    def get_incident_summary(self) -> Dict:
        """获取事件摘要"""
        return {
            "id": self.incident["id"],
            "status": self.incident["status"],
            "severity": self.incident["severity"],
            "detected_at": self.incident["detected_at"],
            "events_count": len(self.incident["timeline"])
        }


# 使用示例
def demonstrate_incident_response():
    """演示安全事件响应流程"""
    client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
    response = SecurityIncidentResponse(client)

    # 步骤1：初始化事件
    incident = response.initialize_incident(
        description="监控发现异常登录行为：多个账户在短时间内从不同地理位置登录",
        severity="HIGH"
    )
    print(f"事件已创建: {incident['id']}")
    print(f"初步评估:\n{incident['initial_assessment']}\n")

    # 步骤2：调查阶段
    evidence = [
        "日志显示：账户A在2小时内从德国、中国、巴西三地登录",
        "会话令牌被转发至外部IP地址45.xxx.xxx.xxx",
        "同一IP段还尝试访问了其他15个账户",
        "用户报告收到可疑的密码重置邮件"
    ]

    investigation = response.investigate(evidence)
    print(f"调查结果:\n{investigation['analysis']}\n")

    # 步骤3：制定修复计划
    remediation = response.develop_remediation_plan()
    print(f"修复计划:\n{remediation['plan']}\n")

    # 步骤4：生成最终报告
    final_report = response.generate_post_incident_report()
    print(f"最终报告:\n{final_report}")

    # 获取事件摘要
    print(f"\n事件摘要: {response.get_incident_summary()}")


if __name__ == "__main__":
    demonstrate_incident_response()