个人信息泄漏检测技术架构：如何实现隐私安全的API查询系统

【免费下载链接】leak-check个人信息 “泄漏” 检测接口项目地址: https://gitcode.com/gh_mirrors/le/leak-check

在数字化时代，个人信息安全已成为每个互联网用户必须面对的现实挑战。当您的手机号、邮箱、身份证号等敏感信息在暗网或公开数据集中流通时，如何安全地检测这些泄漏风险而不暴露更多隐私？leak-check项目提供了一个技术解决方案：一个基于FastAPI构建的个人信息泄漏检测接口，采用脱敏聚合和隐私优先的设计理念，在保护用户隐私的同时提供有效的安全风险评估。

问题导向：个人信息泄漏检测的技术挑战

个人信息泄漏检测面临多重技术挑战。首先，查询过程本身可能成为新的隐私泄露点，用户需要查询自己的信息是否被泄露，但查询行为本身就可能暴露用户的关注点。其次，查询结果可能包含敏感信息，如果直接返回原始数据，反而会造成二次泄露。第三，不同数据源格式各异，需要统一的聚合处理机制。最后，系统需要平衡查询效率和隐私保护，确保在保护用户隐私的同时提供准确的检测结果。

leak-check项目正是针对这些挑战而设计的，它通过3层架构解决了个人信息泄漏检测的核心痛点：隐私保护、数据聚合和接口安全。

技术理念：隐私优先的泄漏检测架构

leak-check采用"查询不记录、结果脱敏、数据聚合"三大核心原则。这一设计理念确保系统在技术层面实现隐私保护：

1. 无痕查询：系统不记录任何查询日志，避免查询行为本身成为隐私泄露源
2. 脱敏响应：所有查询结果都经过严格的脱敏处理，防止二次泄露
3. 聚合展示：相同信息的不同泄漏记录被聚合处理，减少信息冗余

项目的技术架构基于FastAPI + SQLAlchemy + SQLite的轻量级组合，这种技术选型既保证了高性能响应，又降低了部署复杂度。FastAPI的异步特性确保了高并发查询下的系统稳定性，而SQLite的轻量级特性使得项目可以轻松部署在各种环境中。

核心架构：3层脱敏聚合技术实现

数据模型设计

在models/database.py中，项目定义了简洁而完整的数据模型：

class Person(Base): __tablename__ = "person" rowid: Mapped[int] = mapped_column(Integer, primary_key=True) id: Mapped[Optional[str]] # 身份证号 name: Mapped[Optional[str]] # 姓名 phone: Mapped[Optional[str]] # 手机号 email: Mapped[Optional[str]] # 邮箱 qq: Mapped[Optional[int]] # QQ号 # ... 其他字段 source_id: Mapped[int] = mapped_column(ForeignKey("source.id"))

这种设计支持灵活的数据存储，同时通过外键关联到数据来源表，便于追踪泄漏数据的原始出处。

智能类型识别机制

models/request.py中的ModelRequestQuery类实现了智能的类型识别功能：

@model_validator(mode="before") @classmethod def validate_and_detect(cls, values): q_ = values.get("q") cleaned = re.sub(r"[ \-\(\)]", "", q_) # 手机号识别（支持国内和国际格式） cn_phone_pattern = r"^1\d{10}$" intl_phone_pattern = r"^\+\d{6,15}$" phone_pattern = rf"(?:{cn_phone_pattern})|(?:{intl_phone_pattern})" # 邮箱识别 email_pattern = r"^[\w\.-]+@[\w\.-]+\.\w+$" # 身份证识别（支持大陆和台湾格式） mainland_id_pattern = r"^\d{17}[\dXx]$" taiwan_id_pattern = r"^[A-Z][12]\d{8}$" id_pattern = rf"(?:{mainland_id_pattern})|(?:{taiwan_id_pattern})" # 自动识别类型 if re.fullmatch(phone_pattern, cleaned): values["type"] = "phone" elif re.fullmatch(email_pattern, q_): values["type"] = "email" elif re.fullmatch(id_pattern, cleaned): values["type"] = "id" elif re.fullmatch(qq_pattern, cleaned): values["type"] = "qq"

这种自动识别机制简化了API调用，用户只需提供查询内容，系统自动判断查询类型，提高了用户体验。

脱敏处理核心算法

lib/masking.py实现了多种脱敏算法，针对不同数据类型采用不同的脱敏策略：

def mask_phone(v: str) -> str: if not v or len(v) < 7: return "***" return v[:2] + "*******" + v[-2:] def mask_email(v: str) -> str: if not v or "@" not in v: return "***" name, domain = v.split("@", 1) if len(name) <= 2: name_mask = name[0] + "***" else: name_mask = name[:2] + "***" return f"{name_mask}@{domain}" def mask_id(v: str) -> str: if not v or len(v) < 10: return "***" return v[:2] + "********" + v[-2:] def mask_name(v: str) -> str: if not v: return "***" if len(v) == 1: return "*" return "*" + v[-1]

这些脱敏算法确保了即使查询结果被截获，攻击者也无法获得完整的敏感信息。例如，手机号13800138000会被脱敏为13*******00，邮箱user@example.com会被脱敏为us***@example.com。

数据聚合处理

lib/aggregation.py提供了数据清洗和去重功能：

def clean_str_set(values: Iterable[Optional[str]]) -> list[str]: return list({ v.strip() for v in values if v and str(v).strip() }) def clean_int_set(values: Iterable) -> list[int]: result = set() for v in values: try: if v is not None and str(v).strip(): result.add(int(v)) except (TypeError, ValueError): continue return list(result)

这种聚合处理避免了重复信息的冗余展示，同时确保了数据格式的统一性。

API接口设计与实现

核心查询接口

main.py中的/dig/masking接口是系统的核心：

@app.post( "/dig/masking", summary="查询个人信息'泄漏'记录 - 脱敏", response_model=ModelResponsePersonAggregatedMasking, ) def get_person_by_dig(body: ModelRequestQuery, session: SessionDep): persons: Sequence[Person] = [] match body.type: case "phone": persons = crud.read_persons_by_dig(session, phone_=body.q) case "qq": persons = crud.read_persons_by_dig(session, qq_=int(body.q)) case "email": persons = crud.read_persons_by_dig(session, email_=body.q) case "id": persons = crud.read_persons_by_dig(session, id_=body.q.upper()) # 脱敏聚合处理 aggregated = ModelResponsePersonAggregatedMasking( id=mask_list("id", (p.id for p in persons)), name=mask_list("name", (p.name for p in persons)), phone=mask_list("phone", (p.phone for p in persons)), email=mask_list("email", (p.email for p in persons)), # ... 其他字段处理 source=list({ p.source_obj.source for p in persons if p.source_obj and p.source_obj.source }) ) return aggregated

这个接口支持四种查询类型：手机号、QQ号、邮箱和身份证号。查询结果会经过完整的脱敏处理，确保隐私安全。

响应数据结构

models/response.py定义了清晰的响应模型：

class ModelResponsePersonAggregatedMasking(BaseModel): id: List[str] # 脱敏后的身份证号列表 name: List[str] # 脱敏后的姓名列表 phone: List[str] # 脱敏后的手机号列表 email: List[str] # 脱敏后的邮箱列表 qq: List[str] # 脱敏后的QQ号列表 weibo: List[str] # 脱敏后的微博号列表 address: List[str] # 脱敏后的地址列表 car: List[str] # 脱敏后的车辆信息列表 contact: List[str] # 脱敏后的联系人列表 company: List[str] # 脱敏后的公司信息列表 source: List[str] # 数据来源（不脱敏）

这种结构化的响应格式便于客户端解析和处理，同时保持了数据的完整性。

部署实践：快速搭建检测系统

一键安装部署

项目提供了完整的安装脚本install.sh，支持快速部署：

# 使用安装脚本快速部署 curl -LsSf https://raw.githubusercontent.com/garinasset/leak-check/refs/heads/main/install.sh | bash

安装脚本会自动完成以下步骤：

1. 检测并安装uv（Python包管理工具）
2. 克隆项目代码
3. 配置示例数据库
4. 安装Python 3.14环境
5. 创建虚拟环境并安装依赖

数据库初始化

项目采用SQLite作为数据库，支持快速部署和迁移：

-- 创建数据来源表 CREATE TABLE source ( id INTEGER PRIMARY KEY, source TEXT DEFAULT NULL ); -- 创建个人信息表 CREATE TABLE person( id TEXT DEFAULT NULL, name TEXT DEFAULT NULL, receiver TEXT DEFAULT NULL, nickname TEXT DEFAULT NULL, phone TEXT DEFAULT NULL, address TEXT DEFAULT NULL, car TEXT DEFAULT NULL, email TEXT DEFAULT NULL, qq INTEGER DEFAULT NULL, weibo INTEGER DEFAULT NULL, contact TEXT DEFAULT NULL, company TEXT DEFAULT NULL, source_id INTEGER DEFAULT 0, FOREIGN KEY (source_id) REFERENCES source(id) );

生产环境配置

项目提供了systemd服务配置文件uvicorn-leak-check.service，支持生产环境部署：

# 复制服务配置文件 sudo cp uvicorn-leak-check.service /etc/systemd/system/ # 启用并启动服务 sudo systemctl enable uvicorn-leak-check.service sudo systemctl start uvicorn-leak-check.service

API使用指南与最佳实践

基础查询示例

使用curl进行个人信息泄漏检测：

# 查询手机号泄漏情况 curl -X POST "http://localhost:8000/leak-check/dig/masking" \ -H "Content-Type: application/json" \ -d '{"q": "13800138000"}' # 查询邮箱泄漏情况 curl -X POST "http://localhost:8000/leak-check/dig/masking" \ -H "Content-Type: application/json" \ -d '{"q": "user@example.com"}' # 查询身份证号泄漏情况 curl -X POST "http://localhost:8000/leak-check/dig/masking" \ -H "Content-Type: application/json" \ -d '{"q": "110101199001011234"}' # 查询QQ号泄漏情况 curl -X POST "http://localhost:8000/leak-check/dig/masking" \ -H "Content-Type: application/json" \ -d '{"q": "123456789"}'

Python客户端集成

import requests import json class LeakCheckClient: def __init__(self, base_url="http://localhost:8000/leak-check"): self.base_url = base_url def check_phone(self, phone_number): """检查手机号泄漏情况""" response = requests.post( f"{self.base_url}/dig/masking", json={"q": phone_number} ) return response.json() def check_email(self, email): """检查邮箱泄漏情况""" response = requests.post( f"{self.base_url}/dig/masking", json={"q": email} ) return response.json() def check_id_card(self, id_card): """检查身份证号泄漏情况""" response = requests.post( f"{self.base_url}/dig/masking", json={"q": id_card} ) return response.json() def check_qq(self, qq_number): """检查QQ号泄漏情况""" response = requests.post( f"{self.base_url}/dig/masking", json={"q": str(qq_number)} ) return response.json() # 使用示例 client = LeakCheckClient() result = client.check_phone("13800138000") print(f"手机号泄漏情况: {result}")

响应结果解析

查询结果示例：

{ "id": ["11**********1234"], "name": ["*明", "*华"], "phone": ["13*******00", "18*******88"], "email": ["us***@example.com", "te***@test.com"], "qq": ["*****6789"], "weibo": ["*****4321"], "address": ["北京****", "上海****"], "car": ["宝马****"], "contact": ["*先生"], "company": ["科技****"], "source": ["数据源A", "数据源B"] }

性能优化与安全考虑

查询性能优化

1. 数据库索引优化：在phone、email、id、qq字段上建立索引，提高查询速度
2. 连接池管理：SQLAlchemy连接池配置优化，支持高并发查询
3. 缓存策略：可考虑添加Redis缓存层，缓存热门查询结果

安全增强措施

1. 请求频率限制：防止恶意爬虫和暴力查询
2. IP黑名单机制：识别并阻止可疑IP地址
3. 查询内容验证：增强输入验证，防止SQL注入攻击
4. HTTPS强制：生产环境必须启用HTTPS加密传输

隐私保护强化

1. 查询日志脱敏：即使需要记录日志，也要对查询内容进行脱敏
2. 数据生命周期管理：定期清理过期数据
3. 访问控制：基于API密钥的访问控制机制

扩展应用场景与技术集成

企业安全监控平台集成

leak-check可以作为企业安全监控平台的核心组件：

class EnterpriseSecurityMonitor: def __init__(self, leak_check_client): self.client = leak_check_client self.employee_data = self.load_employee_data() def monitor_employee_leaks(self): """监控员工信息泄漏情况""" alerts = [] for employee in self.employee_data: # 检查手机号 phone_result = self.client.check_phone(employee.phone) if phone_result.get("source"): alerts.append({ "employee": employee.name, "type": "phone", "sources": phone_result["source"] }) # 检查邮箱 email_result = self.client.check_email(employee.email) if email_result.get("source"): alerts.append({ "employee": employee.name, "type": "email", "sources": email_result["source"] }) return alerts def generate_security_report(self): """生成安全报告""" alerts = self.monitor_employee_leaks() report = { "total_employees": len(self.employee_data), "leak_alerts": len(alerts), "details": alerts, "timestamp": datetime.now().isoformat() } return report

个人隐私保护工具集成

开发基于leak-check的个人隐私保护浏览器插件：

// 浏览器插件示例 class PrivacyGuardExtension { constructor() { this.apiEndpoint = "https://leak-check.yourdomain.com"; } async checkCurrentPage() { // 提取页面中的个人信息 const personalInfo = this.extractPersonalInfo(); // 批量检查泄漏情况 const results = await Promise.all( personalInfo.map(info => this.checkLeak(info)) ); // 显示检查结果 this.displayResults(results); } extractPersonalInfo() { // 从页面中提取手机号、邮箱等信息 const phoneNumbers = this.extractPhoneNumbers(); const emails = this.extractEmails(); return [...phoneNumbers, ...emails]; } async checkLeak(info) { const response = await fetch(`${this.apiEndpoint}/dig/masking`, { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({q: info}) }); return await response.json(); } }

数据源扩展架构

支持多数据源集成：

class MultiSourceLeakChecker: def __init__(self): self.sources = { 'internal': InternalDataSource(), 'external_api': ExternalAPIDataSource(), 'darkweb_monitor': DarkwebMonitorSource() } async def check_across_sources(self, query, query_type): """跨多个数据源检查泄漏情况""" results = {} for source_name, source in self.sources.items(): try: source_result = await source.query(query, query_type) if source_result: results[source_name] = source_result except Exception as e: print(f"数据源 {source_name} 查询失败: {e}") # 聚合所有数据源的结果 aggregated = self.aggregate_results(results) return aggregated def aggregate_results(self, results): """聚合多个数据源的结果""" aggregated = { 'id': set(), 'name': set(), 'phone': set(), 'email': set(), 'source': set() } for source_name, source_results in results.items(): for field in aggregated.keys(): if field in source_results: aggregated[field].update(source_results[field]) aggregated['source'].add(source_name) # 转换为列表并脱敏 return { field: list(values) if field != 'source' else list(values) for field, values in aggregated.items() }

技术价值与行业影响

技术创新的核心价值

leak-check项目的技术价值体现在多个层面：

1. 隐私保护技术创新：将脱敏技术从简单的数据展示扩展到查询响应全流程，实现了真正的隐私优先设计
2. 架构轻量化设计：采用SQLite + FastAPI的组合，在保证性能的同时大幅降低部署复杂度
3. 智能识别算法：自动识别查询内容类型，简化了API调用流程
4. 模块化设计：清晰的模块划分便于功能扩展和维护

对安全行业的贡献

1. 标准化接口：为个人信息泄漏检测提供了标准化的API接口设计参考
2. 开源参考实现：为安全厂商和个人开发者提供了完整的参考实现
3. 隐私保护最佳实践：展示了如何在保护隐私的前提下提供有效的安全服务
4. 技术透明度：完全开源的代码增强了系统的可信度

未来发展方向

1. 多语言支持：扩展支持更多个人信息类型和国际格式
2. 机器学习集成：使用机器学习算法识别新型泄漏模式
3. 实时监控：建立实时数据泄漏监控和预警系统
4. 区块链存证：使用区块链技术存储查询记录，增强审计能力
5. 联邦学习应用：在不暴露原始数据的前提下进行联合分析

总结：构建隐私安全的数字未来

leak-check项目展示了如何在技术层面平衡个人信息查询需求与隐私保护要求。通过3层脱敏聚合架构、智能类型识别机制和隐私优先的设计理念，项目为个人信息泄漏检测提供了一个可靠的技术解决方案。

在数据泄露事件频发的今天，类似leak-check这样的工具不仅具有技术价值，更体现了对用户隐私权的尊重。项目的开源特性使其可以成为更多安全产品的基础组件，推动整个行业向更加隐私友好的方向发展。

随着GDPR、个人信息保护法等法规的出台，个人信息保护已成为全球共识。leak-check项目为开发者提供了一个优秀的技术范例，展示了如何在遵守法规的前提下提供有价值的服务。无论是个人开发者构建隐私工具，还是企业构建安全监控系统，都可以从这个项目中获得技术灵感和实现参考。

在数字化浪潮中，技术应该是保护隐私的盾牌，而不是侵犯隐私的利剑。leak-check项目正是这一理念的实践者，通过技术创新为构建更加安全、隐私的数字世界贡献力量。

【免费下载链接】leak-check个人信息 “泄漏” 检测接口项目地址: https://gitcode.com/gh_mirrors/le/leak-check