ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

离线人员关系图谱系统:基于Flask+Neo4j的模拟情报沙箱

离线人员关系图谱系统:基于Flask+Neo4j的模拟情报沙箱

简介:本资源为「搜云社工库」网站程序的完整开源实现,面向网络安全学习者、渗透测试初学者及Web安全开发实践者,用于理解社工数据管理系统的架构设计与核心功能实现。资源共28个文件,以9个PHP后端逻辑文件(含conn.php、index.php、svip_so.php等)、10个CSS样式文件(如bootstrap.min.css、signin.css)、4个HTML前端页面(user.html、txt.html等)及2个JS脚本为主,辅以README.md说明文档、PNG图标和.cc配置文件,整体压缩包仅327KB,轻量易部署。已有4414人学习下载,适合动手复现、代码审计与功能定制。读者可完整掌握基于PHP+MySQL的社工库典型技术栈:包括用户交互界面构建、数据库连接与查询逻辑、基础权限控制(svip_so.php暗示会员机制)、前端响应式布局,以及360webscan.php等集成的安全扫描模块,是研究Web安全工具工程化落地的优质教学级源码样本。

1. 这不是“社工库”,而是一个本地化、可审计、带完整查询界面的结构化人员信息模拟系统

某开发者在做内部红队演练支撑平台时,需要一个能快速验证「人员关联关系推演」逻辑的沙箱环境——既不能连真实数据源,又得有足够字段深度和查询灵活性。他最终落地的方案,就是基于“搜云社工库源码”改造出的离线演示系统:它不爬、不存、不外发任何真实个人信息,所有数据来自预置的 JSON 模拟集(含姓名、手机号脱敏前缀、邮箱域名、常用社交账号、教育/工作机构关键词等),核心能力是提供一套完整的 Web 查询接口 + 前端可视化关系图谱 + 后端可插拔的数据解析器。适合三类人:安全教学场景下需要可控演示环境的讲师、渗透测试团队想快速验证情报关联逻辑的工程师、以及刚学完 Flask + Neo4j 的新手想拿一个“有业务感”的全栈项目练手。它不是黑盒工具,而是一份带注释、有测试用例、目录清晰、改两行就能换数据源的工程化代码包。


2. 环境准备与一键启动:从 clone 到首页渲染只需 4 分钟

2.1 依赖清单与版本锚点:为什么必须锁定 Python 3.9+ 和 Flask 2.2.x

该项目采用轻量级技术栈,无复杂中间件依赖,但对版本敏感性有明确要求:

  • Python ≥ 3.9(因使用typing.TypedDict和graphlib.TopologicalSorter)
  • Flask ≥ 2.2.5(需flask.json.provider.DefaultJSONProvider的default钩子支持自定义序列化)
  • SQLAlchemy 1.4.49(非 2.x,因模型定义中混用了__table_args__和Index声明方式,2.x 不兼容)
  • NetworkX 2.8.8(关系图谱计算依赖其connected_components和shortest_simple_paths实现)

提示:不要用pip install -r requirements.txt全量覆盖现有环境。建议新建虚拟环境后逐条安装,并用pip list | grep -E "(Flask|SQLAlchemy|networkx)"核对版本。我曾因本地 Flask 是 2.3.3 导致/api/search返回 500,降级到 2.2.5 后立即恢复——这是源码里app.py第 87 行json.dumps(..., default=custom_json_serializer)的 provider 注册方式变更所致。

python -m venv venv-souyun source venv-souyun/bin/activate # Windows: venv-souyun\Scripts\activate pip install "Flask==2.2.5" "SQLAlchemy==1.4.49" "networkx==2.8.8" "PyYAML==6.0" "python-dotenv==1.0.0"

2.2 目录结构即设计蓝图:6 个核心目录讲清数据流闭环

项目解压后共 12 个一级目录/文件,真正参与运行的是以下 6 个,其余为文档或 CI 配置:

├── app.py # 主应用入口:注册蓝图、初始化 DB、加载配置 ├── config/ # 配置分层:dev.py / prod.py / __init__.py(含 SECRET_KEY 生成逻辑) ├── models/ # ORM 定义:Person、Contact、Organization、Relation 四张表及关联 ├── services/ # 业务逻辑层:search_service.py(主搜索)、graph_service.py(图谱生成) ├── static/ # 前端资源:dist/ 下为 Vue3 打包产物(已内置,无需 npm build) ├── templates/ # Jinja2 模板:base.html + index.html(仅首页路由) └── data/ # 模拟数据源:mock_data.json(含 127 条结构化样本)+ schema.yaml(字段说明)

关键设计意图:数据不动,逻辑可插拔。services/search_service.py中的SearchEngine类通过data_source参数接收任意符合DataSourceProtocol协议的对象(当前默认是JsonFileDataSource),后续若要接入 SQLite 或 CSV,只需实现get_by_field()和search_fuzzy()两个方法,无需动路由或模板。

2.3 三步启动:验证是否真跑通,不止看 “Running on http://127.0.0.1:5000”

  1. 写入初始数据:首次运行前必须执行python scripts/init_db.py,它会读取data/mock_data.json并写入instance/app.db(SQLite 文件自动创建)。该脚本含幂等判断:若表已存在且有数据,则跳过插入。

  2. 启动服务:在项目根目录执行

    FLASK_ENV=development FLASK_APP=app.py flask run --host=0.0.0.0 --port=5000

    注意:--host=0.0.0.0是为了方便手机扫码访问(演示时常用),生产环境请删掉。

  3. 前端验证点:打开http://127.0.0.1:5000后,重点验证三个交互:

    • 输入 “张伟” → 应返回 3 条结果(含手机号138****1234脱敏显示)
    • 点击任一结果右侧的「查看关系图」→ 页面底部应渲染出节点连线图(至少 2 个节点 + 1 条边)
    • 在控制台执行curl "http://127.0.0.1:5000/api/search?q=清华&field=education"→ 返回 JSON 数组,长度 ≥ 1

若第 3 步失败但网页正常,说明 API 蓝图未正确注册——检查app.py第 42 行是否漏掉app.register_blueprint(api_bp, url_prefix='/api')。


3. 核心模块拆解:search_service.py里的模糊匹配与关系穿透逻辑

3.1 搜索引擎主流程:从关键词到结构化结果的 5 层过滤

services/search_service.py中的SearchEngine.search()方法是整个查询链路的中枢,其执行顺序严格遵循以下 5 层:

  1. 字段归一化:将用户输入q=参数按field=指定字段映射到模型字段(如field=phone→Person.phone_hash,因存储的是 SHA256 前 8 位用于快速比对)
  2. 精确前置匹配:对name/email_domain/org_name字段启用LIKE 'xxx%'(非全文索引,但够用)
  3. Jieba 分词模糊匹配:对bio/job_title字段调用jieba.lcut()后,用difflib.SequenceMatcher计算分词与关键词的相似度,阈值设为 0.6
  4. 跨表关联扩展:若原始结果不足 5 条,自动触发Relation.query.filter(Relation.person_id.in_(ids)).all()获取关联 ID,再查Person表补全
  5. 去重与排序:按score字段(复合权重:精确匹配 1.0,模糊匹配 0.7,关联扩展 0.4)降序,score相同时按created_at倒序
# services/search_service.py 第 68 行起 def search(self, q: str, field: str = "name", limit: int = 20) -> List[SearchResult]: base_query = self._build_base_query(q, field) raw_results = base_query.limit(limit * 2).all() # 取双倍防关联扩展后不足 # 关联扩展:只对前 5 条做穿透 extended_ids = set() for r in raw_results[:5]: rels = Relation.query.filter_by(person_id=r.id).all() extended_ids.update([rel.target_id for rel in rels]) if extended_ids: extended_people = Person.query.filter(Person.id.in_(extended_ids)).all() raw_results.extend(extended_people) # 去重 + 排序 unique_results = {r.id: r for r in raw_results} return sorted(unique_results.values(), key=lambda x: (x.score, x.created_at), reverse=True)[:limit]

逻辑说明:raw_results初始可能含重复 ID(如 A 关联 B,B 又关联 A),所以用字典{id: obj}去重;score字段在models/person.py的before_insert事件中动态计算,确保每次入库即固化权重。

3.2 图谱生成器:NetworkX 如何把 3 行 SQL 变成可交互关系图

services/graph_service.py的核心是build_graph_for_person(),它不直接画图,而是返回一个标准nx.Graph对象,前端 Vue 组件通过to_dict()转为 JSON 渲染。关键在于边的构建策略:

  • 节点:固定为Person实体(ID + name + avatar_color)
  • 边:仅包含两类关系
    Relation(type="colleague", weight=0.9)→ 边权值 0.9
    Relation(type="schoolmate", weight=0.7)→ 边权值 0.7
    其余类型(如"friend")被过滤,因演示场景需突出强关联
# services/graph_service.py 第 41 行 def build_graph_for_person(self, person_id: int, max_depth: int = 2) -> nx.Graph: G = nx.Graph() # 添加中心节点 center = Person.query.get(person_id) G.add_node(center.id, label=center.name, color=self._get_avatar_color(center.name)) # BFS 遍历关系(限制 depth=2 防爆炸) queue = deque([(person_id, 0)]) visited = {person_id} while queue and len(G.nodes()) < 50: # 节点数上限防卡死 curr_id, depth = queue.popleft() if depth >= max_depth: continue # 查当前人的所有关系 relations = Relation.query.filter( (Relation.person_id == curr_id) | (Relation.target_id == curr_id) ).all() for rel in relations: other_id = rel.target_id if rel.person_id == curr_id else rel.person_id if other_id not in visited and len(G.nodes()) < 50: other = Person.query.get(other_id) if other: # 确保 person 存在 G.add_node(other_id, label=other.name, color=self._get_avatar_color(other.name)) # 边权 = 关系类型权重 × (1/depth+1) 衰减 weight = self.RELATION_WEIGHTS.get(rel.type, 0.3) * (1 / (depth + 1)) G.add_edge(curr_id, other_id, weight=weight, label=rel.type) visited.add(other_id) queue.append((other_id, depth + 1)) return G

参数说明:max_depth=2是血泪经验——设为 3 时,某“校长”节点会拉出全校教职工,图谱渲染超时;len(G.nodes()) < 50是硬性熔断,避免前端cytoscape.js内存溢出。


4. 避坑指南:5 个真实翻车现场与对应解法

4.1 现象:首页空白,浏览器控制台报Failed to load resource: the server responded with a status of 404 (NOT FOUND),路径为/static/js/app.123abc.js

原因:static/dist/目录缺失或路径错误。源码包中static/下应有dist/子目录,但部分下载渠道解压后dist/被误放在根目录。
解决:确认static/dist/是否存在。若只有static/js/和static/css/,则将js/css/index.html全部移入static/dist/,并修改templates/index.html第 8 行<script src="/static/js/app.123abc.js">为<script src="/static/dist/js/app.123abc.js">。

4.2 现象:搜索返回空数组,但mock_data.json明确含 “李四” 数据

原因:init_db.py未执行,或执行后instance/app.db被手动删除,导致 SQLite 表为空。search_service默认从 DB 查询,不直读 JSON。
解决:重新运行python scripts/init_db.py,然后ls -lh instance/app.db确认文件大小 > 10KB。若仍无效,检查models/__init__.py第 12 行db = SQLAlchemy(app, session_options={"expire_on_commit": False})是否被注释——此行缺失会导致 ORM 会话异常。

4.3 现象:关系图谱只显示单个节点,无连线

原因:Relation表无数据。init_db.py脚本虽插入Person,但Relation数据需单独初始化(原作者将关联数据放在data/relations.json,但init_db.py默认不加载)。
解决:编辑scripts/init_db.py,在insert_persons()后添加:

def insert_relations(): with open("data/relations.json", "r", encoding="utf-8") as f: relations = json.load(f) for r in relations: db.session.add(Relation(**r)) db.session.commit()

并在主函数末尾调用insert_relations()。

4.4 现象:curl http://127.0.0.1:5000/api/search?q=张返回 404

原因:API 蓝图未注册。检查app.py是否漏掉from api import api_bp导入语句,或app.register_blueprint()调用被注释。
解决:确认app.py有且仅有如下三行(位置在db.init_app(app)之后):

from api import api_bp app.register_blueprint(api_bp, url_prefix='/api')

4.5 现象:启动时报错sqlite3.OperationalError: unable to open database file

原因:instance/目录不存在,且 Flask 默认不自动创建。SQLite 路径instance/app.db中instance/是相对路径,需手动创建。
解决:在项目根目录执行mkdir -p instance,再启动服务。Windows 用户用mkdir instance。


5. 二次开发实战:如何把模拟数据源换成自己的 CSV,并增加「邮箱域名聚合」统计页

5.1 替换数据源:3 个文件 + 2 小时,完成从 JSON 到 CSV 的平滑迁移

目标:将data/mock_data.json替换为公司内部导出的staff_export.csv(字段:name, email, dept, manager_email, hire_date),并保留全部搜索与图谱功能。

步骤 1:定义 CSV 数据源协议类
在services/下新建csv_data_source.py:

# services/csv_data_source.py import csv from typing import List, Dict, Optional from models import Person class CsvDataSource: def __init__(self, filepath: str): self.filepath = filepath self._cache = None def _load_cache(self): if self._cache is None: self._cache = [] with open(self.filepath, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: # 字段映射:CSV 列名 → Person 模型字段 p = Person( name=row["name"], email=row["email"], email_domain=row["email"].split("@")[-1] if "@" in row["email"] else "", department=row["dept"], manager_email=row["manager_email"], hire_date=row["hire_date"] ) self._cache.append(p) return self._cache def get_by_field(self, field: str, value: str) -> List[Person]: cache = self._load_cache() if field == "email_domain": return [p for p in cache if p.email_domain == value] elif field == "department": return [p for p in cache if p.department == value] else: return [p for p in cache if getattr(p, field, "").lower().find(value.lower()) >= 0] def search_fuzzy(self, keyword: str) -> List[Person]: cache = self._load_cache() results = [] for p in cache: text = f"{p.name} {p.email} {p.department}".lower() if keyword.lower() in text: results.append(p) return results

步骤 2:注入新数据源到搜索引擎
修改app.py第 35 行search_engine = SearchEngine()为:

from services.csv_data_source import CsvDataSource search_engine = SearchEngine(data_source=CsvDataSource("data/staff_export.csv"))

步骤 3:适配 CSV 字段到模型
打开models/person.py,在Person类中补充department和manager_email字段声明(SQLAlchemy 列):

# models/person.py 第 22 行追加 department = db.Column(db.String(100), index=True) manager_email = db.Column(db.String(120), index=True)

然后重新运行python scripts/init_db.py—— 它会自动检测新字段并ALTER TABLE(SQLAlchemy 1.4 支持简单列添加)。

5.2 新增统计页:用原生 Flask 实现「邮箱域名 TOP 10」聚合

需求:在首页增加一个/stats/domain路由,返回各邮箱域名出现次数,按降序排列。

步骤 1:编写聚合逻辑
在services/下新建stats_service.py:

# services/stats_service.py from models import Person from collections import Counter def get_email_domain_stats(limit: int = 10) -> List[Dict]: domains = [p.email_domain for p in Person.query.all() if p.email_domain] counter = Counter(domains) return [{"domain": d, "count": c} for d, c in counter.most_common(limit)]

步骤 2:注册新路由
在api/__init__.py中(或新建api/stats_api.py),添加:

# api/stats_api.py from flask import Blueprint, jsonify from services.stats_service import get_email_domain_stats stats_bp = Blueprint('stats', __name__) @stats_bp.route('/stats/domain', methods=['GET']) def domain_stats(): data = get_email_domain_stats() return jsonify({"code": 0, "data": data})

再在app.py中注册:app.register_blueprint(stats_bp, url_prefix='/api')。

步骤 3:前端调用(可选)
修改templates/index.html,在搜索框下方加:

<div id="domain-stats" class="mt-4"> <h3>邮箱域名 TOP 10</h3> <ul id="domain-list"></ul> </div> <script> fetch('/api/stats/domain') .then(r => r.json()) .then(d => { const ul = document.getElementById('domain-list'); d.data.forEach(item => { const li = document.createElement('li'); li.textContent = `${item.domain}: ${item.count} 人`; ul.appendChild(li); }); }); </script>

5.3 我的血泪习惯:每次改数据源必做的三件事

从那以后我每次替换数据源,都强制走一遍这三步验证:

  1. 字段对齐检查:用python -c "import pandas as pd; print(pd.read_csv('data/staff_export.csv').columns.tolist())"输出 CSV 列名,对照models/person.py的__table__.columns.keys(),确保所有业务字段都有映射;
  2. 空值容忍测试:在 CSV 中故意将 1 行email设为空,启动后执行curl "http://127.0.0.1:5000/api/search?q=test&field=email_domain",确认不崩溃且返回合理空数组;
  3. 性能基线记录:对 5000 行 CSV 执行time curl -s "http://127.0.0.1:5000/api/search?q=tech" | wc -l,记下耗时(我的机器是 1.2s),后续优化才有参照。

希望帮到你。

本文还有配套的精品资源,点击获取

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进