本文目录导读：
这是一个非常专业且在网络安全领域很有价值的问题，在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画雷达图或者热力图展示攻击向量。
如果你有具体的日志格式（比如Suricata的EVEE JSON、Windows Security Logs），告诉我,我可以帮你写对应的解析和特征提取代码。
