rajapalagummi/Cybersecurity-Threat-Detection

# 网络安全威胁检测与网络异常情报平台 ## PyTorch Autoencoder + 图分析 + 实时攻击注入 | 10 种攻击类型 ## 概述 每个组织的网络每秒都会产生数千个事件。大多数是正常的，少数是攻击。面临的挑战在于如何在造成损害之前，实时且自动地将它们区分开来。 本项目构建了一个生产级的网络安全分析平台，该平台使用 PyTorch autoencoder（仅在正常流量上训练，标记偏差）、Isolation Forest 集成评分以及 NetworkX 图分析（用于攻击路径检测）来检测异常行为。一个实时攻击注入器模拟了 10 种不同的攻击类型——每种都有其独特的网络签名——通过 Grafana 仪表盘和 Neo4j 图可视化展示实时检测过程。 ## 实时演示攻击类型 ``` python3 inject_attack.py --type brute_force # SSH/RDP repeated failed logins → success python3 inject_attack.py --type ddos # 20 sources flooding single target python3 inject_attack.py --type port_scan # Sequential port probing (recon) python3 inject_attack.py --type lateral_movement # Compromised host → 8 internal hops python3 inject_attack.py --type data_exfiltration # Large outbound transfers (~38MB) python3 inject_attack.py --type ransomware # SMB spread + file encryption python3 inject_attack.py --type credential_stuffing # 15 rotating IPs, 60 accounts python3 inject_attack.py --type sql_injection # Web → DB anomalous query patterns python3 inject_attack.py --type privilege_escalation # user_012 → admin → root progression python3 inject_attack.py --type c2_beacon # Periodic callbacks (beaconing pattern) python3 inject_attack.py --type all # All 10 sequentially ``` ## 架构 ``` Network Event Simulator (src/network_simulator.py) ↓ Realistic baseline traffic (logins, transfers, DNS, HTTP) ↓ SQLite event store (data/events.db) Attack Injector (inject_attack.py) ↓ 10 attack types with distinct network signatures ↓ Injected into same event stream Anomaly Detection Engine (src/detector.py) ↓ Feature extraction: 8 numerical features per event ↓ PyTorch Autoencoder: trained on normal traffic only ↓ Isolation Forest: ensemble anomaly scoring ↓ Weighted ensemble: 60% autoencoder + 40% IsoForest ↓ Per-event anomaly score [0-1] Graph Analysis (NetworkX + Neo4j) ↓ Directed graph: IP nodes, connection edges ↓ Attack pattern detection: high out-degree, centrality, attack edges ↓ Visual path exploration in Neo4j browser Dashboards ↓ Grafana: real-time anomaly timeline, attack type breakdown, alerts ↓ Neo4j: interactive network graph with attack path highlighting ``` ## 技术实现 ### 1. PyTorch Autoencoder — 无监督异常检测 仅在正常流量上进行训练。学习以低误差重建正常事件。攻击事件具有高重建误差 = 高异常分数。 ``` class NetworkAutoencoder(nn.Module): def __init__(self, input_dim=8): super().__init__() self.encoder = nn.Sequential( nn.Linear(input_dim, 32), nn.ReLU(), nn.Dropout(0.1), nn.Linear(32, 16), nn.ReLU(), nn.Linear(16, 8), nn.ReLU(), ) self.decoder = nn.Sequential( nn.Linear(8, 16), nn.ReLU(), nn.Linear(16, 32), nn.ReLU(), nn.Dropout(0.1), nn.Linear(32, input_dim), ) def reconstruction_error(self, x): recon = self.forward(x) errors = torch.mean((x - recon) ** 2, dim=1) return errors.numpy() ``` **阈值：** 训练重建误差的第 95 个百分位数。高于阈值的事件 = 异常。 ### 2. 特征工程 — 8 个网络特征 ``` features["bytes_sent_log"] = np.log1p(df["bytes_sent"]) features["bytes_recv_log"] = np.log1p(df["bytes_recv"]) features["duration_log"] = np.log1p(df["duration_ms"]) features["dst_port_norm"] = df["dst_port"] / 65535.0 features["src_port_norm"] = df["src_port"] / 65535.0 features["is_external_src"] = ... features["is_external_dst"] = ... features["is_failure"] = (df["status"] == "failed") ``` ### 3. 集成评分 ``` ae_scores = autoencoder.reconstruction_error(X) / threshold iso_scores = -isolation_forest.score_samples(X) final_scores = 0.6 * ae_scores + 0.4 * iso_scores ``` ### 4. 图攻击模式检测 ``` for node in G.nodes(): if G.out_degree(node) > 10: findings.append({"type": "high_out_degree", "severity": "HIGH"}) centrality = nx.betweenness_centrality(G) high_pivots = [(n, c) for n, c in centrality.items() if c > 0.3] ``` ### 5. 攻击签名 — 使每种攻击可被检测的特征 | 攻击 | 签名 | 关键特征 | |---|---|---| | 暴力破解 | 29 次失败 + 1 次成功 | 单个 src_ip 的 is_failure 激增 | | DDoS | 100 个数据包，20 个源，极短持续时间 | bytes_recv ≈ 0，高吞吐量 | | 端口扫描 | 连续端口，极小字节，速度快 | dst_port 多样，小 payload | | 横向移动 | 内部→内部，多种服务 | 双方的 is_external=0 | | 数据泄露 | 500KB-5MB 出站，外部 dst | bytes_sent_log 极高 | | 勒索软件 | SMB dst_port=445，高写入量 | service=smb，高 bytes | | 凭据填充 | 大量用户，轮换 IP，认证失败 | is_failure=1，src_ip 多样 | | SQL 注入 | DB 端口 3306，大量 recv（数据转储） | dst_port=3306，bytes_recv 激增 | | 权限提升 | user→admin→root 路径 | username 模式 | | C2 Beacon | 一致的时间间隔，小 payload | duration_ms 一致 | ## 关键指标 - 处理了 **655 个总事件**（300 个基线 + 355 个攻击） - 具有独特网络签名的 **10 种攻击类型** - 完整注入演示后达到 **54.2% 攻击率** - **PyTorch Autoencoder**：8 维编码，100 个 epochs，最终 loss 为 0.4679 - **集成评分**：60% autoencoder + 40% Isolation Forest - 检测到 **3 种图模式**：高出度、攻击边、中心性枢纽 - **实时**：事件按批次评分，仪表盘每 5 秒刷新一次 - **零付费 API**：完全本地化，无云依赖 ## 运行方式 ``` # 1. 设置 python3 -m venv venv && source venv/bin/activate pip install -r requirements.txt # 2. 运行完整 pipeline python3 main.py # 3. 启动 dashboards docker-compose up -d # Grafana: http://localhost:3001 (凭据见 docker-compose.yml) # Neo4j: http://localhost:7474 (凭据见 docker-compose.yml) # 4. 实时演示 — Terminal 1：连续流量 python3 -c "from src.network_simulator import run_simulator; run_simulator(3.0)" # 5. 实时演示 — Terminal 2：注入攻击 python3 inject_attack.py --type brute_force python3 inject_attack.py --type ddos python3 inject_attack.py --type ransomware python3 inject_attack.py --type all # 6. 加载 Neo4j 图 python3 src/neo4j_loader.py # 在 Neo4j 浏览器中查询：MATCH p=(a)-[r:CONNECTED_TO]->(b) WHERE r.is_attack=true RETURN p LIMIT 50 ``` *由 Raja Palagummi 构建 | rajapalagummi.com | github.com/rajapalagummi*

标签：CISA项目, IP 地址批量处理, PE 加载器, Web报告查看器, 凭据扫描, 图神经网络, 威胁情报, 安全测试, 开发者工具, 异常检测, 插件系统, 攻击性安全, 特权检测, 网络安全, 自编码器, 请求拦截, 逆向工具, 隐私保护