mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-09-29 05:56:38 +08:00
Merge branch 'doc'
This commit is contained in:
@@ -169,9 +169,9 @@ Client -> OpenResty server block -> WAF Lua -> named upstream -> Origin
|
||||
|
||||
网站配置是反向代理聚合边界。一条网站配置可绑定多个域名,并共享站点级流量限制、反向代理和缓存配置。
|
||||
|
||||
WAF 在 OpenResty `access_by_lua_file` 阶段执行。规则来自当前激活版本携带的 `waf_config.json`,全局规则组默认生效,网站可叠加自定义规则组。
|
||||
WAF 在 OpenResty `access_by_lua_file` 阶段执行。规则来自当前激活版本携带的 `waf_config.json`,全局规则组默认生效,网站可叠加自定义规则组。`waf_config.json` 只保存规则组直接 IP 和 IP 组引用 ID;IP 组成员由 Agent 独立同步到本地 `waf_ip_groups.json`,OpenResty Lua 按引用 ID 合并判断。
|
||||
|
||||
WAF IP 组由 Server 管理并在发布时展开到 `waf_config.json`。手动 IP 组直接保存 IP/IP 段列表;自动 IP 组由 Server 定时任务读取请求日志、按单个 IP 聚合指标并执行 Expr 规则;订阅 IP 组由 Server 定时任务同步远程文本或 JSON 源。OpenResty Lua 只读取 Agent 落地的运行时 JSON,不直接访问 Server 数据库、请求日志或远程订阅源。
|
||||
WAF IP 组由 Server 管理。手动 IP 组直接保存 IP/IP 段列表;自动 IP 组由 Server 定时任务读取请求日志、按单个 IP 聚合指标并执行 Expr 规则;订阅 IP 组由 Server 定时任务同步远程文本或 JSON 源。Agent 心跳会上报本地 IP 组 checksum,Server 只返回不一致的 IP 组;Server 侧 IP 组更新时会通过 Agent WebSocket 广播变更组。OpenResty Lua 只读取 Agent 落地的运行时 JSON,不直接访问 Server 数据库、请求日志或远程订阅源。
|
||||
|
||||
## 核心对象
|
||||
|
||||
|
||||
@@ -181,7 +181,8 @@ IP 组约束:
|
||||
* 手动 IP 组由管理端直接维护 IP/IP 段列表。
|
||||
* 自动 IP 组使用 Expr 语法保存自定义规则,由 Server 定时按单个 IP 聚合请求日志并更新 IP 列表。
|
||||
* 订阅 IP 组由 Server 定时从 HTTP/HTTPS URL 同步,支持文本列表和 JSON 映射。
|
||||
* WAF 运行时不访问数据库;发布时将规则组引用的启用 IP 组展开进完整配置版本。
|
||||
* WAF 运行时不访问数据库;发布版本只保存规则组引用的 IP 组 ID,不把 IP 组成员展开进版本快照。
|
||||
* Agent 通过心跳上报本地 IP 组 checksum,Server 仅返回 checksum 不一致的 IP 组;Server 侧 IP 组更新时会通过 Agent WebSocket 主动广播变更组,使节点可在不重新发布配置版本的情况下更新 WAF IP 组内容。
|
||||
|
||||
自动 IP 组首批内置预设规则:
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ Server 发布时必须:
|
||||
2. 读取 Server 侧 OpenResty 主配置、性能参数、缓存参数和必要 Lua 资源。
|
||||
3. 读取域名与证书绑定关系。
|
||||
4. 读取 WAF 全局规则组、自定义规则组、IP 组引用与网站绑定关系。
|
||||
5. 使用自动 IP 组最近一次执行后的 IP 列表,并展开 WAF 规则组引用的启用 IP 组,渲染完整 OpenResty 配置与 WAF 运行时配置。
|
||||
5. 保留 WAF 规则组引用的 IP 组 ID,渲染完整 OpenResty 配置与 WAF 运行时配置;IP 组成员不进入发布版本。
|
||||
6. 计算 `checksum`。
|
||||
7. 写入 `config_versions`。
|
||||
8. 切换激活版本。
|
||||
@@ -70,4 +70,10 @@ Agent 发现新版本后会:
|
||||
* Agent API 固定使用节点专属 `agent_token`,首次接入可使用 `discovery_token`。
|
||||
* Server 不提供远程 shell 或任意命令执行入口。
|
||||
* 配置版本必须保存完整快照、渲染结果和 `checksum`。
|
||||
* WAF 规则组、IP 组快照和网站绑定关系必须随完整配置版本进入快照与 checksum,回滚时不得依赖当前可变 WAF 配置。
|
||||
* WAF 规则组、IP 组引用 ID 和网站绑定关系必须随完整配置版本进入快照与 checksum;IP 组成员由 Agent 独立按 checksum 差异同步,不受版本回滚影响。
|
||||
|
||||
## WAF IP 组运行时同步
|
||||
|
||||
WAF IP 组成员不纳入配置版本。发布版本只包含规则组直接 IP 与 `ip_whitelist_group_ids` / `ip_blacklist_group_ids`。Agent 应用版本后会从渲染出的 `waf_config.json` 中提取引用 ID,并向 Server 请求缺失或 checksum 不一致的 IP 组数据。
|
||||
|
||||
Agent 后续心跳会携带本地 IP 组 checksum。Server 根据当前激活版本引用的 IP 组 ID 对比 checksum,只返回差异组,避免每次心跳传输全部 IP 组。Server 在手动更新、订阅同步或自动规则执行后,会通过 Agent WebSocket 广播发生变化的 IP 组;WS 不可用时,下一次 HTTP heartbeat 仍会按 checksum 差异补齐。
|
||||
|
||||
+2
-2
@@ -83,11 +83,11 @@ HTTPS 按域名绑定证书,而不是按整个网站统一强制启用。
|
||||
* WAF 页面维护全局规则组和自定义规则组。全局规则组始终应用到全部网站;自定义规则组可以在规则组内一键选择网站,也可以在网站详情的 `WAF` 分区绑定。
|
||||
* 点击 WAF 页面中的 **管理 IP 组** 可以进入独立 IP 组页面。手动 IP 组直接维护 IP/IP 段;自动 IP 组使用 Expr 规则按单个 IP 聚合请求日志并定时更新名单;订阅 IP 组可从远程文本或 JSON 源定时同步。
|
||||
* 自动 IP 组页面提供两个预设:单个 IP 请求数大于 100 且 404 占比不低于 80%;单个 IP 通过 IP 地址访问次数大于 50 且该访问占比大于 50%。保存前可点击 **测试规则** 查看当前日志窗口命中的 IP,保存后可点击 **立即执行** 更新组内名单,语法见 [WAF 自动 IP 组规则语法](./waf-ip-group-expr.md)。
|
||||
* 在 WAF 规则组的黑白名单中,IP 维度既可以直接添加 IP/IP 段,也可以引用已有 IP 组。发布时 Server 会把启用 IP 组展开到 WAF 运行时配置。
|
||||
* 在 WAF 规则组的黑白名单中,IP 维度既可以直接添加 IP/IP 段,也可以引用已有 IP 组。发布时版本只携带 IP 组引用 ID;Agent 会按 checksum 差异同步 IP 组成员,并在 Server 通过 WebSocket 广播 IP 组更新时实时落地到节点。
|
||||
* `PoW` 是规则组内的一个配置 Tab,位于 `黑白名单` 与 `拦截返回` 之间,复用站点已有 PoW 执行逻辑,可将当前 PoW 配置应用到全部网站或当前规则组绑定的网站。
|
||||
* 网站详情页不再单独编辑 PoW 规则,只展示全局 WAF 规则组并绑定自定义 WAF 规则组。PoW 的启用范围和规则内容应回到 WAF 页面统一维护。
|
||||
|
||||
WAF、IP 组或 PoW 配置修改后,都需要重新发布并激活配置版本,Agent 才会拉取并应用到 OpenResty。
|
||||
WAF 规则组、网站绑定或 PoW 配置修改后,需要重新发布并激活配置版本,Agent 才会拉取并应用到 OpenResty。IP 组成员变化不需要重新发布版本;在线 Agent 会通过 WebSocket 增量更新,离线或未升级 WS 的 Agent 会在下一次心跳中按 checksum 差异补齐。
|
||||
|
||||
## 发布、激活与回滚
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# WAF 自动 IP 组规则语法
|
||||
|
||||
自动 IP 组用于从请求日志中按单个客户端 IP 聚合指标,再用 Expr 表达式判断是否把该 IP 加入组内名单。自动 IP 组可以被 WAF 规则组的 IP 黑名单或白名单引用;发布配置时,Server 会把启用 IP 组展开到 `waf_config.json`。
|
||||
自动 IP 组用于从请求日志中按单个客户端 IP 聚合指标,再用 Expr 表达式判断是否把该 IP 加入组内名单。自动 IP 组可以被 WAF 规则组的 IP 黑名单或白名单引用;发布配置时,Server 只把 IP 组引用 ID 写入 `waf_config.json`,IP 组成员由 Agent 独立同步到本地运行时文件。
|
||||
|
||||
## 配置结构
|
||||
|
||||
@@ -158,4 +158,4 @@ IP 直连访问异常:
|
||||
|
||||
先用较短的回看窗口和较高阈值观察命中结果,再逐步调整阈值。管理端 IP 组页面支持在保存前点击 **测试规则**,直接查看当前回看窗口内命中的 IP;自动 IP 组真正执行后会覆盖该组的 IP 列表。如果要长期保留某些地址,建议放入手动 IP 组,并在 WAF 规则组中同时引用手动组和自动组。
|
||||
|
||||
自动 IP 组更新后不会立即改变 Agent 上的运行时配置。需要重新发布并激活配置版本,Agent 才会拉取新的 `waf_config.json`。
|
||||
自动 IP 组更新后不需要重新发布配置版本。在线 Agent 会通过 WebSocket 收到变更 IP 组并更新本地 `waf_ip_groups.json`;WebSocket 不可用时,Agent 会在下一次心跳中上报本地 IP 组 checksum,Server 只返回 checksum 不一致的 IP 组。
|
||||
|
||||
@@ -202,6 +202,7 @@ v1-v7 视为历史初始基线,不再维护逐版本升级文件。从 v8 起
|
||||
* 同时读取 OpenResty 主配置参数、反代性能参数与缓存参数。
|
||||
* 读取 WAF 规则组、规则组引用的 IP 组与网站绑定关系,并在发布快照中保存可回放数据。
|
||||
* 自动型 WAF IP 组只能由 Server 定时任务读取请求日志并执行 Expr 布尔规则,OpenResty Lua 与 Agent 不得直接访问请求日志库或执行自动挖掘逻辑。
|
||||
* 发布版本不得展开 WAF IP 组成员;Agent 必须通过独立的 IP 组 checksum 差异同步和 WebSocket 增量广播维护本地 `waf_ip_groups.json`。
|
||||
* **内网穿透配置扩展**:区分上游类型,为 `upstream_type = 'tunnel'` 的代理规则生成独立的 tunnel 配置数据。
|
||||
* OpenResty 侧:将 tunnel 上游自动渲染为 `http://127.0.0.1:{relay_vhost_port}`,必须保留原始 `Host` 请求头。
|
||||
* Tunnel 侧:为每个 Client 生成完整的 relay 列表与 frpc 代理定义(frpc proxy 配置)。
|
||||
@@ -227,6 +228,7 @@ Agent 必须满足:
|
||||
* 发现新版本时先备份旧文件。
|
||||
* 写入主配置、路由配置与必要证书文件。
|
||||
* 写入 WAF/PoW 运行时配置,并确保 WAF Lua 资源由 Agent 统一管理。
|
||||
* WAF IP 组同步必须按组增量更新,不得在每次心跳或每次同步中传输全部 IP 组。
|
||||
* 写入新配置后执行 `openresty -t -c <main_config_path>`,再 reload;reload 发现运行时未启动时允许直接启动 OpenResty。
|
||||
* 周期性运行时健康检查不得调用 `openresty -t`,避免健康探针触发 upstream 域名同步解析;应优先请求本地 `openresty_observability_port` 上的 `/openflare/stub_status`,以 HTTP `200 OK` 作为 OpenResty 主进程和 worker 正在提供服务的判断依据。
|
||||
* 新配置激活失败时必须先尝试用目标配置恢复运行,再回滚到旧配置并重新拉起 OpenResty。
|
||||
|
||||
@@ -69,6 +69,43 @@ Agent 正式请求统一使用节点专属 `agent_token`,首次接入可使用
|
||||
X-Agent-Token: <token>
|
||||
```
|
||||
|
||||
### Agent WAF IP 组同步
|
||||
|
||||
Agent 心跳 payload 可携带本地 WAF IP 组 checksum:
|
||||
|
||||
```json
|
||||
{
|
||||
"waf_ip_group_checksums": {
|
||||
"1": "sha256..."
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Server 会根据当前激活版本引用的 IP 组 ID 对比 checksum,并在心跳响应顶层返回差异组:
|
||||
|
||||
```json
|
||||
{
|
||||
"waf_ip_groups": [
|
||||
{
|
||||
"id": 1,
|
||||
"name": "自动黑名单",
|
||||
"type": "automatic",
|
||||
"enabled": true,
|
||||
"ip_list": ["203.0.113.10"],
|
||||
"checksum": "sha256..."
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
Agent 也可以在应用新版本后主动请求差异同步:
|
||||
|
||||
| 方法 | 路径 | 说明 |
|
||||
| --- | --- | --- |
|
||||
| `POST` | `/api/agent/waf/ip-groups/sync` | 根据 Agent 上报的 `ids` 与 `checksums` 返回不一致的 IP 组 |
|
||||
|
||||
当 Server 侧 IP 组更新时,已连接的 Agent WebSocket 会收到 `type = "waf_ip_groups"` 的消息,payload 为发生变化的 IP 组数组。Agent 应只更新收到的组,不要求 Server 每次下发全部 IP 组。
|
||||
|
||||
日志中不得打印完整 Token。
|
||||
|
||||
## Swagger
|
||||
|
||||
@@ -24,6 +24,8 @@ type SyncService interface {
|
||||
SyncOnStartup(ctx context.Context, target *protocol.ActiveConfigMeta) error
|
||||
SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
|
||||
ForceSyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
|
||||
WAFIPGroupChecksums() (map[string]string, error)
|
||||
ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) error
|
||||
}
|
||||
|
||||
type Updater interface {
|
||||
@@ -161,6 +163,7 @@ func (r *Runner) performHeartbeatCycle(ctx context.Context, nodeID string, start
|
||||
}
|
||||
slog.Debug("agent heartbeat succeeded", "mode", mode, "node_id", nodeID)
|
||||
changed := r.applySettings(heartbeatResult.AgentSettings)
|
||||
r.applyWAFIPGroups(ctx, heartbeatResult.WAFIPGroups)
|
||||
if startup {
|
||||
if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil {
|
||||
r.recordSyncError(err)
|
||||
@@ -307,6 +310,14 @@ func (r *Runner) handleWebSocketMessage(ctx context.Context, message protocol.WS
|
||||
slog.Error("agent ws triggered force sync failed", "version", target.Version, "error", err)
|
||||
}
|
||||
return false, nil
|
||||
case protocol.WSMessageTypeWAFIPGroups:
|
||||
var groups []protocol.WAFIPGroup
|
||||
if err := json.Unmarshal(message.Payload, &groups); err != nil {
|
||||
slog.Debug("agent ws waf ip groups decode failed", "error", err)
|
||||
return false, nil
|
||||
}
|
||||
r.applyWAFIPGroups(ctx, groups)
|
||||
return false, nil
|
||||
case protocol.WSMessageTypePing:
|
||||
slog.Debug("agent ws ping received")
|
||||
return false, conn.SendPong()
|
||||
@@ -470,6 +481,7 @@ func (r *Runner) tryRegister(ctx context.Context, nodeID *string) error {
|
||||
heartbeatResult = &protocol.HeartbeatResult{}
|
||||
}
|
||||
r.applySettings(heartbeatResult.AgentSettings)
|
||||
r.applyWAFIPGroups(ctx, heartbeatResult.WAFIPGroups)
|
||||
if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil {
|
||||
r.recordSyncError(err)
|
||||
slog.Error("agent post-register startup sync failed", "error", err)
|
||||
@@ -563,7 +575,7 @@ func (r *Runner) nodePayload(nodeID string) protocol.NodePayload {
|
||||
}
|
||||
metricSnapshot := observability.BuildSnapshot(r.Config, r.StateStore, managedOpenRestyMetrics)
|
||||
healthEvents := observability.BuildHealthEvents(snapshot)
|
||||
return protocol.NodePayload{
|
||||
payload := protocol.NodePayload{
|
||||
NodeID: nodeID,
|
||||
Name: r.Config.NodeName,
|
||||
IP: r.Config.NodeIP,
|
||||
@@ -579,6 +591,25 @@ func (r *Runner) nodePayload(nodeID string) protocol.NodePayload {
|
||||
AccessLogs: accessLogs,
|
||||
HealthEvents: healthEvents,
|
||||
}
|
||||
if r.SyncService != nil {
|
||||
checksums, err := r.SyncService.WAFIPGroupChecksums()
|
||||
if err != nil {
|
||||
slog.Debug("load local waf ip group checksums failed", "error", err)
|
||||
} else if len(checksums) > 0 {
|
||||
payload.WAFIPGroupChecksums = checksums
|
||||
}
|
||||
}
|
||||
return payload
|
||||
}
|
||||
|
||||
func (r *Runner) applyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) {
|
||||
if len(groups) == 0 || r.SyncService == nil {
|
||||
return
|
||||
}
|
||||
if err := r.SyncService.ApplyWAFIPGroups(ctx, groups); err != nil {
|
||||
r.recordSyncError(err)
|
||||
slog.Error("agent apply waf ip groups failed", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Runner) prepareHeartbeatPayload(nodeID string) (protocol.NodePayload, []int64) {
|
||||
|
||||
@@ -70,6 +70,8 @@ type fakeSyncService struct {
|
||||
syncOnceCalls int
|
||||
lastTarget *protocol.ActiveConfigMeta
|
||||
onSyncOnceCall func(int)
|
||||
wafChecksums map[string]string
|
||||
wafGroups []protocol.WAFIPGroup
|
||||
}
|
||||
|
||||
type fakeRuntimeManager struct {
|
||||
@@ -135,6 +137,20 @@ func (f *fakeSyncService) ForceSyncOnce(ctx context.Context, target *protocol.Ac
|
||||
return f.syncOnceErr
|
||||
}
|
||||
|
||||
func (f *fakeSyncService) WAFIPGroupChecksums() (map[string]string, error) {
|
||||
if f.wafChecksums == nil {
|
||||
return map[string]string{}, nil
|
||||
}
|
||||
return f.wafChecksums, nil
|
||||
}
|
||||
|
||||
func (f *fakeSyncService) ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.wafGroups = append(f.wafGroups, groups...)
|
||||
return nil
|
||||
}
|
||||
|
||||
type fakeWebSocketConnection struct {
|
||||
pongCalls int
|
||||
}
|
||||
|
||||
@@ -54,6 +54,7 @@ func (c *Client) Heartbeat(ctx context.Context, payload protocol.NodePayload) (*
|
||||
return &protocol.HeartbeatResult{
|
||||
AgentSettings: resp.AgentSettings,
|
||||
ActiveConfig: resp.ActiveConfig,
|
||||
WAFIPGroups: resp.WAFIPGroups,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -74,6 +75,17 @@ func (c *Client) ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPa
|
||||
return c.postJSON(ctx, "/api/agent/apply-logs", payload, nil)
|
||||
}
|
||||
|
||||
func (c *Client) SyncWAFIPGroups(ctx context.Context, payload protocol.WAFIPGroupSyncRequest) (*protocol.WAFIPGroupSyncResponse, error) {
|
||||
resp := protocol.APIResponse[protocol.WAFIPGroupSyncResponse]{}
|
||||
if err := c.postJSON(ctx, "/api/agent/waf/ip-groups/sync", payload, &resp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !resp.Success {
|
||||
return nil, errors.New(resp.Message)
|
||||
}
|
||||
return &resp.Data, nil
|
||||
}
|
||||
|
||||
func (c *Client) SetToken(token string) {
|
||||
c.token = strings.TrimSpace(token)
|
||||
slog.Debug("http client token updated")
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
@@ -27,6 +28,7 @@ import (
|
||||
|
||||
const RuntimeConfigDirPlaceholder = "__OPENFLARE_RUNTIME_CONFIG_DIR__"
|
||||
const ResolverDirectivePlaceholder = "__OPENFLARE_RESOLVER_DIRECTIVE__"
|
||||
const WAFIPGroupsConfigFileName = "waf_ip_groups.json"
|
||||
|
||||
type Executor interface {
|
||||
Test(ctx context.Context) error
|
||||
@@ -190,6 +192,10 @@ type ApplyOutcome struct {
|
||||
Message string
|
||||
}
|
||||
|
||||
type wafIPGroupsRuntimeConfig struct {
|
||||
Groups map[string]protocol.WAFIPGroup `json:"groups"`
|
||||
}
|
||||
|
||||
func (m *Manager) Apply(ctx context.Context, mainConfig string, routeConfig string, supportFiles []protocol.SupportFile) ApplyOutcome {
|
||||
slog.Info("openresty apply started", "main_config", m.MainConfigPath, "route_config", m.RouteConfigPath, "cert_files", len(supportFiles))
|
||||
backup, err := m.backup()
|
||||
@@ -425,6 +431,77 @@ func (m *Manager) CurrentChecksum() (string, error) {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (m *Manager) WAFIPGroupChecksums() (map[string]string, error) {
|
||||
config, err := m.readWAFIPGroupsRuntimeConfig()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result := make(map[string]string, len(config.Groups))
|
||||
for id, group := range config.Groups {
|
||||
if strings.TrimSpace(group.Checksum) != "" {
|
||||
result[id] = strings.TrimSpace(group.Checksum)
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (m *Manager) SyncWAFIPGroups(groups []protocol.WAFIPGroup) error {
|
||||
if m.RuntimeConfigDir == "" || len(groups) == 0 {
|
||||
return nil
|
||||
}
|
||||
config, err := m.readWAFIPGroupsRuntimeConfig()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if config.Groups == nil {
|
||||
config.Groups = make(map[string]protocol.WAFIPGroup)
|
||||
}
|
||||
for _, group := range groups {
|
||||
if group.ID == 0 {
|
||||
continue
|
||||
}
|
||||
config.Groups[fmt.Sprintf("%d", group.ID)] = group
|
||||
}
|
||||
data, err := json.Marshal(config)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.MkdirAll(m.RuntimeConfigDir, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
path := filepath.Join(m.RuntimeConfigDir, WAFIPGroupsConfigFileName)
|
||||
if err := os.WriteFile(path, data, 0o644); err != nil {
|
||||
return fmt.Errorf("write %s: %w", WAFIPGroupsConfigFileName, err)
|
||||
}
|
||||
slog.Info("synced waf ip groups", "path", path, "group_count", len(groups))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) readWAFIPGroupsRuntimeConfig() (*wafIPGroupsRuntimeConfig, error) {
|
||||
config := &wafIPGroupsRuntimeConfig{Groups: map[string]protocol.WAFIPGroup{}}
|
||||
if m.RuntimeConfigDir == "" {
|
||||
return config, nil
|
||||
}
|
||||
path := filepath.Join(m.RuntimeConfigDir, WAFIPGroupsConfigFileName)
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return config, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if len(data) == 0 {
|
||||
return config, nil
|
||||
}
|
||||
if err := json.Unmarshal(data, config); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if config.Groups == nil {
|
||||
config.Groups = map[string]protocol.WAFIPGroup{}
|
||||
}
|
||||
return config, nil
|
||||
}
|
||||
|
||||
type ExecutorOptions struct {
|
||||
NginxPath string
|
||||
MainConfigPath string
|
||||
|
||||
@@ -915,6 +915,37 @@ func TestManagerApplyRejectsCertFilePathTraversal(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerSyncWAFIPGroupsWritesDeltaRuntimeFile(t *testing.T) {
|
||||
manager := &Manager{RuntimeConfigDir: t.TempDir()}
|
||||
|
||||
if err := manager.SyncWAFIPGroups([]protocol.WAFIPGroup{
|
||||
{ID: 1, Enabled: true, IPList: []string{"203.0.113.10"}, Checksum: "sum-1"},
|
||||
}); err != nil {
|
||||
t.Fatalf("SyncWAFIPGroups failed: %v", err)
|
||||
}
|
||||
if err := manager.SyncWAFIPGroups([]protocol.WAFIPGroup{
|
||||
{ID: 2, Enabled: true, IPList: []string{"198.51.100.10"}, Checksum: "sum-2"},
|
||||
}); err != nil {
|
||||
t.Fatalf("SyncWAFIPGroups second delta failed: %v", err)
|
||||
}
|
||||
|
||||
checksums, err := manager.WAFIPGroupChecksums()
|
||||
if err != nil {
|
||||
t.Fatalf("WAFIPGroupChecksums failed: %v", err)
|
||||
}
|
||||
if checksums["1"] != "sum-1" || checksums["2"] != "sum-2" {
|
||||
t.Fatalf("expected merged checksums, got %#v", checksums)
|
||||
}
|
||||
data, err := os.ReadFile(filepath.Join(manager.RuntimeConfigDir, WAFIPGroupsConfigFileName))
|
||||
if err != nil {
|
||||
t.Fatalf("failed to read runtime ip group file: %v", err)
|
||||
}
|
||||
text := string(data)
|
||||
if !strings.Contains(text, "203.0.113.10") || !strings.Contains(text, "198.51.100.10") {
|
||||
t.Fatalf("expected runtime file to keep both groups, got %s", text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestObservabilityListenAddress(t *testing.T) {
|
||||
if got := ObservabilityListenAddress(18081); got != "127.0.0.1:18081" {
|
||||
t.Fatalf("unexpected default observability listen address: %s", got)
|
||||
|
||||
@@ -49,6 +49,36 @@ local function load_config()
|
||||
return nil
|
||||
end
|
||||
|
||||
local function load_ip_groups()
|
||||
local paths = {
|
||||
"__OPENFLARE_RUNTIME_CONFIG_DIR__/waf_ip_groups.json",
|
||||
"/etc/nginx/openflare-lua/waf_ip_groups.json",
|
||||
"/usr/local/openresty/nginx/conf/waf_ip_groups.json"
|
||||
}
|
||||
for _, path in ipairs(paths) do
|
||||
local content = read_file(path)
|
||||
if content and content ~= "" then
|
||||
local hash = ngx.md5(content)
|
||||
if config_dict:get("_ip_groups_hash") == hash then
|
||||
local cached = config_dict:get("_ip_groups_json")
|
||||
if cached then
|
||||
local decoded = cjson.decode(cached)
|
||||
if decoded then
|
||||
return decoded
|
||||
end
|
||||
end
|
||||
end
|
||||
local decoded = cjson.decode(content)
|
||||
if decoded then
|
||||
config_dict:set("_ip_groups_hash", hash, 0)
|
||||
config_dict:set("_ip_groups_json", content, 0)
|
||||
return decoded
|
||||
end
|
||||
end
|
||||
end
|
||||
return { groups = {} }
|
||||
end
|
||||
|
||||
local function list_contains(items, value)
|
||||
if not items or type(items) ~= "table" or not value or value == "" then
|
||||
return false
|
||||
@@ -109,6 +139,20 @@ local function ip_matches(items, ip)
|
||||
return false
|
||||
end
|
||||
|
||||
local function ip_matches_group_ids(group_ids, ip, ip_groups_config)
|
||||
if not group_ids or type(group_ids) ~= "table" or not ip or ip == "" then
|
||||
return false
|
||||
end
|
||||
local groups = (ip_groups_config or {}).groups or {}
|
||||
for _, id in ipairs(group_ids) do
|
||||
local group = groups[tostring(id)]
|
||||
if group and group.enabled and ip_matches(group.ip_list, ip) then
|
||||
return true
|
||||
end
|
||||
end
|
||||
return false
|
||||
end
|
||||
|
||||
local function lookup_country(ip)
|
||||
local ok, maxminddb = pcall(require, "resty.maxminddb")
|
||||
if not ok or not maxminddb then
|
||||
@@ -181,6 +225,7 @@ end
|
||||
|
||||
local ip = ngx.var.remote_addr or ""
|
||||
local groups = active_groups(config)
|
||||
local ip_groups_config = load_ip_groups()
|
||||
if #groups == 0 then
|
||||
if config_dict:add("_empty_groups_logged", true, 60) then
|
||||
ngx.log(ngx.WARN, "openflare waf has no active rule group for site: ", ngx.var.openflare_waf_site or "")
|
||||
@@ -189,7 +234,7 @@ if #groups == 0 then
|
||||
end
|
||||
|
||||
for _, group in ipairs(groups) do
|
||||
if ip_matches(group.ip_whitelist, ip) then
|
||||
if ip_matches(group.ip_whitelist, ip) or ip_matches_group_ids(group.ip_whitelist_group_ids, ip, ip_groups_config) then
|
||||
return
|
||||
end
|
||||
end
|
||||
@@ -205,7 +250,7 @@ for _, group in ipairs(groups) do
|
||||
end
|
||||
|
||||
for _, group in ipairs(groups) do
|
||||
if ip_matches(group.ip_blacklist, ip) then
|
||||
if ip_matches(group.ip_blacklist, ip) or ip_matches_group_ids(group.ip_blacklist_group_ids, ip, ip_groups_config) then
|
||||
return exit_with_group(group)
|
||||
end
|
||||
end
|
||||
|
||||
@@ -14,11 +14,13 @@ type HeartbeatAPIResponse struct {
|
||||
Data any `json:"data"`
|
||||
AgentSettings *AgentSettings `json:"agent_settings,omitempty"`
|
||||
ActiveConfig *ActiveConfigMeta `json:"active_config,omitempty"`
|
||||
WAFIPGroups []WAFIPGroup `json:"waf_ip_groups,omitempty"`
|
||||
}
|
||||
|
||||
type HeartbeatResult struct {
|
||||
AgentSettings *AgentSettings
|
||||
ActiveConfig *ActiveConfigMeta
|
||||
WAFIPGroups []WAFIPGroup
|
||||
}
|
||||
|
||||
type AgentSettings struct {
|
||||
@@ -37,6 +39,7 @@ const (
|
||||
WSMessageTypeSettings = "settings"
|
||||
WSMessageTypeActiveConfig = "active_config"
|
||||
WSMessageTypeForceSyncConfig = "force_sync_config"
|
||||
WSMessageTypeWAFIPGroups = "waf_ip_groups"
|
||||
WSMessageTypePing = "ping"
|
||||
WSMessageTypePong = "pong"
|
||||
)
|
||||
@@ -81,6 +84,7 @@ type NodePayload struct {
|
||||
AccessLogs []NodeAccessLog `json:"access_logs,omitempty"`
|
||||
BufferedObservability []BufferedObservabilityRecord `json:"buffered_observability,omitempty"`
|
||||
HealthEvents []NodeHealthEvent `json:"health_events"`
|
||||
WAFIPGroupChecksums map[string]string `json:"waf_ip_group_checksums,omitempty"`
|
||||
}
|
||||
|
||||
type NodeSystemProfile struct {
|
||||
@@ -177,6 +181,24 @@ type ActiveConfigMeta struct {
|
||||
Checksum string `json:"checksum"`
|
||||
}
|
||||
|
||||
type WAFIPGroup struct {
|
||||
ID uint `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Type string `json:"type"`
|
||||
Enabled bool `json:"enabled"`
|
||||
IPList []string `json:"ip_list"`
|
||||
Checksum string `json:"checksum"`
|
||||
}
|
||||
|
||||
type WAFIPGroupSyncRequest struct {
|
||||
IDs []uint `json:"ids,omitempty"`
|
||||
Checksums map[string]string `json:"checksums,omitempty"`
|
||||
}
|
||||
|
||||
type WAFIPGroupSyncResponse struct {
|
||||
Groups []WAFIPGroup `json:"groups"`
|
||||
}
|
||||
|
||||
type SupportFile struct {
|
||||
Path string `json:"path"`
|
||||
Content string `json:"content"`
|
||||
|
||||
@@ -4,10 +4,12 @@ import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
openrestyrender "openflare/utils/render/openresty"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"openflare-agent/internal/nginx"
|
||||
@@ -24,6 +26,7 @@ const (
|
||||
type ConfigClient interface {
|
||||
GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error)
|
||||
ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPayload) error
|
||||
SyncWAFIPGroups(ctx context.Context, payload protocol.WAFIPGroupSyncRequest) (*protocol.WAFIPGroupSyncResponse, error)
|
||||
}
|
||||
|
||||
type NginxManager interface {
|
||||
@@ -31,6 +34,8 @@ type NginxManager interface {
|
||||
EnsureRuntime(ctx context.Context, recreate bool) error
|
||||
EnsureSafeFallbackRuntime(ctx context.Context, reason string) error
|
||||
CurrentChecksum() (string, error)
|
||||
WAFIPGroupChecksums() (map[string]string, error)
|
||||
SyncWAFIPGroups(groups []protocol.WAFIPGroup) error
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
@@ -162,6 +167,20 @@ func (s *Service) ForceSyncOnce(ctx context.Context, target *protocol.ActiveConf
|
||||
return s.applyIfNeeded(ctx, "force", true, snapshot, currentChecksum, target, config)
|
||||
}
|
||||
|
||||
func (s *Service) WAFIPGroupChecksums() (map[string]string, error) {
|
||||
if s.nginxManager == nil {
|
||||
return map[string]string{}, nil
|
||||
}
|
||||
return s.nginxManager.WAFIPGroupChecksums()
|
||||
}
|
||||
|
||||
func (s *Service) ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) error {
|
||||
if len(groups) == 0 || s.nginxManager == nil {
|
||||
return nil
|
||||
}
|
||||
return s.nginxManager.SyncWAFIPGroups(groups)
|
||||
}
|
||||
|
||||
func (s *Service) applyIfNeeded(ctx context.Context, mode string, startup bool, snapshot *state.Snapshot, currentChecksum string, target *protocol.ActiveConfigMeta, config *protocol.ActiveConfigResponse) error {
|
||||
if currentChecksum == config.Checksum && !startup {
|
||||
slog.Debug("local openresty config already up to date", "mode", mode, "version", config.Version)
|
||||
@@ -273,10 +292,36 @@ func (s *Service) applyIfNeeded(ctx context.Context, mode string, startup bool,
|
||||
slog.Warn("failed apply log reported", "version", config.Version)
|
||||
return outcomeError(config.Version, message)
|
||||
}
|
||||
if err := s.syncReferencedWAFIPGroups(ctx, rendered.supportFiles); err != nil {
|
||||
slog.Error("sync referenced waf ip groups failed", "version", config.Version, "error", err)
|
||||
return err
|
||||
}
|
||||
slog.Debug("apply log reported", "version", config.Version, "result", reportResult)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) syncReferencedWAFIPGroups(ctx context.Context, supportFiles []protocol.SupportFile) error {
|
||||
ids := referencedWAFIPGroupIDs(supportFiles)
|
||||
if len(ids) == 0 {
|
||||
return nil
|
||||
}
|
||||
checksums, err := s.WAFIPGroupChecksums()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
response, err := s.client.SyncWAFIPGroups(ctx, protocol.WAFIPGroupSyncRequest{
|
||||
IDs: ids,
|
||||
Checksums: checksums,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if response == nil || len(response.Groups) == 0 {
|
||||
return nil
|
||||
}
|
||||
return s.ApplyWAFIPGroups(ctx, response.Groups)
|
||||
}
|
||||
|
||||
type renderedActiveConfig struct {
|
||||
mainConfig string
|
||||
routeConfig string
|
||||
@@ -326,6 +371,48 @@ func fromOpenRestySupportFiles(files []openrestyrender.SupportFile) []protocol.S
|
||||
return result
|
||||
}
|
||||
|
||||
func referencedWAFIPGroupIDs(supportFiles []protocol.SupportFile) []uint {
|
||||
var content string
|
||||
for _, file := range supportFiles {
|
||||
if file.Path == "waf_config.json" {
|
||||
content = strings.TrimSpace(file.Content)
|
||||
break
|
||||
}
|
||||
}
|
||||
if content == "" {
|
||||
return []uint{}
|
||||
}
|
||||
var payload struct {
|
||||
RuleGroups []struct {
|
||||
IPWhitelistGroups []uint `json:"ip_whitelist_group_ids"`
|
||||
IPBlacklistGroups []uint `json:"ip_blacklist_group_ids"`
|
||||
} `json:"rule_groups"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(content), &payload); err != nil {
|
||||
slog.Debug("decode waf_config.json for ip group references failed", "error", err)
|
||||
return []uint{}
|
||||
}
|
||||
seen := make(map[uint]struct{})
|
||||
for _, group := range payload.RuleGroups {
|
||||
for _, id := range group.IPWhitelistGroups {
|
||||
if id > 0 {
|
||||
seen[id] = struct{}{}
|
||||
}
|
||||
}
|
||||
for _, id := range group.IPBlacklistGroups {
|
||||
if id > 0 {
|
||||
seen[id] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
ids := make([]uint, 0, len(seen))
|
||||
for id := range seen {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
|
||||
return ids
|
||||
}
|
||||
|
||||
func shouldReportNoopApply(snapshot *state.Snapshot, version string, checksum string) bool {
|
||||
if snapshot == nil {
|
||||
return false
|
||||
|
||||
@@ -72,6 +72,10 @@ func (f *fakeClient) ReportApplyLog(ctx context.Context, payload protocol.ApplyL
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeClient) SyncWAFIPGroups(ctx context.Context, payload protocol.WAFIPGroupSyncRequest) (*protocol.WAFIPGroupSyncResponse, error) {
|
||||
return &protocol.WAFIPGroupSyncResponse{}, nil
|
||||
}
|
||||
|
||||
func (m *fakeManager) Apply(ctx context.Context, mainConfig string, routeConfig string, supportFiles []protocol.SupportFile) nginx.ApplyOutcome {
|
||||
m.applyMainContents = append(m.applyMainContents, mainConfig)
|
||||
m.applyRouteContents = append(m.applyRouteContents, routeConfig)
|
||||
@@ -96,6 +100,14 @@ func (m *fakeManager) CurrentChecksum() (string, error) {
|
||||
return m.currentChecksum, m.currentChecksumErr
|
||||
}
|
||||
|
||||
func (m *fakeManager) WAFIPGroupChecksums() (map[string]string, error) {
|
||||
return map[string]string{}, nil
|
||||
}
|
||||
|
||||
func (m *fakeManager) SyncWAFIPGroups(groups []protocol.WAFIPGroup) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestSyncOnceSuccess(t *testing.T) {
|
||||
client := &fakeClient{
|
||||
config: protocol.ActiveConfigResponse{
|
||||
|
||||
@@ -78,9 +78,33 @@ func AgentHeartbeat(c *gin.Context) {
|
||||
respondSuccessWithExtras(c, node.Node, gin.H{
|
||||
"agent_settings": node.AgentSettings,
|
||||
"active_config": node.ActiveConfig,
|
||||
"waf_ip_groups": node.WAFIPGroups,
|
||||
})
|
||||
}
|
||||
|
||||
// AgentSyncWAFIPGroups godoc
|
||||
// @Summary Sync WAF IP groups for agent
|
||||
// @Tags Agent
|
||||
// @Accept json
|
||||
// @Produce json
|
||||
// @Security AgentTokenAuth
|
||||
// @Param payload body service.AgentWAFIPGroupSyncInput true "WAF IP group sync payload"
|
||||
// @Success 200 {object} map[string]interface{}
|
||||
// @Failure 400 {object} map[string]interface{}
|
||||
// @Router /api/agent/waf/ip-groups/sync [post]
|
||||
func AgentSyncWAFIPGroups(c *gin.Context) {
|
||||
var input service.AgentWAFIPGroupSyncInput
|
||||
if !bindJSON(c, &input) {
|
||||
return
|
||||
}
|
||||
result, err := service.SyncWAFIPGroupsForAgent(input)
|
||||
if err != nil {
|
||||
respondFailure(c, err.Error())
|
||||
return
|
||||
}
|
||||
respondSuccess(c, result)
|
||||
}
|
||||
|
||||
// AgentGetActiveConfig godoc
|
||||
// @Summary Get active config for agent
|
||||
// @Tags Agent
|
||||
@@ -254,12 +278,17 @@ func handleAgentWSStatus(c *gin.Context, node *model.Node, message service.Agent
|
||||
if response.ActiveConfig != nil {
|
||||
activeConfigSent = service.SendAgentWSActiveConfig(node.NodeID, response.ActiveConfig)
|
||||
}
|
||||
wafIPGroupsSent := false
|
||||
if len(response.WAFIPGroups) > 0 {
|
||||
wafIPGroupsSent = service.SendAgentWSWAFIPGroups(node.NodeID, response.WAFIPGroups)
|
||||
}
|
||||
slog.Debug("agent ws status processed",
|
||||
"node_id", node.NodeID,
|
||||
"current_version", payload.CurrentVersion,
|
||||
"openresty_status", payload.OpenrestyStatus,
|
||||
"settings_sent", settingsSent,
|
||||
"active_config_sent", activeConfigSent,
|
||||
"waf_ip_groups_sent", wafIPGroupsSent,
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -222,6 +222,7 @@ func SetApiRouter(router *gin.Engine) {
|
||||
authorizedRoute.GET("/ws", controller.AgentWebSocket)
|
||||
authorizedRoute.POST("/nodes/heartbeat", controller.AgentHeartbeat)
|
||||
authorizedRoute.GET("/config-versions/active", controller.AgentGetActiveConfig)
|
||||
authorizedRoute.POST("/waf/ip-groups/sync", controller.AgentSyncWAFIPGroups)
|
||||
authorizedRoute.POST("/apply-logs", controller.AgentReportApplyLog)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -41,6 +41,7 @@ type AgentNodePayload struct {
|
||||
AccessLogs []AgentNodeAccessLog `json:"access_logs,omitempty"`
|
||||
BufferedObservability []AgentBufferedObservabilityRecord `json:"buffered_observability,omitempty"`
|
||||
HealthEvents []AgentNodeHealthEvent `json:"health_events"`
|
||||
WAFIPGroupChecksums map[string]string `json:"waf_ip_group_checksums,omitempty"`
|
||||
}
|
||||
|
||||
type ApplyLogPayload struct {
|
||||
@@ -107,6 +108,25 @@ type HeartbeatResponse struct {
|
||||
Node *model.Node `json:"node"`
|
||||
AgentSettings *AgentSettings `json:"agent_settings"`
|
||||
ActiveConfig *ActiveConfigMeta `json:"active_config"`
|
||||
WAFIPGroups []AgentWAFIPGroup `json:"waf_ip_groups,omitempty"`
|
||||
}
|
||||
|
||||
type AgentWAFIPGroup struct {
|
||||
ID uint `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Type string `json:"type"`
|
||||
Enabled bool `json:"enabled"`
|
||||
IPList []string `json:"ip_list"`
|
||||
Checksum string `json:"checksum"`
|
||||
}
|
||||
|
||||
type AgentWAFIPGroupSyncInput struct {
|
||||
IDs []uint `json:"ids"`
|
||||
Checksums map[string]string `json:"checksums"`
|
||||
}
|
||||
|
||||
type AgentWAFIPGroupSyncResult struct {
|
||||
Groups []AgentWAFIPGroup `json:"groups"`
|
||||
}
|
||||
|
||||
type NodeView struct {
|
||||
@@ -183,10 +203,15 @@ func HeartbeatNode(node *model.Node, payload AgentNodePayload) (*HeartbeatRespon
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, err
|
||||
}
|
||||
wafIPGroups, err := ChangedWAFIPGroupsForAgent(nil, payload.WAFIPGroupChecksums)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &HeartbeatResponse{
|
||||
Node: node,
|
||||
AgentSettings: buildAgentSettings(node, updateNow, updateChannel.String(), updateTag, restartOpenrestyNow),
|
||||
ActiveConfig: activeConfig,
|
||||
WAFIPGroups: wafIPGroups,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ package service
|
||||
import (
|
||||
"errors"
|
||||
"openflare/model"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -74,6 +75,75 @@ func TestGetActiveConfigForAgentIncludesWAFConfig(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestChangedWAFIPGroupsForAgentReturnsChecksumDelta(t *testing.T) {
|
||||
setupServiceTestDB(t)
|
||||
|
||||
route, err := CreateProxyRoute(ProxyRouteInput{
|
||||
SiteName: "agent-waf-ip-group",
|
||||
Domains: []string{"agent-waf-ip-group.example.com"},
|
||||
OriginURL: "https://origin.internal",
|
||||
Enabled: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateProxyRoute failed: %v", err)
|
||||
}
|
||||
ipGroup, err := CreateWAFIPGroup(WAFIPGroupInput{
|
||||
Name: "agent runtime group",
|
||||
Type: WAFIPGroupTypeManual,
|
||||
Enabled: true,
|
||||
IPList: []string{"203.0.113.44"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateWAFIPGroup failed: %v", err)
|
||||
}
|
||||
ruleGroup, err := CreateWAFRuleGroup(WAFRuleGroupInput{
|
||||
Name: "agent refs",
|
||||
Enabled: true,
|
||||
IPBlacklistGroups: []uint{ipGroup.ID},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateWAFRuleGroup failed: %v", err)
|
||||
}
|
||||
if _, err = ReplaceWAFSiteRuleGroups(route.ID, []uint{ruleGroup.ID}); err != nil {
|
||||
t.Fatalf("ReplaceWAFSiteRuleGroups failed: %v", err)
|
||||
}
|
||||
if _, err = PublishConfigVersion("root", false); err != nil {
|
||||
t.Fatalf("PublishConfigVersion failed: %v", err)
|
||||
}
|
||||
|
||||
groups, err := ChangedWAFIPGroupsForAgent(nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("ChangedWAFIPGroupsForAgent failed: %v", err)
|
||||
}
|
||||
if len(groups) != 1 || groups[0].ID != ipGroup.ID || groups[0].IPList[0] != "203.0.113.44" || groups[0].Checksum == "" {
|
||||
t.Fatalf("unexpected changed groups: %#v", groups)
|
||||
}
|
||||
groupKey := strconv.FormatUint(uint64(ipGroup.ID), 10)
|
||||
same, err := ChangedWAFIPGroupsForAgent(nil, map[string]string{groupKey: groups[0].Checksum})
|
||||
if err != nil {
|
||||
t.Fatalf("ChangedWAFIPGroupsForAgent with checksum failed: %v", err)
|
||||
}
|
||||
if len(same) != 0 {
|
||||
t.Fatalf("expected no delta for matching checksum, got %#v", same)
|
||||
}
|
||||
updated, err := UpdateWAFIPGroup(ipGroup.ID, WAFIPGroupInput{
|
||||
Name: "agent runtime group",
|
||||
Type: WAFIPGroupTypeManual,
|
||||
Enabled: true,
|
||||
IPList: []string{"203.0.113.45"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("UpdateWAFIPGroup failed: %v", err)
|
||||
}
|
||||
delta, err := ChangedWAFIPGroupsForAgent(nil, map[string]string{groupKey: groups[0].Checksum})
|
||||
if err != nil {
|
||||
t.Fatalf("ChangedWAFIPGroupsForAgent after update failed: %v", err)
|
||||
}
|
||||
if len(delta) != 1 || delta[0].ID != updated.ID || delta[0].IPList[0] != "203.0.113.45" || delta[0].Checksum == groups[0].Checksum {
|
||||
t.Fatalf("expected updated group delta, got %#v", delta)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetActiveConfigForAgentUsesTenMinutePoWSessionDefault(t *testing.T) {
|
||||
setupServiceTestDB(t)
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ const (
|
||||
AgentWSMessageTypeSettings = "settings"
|
||||
AgentWSMessageTypeActiveConfig = "active_config"
|
||||
AgentWSMessageTypeForceSyncConfig = "force_sync_config"
|
||||
AgentWSMessageTypeWAFIPGroups = "waf_ip_groups"
|
||||
AgentWSMessageTypePing = "ping"
|
||||
AgentWSMessageTypePong = "pong"
|
||||
|
||||
@@ -77,6 +78,16 @@ func SendAgentWSForceSyncConfig(nodeID string, activeConfig *ActiveConfigMeta) b
|
||||
})
|
||||
}
|
||||
|
||||
func SendAgentWSWAFIPGroups(nodeID string, groups []AgentWAFIPGroup) bool {
|
||||
if len(groups) == 0 {
|
||||
return false
|
||||
}
|
||||
return DefaultAgentWSHub.SendMessage(nodeID, WSMessage{
|
||||
Type: AgentWSMessageTypeWAFIPGroups,
|
||||
Payload: groups,
|
||||
})
|
||||
}
|
||||
|
||||
func SendAgentWSPong(nodeID string) bool {
|
||||
return DefaultAgentWSHub.SendMessage(nodeID, WSMessage{
|
||||
Type: AgentWSMessageTypePong,
|
||||
@@ -111,3 +122,20 @@ func BroadcastAgentWSActiveConfig(activeConfig *ActiveConfigMeta) AgentWSBroadca
|
||||
)
|
||||
return result
|
||||
}
|
||||
|
||||
func BroadcastAgentWSWAFIPGroups(groups []AgentWAFIPGroup) WSBroadcastResult {
|
||||
if len(groups) == 0 {
|
||||
return WSBroadcastResult{}
|
||||
}
|
||||
result := DefaultAgentWSHub.Broadcast(WSMessage{
|
||||
Type: AgentWSMessageTypeWAFIPGroups,
|
||||
Payload: groups,
|
||||
})
|
||||
slog.Debug("agent ws broadcast waf ip groups",
|
||||
"group_count", len(groups),
|
||||
"client_count", result.ClientCount,
|
||||
"success_count", result.SuccessCount,
|
||||
"failed_nodes", result.FailedIDs,
|
||||
)
|
||||
return result
|
||||
}
|
||||
|
||||
@@ -652,16 +652,11 @@ func buildSnapshotWAFIPGroups(ruleGroups []snapshotWAFRuleGroup) ([]snapshotWAFI
|
||||
if group == nil {
|
||||
return nil, fmt.Errorf("IP 组 %d 不存在", id)
|
||||
}
|
||||
ips, err := decodeStringList(group.IPList)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("IP 组 %s 列表无效: %w", group.Name, err)
|
||||
}
|
||||
snapshots = append(snapshots, snapshotWAFIPGroup{
|
||||
ID: group.ID,
|
||||
Name: group.Name,
|
||||
Type: group.Type,
|
||||
Enabled: group.Enabled,
|
||||
IPList: ips,
|
||||
})
|
||||
}
|
||||
return snapshots, nil
|
||||
|
||||
@@ -2,10 +2,13 @@ package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
@@ -178,7 +181,11 @@ func CreateWAFIPGroup(input WAFIPGroupInput) (*WAFIPGroupView, error) {
|
||||
if err := group.Insert(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return GetWAFIPGroup(group.ID)
|
||||
view, err := GetWAFIPGroup(group.ID)
|
||||
if err == nil {
|
||||
broadcastWAFIPGroupToAgents(group.ID)
|
||||
}
|
||||
return view, err
|
||||
}
|
||||
|
||||
func UpdateWAFIPGroup(id uint, input WAFIPGroupInput) (*WAFIPGroupView, error) {
|
||||
@@ -193,7 +200,11 @@ func UpdateWAFIPGroup(id uint, input WAFIPGroupInput) (*WAFIPGroupView, error) {
|
||||
if err := group.Update(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return GetWAFIPGroup(group.ID)
|
||||
view, err := GetWAFIPGroup(group.ID)
|
||||
if err == nil {
|
||||
broadcastWAFIPGroupToAgents(group.ID)
|
||||
}
|
||||
return view, err
|
||||
}
|
||||
|
||||
func DeleteWAFIPGroup(id uint) error {
|
||||
@@ -364,6 +375,151 @@ func buildWAFIPGroupView(group *model.WAFIPGroup, referenceCount int) (WAFIPGrou
|
||||
return view, nil
|
||||
}
|
||||
|
||||
func ChangedWAFIPGroupsForAgent(ids []uint, checksums map[string]string) ([]AgentWAFIPGroup, error) {
|
||||
targetIDs := uniqueUintIDs(ids)
|
||||
if len(targetIDs) == 0 {
|
||||
activeIDs, err := activeConfigWAFIPGroupIDs()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
targetIDs = activeIDs
|
||||
}
|
||||
if len(targetIDs) == 0 {
|
||||
return []AgentWAFIPGroup{}, nil
|
||||
}
|
||||
groups, err := buildAgentWAFIPGroups(targetIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
changed := make([]AgentWAFIPGroup, 0, len(groups))
|
||||
for _, group := range groups {
|
||||
if strings.TrimSpace(checksums[fmt.Sprintf("%d", group.ID)]) == group.Checksum {
|
||||
continue
|
||||
}
|
||||
changed = append(changed, group)
|
||||
}
|
||||
return changed, nil
|
||||
}
|
||||
|
||||
func SyncWAFIPGroupsForAgent(input AgentWAFIPGroupSyncInput) (*AgentWAFIPGroupSyncResult, error) {
|
||||
groups, err := ChangedWAFIPGroupsForAgent(input.IDs, input.Checksums)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &AgentWAFIPGroupSyncResult{Groups: groups}, nil
|
||||
}
|
||||
|
||||
func buildAgentWAFIPGroups(ids []uint) ([]AgentWAFIPGroup, error) {
|
||||
ids = uniqueUintIDs(ids)
|
||||
if len(ids) == 0 {
|
||||
return []AgentWAFIPGroup{}, nil
|
||||
}
|
||||
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
|
||||
groups, err := model.ListWAFIPGroupsByIDs(ids)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
groupByID := make(map[uint]*model.WAFIPGroup, len(groups))
|
||||
for _, group := range groups {
|
||||
groupByID[group.ID] = group
|
||||
}
|
||||
result := make([]AgentWAFIPGroup, 0, len(ids))
|
||||
for _, id := range ids {
|
||||
group := groupByID[id]
|
||||
if group == nil {
|
||||
continue
|
||||
}
|
||||
agentGroup, err := buildAgentWAFIPGroup(group)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result = append(result, agentGroup)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func buildAgentWAFIPGroup(group *model.WAFIPGroup) (AgentWAFIPGroup, error) {
|
||||
if group == nil {
|
||||
return AgentWAFIPGroup{}, errors.New("IP 组不存在")
|
||||
}
|
||||
ips, err := decodeStringList(group.IPList)
|
||||
if err != nil {
|
||||
return AgentWAFIPGroup{}, err
|
||||
}
|
||||
if !group.Enabled {
|
||||
ips = []string{}
|
||||
}
|
||||
agentGroup := AgentWAFIPGroup{
|
||||
ID: group.ID,
|
||||
Name: group.Name,
|
||||
Type: group.Type,
|
||||
Enabled: group.Enabled,
|
||||
IPList: ips,
|
||||
}
|
||||
agentGroup.Checksum = checksumAgentWAFIPGroup(agentGroup)
|
||||
return agentGroup, nil
|
||||
}
|
||||
|
||||
func checksumAgentWAFIPGroup(group AgentWAFIPGroup) string {
|
||||
payload := struct {
|
||||
ID uint `json:"id"`
|
||||
Enabled bool `json:"enabled"`
|
||||
IPList []string `json:"ip_list"`
|
||||
}{
|
||||
ID: group.ID,
|
||||
Enabled: group.Enabled,
|
||||
IPList: append([]string{}, group.IPList...),
|
||||
}
|
||||
sort.Strings(payload.IPList)
|
||||
data, _ := json.Marshal(payload)
|
||||
sum := sha256.Sum256(data)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func activeConfigWAFIPGroupIDs() ([]uint, error) {
|
||||
version, err := model.GetActiveConfigVersion()
|
||||
if err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return []uint{}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
snapshot, err := parseSnapshotDocument(version.SnapshotJSON)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
idSet := make(map[uint]struct{})
|
||||
for _, group := range snapshot.WAF.RuleGroups {
|
||||
for _, id := range group.IPWhitelistGroups {
|
||||
if id > 0 {
|
||||
idSet[id] = struct{}{}
|
||||
}
|
||||
}
|
||||
for _, id := range group.IPBlacklistGroups {
|
||||
if id > 0 {
|
||||
idSet[id] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
ids := make([]uint, 0, len(idSet))
|
||||
for id := range idSet {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
func broadcastWAFIPGroupToAgents(id uint) {
|
||||
groups, err := buildAgentWAFIPGroups([]uint{id})
|
||||
if err != nil || len(groups) == 0 {
|
||||
if err != nil {
|
||||
slog.Debug("build waf ip group broadcast payload failed", "id", id, "error", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
BroadcastAgentWSWAFIPGroups(groups)
|
||||
}
|
||||
|
||||
func syncWAFIPGroup(group *model.WAFIPGroup, now time.Time) (*WAFIPGroupSyncResult, error) {
|
||||
if group == nil {
|
||||
return nil, errors.New("IP 组不存在")
|
||||
@@ -399,6 +555,7 @@ func syncWAFIPGroupSubscription(group *model.WAFIPGroup, now time.Time) (*WAFIPG
|
||||
if err := group.UpdateSyncResult(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
broadcastWAFIPGroupToAgents(group.ID)
|
||||
view, err := GetWAFIPGroup(group.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -481,6 +638,7 @@ func syncWAFIPGroupAutomatic(group *model.WAFIPGroup, now time.Time) (*WAFIPGrou
|
||||
if err := group.UpdateSyncResult(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
broadcastWAFIPGroupToAgents(group.ID)
|
||||
view, err := GetWAFIPGroup(group.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -305,7 +305,7 @@ func TestWAFIPGroupAutomaticRejectsInvalidExpr(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPublishConfigVersionExpandsWAFIPGroupReferences(t *testing.T) {
|
||||
func TestPublishConfigVersionKeepsWAFIPGroupReferences(t *testing.T) {
|
||||
setupServiceTestDB(t)
|
||||
|
||||
route, err := CreateProxyRoute(ProxyRouteInput{
|
||||
@@ -345,18 +345,26 @@ func TestPublishConfigVersionExpandsWAFIPGroupReferences(t *testing.T) {
|
||||
if !strings.Contains(result.Version.SnapshotJSON, `"ip_groups"`) {
|
||||
t.Fatal("expected snapshot to include waf ip groups")
|
||||
}
|
||||
if strings.Contains(result.Version.SnapshotJSON, "203.0.113.30") {
|
||||
t.Fatal("expected snapshot to avoid embedding waf ip group members")
|
||||
}
|
||||
var files []SupportFile
|
||||
if err = json.Unmarshal([]byte(result.Version.SupportFilesJSON), &files); err != nil {
|
||||
t.Fatalf("decode support files failed: %v", err)
|
||||
}
|
||||
found := false
|
||||
foundReference := false
|
||||
for _, file := range files {
|
||||
if file.Path == "waf_config.json" && strings.Contains(file.Content, "203.0.113.30") {
|
||||
found = true
|
||||
if file.Path == "waf_config.json" {
|
||||
if strings.Contains(file.Content, "203.0.113.30") {
|
||||
t.Fatalf("expected waf_config.json to avoid expanded IP group members, got %s", file.Content)
|
||||
}
|
||||
if strings.Contains(file.Content, `"ip_blacklist_group_ids":[`) {
|
||||
foundReference = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("expected expanded IP group in waf_config.json, got %#v", files)
|
||||
if !foundReference {
|
||||
t.Fatalf("expected IP group reference in waf_config.json, got %#v", files)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -186,6 +186,8 @@ func RenderWAFConfig(snapshot WAFDocument) (string, error) {
|
||||
BlockResponseBody string `json:"block_response_body"`
|
||||
IPWhitelist []string `json:"ip_whitelist"`
|
||||
IPBlacklist []string `json:"ip_blacklist"`
|
||||
IPWhitelistGroups []uint `json:"ip_whitelist_group_ids,omitempty"`
|
||||
IPBlacklistGroups []uint `json:"ip_blacklist_group_ids,omitempty"`
|
||||
CountryWhitelist []string `json:"country_whitelist"`
|
||||
CountryBlacklist []string `json:"country_blacklist"`
|
||||
RegionWhitelist []string `json:"region_whitelist"`
|
||||
@@ -199,10 +201,6 @@ func RenderWAFConfig(snapshot WAFDocument) (string, error) {
|
||||
groups := make([]wafRuntimeRuleGroup, 0, len(snapshot.RuleGroups))
|
||||
globalGroupIDs := make([]uint, 0)
|
||||
enabledGroupIDs := make(map[uint]struct{}, len(snapshot.RuleGroups))
|
||||
ipGroupsByID := make(map[uint]WAFIPGroup, len(snapshot.IPGroups))
|
||||
for _, group := range snapshot.IPGroups {
|
||||
ipGroupsByID[group.ID] = group
|
||||
}
|
||||
for _, group := range snapshot.RuleGroups {
|
||||
if !group.Enabled {
|
||||
continue
|
||||
@@ -221,8 +219,10 @@ func RenderWAFConfig(snapshot WAFDocument) (string, error) {
|
||||
IsGlobal: group.IsGlobal,
|
||||
BlockStatusCode: statusCode,
|
||||
BlockResponseBody: group.BlockResponseBody,
|
||||
IPWhitelist: expandWAFIPGroups(group.IPWhitelist, group.IPWhitelistGroups, ipGroupsByID),
|
||||
IPBlacklist: expandWAFIPGroups(group.IPBlacklist, group.IPBlacklistGroups, ipGroupsByID),
|
||||
IPWhitelist: sortedUniqueStrings(group.IPWhitelist),
|
||||
IPBlacklist: sortedUniqueStrings(group.IPBlacklist),
|
||||
IPWhitelistGroups: sortedUniqueUintIDs(group.IPWhitelistGroups),
|
||||
IPBlacklistGroups: sortedUniqueUintIDs(group.IPBlacklistGroups),
|
||||
CountryWhitelist: group.CountryWhitelist,
|
||||
CountryBlacklist: group.CountryBlacklist,
|
||||
RegionWhitelist: group.RegionWhitelist,
|
||||
@@ -250,20 +250,19 @@ func RenderWAFConfig(snapshot WAFDocument) (string, error) {
|
||||
return string(data), err
|
||||
}
|
||||
|
||||
func expandWAFIPGroups(direct []string, groupIDs []uint, ipGroupsByID map[uint]WAFIPGroup) []string {
|
||||
items := append([]string{}, direct...)
|
||||
for _, id := range groupIDs {
|
||||
group, ok := ipGroupsByID[id]
|
||||
if !ok || !group.Enabled {
|
||||
continue
|
||||
}
|
||||
items = append(items, group.IPList...)
|
||||
}
|
||||
func sortedUniqueStrings(values []string) []string {
|
||||
items := append([]string{}, values...)
|
||||
items = uniqueStrings(items)
|
||||
sort.Strings(items)
|
||||
return items
|
||||
}
|
||||
|
||||
func sortedUniqueUintIDs(values []uint) []uint {
|
||||
items := uniqueUintIDs(values)
|
||||
sort.Slice(items, func(i, j int) bool { return items[i] < items[j] })
|
||||
return items
|
||||
}
|
||||
|
||||
func ChecksumBundle(mainConfig string, routeConfig string, supportFiles []SupportFile) string {
|
||||
var builder strings.Builder
|
||||
builder.WriteString(mainConfig)
|
||||
|
||||
@@ -188,7 +188,7 @@ export function RuleEntryModal({
|
||||
选择 IP 组
|
||||
</h3>
|
||||
<p className="mt-1 text-xs leading-5 text-[var(--foreground-secondary)]">
|
||||
被引用的 IP 组会在发布配置时展开到 WAF 运行时名单。
|
||||
发布版本只保存引用 ID,IP 组成员由 Agent 按 checksum 差异同步。
|
||||
</p>
|
||||
</div>
|
||||
<span className="rounded-full border border-[var(--border-default)] px-2.5 py-1 text-xs font-medium text-[var(--foreground-secondary)]">
|
||||
|
||||
@@ -253,8 +253,8 @@ describe('WAF IP groups', () => {
|
||||
await userEvent.click(screen.getByRole('button', { name: /测试规则/ }));
|
||||
|
||||
expect(await screen.findByText('命中 2 个 IP。')).toBeInTheDocument();
|
||||
expect(screen.getByText('203.0.113.10')).toBeInTheDocument();
|
||||
expect(screen.getByText('203.0.113.11')).toBeInTheDocument();
|
||||
expect(screen.getByText(/203\.0\.113\.10/)).toBeInTheDocument();
|
||||
expect(screen.getByText(/203\.0\.113\.11/)).toBeInTheDocument();
|
||||
expect(testMock).toHaveBeenCalledWith({
|
||||
auto_config: expect.objectContaining({
|
||||
lookback_minutes: 60,
|
||||
|
||||
Reference in New Issue
Block a user