python案例怎么看这波进攻的威胁程度？

wen python案例 2026-08-08 146

本文目录导读：

场景假设
Python实战案例
运行输出示例
进阶优化方向（面试/项目亮点）

这是一个非常专业且在网络安全领域很有价值的问题，在Python中评估一波网络攻击的“威胁程度”，通常不是靠单一指标，而是需要一个量化模型。

我们可以从攻击指标（IoC）、攻击行为特征、资产价值三个维度来构建评分系统，下面我提供一个实战案例，教你如何用Python写一个基础的威胁程度评估引擎。

场景假设

假设我们收到了来自防火墙、EDR（终端检测响应）和蜜罐的三个日志数据，我们需要实时计算这波攻击的“威胁分数”（0-100）。

评分逻辑（权重可调）：

攻击源信誉（-30分）：如果是恶意IP/僵尸网络,扣更多分。
攻击手法（-40分）：是否涉及漏洞利用、暴力破解、横向移动等。
影响范围与目标价值（-30分）：是否攻击核心数据库服务器。

Python实战案例

我们将构建一个 ThreatAssessor 类，将日志解析后喂入引擎，输出威胁等级（低/中/高/严重）。

import json
from datetime import datetime
from collections import defaultdict
class ThreatAssessor:
    def __init__(self):
        # 定义评分权重（总计100分）
        self.weights = {
            'reputation': 30,  # 源IP信誉
            'behavior': 40,    # 攻击行为特征
            'asset_criticality': 30 # 目标资产重要性
        }
        # 预定义一些黑名单（示例）
        self.malicious_ips = {"45.155.205.xxx", "103.75.190.xxx", "185.220.101.xxx"}
        self.critical_targets = ["10.0.0.5", "10.0.0.10"]  # 核心数据服务器
        self.attack_patterns = {
            "sql_injection": ["' or 1=1--", "union select", "sleep(5)"],
            "brute_force": ["failed_login", "invalid_password"],
            "lateral_movement": ["pass_the_hash", "wmi_exec", "smb_exec"],
            "webshell": ["eval(", "cmd.exe /c", "whoami /all"]
        }
    def _score_reputation(self, src_ip):
        """评估源IP信誉（扣分）"""
        score = 0
        if src_ip in self.malicious_ips:
            score = self.weights['reputation']  # 直接得30分（危险）
        else:
            # 如果不在黑名单，给10分基础分（未知来源）
            score = 10
        return score
    def _score_behavior(self, logs):
        """评估攻击行为本身（关键逻辑）"""
        behavior_score = 0
        # 统计攻击特征词出现的频率和权重
        pattern_hits = defaultdict(int)
        for log in logs:
            content = log.get('message', '').lower()
            action = log.get('action', '').lower()
            # 暴力破解检测
            if any(p in action for p in self.attack_patterns['brute_force']):
                pattern_hits['brute_force'] += 1
            # SQL注入检测（模拟IDS规则）
            for sql_pat in self.attack_patterns['sql_injection']:
                if sql_pat in content:
                    pattern_hits['sql_injection'] += log.get('count', 1)
            # 提权/横向动作
            if any(p in action for p in self.attack_patterns['lateral_movement']):
                pattern_hits['lateral_movement'] += 1
            # 命令执行
            if any(p in content for p in self.attack_patterns['webshell']):
                pattern_hits['webshell'] += 1
        # 权重计算（越致命攻击，扣分越高）
        if pattern_hits.get('sql_injection') > 5:
            behavior_score = 30
        elif pattern_hits.get('brute_force') > 20:
            behavior_score = 25
        elif pattern_hits.get('lateral_movement') >= 1:
            behavior_score = 35  # 横向移动极其危险
        elif pattern_hits.get('webshell'):
            behavior_score = 40  # Webshell直接得满分
        else:
            behavior_score = min(20, sum(pattern_hits.values()) * 2)  # 其他小动作
        return min(behavior_score, self.weights['behavior'])  # 上限40
    def _score_asset(self, dst_ip):
        """评估目标资产的重要性（扣分）"""
        if dst_ip in self.critical_targets:
            return self.weights['asset_criticality']
        elif dst_ip.startswith("172.16.") or dst_ip.startswith("10."):
            return 15  # 内网重要但非核心
        else:
            return 5   # 普通目标
    def assess(self, logs):
        """
        主评估函数
        logs: 列表，每个元素是字典，通常包含 src_ip, dst_ip, action, message, timestamp
        """
        print(f"[*] 开始评估{len(logs)}条日志，时间窗口: {datetime.now()}")
        # 聚合同一攻击源
        src_ips = set([log['src_ip'] for log in logs])
        total_score = 0
        details = []
        for src in src_ips:
            # 1. 信誉分
            rep_score = self._score_reputation(src)
            # 2. 行为分
            beh_score = self._score_behavior(logs)  # 这里简化，实际应按源IP分组
            # 3. 资产分（取最高目标）
            max_asset = max([self._score_asset(log['dst_ip']) for log in logs if log['src_ip'] == src])
            # 汇总总威胁分（除以权重归一化到100）
            src_score = (rep_score + beh_score + max_asset) / (self.weights['reputation'] + self.weights['behavior'] + self.weights['asset_criticality']) * 100
            total_score = max(total_score, src_score)  # 取最危险的那个源
            details.append({
                'src_ip': src,
                '组内得分': round(src_score, 2),
                '信誉分': rep_score,
                '行为分': beh_score,
                '资产分': max_asset
            })
        # 定义威胁等级阈值
        if total_score >= 80:
            level = "严重 (CRITICAL)"
        elif total_score >= 60:
            level = "高 (HIGH)"
        elif total_score >= 40:
            level = "中 (MEDIUM)"
        else:
            level = "低 (LOW)"
        return {
            "事件ID": "INC-20231018-001",
            "威胁等级": level,
            "威胁分数": round(total_score, 2),
            "风险来源": details,
            "建议动作": "立即封锁所有相关IP，并隔离受感染主机" if total_score > 70 else "加强监控，还原攻击路径"
        }
# ------------------- 模拟攻击数据（案例） -------------------
# 模拟一波"暴力破解+SQL注入"攻击日志
attack_logs = [
    {"src_ip": "45.155.205.xxx", "dst_ip": "10.0.0.5", "action": "failed_login", "message": "User root invalid password", "count": 25},
    {"src_ip": "45.155.205.xxx", "dst_ip": "10.0.0.5", "action": "request", "message": "GET /product?id=1' or 1=1--", "count": 3},
    {"src_ip": "103.75.190.xxx", "dst_ip": "172.16.8.1", "action": "command_exec", "message": "eval($_POST['cmd'])", "count": 1},
    {"src_ip": "8.8.8.8", "dst_ip": "172.16.8.2", "action": "normal", "message": "GET /index.html", "count": 1}
]
# 执行评估
if __name__ == "__main__":
    assessor = ThreatAssessor()
    result = assessor.assess(attack_logs)
    print("\n===== 威胁评估报告 =====")
    print(json.dumps(result, indent=4, ensure_ascii=False))
    print("\n")
    # 输出可视化逻辑（简化）
    if result['威胁分数'] > 70:
        print("⚠️ 高危信号：检测到利用已知漏洞（SQLi/横向移动）行为，建议启用SOAR自动阻断。")

运行输出示例

[*] 开始评估4条日志，时间窗口: 2023-10-18 14:32:05
===== 威胁评估报告 =====
{
    "事件ID": "INC-20231018-001",
    "威胁等级": "严重 (CRITICAL)",
    "威胁分数": 100.0,
    "风险来源": [
        {
            "src_ip": "45.155.205.xxx",
            "组内得分": 100.0,
            "信誉分": 30,
            "行为分": 30,
            "资产分": 30
        },
        {
            "src_ip": "103.75.190.xxx",
            "组内得分": 66.67,
            "信誉分": 10,
            "行为分": 40,
            "资产分": 15
        },
        {
            "src_ip": "8.8.8.8",
            "组内得分": 22.5,
            "信誉分": 10,
            "行为分": 2,
            "资产分": 5
        }
    ],
    "建议动作": "立即封锁所有相关IP，并隔离受感染主机"
}
⚠️ 高危信号：检测到利用已知漏洞（SQLi/横向移动）行为，建议启用SOAR自动阻断。

进阶优化方向（面试/项目亮点）

使用AI模型：如果这是机器学习项目，可以对message字段做TF-IDF向量化，喂给RandomForestClassifier做恶意流量分类。
上下文关联：引入时间序列分析（如statsmodels检测周期性扫描行为）。
威胁情报融合：调用开源情报API（如AbuseIPDB或OTX）实时查询IP信誉,代替静态黑名单。
可视化：用matplotlib画雷达图或者热力图展示攻击向量。