AI算法用于日志异常分析（Python）

1
#验证基础包功能
2
cat > ~/log-lab/exp4/scripts/verify_env.py <<'PYEOF'
3
#!/usr/bin/env python3
4
"""基础环境安装检验脚本"""
5
import sys
6
def test_imports():
7
"""测试关键模块导入"""
8
print("=" * 50)
9
print("基础环境安装检验")
10
print("=" * 50)
11
tests = [
12
("pandas", "pd"),
13
("numpy", "np"),
14
("sklearn", "sklearn"),
15
("matplotlib", "plt"),
16
("seaborn", "sns"),
17
]
18
all_passed = True
19
for module, alias in tests:
20
try:
21
__import__(module)
22
print(f"✓ {module:15} 导入成功")
23
except ImportError as e:
24
print(f"✗ {module:15} 导入失败: {e}")
25
all_passed = False
26
print("\n" + "=" * 50)
27
if all_passed:
28
print("✓ 所有检验通过！基础环境安装成功。")
29
print("=" * 50)
30
return 0
31
else:
32
print("✗ 部分检验失败，请检查安装。")
33
print("=" * 50)
34
return 1
35
if __name__ == "__main__":
36
sys.exit(test_imports())
37
PYEOF

1
# 创建解析脚本
2
cat > ~/log-lab/exp4/scripts/drain_parser.py <<'PYEOF'
3
#!/usr/bin/env python3
4
"""
5
Drain日志解析器 - 纯Python实现
6
参考：Drain: An Online Log Parsing Approach with Fixed Depth Tree (ICWS 2017)
7
"""
8
import re
9
from collections import defaultdict
10
class DrainNode:
11
"""解析树节点"""
12
def __init__(self, depth=0):
13
self.depth = depth
14
self.children = {}      # token -> DrainNode
15
self.template_id = None  # 叶子节点存储模板ID
16
class DrainParser:
17
"""
18
Drain算法实现
19
"""
20
def __init__(self, depth=4, sim_threshold=0.5, max_children=100):
21
self.depth = depth
22
self.sim_threshold = sim_threshold
23
self.max_children = max_children
24
self.root = DrainNode(depth=0)
25
self.templates = {}      # id -> token列表
26
self.template_str = {}   # id -> 字符串
27
self.template_count = defaultdict(int)
28
self.next_id = 0
29
def _preprocess(self, log_line):
30
"""预处理：替换变量"""
31
# 替换IP地址
32
line = re.sub(r'\d+\.\d+\.\d+\.\d+(:\d+)?', '<IP>', log_line)
33
# 替换十六进制
34
line = re.sub(r'0x[0-9a-fA-F]+', '<HEX>', line)
35
# 替换长数字ID
36
line = re.sub(r'\b\d{4,}\b', '<ID>', line)
37
# 替换普通数字
38
line = re.sub(r'\b\d+\b', '<*>', line)
39
return line
40
def _get_first_token(self, tokens):
41
"""第一层分组key"""
42
if len(tokens) >= 2:
43
return f"{tokens[0]} {tokens[1]}"
44
return tokens[0] if tokens else ""
45
def _similarity(self, t1, t2):
46
"""计算两个token列表的相似度"""
47
if len(t1) != len(t2):
48
return 0.0
49
match = sum(1 for a, b in zip(t1, t2) if a == b)
50
return match / len(t1)
51
def _merge(self, template, tokens):
52
"""合并模板"""
53
return ['<*>' if a != b else a for a, b in zip(template, tokens)]
54
def parse(self, log_lines):
55
"""解析日志列表"""
56
results = []
57
for idx, line in enumerate(log_lines):
58
# 预处理
59
preprocessed = self._preprocess(line)
60
tokens = preprocessed.split()
61
if not tokens:
62
results.append({
63
'log_id': idx,
64
'log_line': line,
65
'template_id': -1,
66
'template': '',
67
'similarity': 0.0
68
})
69
continue
70
# 第一层分组
71
first_token = self._get_first_token(tokens)
72
if first_token not in self.root.children:
73
self.root.children[first_token] = DrainNode(depth=1)
74
# 在树中搜索最佳匹配
75
best_sim = -1
76
best_tid = None
77
# 简化：线性搜索所有模板（小数据量足够）
78
for tid, template in self.templates.items():
79
# 快速过滤：长度相同才比较
80
if len(template) != len(tokens):
81
continue
82
sim = self._similarity(template, tokens)
83
if sim > best_sim:
84
best_sim = sim
85
best_tid = tid
86
if best_sim >= self.sim_threshold and best_tid is not None:
87
# 匹配成功，更新模板
88
merged = self._merge(self.templates[best_tid], tokens)
89
self.templates[best_tid] = merged
90
self.template_str[best_tid] = ' '.join(merged)
91
self.template_count[best_tid] += 1
92
results.append({
93
'log_id': idx,
94
'log_line': line,
95
'template_id': best_tid,
96
'template': self.template_str[best_tid],
97
'similarity': best_sim
98
})
99
else:
100
# 创建新模板
101
tid = self.next_id
102
self.next_id += 1
103
self.templates[tid] = tokens
104
self.template_str[tid] = ' '.join(tokens)
105
self.template_count[tid] = 1
106
results.append({
107
'log_id': idx,
108
'log_line': line,
109
'template_id': tid,
110
'template': self.template_str[tid],
111
'similarity': 0.0
112
})
113
return results
114
def get_stats(self):
115
"""获取模板统计信息"""
116
stats = []
117
for tid in sorted(self.templates.keys()):
118
stats.append({
119
'template_id': tid,
120
'template': self.template_str[tid],
121
'count': self.template_count[tid]
122
})
123
return sorted(stats, key=lambda x: x['count'], reverse=True)
124
if __name__ == "__main__":
125
print("✓ Drain解析器模块可用")
126
PYEOF

1
cat > ~/log-lab/exp4/scripts/create_sample_data.py <<'PYEOF'
2
#!/usr/bin/env python3
3
"""创建示例日志数据集"""
4
import pandas as pd
5
import random
6
import os
7
from datetime import datetime, timedelta
8
print("=" * 50)
9
print("创建示例日志数据集")
10
print("=" * 50)
11
# 创建HDFS模拟日志
12
print("\n[1/2] 创建HDFS模拟日志...")
13
hdfs_templates = [
14
"INFO dfs.DataNode: Receiving block blk_{id} src: /10.250.19.102:50010 dest: /10.250.19.102:50010",
15
"INFO dfs.DataNode: Received block blk_{id} of size 67108864 from /10.251.42.12",
16
"INFO dfs.FSNamesystem: BLOCK* NameSystem.allocateBlock: /user/hdfs/file_{id}.blk_{id}",
17
"WARN dfs.DataNode: Failed to transfer blk_{id} to /10.251.42.13:50010, got java.io.IOException",
18
"ERROR dfs.DataNode: Exception in receiveBlock for block blk_{id} java.io.EOFException",
19
"INFO dfs.FSNamesystem: BLOCK* NameSystem.addStoredBlock: blockMap updated: 10.250.19.102:50010 is added to blk_{id}",
20
"WARN dfs.FSNamesystem: BLOCK* NameSystem.addToInvalidates: blk_{id} is added to invalidSet of 10.251.42.12:50010",
21
"ERROR dfs.FSNamesystem: BLOCK* NameSystem.addStoredBlock: Redundant addStoredBlock request received for blk_{id}",
22
]
23
normal_logs = []
24
anomaly_logs = []
25
# 生成正常日志（90%）
26
for i in range(900):
27
template = random.choice(hdfs_templates[:4])
28
log_id = f"{random.randint(1000000000, 9999999999)}"
29
log_line = template.format(id=log_id)
30
normal_logs.append({
31
'log_line': log_line,
32
'Label': 0,
33
'timestamp': datetime.now() - timedelta(minutes=random.randint(0, 1440))
34
})
35
# 生成异常日志（10%）
36
for i in range(100):
37
template = random.choice(hdfs_templates[4:])
38
log_id = f"{random.randint(1000000000, 9999999999)}"
39
log_line = template.format(id=log_id)
40
anomaly_logs.append({
41
'log_line': log_line,
42
'Label': 1,
43
'timestamp': datetime.now() - timedelta(minutes=random.randint(0, 1440))
44
})
45
hdfs_df = pd.DataFrame(normal_logs + anomaly_logs)
46
hdfs_df.to_csv('~/log-lab/exp4/data/hdfs_logs.csv', index=False)
47
print(f"✓ HDFS日志: {len(hdfs_df)} 条（正常:{len(normal_logs)}, 异常:{len(anomaly_logs)}）")
48
print("\n[2/2] 数据集准备完成！")
49
print("=" * 50)
50
PYEOF

1
cat > ~/log-lab/exp4/scripts/parse_logs.py <<'PYEOF'
2
#!/usr/bin/env python3
3
"""
4
日志解析实验：使用Drain算法解析日志模板
5
"""
6
import sys
7
sys.path.insert(0, '/home/kali/log-lab/exp4/scripts')
8
import pandas as pd
9
from drain_parser import DrainParser
10
print("=" * 60)
11
print("任务4：日志解析（Drain算法）")
12
print("=" * 60)
13
# 加载数据
14
print("\n[1/4] 加载日志数据...")
15
df = pd.read_csv('~/log-lab/exp4/data/hdfs_logs.csv')
16
print(f"✓ 加载完成，共 {len(df)} 条日志")
17
# 查看原始日志
18
print("\n[2/4] 原始日志样本（前5条）:")
19
for i, log in enumerate(df['log_line'].head(5)):
20
print(f"  [{i}] {log[:80]}...")
21
# 配置解析器
22
print("\n[3/4] 配置Drain解析器...")
23
sim_th = 0.5
24
parser = DrainParser(sim_threshold=sim_th)
25
print(f"  相似度阈值: {sim_th}")
26
# 执行解析
27
print("\n[4/4] 执行日志解析...")
28
results = parser.parse(df['log_line'].tolist())
29
# 转换为DataFrame
30
parsed_result = pd.DataFrame(results)
31
# 合并标签
32
parsed_result['Label'] = df['Label'].values
33
# 分析结果
34
print("\n" + "=" * 60)
35
print("解析结果分析")
36
print("=" * 60)
37
template_count = parsed_result['template'].nunique()
38
print(f"\n发现 {template_count} 种日志模板")
39
print("\nTop 10 日志模板及出现频率:")
40
template_stats = parsed_result['template'].value_counts().head(10)
41
for idx, (template, count) in enumerate(template_stats.items(), 1):
42
# 判断是否为异常模板
43
template_logs = parsed_result[parsed_result['template'] == template]
44
anomaly_ratio = template_logs['Label'].mean()
45
status = "⚠️ 异常" if anomaly_ratio > 0.5 else "✓ 正常"
46
print(f"\n  [{idx}] {status} 出现 {count} 次 (异常率:{anomaly_ratio:.1%})")
47
print(f"      模板: {template[:100]}...")
48
# 保存结果
49
parsed_result.to_csv('~/log-lab/exp4/data/parsed_logs.csv', index=False)
50
print(f"\n✓ 解析结果已保存: ~/log-lab/exp4/data/parsed_logs.csv")
51
# 参数调优实验
52
print("\n" + "=" * 60)
53
print("学生任务：")
54
print("1. 什么是日志模板？为什么需要解析？")
55
print("   - 日志模板是将具体日志抽象为通用模式")
56
print("   - 目的是减少日志种类，发现异常模式")
57
print("2. 调整sim_th参数（0.3 vs 0.7），观察模板数量变化")
58
print("=" * 60)
59
PYEOF

1
cat > ~/log-lab/exp4/scripts/anomaly_detection.py <<'PYEOF'
2
#!/usr/bin/env python3
3
"""
4
异常检测实验：使用Isolation Forest检测日志异常
5
"""
6
import pandas as pd
7
import numpy as np
8
import os
9
from sklearn.feature_extraction.text import TfidfVectorizer
10
from sklearn.ensemble import IsolationForest
11
from sklearn.metrics import classification_report, confusion_matrix
12
import matplotlib
13
matplotlib.use('Agg')
14
import matplotlib.pyplot as plt
15
# 设置中文字体（避免警告）
16
plt.rcParams['font.sans-serif'] = ['DejaVu Sans']
17
plt.rcParams['axes.unicode_minus'] = False
18
# 使用绝对路径
19
OUTPUT_DIR = os.path.expanduser('~/log-lab/exp4/output')
20
DATA_DIR = os.path.expanduser('~/log-lab/exp4/data')
21
os.makedirs(OUTPUT_DIR, exist_ok=True)
22
print("=" * 60)
23
print("任务5：异常检测（Isolation Forest）")
24
print("=" * 60)
25
# 加载解析后的数据
26
print("\n[1/6] 加载数据...")
27
df = pd.read_csv(f'{DATA_DIR}/parsed_logs.csv')
28
print(f"Loaded: {len(df)} logs")
29
# 查看标签分布
30
label_dist = df['Label'].value_counts()
31
print(f"\n标签分布:")
32
print(f"  正常(0): {label_dist.get(0, 0)}")
33
print(f"  异常(1): {label_dist.get(1, 0)}")
34
# 特征提取
35
print("\n[2/6] 特征提取（TF-IDF）...")
36
vectorizer = TfidfVectorizer(
37
max_features=50,
38
ngram_range=(1, 2),
39
stop_words='english'
40
)
41
X = vectorizer.fit_transform(df['log_line'])
42
print(f"Feature matrix shape: {X.shape}")
43
# 训练异常检测模型
44
print("\n[3/6] 训练Isolation Forest模型...")
45
contamination_rate = 0.1
46
model = IsolationForest(
47
contamination=contamination_rate,
48
random_state=42,
49
n_estimators=100
50
)
51
# 拟合并预测
52
predictions = model.fit_predict(X)
53
# 转换标签：1->0(正常), -1->1(异常)
54
df['predicted_anomaly'] = np.where(predictions == -1, 1, 0)
55
print(f"Training complete")
56
print(f"  Detected anomalies: {df['predicted_anomaly'].sum()}")
57
print(f"  Anomaly rate: {df['predicted_anomaly'].mean()*100:.2f}%")
58
# 评估模型
59
print("\n[4/6] 模型评估...")
60
print("\nClassification Report:")
61
print(classification_report(
62
df['Label'],
63
df['predicted_anomaly'],
64
target_names=['Normal', 'Anomaly']
65
))
66
# 混淆矩阵
67
cm = confusion_matrix(df['Label'], df['predicted_anomaly'])
68
print("\nConfusion Matrix:")
69
print(f"                 Pred Normal  Pred Anomaly")
70
print(f"Actual Normal    {cm[0,0]:8}     {cm[0,1]:8}")
71
print(f"Actual Anomaly   {cm[1,0]:8}     {cm[1,1]:8}")
72
# 可视化混淆矩阵（使用英文避免字体问题）
73
plt.figure(figsize=(8, 6))
74
plt.imshow(cm, interpolation='nearest', cmap=plt.cm.Blues)
75
plt.title('Anomaly Detection Confusion Matrix', fontsize=14)
76
plt.colorbar()
77
tick_marks = np.arange(2)
78
plt.xticks(tick_marks, ['Pred Normal', 'Pred Anomaly'])
79
plt.yticks(tick_marks, ['Actual Normal', 'Actual Anomaly'])
80
plt.ylabel('True Label', fontsize=12)
81
plt.xlabel('Predicted Label', fontsize=12)
82
for i in range(2):
83
for j in range(2):
84
plt.text(j, i, format(cm[i, j], 'd'),
85
horizontalalignment="center",
86
color="white" if cm[i, j] > cm.max()/2 else "black")
87
plt.tight_layout()
88
# 使用绝对路径保存
89
output_file = f'{OUTPUT_DIR}/confusion_matrix.png'
90
plt.savefig(output_file, dpi=150)
91
print(f"\nSaved: {output_file}")
92
# 查看检测到的异常样本
93
print("\n[5/6] 异常样本分析...")
94
anomalies = df[df['predicted_anomaly'] == 1]
95
print(f"\nDetected {len(anomalies)} anomalies, samples:")
96
for idx, row in anomalies.head(5).iterrows():
97
true_label = "True Anomaly" if row['Label'] == 1 else "False Positive"
98
print(f"\n  [{idx}] {true_label}")
99
print(f"      {row['log_line'][:100]}...")
100
# 异常类型聚类分析
101
print("\n[6/6] 异常类型分布（基于模板）...")
102
if 'template' in df.columns:
103
anomaly_templates = anomalies['template'].value_counts().head(5)
104
print("\nTop 5 anomaly templates:")
105
for template, count in anomaly_templates.items():
106
print(f"  {count} times: {template[:80]}...")
107
# 保存结果
108
df.to_csv(f'{DATA_DIR}/anomaly_results.csv', index=False)
109
print(f"\nSaved: {DATA_DIR}/anomaly_results.csv")
110
print("\n" + "=" * 60)
111
print("Experiment completed!")
112
print("=" * 60)
113
PYEOF

1
cat > docker-compose.yml <<'EOF'
2
services:
3
dvwa:
4
image: vulnerables/web-dvwa
5
container_name: exp4-dvwa
6
ports:
7
- "8081:80"
8
environment:
9
- MYSQL_PASS=password
10
restart: unless-stopped
11
networks:
12
- exp4-net
13
logging:
14
driver: "json-file"
15
options:
16
max-size: "10m"
17
max-file: "3"
18
tag: "{{.Name}}"
19
loki:
20
image: grafana/loki:2.9.4
21
container_name: exp4-loki
22
ports:
23
- "3101:3100"
24
volumes:
25
- ./loki-config.yml:/etc/loki/local-config.yaml
26
- loki-data:/loki
27
command: -config.file=/etc/loki/local-config.yaml
28
restart: unless-stopped
29
networks:
30
- exp4-net
31
promtail:
32
image: grafana/promtail:2.9.4
33
container_name: exp4-promtail
34
volumes:
35
- /var/lib/docker/containers:/var/lib/docker/containers:ro
36
- /var/run/docker.sock:/var/run/docker.sock:ro
37
- ./promtail-config.yml:/etc/promtail/config.yml
38
command: -config.file=/etc/promtail/config.yml
39
restart: unless-stopped
40
networks:
41
- exp4-net
42
depends_on:
43
- loki
44
grafana:
45
image: grafana/grafana:10.3.1
46
container_name: exp4-grafana
47
ports:
48
- "3001:3000"
49
environment:
50
- GF_SECURITY_ADMIN_USER=admin
51
- GF_SECURITY_ADMIN_PASSWORD=admin123
52
- GF_USERS_ALLOW_SIGN_UP=false
53
volumes:
54
- grafana-data:/var/lib/grafana
55
- ./grafana-provisioning:/etc/grafana/provisioning
56
restart: unless-stopped
57
networks:
58
- exp4-net
59
depends_on:
60
- loki
61
networks:
62
exp4-net:
63
driver: bridge
64
volumes:
65
loki-data:
66
grafana-data:
67
EOF

1
cat > promtail-config.yml <<'EOF'
2
server:
3
http_listen_port: 9080
4
grpc_listen_port: 0
5
positions:
6
filename: /tmp/positions-exp4.yaml
7
clients:
8
- url: http://loki:3100/loki/api/v1/push
9
scrape_configs:
10
- job_name: docker-logs
11
static_configs:
12
- targets:
13
- localhost
14
labels:
15
job: docker-logs
16
__path__: /var/lib/docker/containers/*/*-json.log
17
pipeline_stages:
18
- json:
19
expressions:
20
log: log
21
stream: stream
22
time: time
23
attrs: attrs
24
- json:
25
source: attrs
26
expressions:
27
tag: tag
28
- regex:
29
source: tag
30
expression: '^(?P<container_name>[a-zA-Z0-9_-]+)$'
31
- labels:
32
container_name:
33
- timestamp:
34
source: time
35
format: RFC3339Nano
36
- output:
37
source: log
38
EOF

1
cat > loki-config.yml <<'EOF'
2
auth_enabled: false
3
server:
4
http_listen_port: 3100
5
grpc_listen_port: 9096
6
common:
7
instance_addr: 127.0.0.1
8
path_prefix: /loki
9
storage:
10
filesystem:
11
chunks_directory: /loki/chunks
12
rules_directory: /loki/rules
13
replication_factor: 1
14
ring:
15
kvstore:
16
store: inmemory
17
schema_config:
18
configs:
19
- from: 2020-10-24
20
store: boltdb-shipper
21
object_store: filesystem
22
schema: v11
23
index:
24
prefix: index_
25
period: 24h
26
limits_config:
27
reject_old_samples: true
28
reject_old_samples_max_age: 168h
29
EOF

1
cat > ~/log-lab/exp4/scripts/export_dvwa_from_loki.py <<'PYEOF'
2
#!/usr/bin/env python3
3
"""
4
从实验4专用Loki导出DVWA攻击日志（分批查询版）
5
自动切分时间窗口，避免单批次过大导致HTTP 400
6
"""
7
import urllib.request
8
import urllib.parse
9
import json
10
import pandas as pd
11
import os
12
from datetime import datetime, timedelta, timezone
13
print("=" * 60)
14
print("任务6.3：从Loki导出DVWA攻击日志（分批查询版）")
15
print("=" * 60)
16
DATA_DIR = os.path.expanduser('~/log-lab/exp4/data')
17
os.makedirs(DATA_DIR, exist_ok=True)
18
LOKI_URL = "http://localhost:3101"
19
BATCH_LIMIT = 5000          # Loki 默认上限
20
TIME_SLICE_MINUTES = 30     # 每批查询30分钟窗口
21
def loki_get(endpoint, params):
22
query_string = urllib.parse.urlencode(params)
23
url = f"{LOKI_URL}{endpoint}?{query_string}"
24
try:
25
req = urllib.request.Request(url, method='GET')
26
req.add_header('Accept', 'application/json')
27
with urllib.request.urlopen(req, timeout=60) as resp:
28
return resp.read().decode('utf-8'), resp.status
29
except urllib.error.HTTPError as e:
30
body = e.read().decode('utf-8') if e.read() else ""
31
return body, e.code
32
except Exception as e:
33
return str(e), 0
34
# [1/4] 验证Loki
35
print("\n[1/4] 验证Loki服务...")
36
body, code = loki_get("/ready", {})
37
if code != 200:
38
print(f"  ✗ Loki未就绪 (HTTP {code})")
39
exit(1)
40
print(f"  ✓ Loki就绪")
41
# [2/4] 确定攻击发生的时间范围
42
print("\n[2/4] 探测攻击日志时间范围...")
43
# 先查最近1条，看最新时间
44
body, code = loki_get("/loki/api/v1/query", {
45
'query': '{job="docker-logs", container_name="exp4-dvwa"}',
46
'limit': 1
47
})
48
if code == 200:
49
data = json.loads(body)
50
if data.get('data', {}).get('result'):
51
vals = data['data']['result'][0].get('values', [])
52
if vals:
53
latest_ts_ns = int(vals[0][0])
54
latest_dt = datetime.fromtimestamp(latest_ts_ns/1e9, tz=timezone.utc)
55
print(f"  最新日志时间: {latest_dt.strftime('%Y-%m-%d %H:%M:%S')}")
56
else:
57
latest_dt = datetime.now(timezone.utc)
58
print(f"  无数据，使用当前时间")
59
else:
60
latest_dt = datetime.now(timezone.utc)
61
else:
62
latest_dt = datetime.now(timezone.utc)
63
# 攻击大概在最近6小时内发生
64
end_time = latest_dt
65
start_time = end_time - timedelta(hours=6)
66
print(f"  查询范围: {start_time.strftime('%H:%M')} ~ {end_time.strftime('%H:%M')}")
67
# [3/4] 分批查询
68
print(f"\n[3/4] 分批查询（每批{BATCH_LIMIT}条，窗口{TIME_SLICE_MINUTES}分钟）...")
69
all_logs = []
70
current_end = end_time
71
batch_num = 0
72
while current_end > start_time:
73
current_start = max(current_end - timedelta(minutes=TIME_SLICE_MINUTES), start_time)
74
start_ns = int(current_start.timestamp() * 1e9)
75
end_ns = int(current_end.timestamp() * 1e9)
76
body, code = loki_get("/loki/api/v1/query_range", {
77
'query': '{job="docker-logs", container_name="exp4-dvwa"}',
78
'start': str(start_ns),
79
'end': str(end_ns),
80
'limit': BATCH_LIMIT,
81
'step': 1
82
})
83
if code != 200:
84
print(f"  批次 {batch_num+1} 失败 (HTTP {code}): {body[:100]}...")
85
break
86
data = json.loads(body)
87
batch_logs = []
88
for stream in data.get('data', {}).get('result', []):
89
labels = stream.get('stream', {})
90
for ts, line in stream.get('values', []):
91
ts_sec = int(ts) / 1e9
92
dt = datetime.fromtimestamp(ts_sec, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
93
batch_logs.append({
94
'timestamp': dt,
95
'log_line': line,
96
'container_name': labels.get('container_name', ''),
97
'job': labels.get('job', '')
98
})
99
batch_num += 1
100
print(f"  批次 {batch_num}: {current_start.strftime('%H:%M')}~{current_end.strftime('%H:%M')} → {len(batch_logs)} 条")
101
all_logs.extend(batch_logs)
102
# 如果本批次满5000条，说明可能还有更多，缩小窗口继续
103
# 如果不足5000条，说明该时段已捞完，跳到下一时段
104
if len(batch_logs) < BATCH_LIMIT:
105
current_end = current_start
106
else:
107
# 满5000条，用最后一条的时间戳作为下一批次的end，避免重叠
108
last_ts = batch_logs[-1]['timestamp']
109
current_end = datetime.strptime(last_ts, '%Y-%m-%d %H:%M:%S').replace(tzinfo=timezone.utc)
110
# 稍微回退一点避免漏掉边界
111
current_end = current_end - timedelta(seconds=1)
112
# 安全上限：最多查20批
113
if batch_num >= 20:
114
print("  ⚠ 达到批次上限，停止查询")
115
break
116
print(f"\n  ✓ 总计获取: {len(all_logs)} 条日志")
117
# [4/4] 保存与过滤分析
118
print("\n[4/4] 保存结果...")
119
df = pd.DataFrame(all_logs)
120
output_file = f'{DATA_DIR}/dvwa_logs_loki.csv'
121
df.to_csv(output_file, index=False)
122
print(f"  ✓ 已保存: {output_file}")
123
# 分析日志构成
124
import re
125
access_logs = [l for l in all_logs if re.match(r'^\d+\.\d+\.\d+\.\d+\s+\S+\s+\S+\s+\[\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2}', l['log_line'])]
126
error_logs = [l for l in all_logs if '[:error]' in l['log_line']]
127
other_logs = [l for l in all_logs if l not in access_logs and l not in error_logs]
128
print(f"\n日志构成分析:")
129
print(f"  Apache Access Log: {len(access_logs)} 条")
130
print(f"  Apache Error Log:  {len(error_logs)} 条")
131
print(f"  其他/Supervisord:  {len(other_logs)} 条")
132
# 攻击特征检查
133
sql_count = sum(1 for l in access_logs if 'vulnerabilities/sqli' in l['log_line'])
134
brute_count = sum(1 for l in access_logs if 'vulnerabilities/brute' in l['log_line'])
135
dirb_ua = sum(1 for l in access_logs if 'dirb' in l['log_line'].lower())
136
print(f"\n攻击特征检查（Access Log中）:")
137
print(f"  SQL注入 (/sqli):     {sql_count} 条")
138
print(f"  暴力破解 (/brute):   {brute_count} 条")
139
print(f"  Dirb扫描器UA:        {dirb_ua} 条")
140
if sql_count == 0 and brute_count == 0 and dirb_ua == 0:
141
print("\n  ⚠ 未检测到攻击特征！请确认:")
142
print("    1. 已在 http://localhost:8081 执行SQLMap/Hydra/Dirb攻击")
143
print("    2. 攻击发生在最近6小时内")
144
print("    3. DVWA Security Level设置为Low")
145
print(f"\nAccess Log样本（前3条）:")
146
for i, row in enumerate(access_logs[:3], 1):
147
print(f"  [{i}] {row['log_line'][:80]}...")
148
print("\n" + "=" * 60)
149
print("Loki日志导出完成！")
150
print("=" * 60)
151
PYEOF