ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

20行代码为MailAccess写一个OSINT模块:模块契约与自动发现机制开发者教程

20行代码为MailAccess写一个OSINT模块:模块契约与自动发现机制开发者教程

【免费下载链接】MailAccess

Free email OSINT tool, 5000+ platforms, identity clustering, breach detection. No API keys required. pip install mailaccess

项目地址:https://gitcode.com/gh_mirrors/mai/MailAccess
点击查看免费下载

MailAccess 是一个开源免费的邮件 OSINT 工具(pip install mailaccess即可运行),内置 82 个 OSINT 模块,可跨 5,300+ 平台完成身份聚类、数据泄露检测等调查,绝大多数模块无需 API Key。本文面向开发者,带你拆解它的模块契约与自动发现机制:只需一个约 20 行的 Python 文件,就能把你的数据源接入整条调查流水线,重启服务即被自动注册,全程零手动接线。

一、为什么 MailAccess 的模块设计值得学

MailAccess 的架构哲学是"数据源即插件"。一次investigate命令触发时,引擎会把所有已注册模块并发调度执行,各自产出标准化证据(finding),最终聚合成一份带置信度和溯源的报告。

对贡献者来说,这意味着:你不需要理解整个引擎,只需遵守一份极简契约。契约定义在 backend/modules/base.py,全文不到 100 行,是所有 OSINT 模块必须满足的"接口协议"。

二、模块契约:BaseModule 的四个类属性 + 一个方法

类属性:模块的"身份证"

继承BaseModule后,你要在类级别(不是__init__里)声明:

属性作用默认值
name唯一 slug,用于 API 响应和数据库记录必填
description一行说明,会展示在GET /api/modules/列表里必填
requires_key为True时,缺少 API Key 会自动跳过False
email_linked查找键就是邮箱本身时为True,会盖印到每条 finding 上False
priority调度优先级100

唯一方法:async run(email)

@abstractmethod async def run(self, email: str) -> ModuleResult: """Run the module against *email* and return a ModuleResult."""

返回值是ModuleResult数据类,包含四个字段:status(状态)、findings(发现列表)、metadata(模块级补充信息)、errors(人类可读错误)。状态枚举ModuleStatus有五种:

  • SUCCESS— 正常完成(可能有无发现)
  • SUCCESS_EMPTY— 正常完成且无发现
  • PARTIAL— 部分成功(例如部分请求被限流)
  • FAILED— 模块内部失败
  • SKIPPED— 因缺 Key 或配置关闭而主动跳过

💡 一个细节:BaseModule.__init_subclass__会在类创建时自动给你的run包一层校验——如果你返回了None或不是ModuleResult的对象,会直接抛AssertionError。这层"守门人"保证流水线里的脏数据无处可藏(源码见 backend/modules/base.py)。

三、20行代码写一个完整模块

按 CONTRIBUTING.md 的契约,一个最小可用的 OSINT 模块长这样:

from __future__ import annotations from ..core.http_client import build_client from .base import BaseModule, ModuleResult, ModuleStatus class ExampleModule(BaseModule): name = "example_check" # 1. 唯一 slug description = "Check Example service." # 2. 一行用途说明 requires_key = False # 3. 是否需要 API Key async def run(self, email: str) -> ModuleResult: # 4. 异步入口 findings: list[dict] = [] async with build_client(timeout=10.0) as client: try: resp = await client.get( f"https://example.com/api/check/{email}" ) if resp.status_code == 200 and resp.json().get("found"): findings.append({ "platform": "example_service", "profile_url": f"https://example.com/{email}", "confidence": "high", "metadata": {"note": "direct email match"}, }) except Exception as exc: # 5. 永不抛出,转成 errors return ModuleResult( status=ModuleStatus.FAILED, errors=[str(exc)] ) return ModuleResult(status=ModuleStatus.SUCCESS, findings=findings)

五条铁律(摘自 CONTRIBUTING.md):

  1. run()永不抛异常——捕获一切,用FAILED+errors返回;
  2. 缺 Key 尽早返回SKIPPED,而不是FAILED;
  3. 出站 HTTP 一律走build_client()(backend/core/http_client.py),它统一管理代理与限流;
  4. 每个模块有超时上限——默认 30 秒(MODULE_TIMEOUT_SECONDS,见 backend/config.py),引擎用asyncio.wait_for强制执行;
  5. 阻塞库要用asyncio.to_thread()包装,别卡死事件循环。

真实示例可参考 backend/modules/hackernews.py:错误收集、限流设置(rate_limiter.set_delay)、finding 去重一应俱全,是很好的"范本源码"。

四、自动发现机制:文件放对位置,重启即注册

这是 MailAccess 最优雅的设计。backend/modules/init.py 里的注册器在首次访问时扫描backend/modules/目录:

def _discover() -> None: package_dir = Path(__file__).parent for _finder, module_name, _ispkg in pkgutil.iter_modules([str(package_dir)]): if module_name in _SKIP: continue mod = importlib.import_module(f".{module_name}", package=__name__) for attr_name in dir(mod): obj = getattr(mod, attr_name) if (isinstance(obj, type) and issubclass(obj, BaseModule) and obj is not BaseModule and hasattr(obj, "name")): _registry[obj.name] = obj

翻译成白话:任何.py文件,只要定义了继承BaseModule且带name属性的类,就会被pkgutil.iter_modules扫出来、导入、注册进全局_registry——没有导入列表要维护,没有工厂方法要改。

三个细节值得注意:

  • 懒加载:发现逻辑被延迟到第一次需要模块列表时才执行(约 1 秒+ 的导入成本),避免拖慢服务冷启动和/health探活;
  • name 必须全局唯一,重复会静默覆盖先到者;
  • 验证很简单:重启服务后跑mailaccess modules,或用 APIGET /api/modules/(backend/api/routes/modules.py)——你的模块名出现了,就说明契约通过。

五、模块如何被引擎调度执行

注册只是入场券。真正执行发生在 backend/core/engine.py 的InvestigationEngine里:

  1. 引擎按阶段 DAG 组织执行(backend/core/phases.py),主阶段一次性拿到get_all_modules()的全部类;
  2. 用asyncio.Semaphore(max_concurrency)控制并发,默认10 个模块同时跑(MAX_CONCURRENT_MODULES),全局出站请求还有 40 的总闸门;
  3. 单模块超时默认 30 秒,超时会以FAILED记录但不影响整次调查;
  4. 你的 finding 会经过证据标注(trace/label)、跨模块去重、身份图聚合,最终进入报告、导出文件和数据库。

也就是说,一个写对契约的模块,天然获得了:并发调度、超时保护、错误隔离、证据链标注和持久化——这些你一行代码都不用写。

六、开发自查清单

提交前对照 CONTRIBUTING.md 的 PR Checklist:

  • name唯一,description一目了然
  • 缺 Key 时返回SKIPPED,不返回FAILED
  • 无阻塞 I/O 直跑事件循环;HTTP 走build_client()
  • 无硬编码凭据;不新增第三方依赖
  • pytest通过,测试随 PR 提交
  • 更新了 docs/modules.md 的模块表与 finding schema

落地路径只有一条:把文件放到backend/modules/my_module.py,下次服务启动自动注册。更多模块的 finding schema 细节(如gravatar、hibp、xposedornot等 82 个模块的完整字段定义),可查阅官方文档:docs/modules.md 的 "Adding a Module" 章节。

从 20 行代码到一条完整证据链,这就是 MailAccess 模块体系的开发体验:契约极小、发现自动、调度托管。

【免费下载链接】MailAccess

Free email OSINT tool, 5000+ platforms, identity clustering, breach detection. No API keys required. pip install mailaccess

项目地址:https://gitcode.com/gh_mirrors/mai/MailAccess
点击查看免费下载

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

返回列表