kev365/openrelik-worker-opensearch

# openrelic-worker-opensearch 用于将 workflow 结果索引到 OpenSearch 并上传至 Timesketch 的 OpenRelik worker。通过单个 Docker 镜像提供两个 Celery 任务。 ## 搜索指南 有关查询已上传数据的技巧，请参阅： - [在 OpenSearch Dashboards 中搜索](docs/searching-opensearch.md) - [在 Timesketch 中搜索](docs/searching-timesketch.md) ## 任务 ### OpenSearch 上传 通过 bulk API 将来自上游 workflow 文件的文档索引到 OpenSearch 索引中。 - 自动检测文件格式：JSONL、JSON、CSV、TSV、Parquet 及其 gzip 压缩变体。 - 添加 OpenRelik 上下文字段（`openrelik_workflow_id`、`openrelik_source_file`、`openrelik_source_path`）。 - 可选的时间线功能：根据 datetime 字段将行展开为包含 `datetime` 和 `timestamp_desc` 列、适用于时间线的文档。 - 可选地在 OpenSearch Dashboards 中进行索引别名关联和创建 index pattern。 ### Timesketch 上传 将文档上传到 Timesketch，并完成 sketch 和 timeline 的注册。 - 创建或复用 sketch 和 timeline。 - 从源数据列中自动检测 `datetime`、`message` 和 `timestamp_desc` 字段。 - 支持包含多个 datetime 字段的时间线处理（每一行的每个字段代表一个事件）。 - 在上传前将时间戳转换为 ISO8601 UTC 格式。 - 针对多文件上传的 timeline 名称前缀模式，或用于将文件合并为单个 timeline 的合并模式。 通过 `upload_method` 可以使用两种上传方法： | 方法 | 工作原理 | 适用场景 | |---|---|---| | `stream`（默认） | 通过 `ImportStreamer` 将文档流式传输至 Timesketch | 默认适用于所有情况。Timesketch 负责管理索引及其 mapping | | `direct_index` | 使用选定的 mapping 批量索引到 OpenSearch 中，然后通过 `generate_timeline_from_es_index` 将完成的索引注册到 Timesketch | 适用于宽表数据源（大约 300+ 个字段），或需要显式指定字段类型的情况 | #### 为什么会有 `direct_index` 在 `stream` 路径中，Timesketch 会根据每个文档的顶层键数量自行设定索引字段限制。有两个因素会导致这种估算偏低：每个字符串都会同时映射 `keyword` 和 `wildcard` 子字段，因此一个源字段会占用三个 mapping 条目而不是一个；此外，嵌套对象会被计算为单个键，而 OpenSearch 会映射每一个叶节点。在 OpenSearch 默认限制总计 1000 个字段的情况下，实际的上限大约只有 333 个源字段。 超过此上限时，上传请求可能会被接受，但实际上不会建立任何索引。详见 [google/timesketch#3889](https://github.com/google/timesketch/issues/3889)。 `direct_index` 通过直接在此处创建索引来规避此问题，这样 mapping 及其 `index.mapping.total_fields.limit` 将来自于你选定的 mapping 文件，而不是自动推断出来的。它还允许你固定那些值不一致的字段类型，而不是让 OpenSearch 根据恰好最先被索引的文档进行推断，从而导致后续文档被拒绝。 注意： - `direct_index` 直接写入 OpenSearch，因此 `opensearch_url` **必须指向 Timesketch 读取数据的同一个集群**，否则注册的 timeline 将是空的。 - 索引名称派生格式为 `ts__`。如果该索引已存在，注册将会失败，因为 Timesketch 不会接管它已经知道的索引。 - 如果有任何文档被拒绝，任务将失败且不会注册 timeline，而不是提供一个悄悄遗漏事件的 timeline。 ## 环境变量 ### OpenSearch 变量 | 变量 | 描述 | |---|---| | `OPENSEARCH_URL` | OpenSearch URL（默认：`https://opensearch:9200`） | | `OPENSEARCH_USERNAME` | Basic auth 用户名（可选） | | `OPENSEARCH_PASSWORD` | Basic auth 密码（可选） | | `OPENSEARCH_API_KEY` | API key 认证（可选，优先级高于 Basic auth） | | `OPENSEARCH_VERIFY_CERTS` | TLS 证书验证（默认：`true`） | | `OPENSEARCH_DASHBOARDS_URL` | 用于创建 index pattern 的 Dashboards URL（可选，例如 `http://opensearch-dashboards:5601`） | ### Timesketch 变量 | 变量 | 描述 | |---|---| | `TIMESKETCH_SERVER_URL` | Timesketch API URL（例如 `http://timesketch-web:5000`） | | `TIMESKETCH_SERVER_EXTERNAL_URL` | 供浏览器访问的 URL，用于报告中的 sketch 链接（如果与 `TIMESKETCH_SERVER_URL` 不同）（可选） | | `TIMESKETCH_USERNAME` | Timesketch 用户名（可按任务覆盖） | | `TIMESKETCH_PASSWORD` | Timesketch 密码（可按任务覆盖） | ## 任务配置 ### OpenSearch 上传配置 | 字段 | 类型 | 必填 | 描述 | |---|---|---|---| | `index_name` | text | 是 | 用于存储文档的 OpenSearch 索引名称 | | `mapping` | select | 否 | 索引 mapping：`dynamic`（默认）、内置名称（例如 `kstrike_ual`）或 `custom`（workflow 中的 .mapping 文件） | | `delimiter` | select | 否 | 源文件分隔符。`auto` 会根据文件扩展名自动检测。对于 CSV/TSV 请手动设置 | | `source_timezone` | select | 否 | 源时间戳的时区。在索引前会转换为 UTC | | `datetime_fields` | text | 否 | 用于时间线处理的逗号分隔时间戳字段（例如 `created_at,modified_at`）。每一行将按字段展开为单独的文档 | | `datetime_format` | select | 否 | 时间戳格式。`auto` 会尝试自动解析 | | `index_alias` | text | 否 | 在索引完成后将别名关联到索引 | | `create_index_pattern` | checkbox | 否 | 在 OpenSearch Dashboards 中创建 index pattern（需要设置 `OPENSEARCH_DASHBOARDS_URL` 环境变量） | | `batch_size` | text | 否 | 每次 bulk 请求的文档数（默认：1000，最大：10000） | | `opensearch_url` | text | 否 | OpenSearch URL 覆盖（默认：`http://opensearch:9200`） | | `username` | text | 否 | Basic auth 用户名 | | `password` | text | 否 | Basic auth 密码 | | `api_key` | text | 否 | API key 认证（优先级高于 Basic auth） | | `disable_verify_certs` | checkbox | 否 | 跳过 TLS 证书验证 | ### Timesketch 上传配置 **Sketch：** | 字段 | 类型 | 必填 | 描述 | |---|---|---|---| | `sketch_name` | text | 否 | 新 sketch 的名称（如果设置了 sketch ID 则忽略此项） | | `sketch_id` | text | 否 | 用于添加 timeline 的现有 sketch ID | | `timeline_name` | text | 否 | timeline 名称的前缀（例如 `case42` 会变成 `case42 - filename.txt`）。启用合并后，将用作确切的 timeline 名称 | | `merge_timelines` | checkbox | 否 | 使用上述名称将所有文件上传到一个 timeline 中 | **字段映射：** | 字段 | 类型 | 必填 | 描述 | |---|---|---|---| | `ts_datetime_field` | text | 否 | 逗号分隔的时间戳字段名称（例如 `created_at,modified_at`）。如果源数据包含 `datetime` 列则会自动检测 | | `ts_message_fields` | text | 否 | 将拼接为 Timesketch `message` 的逗号分隔字段（例如 `user,action,resource`）。如果源数据包含 `message` 列则会自动检测 | **源格式：** | 字段 | 类型 | 必填 | 描述 | |---|---|---|---| | `delimiter` | select | 否 | 源文件分隔符。`auto` 会根据文件扩展名自动检测 | | `ts_datetime_format` | select | 否 | 时间戳格式。`auto` 会尝试自动解析。值将被转换为 ISO8601 | | `source_timezone` | select | 否 | 源时间戳的时区。在上传前会转换为 UTC | **连接：** | 字段 | 类型 | 必填 | 描述 | |---|---|---|---| | `timesketch_url` | text | 是 | Timesketch API URL（默认：`http://host.docker.internal`） | | `timesketch_username` | text | 是 | Timesketch 用户名 | | `timesketch_password` | text | 是 | Timesketch 密码 | **上传方法：** | 字段 | 类型 | 必填 | 描述 | |---|---|---|---| | `upload_method` | select | 否 | `stream`（默认）或 `direct_index`。详见 [Timesketch 上传](#timesketch-upload) | | `mapping` | select | 否 | `direct_index` 使用的索引 mapping。在 `stream` 路径下被忽略 | | `opensearch_url` | text | 否 | 仅限 `direct_index`。必须是 Timesketch 读取数据的同一个集群。回退至 `OPENSEARCH_URL` | | `auth_mode` | text | 否 | 仅限 `direct_index`。可选 `none`、`basic` 或 `apikey`。留空则会根据环境自动检测 | | `username` | text | 否 | 仅限 `direct_index`。OpenSearch 用户名，回退至 `OPENSEARCH_USERNAME` | | `password` | text | 否 | 仅限 `direct_index`。OpenSearch 密码，回退至 `OPENSEARCH_PASSWORD` | ## 内置 Mapping | 名称 | 描述 | |---|---| | `kstrike_ual` | KStrike UAL 解析器输出，包含带类型的日期字段（`insertdate`、`lastaccess`） | | `github_audit` | GitHub 审计日志。针对该格式的字段广度将 `total_fields.limit` 提高至 3000，镜像 Timesketch 自有的 `keyword`/`wildcard` 字符串模板以确保搜索行为完全一致，并将 `permission` 固定为 `keyword` 类型，因为它在大多数记录中是字符串，而在少数记录中是布尔值 | 内置 mapping 会从 `src/mappings/*.json` 中自动发现。它们为导出任务以及当 `upload_method` 为 `direct_index` 时的 Timesketch 任务定义了 OpenSearch 索引字段类型。使用 `dynamic` 让 OpenSearch 自动检测类型，或者使用 `custom` 上传你自己的 `.mapping` 文件。 mapping 文件可以在 `mappings` 旁边包含一个 `settings` 块；它将被直接传递给索引创建过程，这就是提高 `total_fields.limit` 的实现方式。 ## 本地开发 ``` # 使用 test 和 timesketch extras 安装依赖 uv sync --group test --extra timesketch # 运行测试 uv run pytest -v ```

标签：Docker, OpenRelik, Python, Timesketch, 安全取证, 安全防御评估, 数据导入, 无后门, 日志索引, 请求拦截, 逆向工具