From 54d7dfb7c99d408f368026f9b5d0027bc519fc78 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 17:24:25 +0800 Subject: [PATCH 01/11] docs: design per-IP rule limits --- .../2026-04-27-per-ip-rule-limits-design.md | 321 ++++++++++++++++++ 1 file changed, 321 insertions(+) create mode 100644 docs/superpowers/specs/2026-04-27-per-ip-rule-limits-design.md diff --git a/docs/superpowers/specs/2026-04-27-per-ip-rule-limits-design.md b/docs/superpowers/specs/2026-04-27-per-ip-rule-limits-design.md new file mode 100644 index 0000000..4a65085 --- /dev/null +++ b/docs/superpowers/specs/2026-04-27-per-ip-rule-limits-design.md @@ -0,0 +1,321 @@ +# 规则每 IP 连接数与限速设计 + +**日期**: 2026-04-27 +**状态**: 待审核 +**作者**: AI Assistant + +## 概述 + +在转发规则的高级设置中新增两类每客户端 IP 限制:每 IP 最大连接数、每 IP 带宽限速。保留现有总量限制语义不变,新增字段只在用户显式配置时生效。 + +实现优先复用 GOST 已有能力:`climiters` 的 `$$ N` 表示每个客户端 IP 独立最大连接数;`limiters` 支持 IP/CIDR 级带宽桶,可用 `0.0.0.0/0` 和 `::/0` 实现默认覆盖所有 IPv4/IPv6 客户端的每 IP 带宽限速。 + +## 背景 + +当前 FLVX 已经支持规则级最大连接数和规则级限速,但这两个限制都是规则总量: + +- `maxConn` 下发为 GOST `climiters` 的 `$ N`,限制整条规则的总并发连接数。 +- `speedId` 下发为 GOST `limiters` 的 `$ in out`,限制整条规则的总带宽。 + +用户需要的是按客户端 IP 隔离的限制,例如每个 IP 最多 5 个连接、每个 IP 最多 10 Mbps,而不是所有客户端共享同一个总量。 + +## GOST 能力确认 + +### 连接数限制 + +`go-gost/x/limiter/conn/conn.go` 已内置以下语义: + +| Key | 含义 | +|-----|------| +| `$` | 全局连接数限制,所有客户端共享一个 limiter | +| `$$` | 每个客户端 IP 独立连接数限制,每个 IP 创建自己的 limiter | +| `IP` / `CIDR` | 指定 IP 或 CIDR 的连接数限制 | + +因此每 IP 连接数无需新增 agent 限制器,只需后端下发 `$$ N`。 + +### 带宽限制 + +`go-gost/x/limiter/traffic/traffic.go` 已内置以下语义: + +| Key | 含义 | +|-----|------| +| `$` | 服务级总带宽限制 | +| `$$` | 连接级带宽限制 | +| `IP` / `CIDR` | 客户端 IP 或 CIDR 级带宽限制 | + +CIDR 级限制使用 generator,为命中的客户端 IP 创建独立 limiter。使用 `0.0.0.0/0` 和 `::/0` 可以覆盖所有 IPv4/IPv6 客户端,实现每 IP 带宽限速。 + +### 现有缺口 + +TCP listener 已在 Accept 后用客户端地址包装连接级 traffic limiter,路径可用于每 IP 带宽。UDP listener 当前只在 PacketConn 上应用服务级 limiter,没有在 `Accept()` 后按客户端 UDP pseudo-connection 包装 limiter,也没有挂接 connection limiter。因此要让 UDP 与 TCP 语义一致,需要补齐 UDP listener 的 per-client wrapper。 + +## 目标 + +1. 保留现有 `maxConn` 和 `speedId` 的总量语义。 +2. 在规则上新增每 IP 最大连接数。 +3. 在规则上新增每 IP 带宽限速。 +4. 同一规则允许同时配置总量限制和每 IP 限制。 +5. 普通用户不能设置或修改限速规则字段,保持现有权限模型。 +6. TCP 和 UDP 入口都尽量遵循相同限制语义。 + +## 非目标 + +1. 不新增按用户组、节点组、国家地区、ASN 的限制。 +2. 不新增请求频率限制;本次“每个 IP 限速”指带宽限速,不是新建连接频率。 +3. 不改变已有 speed limit 规则表的单位和含义。 +4. 不把用户级默认最大连接数改成每 IP 语义;用户级 `maxConn` 继续作为默认总连接数。 + +## 数据模型 + +在 `forward` 表新增两个字段: + +| 字段 | 类型 | 默认 | 说明 | +|------|------|------|------| +| `ip_max_conn` | int | `0` | 每 IP 最大连接数,`0` 表示不启用 | +| `ip_speed_id` | nullable int64 | `NULL` | 每 IP 带宽限速规则 ID,`NULL` 表示不启用 | + +Go 模型新增: + +```go +IPMaxConn int `gorm:"column:ip_max_conn;not null;default:0"` +IPSpeedID sql.NullInt64 `gorm:"column:ip_speed_id"` +``` + +字段会通过现有 auto-migrate 机制创建,保持 SQLite/PostgreSQL 兼容,不使用 SQLite 不兼容的 GORM tags。 + +## API 行为 + +### 创建规则 + +`/forward/create` 新增入参: + +```json +{ + "ipMaxConn": 5, + "ipSpeedId": 123 +} +``` + +规则: + +- `ipMaxConn` 缺省或小于等于 `0` 时按 `0` 存储,不启用每 IP 连接数限制。 +- `ipSpeedId` 缺省或不存在时存为 `NULL`,不启用每 IP 带宽限速。 +- `ipSpeedId` 指向不存在的限速规则时按 `NULL` 处理,沿用现有 `speedId` 的容错策略。 +- 普通用户提交非空 `ipSpeedId` 时返回错误,保持与 `speedId` 一致的权限边界。 + +### 更新规则 + +`/forward/update` 新增入参: + +```json +{ + "ipMaxConn": 5, + "ipSpeedId": 123 +} +``` + +规则: + +- 未提交 `ipMaxConn` 时保留原值;提交空值或 `0` 时清除每 IP 连接数限制。 +- 未提交 `ipSpeedId` 时保留原值;提交 `null` 时清除每 IP 带宽限速。 +- 普通用户不能把 `ipSpeedId` 改成不同的非空值。 +- 更新后重新同步运行时服务和 limiter。 + +### 列表返回 + +`/forward/list` 返回项新增: + +```json +{ + "ipMaxConn": 5, + "ipSpeedId": 123, + "ipSpeedLimitName": "每IP 10Mbps" +} +``` + +`ipSpeedLimitName` 可选,但建议返回,便于前端显示缺失或已删除的限速规则。 + +## 后端运行时同步 + +### 连接数限制器 + +将现有连接限制器构建从单一总量扩展为组合规则。 + +当前行为: + +```json +{ + "name": "rule_conn_limit_42", + "limits": ["$ 100"] +} +``` + +新增行为: + +```json +{ + "name": "rule_conn_limit_42", + "limits": ["$ 100", "$$ 5"] +} +``` + +规则: + +- `maxConn > 0` 时追加 `$ maxConn`。 +- `ipMaxConn > 0` 时追加 `$$ ipMaxConn`。 +- 如果规则未配置 `maxConn` 且用户有 `MaxConn > 0`,继续继承用户级总连接数,追加 `$ user.MaxConn`。 +- 如果两者都没有,则不下发 `climiter`,服务不引用 `climiter`。 +- limiter 名称继续优先使用 `rule_conn_limit_`;只有用户级默认总连接数且规则没有任何连接限制时可继续使用 `user_conn_limit_`,避免不必要的 per-rule limiter。 + +### 带宽限制器 + +将现有规则限速从单一 `speedId` 扩展为组合 limiter。 + +当前行为: + +```json +{ + "name": "123", + "limits": ["$ 1.3MB 1.3MB"] +} +``` + +新增每 IP 行为: + +```json +{ + "name": "rule_traffic_limit_42", + "limits": [ + "$ 1.3MB 1.3MB", + "0.0.0.0/0 1.3MB 1.3MB", + "::/0 1.3MB 1.3MB" + ] +} +``` + +规则: + +- 只有总量 `speedId` 时,保持现有名称和下发路径,服务继续引用 `speedId` 字符串。 +- 只有每 IP `ipSpeedId` 时,创建 `rule_traffic_limit_`,只包含 IPv4/IPv6 CIDR 行。 +- 总量和每 IP 同时存在时,创建 `rule_traffic_limit_`,同时包含 `$` 和 CIDR 行。 +- 如果规则没有 `speedId`,则总量仍可继承 user tunnel 的 `speedId`,保持现有 fallback 语义;当继承的总量限速与 `ipSpeedId` 同时存在时,也使用 `rule_traffic_limit_` 组合 limiter。 +- 每 IP 限速不从 user tunnel 继承,只由规则字段控制。 +- `AddLimiters` 失败且提示已存在时,使用 `UpdateLimiters` 更新。 + +### 服务配置 + +`buildForwardServiceConfigs` 需要从当前 `limiterID *int64` / `cLimiterName string` 扩展为更明确的运行时限制描述,例如: + +```go +type forwardRuntimeLimiters struct { + TrafficLimiter string + ConnLimiter string +} +``` + +服务配置只关心最终引用的 limiter 名称: + +- `service["limiter"] = runtimeLimiters.TrafficLimiter` +- `service["climiter"] = runtimeLimiters.ConnLimiter` + +这样可以把“如何构建 limiter payload”的逻辑和“如何构建 service JSON”的逻辑分开。 + +## Agent/GOST 调整 + +### WebSocket 命令 + +当前 agent WebSocket 已支持: + +- `AddLimiters` / `UpdateLimiters` / `DeleteLimiters` +- `AddCLimiters` / `UpdateCLimiters` / `DeleteCLimiters` + +本设计无需新增命令类型。 + +### UDP listener + +补齐 `go-gost/x/listener/udp/listener.go` 的 `Accept()` 包装逻辑,使 UDP pseudo-connection 与 TCP listener 一致: + +- 对 `l.options.ConnLimiter` 按客户端地址应用连接数限制。 +- 对 `l.options.TrafficLimiter` 按 `conn.RemoteAddr().String()` 应用连接级 traffic wrapper。 + +需要注意 UDP pseudo-connection 的生命周期由内部 UDP listener 的 TTL/keepalive 控制;connection limiter 必须在 pseudo-connection 关闭时释放计数。 + +## 前端设计 + +在 `vite-frontend/src/pages/forward.tsx` 的规则高级设置中新增两个控件: + +1. `每 IP 最大连接数` +- 类型:number input。 +- 文案:`每个客户端 IP 可同时建立的最大连接数;0 或空表示不限制。` +- 字段:`ipMaxConn`。 + +2. `每 IP 限速` +- 类型:Select,复用现有限速规则列表。 +- 文案:`每个客户端 IP 独享该带宽限制;不选择表示不限制。` +- 字段:`ipSpeedId`。 +- 只对管理员显示,保持与 `规则限速` 一致。 + +前端类型需要同步更新: + +- `ForwardApiItem` +- `ForwardMutationPayload` +- `ForwardForm` 或页面内等价类型 + +## 错误处理与兼容性 + +1. 旧数据默认 `ip_max_conn=0`、`ip_speed_id=NULL`,行为与当前版本一致。 +2. 现有 agent 已支持 limiter 命令和 GOST limiter 语法;发布时需要包含 UDP 修复,才能让 TCP/UDP 都获得完整语义。 +3. 节点离线时沿用现有 warning 行为,规则仍可保存,在线节点跳过下发。 +4. 如果每 IP speed limit ID 被删除,更新时按 `NULL` 处理,列表页可提示或自动清除,和现有 `speedId` 行为一致。 +5. 如果 IPv6 CIDR 在某些监听路径未命中,IPv4 行仍正常生效;测试应覆盖 IPv4,IPv6 通过 payload 合同保证下发。 + +## 测试计划 + +### 后端 contract 测试 + +新增或扩展 `go-backend/tests/contract/max_conn_limit_contract_test.go`: + +1. 创建规则时设置 `ipMaxConn=5`,断言 `AddCLimiters` payload 包含 `$$ 5`。 +2. 同时设置 `maxConn=100` 和 `ipMaxConn=5`,断言 payload 包含 `$ 100` 和 `$$ 5`。 +3. 用户级 `MaxConn` 存在且规则 `ipMaxConn=5` 时,断言 payload 包含 `$ userMaxConn` 和 `$$ 5`。 + +新增每 IP 限速 contract 测试: + +1. 创建规则时设置 `ipSpeedId`,断言 `AddLimiters` payload 包含 `0.0.0.0/0 ...` 和 `::/0 ...`。 +2. 同时设置 `speedId` 和 `ipSpeedId`,断言组合 limiter 包含 `$ ...` 与两个 CIDR 行,服务引用 `rule_traffic_limit_`。 +3. 普通用户提交 `ipSpeedId` 返回错误。 + +### Repository/API 测试 + +1. `CreateForwardTx`、`UpdateForward`、列表查询读写 `ip_max_conn` 和 `ip_speed_id`。 +2. `/forward/list` 返回 `ipMaxConn`、`ipSpeedId`。 + +### GOST/x 测试 + +1. `go-gost/x/limiter/conn`:验证 `$$ N` 为不同 IP 创建独立 limiter。 +2. `go-gost/x/limiter/traffic`:验证 `0.0.0.0/0` 为不同 IPv4 创建独立 limiter。 +3. UDP listener:验证 Accept 返回的 UDP pseudo-connection 关闭后释放 connection limiter。 + +### 验证命令 + +```bash +(cd go-backend && go test ./...) +(cd go-gost/x && go test ./limiter/... ./listener/udp/...) +(cd vite-frontend && pnpm run build) +``` + +## 推荐实施顺序 + +1. 后端模型、repo DTO、API 字段读写。 +2. 后端 limiter payload 构建与服务引用重构。 +3. Contract 测试覆盖连接数和带宽 payload。 +4. GOST UDP listener per-client wrapper 与相关测试。 +5. 前端高级设置表单和类型更新。 +6. 运行后端测试、GOST/x 相关测试、前端构建。 + +## 风险 + +1. UDP pseudo-connection 生命周期和 TCP 连接不同,连接数释放必须依赖 Close 包装正确执行。 +2. 总带宽和每 IP 带宽组合时 limiter 名称从纯 speed ID 变为 rule-level 名称,需要确保更新已有规则时不会留下错误引用。 +3. 旧节点如果没有 UDP wrapper 修复,TCP 生效但 UDP 每 IP 语义可能不完整;发布时应要求 agent 同步升级。 +4. 每 IP 带宽是每个入口节点本地独立限制,不是跨节点全局聚合限制。 From 2b76a9f0be5e920045b12a433425740fee0d6aff Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 22:07:04 +0800 Subject: [PATCH 02/11] feat: persist per-IP forward limits --- go-backend/internal/http/handler/mutations.go | 4 +- go-backend/internal/store/model/model.go | 44 ++++++------ go-backend/internal/store/repo/repository.go | 68 +++++++++++++------ .../internal/store/repo/repository_control.go | 2 + .../internal/store/repo/repository_flow.go | 8 +++ .../repository_forward_proxy_protocol_test.go | 65 ++++++++++++++++++ .../store/repo/repository_mutations.go | 28 ++++---- 7 files changed, 167 insertions(+), 52 deletions(-) diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 5541c4b..2189ce2 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -1804,7 +1804,7 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) { maxConn := asInt(req["maxConn"], 0) proxyProtocol := asInt(req["proxyProtocol"], 0) - forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, proxyProtocol) + forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, 0, nil, proxyProtocol) if err != nil { response.WriteJSON(w, response.Err(-2, err.Error())) return @@ -1958,7 +1958,7 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) { maxConn := asInt(req["maxConn"], forward.MaxConn) proxyProtocol := asInt(req["proxyProtocol"], forward.ProxyProtocol) - if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, proxyProtocol); err != nil { + if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, 0, nil, proxyProtocol); err != nil { response.WriteJSON(w, response.Err(-2, err.Error())) return } diff --git a/go-backend/internal/store/model/model.go b/go-backend/internal/store/model/model.go index 356e37a..2bf5d41 100644 --- a/go-backend/internal/store/model/model.go +++ b/go-backend/internal/store/model/model.go @@ -45,6 +45,8 @@ type Forward struct { Inx int `gorm:"not null;default:0"` SpeedID sql.NullInt64 `gorm:"column:speed_id"` MaxConn int `gorm:"column:max_conn;not null;default:0"` + IPMaxConn int `gorm:"column:ip_max_conn;not null;default:0"` + IPSpeedID sql.NullInt64 `gorm:"column:ip_speed_id"` ProxyProtocol int `gorm:"column:proxy_protocol;not null;default:0"` } @@ -428,20 +430,22 @@ type ChainTunnelBackup struct { } type ForwardBackup struct { - ID int64 `json:"id"` - UserID int64 `json:"userId"` - UserName string `json:"userName"` - Name string `json:"name"` - TunnelID int64 `json:"tunnelId"` - RemoteAddr string `json:"remoteAddr"` - Strategy string `json:"strategy"` - InFlow int64 `json:"inFlow"` - OutFlow int64 `json:"outFlow"` - CreatedTime int64 `json:"createdTime"` - UpdatedTime int64 `json:"updatedTime"` - Status int `json:"status"` + ID int64 `json:"id"` + UserID int64 `json:"userId"` + UserName string `json:"userName"` + Name string `json:"name"` + TunnelID int64 `json:"tunnelId"` + RemoteAddr string `json:"remoteAddr"` + Strategy string `json:"strategy"` + InFlow int64 `json:"inFlow"` + OutFlow int64 `json:"outFlow"` + CreatedTime int64 `json:"createdTime"` + UpdatedTime int64 `json:"updatedTime"` + Status int `json:"status"` Inx int `json:"inx"` SpeedID *int64 `json:"speedId,omitempty"` + IPMaxConn int `json:"ipMaxConn,omitempty"` + IPSpeedID *int64 `json:"ipSpeedId,omitempty"` ForwardPorts *[]ForwardPortBackup `json:"forwardPorts,omitempty"` ProxyProtocol int `json:"proxyProtocol"` } @@ -532,16 +536,18 @@ type ImportResult struct { // ForwardRecord is a minimal forward view used by control plane and flow policy. type ForwardRecord struct { - ID int64 - UserID int64 - UserName string - Name string - TunnelID int64 - RemoteAddr string - Strategy string + ID int64 + UserID int64 + UserName string + Name string + TunnelID int64 + RemoteAddr string + Strategy string Status int SpeedID sql.NullInt64 MaxConn int + IPMaxConn int + IPSpeedID sql.NullInt64 ProxyProtocol int } diff --git a/go-backend/internal/store/repo/repository.go b/go-backend/internal/store/repo/repository.go index 92cafc8..5ccf55d 100644 --- a/go-backend/internal/store/repo/repository.go +++ b/go-backend/internal/store/repo/repository.go @@ -865,29 +865,33 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) { } type fwdRow struct { - ID int64 - UserID int64 - UserName string - Name string - TunnelID int64 - TunnelName string - TrafficRatio float64 - RemoteAddr string - Strategy string - InFlow int64 - OutFlow int64 - CreatedTime int64 - Status int - Inx int - SpeedID sql.NullInt64 - MaxConn int - ProxyProtocol int + ID int64 + UserID int64 + UserName string + Name string + TunnelID int64 + TunnelName string + TrafficRatio float64 + RemoteAddr string + Strategy string + InFlow int64 + OutFlow int64 + CreatedTime int64 + Status int + Inx int + SpeedID sql.NullInt64 + MaxConn int + IPMaxConn int + IPSpeedID sql.NullInt64 + IPSpeedLimitName string + ProxyProtocol int } var rows []fwdRow err := r.db.Model(&model.Forward{}). - Select("forward.id, forward.user_id, forward.user_name, forward.name, forward.tunnel_id, COALESCE(tunnel.name, '') AS tunnel_name, COALESCE(tunnel.traffic_ratio, 1.0) AS traffic_ratio, forward.remote_addr, COALESCE(forward.strategy, 'fifo') AS strategy, forward.in_flow, forward.out_flow, forward.created_time, forward.status, forward.inx, forward.speed_id, forward.max_conn, forward.proxy_protocol"). + Select("forward.id, forward.user_id, forward.user_name, forward.name, forward.tunnel_id, COALESCE(tunnel.name, '') AS tunnel_name, COALESCE(tunnel.traffic_ratio, 1.0) AS traffic_ratio, forward.remote_addr, COALESCE(forward.strategy, 'fifo') AS strategy, forward.in_flow, forward.out_flow, forward.created_time, forward.status, forward.inx, forward.speed_id, forward.max_conn, forward.ip_max_conn, forward.ip_speed_id, COALESCE(ip_speed_limit.name, '') AS ip_speed_limit_name, forward.proxy_protocol"). Joins("LEFT JOIN tunnel ON tunnel.id = forward.tunnel_id"). + Joins("LEFT JOIN speed_limit AS ip_speed_limit ON ip_speed_limit.id = forward.ip_speed_id"). Order("forward.inx ASC, forward.id ASC"). Find(&rows).Error if err != nil { @@ -909,11 +913,18 @@ func (r *Repository) ListForwards() ([]map[string]interface{}, error) { "inFlow": row.InFlow, "outFlow": row.OutFlow, "createdTime": row.CreatedTime, "status": row.Status, "inx": int64(row.Inx), "maxConn": row.MaxConn, + "ipMaxConn": row.IPMaxConn, "proxyProtocol": row.ProxyProtocol, } if row.SpeedID.Valid { item["speedId"] = row.SpeedID.Int64 } + if row.IPSpeedID.Valid { + item["ipSpeedId"] = row.IPSpeedID.Int64 + } + if strings.TrimSpace(row.IPSpeedLimitName) != "" { + item["ipSpeedLimitName"] = row.IPSpeedLimitName + } items = append(items, item) } return items, nil @@ -2102,8 +2113,17 @@ func (r *Repository) exportForwards() ([]model.ForwardBackup, error) { TunnelID: f.TunnelID, RemoteAddr: f.RemoteAddr, Strategy: f.Strategy, InFlow: f.InFlow, OutFlow: f.OutFlow, CreatedTime: f.CreatedTime, UpdatedTime: f.UpdatedTime, Status: f.Status, Inx: f.Inx, + IPMaxConn: f.IPMaxConn, ProxyProtocol: f.ProxyProtocol, } + if f.SpeedID.Valid { + v := f.SpeedID.Int64 + b.SpeedID = &v + } + if f.IPSpeedID.Valid { + v := f.IPSpeedID.Int64 + b.IPSpeedID = &v + } ports, err := r.exportForwardPorts(f.ID) if err != nil { return nil, err @@ -2483,6 +2503,13 @@ func importTunnels(tx *gorm.DB, tunnels []model.TunnelBackup, now int64) (int, e return count, nil } +func nullableBackupInt64(v *int64) int64 { + if v == nil { + return 0 + } + return *v +} + func importForwards(tx *gorm.DB, forwards []model.ForwardBackup, now int64) (int, error) { count := 0 for _, f := range forwards { @@ -2500,13 +2527,16 @@ func importForwards(tx *gorm.DB, forwards []model.ForwardBackup, now int64) (int UpdatedTime: now, Status: f.Status, Inx: f.Inx, + SpeedID: sql.NullInt64{Int64: nullableBackupInt64(f.SpeedID), Valid: f.SpeedID != nil && *f.SpeedID > 0}, + IPMaxConn: f.IPMaxConn, + IPSpeedID: sql.NullInt64{Int64: nullableBackupInt64(f.IPSpeedID), Valid: f.IPSpeedID != nil && *f.IPSpeedID > 0}, ProxyProtocol: f.ProxyProtocol, } err := tx.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "id"}}, DoUpdates: clause.AssignmentColumns([]string{ "user_id", "user_name", "name", "tunnel_id", "remote_addr", "strategy", - "in_flow", "out_flow", "updated_time", "status", "inx", "proxy_protocol", + "in_flow", "out_flow", "updated_time", "status", "inx", "speed_id", "ip_max_conn", "ip_speed_id", "proxy_protocol", }), }).Create(&item).Error if err != nil { diff --git a/go-backend/internal/store/repo/repository_control.go b/go-backend/internal/store/repo/repository_control.go index 038bf9d..351f48b 100644 --- a/go-backend/internal/store/repo/repository_control.go +++ b/go-backend/internal/store/repo/repository_control.go @@ -55,6 +55,8 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode Status: f.Status, SpeedID: f.SpeedID, MaxConn: f.MaxConn, + IPMaxConn: f.IPMaxConn, + IPSpeedID: f.IPSpeedID, ProxyProtocol: f.ProxyProtocol, }) } diff --git a/go-backend/internal/store/repo/repository_flow.go b/go-backend/internal/store/repo/repository_flow.go index 4396d14..e67ca28 100644 --- a/go-backend/internal/store/repo/repository_flow.go +++ b/go-backend/internal/store/repo/repository_flow.go @@ -124,6 +124,8 @@ func (r *Repository) ListActiveForwardsByUser(userID int64) ([]model.ForwardReco Status: f.Status, SpeedID: f.SpeedID, MaxConn: f.MaxConn, + IPMaxConn: f.IPMaxConn, + IPSpeedID: f.IPSpeedID, ProxyProtocol: f.ProxyProtocol, }) } @@ -157,6 +159,8 @@ func (r *Repository) ListActiveForwardsByUserTunnel(userID, tunnelID int64) ([]m Status: f.Status, SpeedID: f.SpeedID, MaxConn: f.MaxConn, + IPMaxConn: f.IPMaxConn, + IPSpeedID: f.IPSpeedID, ProxyProtocol: f.ProxyProtocol, }) } @@ -190,6 +194,8 @@ func (r *Repository) ListForwardsByUserAndTunnel(userID, tunnelID int64) ([]mode Status: f.Status, SpeedID: f.SpeedID, MaxConn: f.MaxConn, + IPMaxConn: f.IPMaxConn, + IPSpeedID: f.IPSpeedID, ProxyProtocol: f.ProxyProtocol, }) } @@ -224,6 +230,8 @@ func (r *Repository) GetForwardRecord(forwardID int64) (*model.ForwardRecord, er Status: f.Status, SpeedID: f.SpeedID, MaxConn: f.MaxConn, + IPMaxConn: f.IPMaxConn, + IPSpeedID: f.IPSpeedID, ProxyProtocol: f.ProxyProtocol, } if strings.TrimSpace(fr.Strategy) == "" { diff --git a/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go b/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go index 2918d4f..c11a095 100644 --- a/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go +++ b/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go @@ -1,6 +1,7 @@ package repo import ( + "database/sql" "testing" "time" @@ -151,6 +152,70 @@ func TestListActiveForwardsByUserTunnelIncludesMaxConn(t *testing.T) { } } +func TestForwardRepositoryPersistsPerIPLimits(t *testing.T) { + r, err := Open(":memory:") + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + forwardID, err := r.CreateForwardTx(1, "admin", "per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, []int64{3}, 24000, "", nil, 0, 5, int64(21), 0) + if err != nil { + t.Fatalf("CreateForwardTx: %v", err) + } + record, err := r.GetForwardRecord(forwardID) + if err != nil { + t.Fatalf("GetForwardRecord after create: %v", err) + } + if record.IPMaxConn != 5 { + t.Fatalf("expected created ipMaxConn 5, got %d", record.IPMaxConn) + } + if !record.IPSpeedID.Valid || record.IPSpeedID.Int64 != 21 { + t.Fatalf("expected created ipSpeedId 21, got %+v", record.IPSpeedID) + } + + if err := r.UpdateForward(forwardID, "per-ip-forward", 2, "2.2.2.2:443", "fifo", now+1, nil, 0, 9, int64(22), 0); err != nil { + t.Fatalf("UpdateForward: %v", err) + } + record, err = r.GetForwardRecord(forwardID) + if err != nil { + t.Fatalf("GetForwardRecord after update: %v", err) + } + if record.IPMaxConn != 9 { + t.Fatalf("expected updated ipMaxConn 9, got %d", record.IPMaxConn) + } + if !record.IPSpeedID.Valid || record.IPSpeedID.Int64 != 22 { + t.Fatalf("expected updated ipSpeedId 22, got %+v", record.IPSpeedID) + } + + if err := r.DB().Create(&model.Forward{ + UserID: 4, + UserName: "user", + Name: "listed-per-ip-forward", + TunnelID: 8, + RemoteAddr: "3.3.3.3:443", + Strategy: "fifo", + CreatedTime: now, + UpdatedTime: now, + Status: 1, + IPMaxConn: 11, + IPSpeedID: sql.NullInt64{Int64: 33, Valid: true}, + }).Error; err != nil { + t.Fatalf("create listed forward: %v", err) + } + records, err := r.ListForwardsByTunnel(8) + if err != nil { + t.Fatalf("ListForwardsByTunnel: %v", err) + } + if len(records) != 1 { + t.Fatalf("expected 1 listed record, got %d", len(records)) + } + if records[0].IPMaxConn != 11 || !records[0].IPSpeedID.Valid || records[0].IPSpeedID.Int64 != 33 { + t.Fatalf("expected listed per-IP limits 11/33, got ipMaxConn=%d ipSpeedId=%+v", records[0].IPMaxConn, records[0].IPSpeedID) + } +} + func mustRepoLastInsertID(t *testing.T, r *Repository) int64 { t.Helper() var id int64 diff --git a/go-backend/internal/store/repo/repository_mutations.go b/go-backend/internal/store/repo/repository_mutations.go index 979d187..0f78d09 100644 --- a/go-backend/internal/store/repo/repository_mutations.go +++ b/go-backend/internal/store/repo/repository_mutations.go @@ -695,7 +695,7 @@ func (r *Repository) GetMinForwardPort(forwardID int64) sql.NullInt64 { return p } -func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remoteAddr, strategy string, now int64, speedID interface{}, maxConn int, proxyProtocol int) error { +func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remoteAddr, strategy string, now int64, speedID interface{}, maxConn int, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int) error { if r == nil || r.db == nil { return errors.New("repository not initialized") } @@ -708,6 +708,8 @@ func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remote "strategy": strategy, "speed_id": nullInt64FromInterface(speedID), "max_conn": maxConn, + "ip_max_conn": ipMaxConn, + "ip_speed_id": nullInt64FromInterface(ipSpeedID), "proxy_protocol": proxyProtocol, "updated_time": now, }).Error @@ -790,17 +792,17 @@ func (r *Repository) RollbackForwardFields(id, userID int64, userName, name stri _ = r.db.Model(&model.Forward{}). Where("id = ?", id). Updates(map[string]interface{}{ - "user_id": userID, - "user_name": userName, - "name": name, - "tunnel_id": tunnelID, - "remote_addr": remoteAddr, - "strategy": strategy, - "status": status, - "speed_id": nullInt64FromInterface(speedID), - "max_conn": maxConn, + "user_id": userID, + "user_name": userName, + "name": name, + "tunnel_id": tunnelID, + "remote_addr": remoteAddr, + "strategy": strategy, + "status": status, + "speed_id": nullInt64FromInterface(speedID), + "max_conn": maxConn, "proxy_protocol": proxyProtocol, - "updated_time": now, + "updated_time": now, }).Error } @@ -1260,7 +1262,7 @@ func (r *Repository) EnsureUserTunnelGrant(userID, tunnelID int64) (int64, bool, return ut.ID, true, nil } -func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, now int64, inx int, entryNodeIDs []int64, port int, inIp string, speedID interface{}, maxConn int, proxyProtocol int) (int64, error) { +func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, now int64, inx int, entryNodeIDs []int64, port int, inIp string, speedID interface{}, maxConn int, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int) (int64, error) { if r == nil || r.db == nil { return 0, errors.New("repository not initialized") } @@ -1281,6 +1283,8 @@ func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnel Inx: inx, MaxConn: maxConn, SpeedID: nullInt64FromInterface(speedID), + IPMaxConn: ipMaxConn, + IPSpeedID: nullInt64FromInterface(ipSpeedID), ProxyProtocol: proxyProtocol, } if err := tx.Create(&fwd).Error; err != nil { From a2000e4d986a0cf984f63f0ccf5fbed5a33c60e5 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 22:16:38 +0800 Subject: [PATCH 03/11] fix: preserve per-IP forward limits during rollback --- .../handler/forward_proxy_protocol_test.go | 23 ++++++++++---- go-backend/internal/http/handler/mutations.go | 4 +-- .../repository_forward_proxy_protocol_test.go | 30 +++++++++++++++++++ .../store/repo/repository_mutations.go | 4 ++- 4 files changed, 53 insertions(+), 8 deletions(-) diff --git a/go-backend/internal/http/handler/forward_proxy_protocol_test.go b/go-backend/internal/http/handler/forward_proxy_protocol_test.go index e5db61c..c5fc9e9 100644 --- a/go-backend/internal/http/handler/forward_proxy_protocol_test.go +++ b/go-backend/internal/http/handler/forward_proxy_protocol_test.go @@ -1,6 +1,7 @@ package handler import ( + "database/sql" "testing" "time" @@ -73,6 +74,8 @@ func TestRollbackForwardMutationRestoresProxyProtocol(t *testing.T) { CreatedTime: now, UpdatedTime: now, Status: 1, + IPMaxConn: 5, + IPSpeedID: sql.NullInt64{Int64: 21, Valid: true}, ProxyProtocol: 2, }).Error; err != nil { t.Fatalf("create forward: %v", err) @@ -81,6 +84,8 @@ func TestRollbackForwardMutationRestoresProxyProtocol(t *testing.T) { forwardID := mustLastInsertID(t, r, "rollback-forward") if err := r.DB().Model(&model.Forward{}).Where("id = ?", forwardID).Updates(map[string]interface{}{ "name": "changed-forward", + "ip_max_conn": 0, + "ip_speed_id": nil, "proxy_protocol": 0, "updated_time": now + 1, }).Error; err != nil { @@ -97,14 +102,22 @@ func TestRollbackForwardMutationRestoresProxyProtocol(t *testing.T) { RemoteAddr: "9.9.9.9:443", Strategy: "fifo", Status: 1, + IPMaxConn: 5, + IPSpeedID: sql.NullInt64{Int64: 21, Valid: true}, ProxyProtocol: 2, }, nil) - var proxyProtocol int - if err := r.DB().Raw("SELECT proxy_protocol FROM forward WHERE id = ?", forwardID).Row().Scan(&proxyProtocol); err != nil { - t.Fatalf("query proxy_protocol: %v", err) + var record model.Forward + if err := r.DB().Where("id = ?", forwardID).First(&record).Error; err != nil { + t.Fatalf("query forward: %v", err) } - if proxyProtocol != 2 { - t.Fatalf("expected proxyProtocol restored to 2, got %d", proxyProtocol) + if record.ProxyProtocol != 2 { + t.Fatalf("expected proxyProtocol restored to 2, got %d", record.ProxyProtocol) + } + if record.IPMaxConn != 5 { + t.Fatalf("expected ipMaxConn restored to 5, got %d", record.IPMaxConn) + } + if !record.IPSpeedID.Valid || record.IPSpeedID.Int64 != 21 { + t.Fatalf("expected ipSpeedId restored to 21, got %+v", record.IPSpeedID) } } diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 2189ce2..09781f6 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -1958,7 +1958,7 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) { maxConn := asInt(req["maxConn"], forward.MaxConn) proxyProtocol := asInt(req["proxyProtocol"], forward.ProxyProtocol) - if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, 0, nil, proxyProtocol); err != nil { + if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, forward.IPMaxConn, forward.IPSpeedID, proxyProtocol); err != nil { response.WriteJSON(w, response.Err(-2, err.Error())) return } @@ -4110,7 +4110,7 @@ func (h *Handler) rollbackForwardMutation(oldForward *forwardRecord, oldPorts [] h.repo.RollbackForwardFields( oldForward.ID, oldForward.UserID, oldForward.UserName, oldForward.Name, oldForward.TunnelID, oldForward.RemoteAddr, oldForward.Strategy, oldForward.Status, - oldForward.SpeedID, oldForward.MaxConn, oldForward.ProxyProtocol, + oldForward.SpeedID, oldForward.MaxConn, oldForward.IPMaxConn, oldForward.IPSpeedID, oldForward.ProxyProtocol, time.Now().UnixMilli(), ) diff --git a/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go b/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go index c11a095..19e208a 100644 --- a/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go +++ b/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go @@ -216,6 +216,36 @@ func TestForwardRepositoryPersistsPerIPLimits(t *testing.T) { } } +func TestRollbackForwardFieldsRestoresPerIPLimits(t *testing.T) { + r, err := Open(":memory:") + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + forwardID, err := r.CreateForwardTx(1, "admin", "rollback-per-ip-forward", 2, "1.1.1.1:443", "fifo", now, 1, nil, 0, "", nil, 7, 5, int64(21), 2) + if err != nil { + t.Fatalf("CreateForwardTx: %v", err) + } + if err := r.UpdateForward(forwardID, "rollback-per-ip-forward", 2, "2.2.2.2:443", "fifo", now+1, nil, 0, 0, nil, 0); err != nil { + t.Fatalf("UpdateForward: %v", err) + } + + r.RollbackForwardFields(forwardID, 1, "admin", "rollback-per-ip-forward", 2, "1.1.1.1:443", "fifo", 1, nil, 7, 5, int64(21), 2, now+2) + + record, err := r.GetForwardRecord(forwardID) + if err != nil { + t.Fatalf("GetForwardRecord: %v", err) + } + if record.IPMaxConn != 5 { + t.Fatalf("expected rollback ipMaxConn 5, got %d", record.IPMaxConn) + } + if !record.IPSpeedID.Valid || record.IPSpeedID.Int64 != 21 { + t.Fatalf("expected rollback ipSpeedId 21, got %+v", record.IPSpeedID) + } +} + func mustRepoLastInsertID(t *testing.T, r *Repository) int64 { t.Helper() var id int64 diff --git a/go-backend/internal/store/repo/repository_mutations.go b/go-backend/internal/store/repo/repository_mutations.go index 0f78d09..394ea86 100644 --- a/go-backend/internal/store/repo/repository_mutations.go +++ b/go-backend/internal/store/repo/repository_mutations.go @@ -785,7 +785,7 @@ func (r *Repository) UpdateForwardPortBindIP(forwardID, nodeID int64, port int, Update("in_ip", sql.NullString{String: inIP, Valid: strings.TrimSpace(inIP) != ""}).Error } -func (r *Repository) RollbackForwardFields(id, userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, status int, speedID interface{}, maxConn int, proxyProtocol int, now int64) { +func (r *Repository) RollbackForwardFields(id, userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, status int, speedID interface{}, maxConn int, ipMaxConn int, ipSpeedID interface{}, proxyProtocol int, now int64) { if r == nil || r.db == nil { return } @@ -801,6 +801,8 @@ func (r *Repository) RollbackForwardFields(id, userID int64, userName, name stri "status": status, "speed_id": nullInt64FromInterface(speedID), "max_conn": maxConn, + "ip_max_conn": ipMaxConn, + "ip_speed_id": nullInt64FromInterface(ipSpeedID), "proxy_protocol": proxyProtocol, "updated_time": now, }).Error From 9e8d27d98ef519cadaa893b904692b24b455bbc9 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 22:20:26 +0800 Subject: [PATCH 04/11] feat: expose per-IP forward limit fields --- go-backend/internal/http/handler/mutations.go | 42 +++++++++++- .../tests/contract/forward_contract_test.go | 64 +++++++++++++++++++ 2 files changed, 104 insertions(+), 2 deletions(-) diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 09781f6..4089766 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -1764,6 +1764,18 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) { response.WriteJSON(w, response.Err(-2, err.Error())) return } + if roleID != 0 { + if ipSpeedIDVal, ok := req["ipSpeedId"]; ok && ipSpeedIDVal != nil { + response.WriteJSON(w, response.Err(-1, "普通用户无法设置每 IP 限速规则")) + return + } + } + ipSpeedID := asAnyToInt64Ptr(req["ipSpeedId"]) + ipSpeedID, err = h.normalizeSpeedLimitReference(ipSpeedID) + if err != nil { + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } port := asInt(req["inPort"], 0) if port <= 0 { port = h.pickTunnelPort(tunnelID) @@ -1802,9 +1814,13 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) { userName = "user" } maxConn := asInt(req["maxConn"], 0) + ipMaxConn := asInt(req["ipMaxConn"], 0) + if ipMaxConn < 0 { + ipMaxConn = 0 + } proxyProtocol := asInt(req["proxyProtocol"], 0) - forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, 0, nil, proxyProtocol) + forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn, ipMaxConn, nullableInt(ipSpeedID), proxyProtocol) if err != nil { response.WriteJSON(w, response.Err(-2, err.Error())) return @@ -1903,6 +1919,24 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) { } else if _, ok := req["speedId"]; ok { newSpeedID = sql.NullInt64{Valid: false} } + rawIPSpeedID, hasIPSpeedID := req["ipSpeedId"] + requestedIPSpeedID := asAnyToInt64Ptr(rawIPSpeedID) + if actorRole != 0 && hasIPSpeedID && requestedIPSpeedID != nil && !sameSpeedLimitSelection(forward.IPSpeedID, requestedIPSpeedID) { + response.WriteJSON(w, response.Err(-1, "普通用户无法修改每 IP 限速规则")) + return + } + ipSpeedID := requestedIPSpeedID + ipSpeedID, err = h.normalizeSpeedLimitReference(ipSpeedID) + if err != nil { + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + newIPSpeedID := forward.IPSpeedID + if ipSpeedID != nil { + newIPSpeedID = sql.NullInt64{Int64: *ipSpeedID, Valid: true} + } else if _, ok := req["ipSpeedId"]; ok { + newIPSpeedID = sql.NullInt64{Valid: false} + } port := asInt(req["inPort"], 0) if port <= 0 { @@ -1956,9 +1990,13 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) { } now := time.Now().UnixMilli() maxConn := asInt(req["maxConn"], forward.MaxConn) + ipMaxConn := asInt(req["ipMaxConn"], forward.IPMaxConn) + if ipMaxConn < 0 { + ipMaxConn = 0 + } proxyProtocol := asInt(req["proxyProtocol"], forward.ProxyProtocol) - if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, forward.IPMaxConn, forward.IPSpeedID, proxyProtocol); err != nil { + if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn, ipMaxConn, newIPSpeedID, proxyProtocol); err != nil { response.WriteJSON(w, response.Err(-2, err.Error())) return } diff --git a/go-backend/tests/contract/forward_contract_test.go b/go-backend/tests/contract/forward_contract_test.go index 2bf72fc..909bb51 100644 --- a/go-backend/tests/contract/forward_contract_test.go +++ b/go-backend/tests/contract/forward_contract_test.go @@ -1085,6 +1085,70 @@ func jsonNumber(v int64) string { return strconv.FormatInt(v, 10) } +func TestForwardIPSpeedLimitPermission(t *testing.T) { + secret := "contract-jwt-secret" + router, repo := setupContractRouter(t, secret) + now := time.Now().UnixMilli() + + if err := repo.DB().Exec(` + INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) + VALUES(2, 'normal_user', 'pwd', 1, ?, 99999, 0, 0, 1, 10, ?, ?, 1) + `, now+86400000, now, now).Error; err != nil { + t.Fatalf("insert user: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) + VALUES(12, 'ip-speed-permission-tunnel', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 0) + `, now, now).Error; err != nil { + t.Fatalf("insert tunnel: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO node(id, name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx) + VALUES(20, 'ip-speed-permission-node', 'ip-speed-permission-secret', '10.22.0.1', '10.22.0.1', '', '32200-32210', '', 'v1', 1, 1, 1, ?, ?, 1, '[::]', '[::]', 0) + `, now, now).Error; err != nil { + t.Fatalf("insert node: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) + VALUES(12, 1, 20, 32201, 'round', 1, 'tls') + `).Error; err != nil { + t.Fatalf("insert chain_tunnel: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO speed_limit(id, name, speed, created_time, status) + VALUES(9, 'per-ip-10m', 10, ?, 1) + `, now).Error; err != nil { + t.Fatalf("insert speed limit: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO user_tunnel(user_id, tunnel_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) + VALUES(2, 12, 10, 99999, 0, 0, 1, ?, 1) + `, now+86400000).Error; err != nil { + t.Fatalf("insert user tunnel: %v", err) + } + + userToken, err := auth.GenerateToken(2, "normal_user", 1, secret) + if err != nil { + t.Fatalf("generate user token: %v", err) + } + body, err := json.Marshal(map[string]interface{}{ + "name": "blocked-ip-speed", + "tunnelId": 12, + "remoteAddr": "1.1.1.1:443", + "strategy": "fifo", + "ipSpeedId": 9, + }) + if err != nil { + t.Fatalf("marshal create payload: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(body)) + req.Header.Set("Authorization", userToken) + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + assertCodeMsg(t, res, -1, "普通用户无法设置每 IP 限速规则") +} + func TestNonAdminCannotSetSpeedIdOrPort(t *testing.T) { secret := "contract-jwt-secret-perm" router, repo := setupContractRouter(t, secret) From dec337d46b9aee50fed9b205168c4ba2d522fac1 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 22:26:34 +0800 Subject: [PATCH 05/11] fix: enforce per-IP speed update permissions --- go-backend/internal/http/handler/mutations.go | 31 ++--- .../tests/contract/forward_contract_test.go | 109 ++++++++++++++++++ 2 files changed, 126 insertions(+), 14 deletions(-) diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 4089766..af3a159 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -1921,21 +1921,24 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) { } rawIPSpeedID, hasIPSpeedID := req["ipSpeedId"] requestedIPSpeedID := asAnyToInt64Ptr(rawIPSpeedID) - if actorRole != 0 && hasIPSpeedID && requestedIPSpeedID != nil && !sameSpeedLimitSelection(forward.IPSpeedID, requestedIPSpeedID) { - response.WriteJSON(w, response.Err(-1, "普通用户无法修改每 IP 限速规则")) - return - } - ipSpeedID := requestedIPSpeedID - ipSpeedID, err = h.normalizeSpeedLimitReference(ipSpeedID) - if err != nil { - response.WriteJSON(w, response.Err(-2, err.Error())) - return - } newIPSpeedID := forward.IPSpeedID - if ipSpeedID != nil { - newIPSpeedID = sql.NullInt64{Int64: *ipSpeedID, Valid: true} - } else if _, ok := req["ipSpeedId"]; ok { - newIPSpeedID = sql.NullInt64{Valid: false} + if actorRole != 0 { + if hasIPSpeedID && !sameSpeedLimitSelection(forward.IPSpeedID, requestedIPSpeedID) { + response.WriteJSON(w, response.Err(-1, "普通用户无法修改每 IP 限速规则")) + return + } + } else { + ipSpeedID := requestedIPSpeedID + ipSpeedID, err = h.normalizeSpeedLimitReference(ipSpeedID) + if err != nil { + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + if ipSpeedID != nil { + newIPSpeedID = sql.NullInt64{Int64: *ipSpeedID, Valid: true} + } else if hasIPSpeedID { + newIPSpeedID = sql.NullInt64{Valid: false} + } } port := asInt(req["inPort"], 0) diff --git a/go-backend/tests/contract/forward_contract_test.go b/go-backend/tests/contract/forward_contract_test.go index 909bb51..6007fe3 100644 --- a/go-backend/tests/contract/forward_contract_test.go +++ b/go-backend/tests/contract/forward_contract_test.go @@ -1149,6 +1149,115 @@ func TestForwardIPSpeedLimitPermission(t *testing.T) { assertCodeMsg(t, res, -1, "普通用户无法设置每 IP 限速规则") } +func TestForwardIPSpeedLimitUpdatePermission(t *testing.T) { + secret := "contract-jwt-secret" + router, repo := setupContractRouter(t, secret) + server := httptest.NewServer(router) + defer server.Close() + now := time.Now().UnixMilli() + + if err := repo.DB().Exec(` + INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status) + VALUES(2, 'normal_user_ip_update', 'pwd', 1, ?, 99999, 0, 0, 1, 10, ?, ?, 1) + `, now+86400000, now, now).Error; err != nil { + t.Fatalf("insert user: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) + VALUES(13, 'ip-speed-update-permission-tunnel', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 0) + `, now, now).Error; err != nil { + t.Fatalf("insert tunnel: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO node(id, name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx) + VALUES(21, 'ip-speed-update-permission-node', 'ip-speed-update-permission-secret', '10.22.0.2', '10.22.0.2', '', '32300-32310', '', 'v1', 1, 1, 1, ?, ?, 1, '[::]', '[::]', 0) + `, now, now).Error; err != nil { + t.Fatalf("insert node: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) + VALUES(13, 1, 21, 32301, 'round', 1, 'tls') + `).Error; err != nil { + t.Fatalf("insert chain_tunnel: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO speed_limit(id, name, speed, created_time, status) + VALUES(10, 'per-ip-10m-update', 10, ?, 1), (11, 'per-ip-20m-update', 20, ?, 1) + `, now, now).Error; err != nil { + t.Fatalf("insert speed limits: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO user_tunnel(user_id, tunnel_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) + VALUES(2, 13, 10, 99999, 0, 0, 1, ?, 1) + `, now+86400000).Error; err != nil { + t.Fatalf("insert user tunnel: %v", err) + } + if err := repo.DB().Exec(` + INSERT INTO forward(id, user_id, user_name, name, tunnel_id, remote_addr, strategy, ip_speed_id, in_flow, out_flow, created_time, updated_time, status, inx) + VALUES(30, 2, 'normal_user_ip_update', 'ip-speed-update-forward', 13, '1.1.1.1:443', 'fifo', 10, 0, 0, ?, ?, 1, 0) + `, now, now).Error; err != nil { + t.Fatalf("insert forward: %v", err) + } + + userToken, err := auth.GenerateToken(2, "normal_user_ip_update", 1, secret) + if err != nil { + t.Fatalf("generate user token: %v", err) + } + stopNode := startMockNodeSession(t, server.URL, "ip-speed-update-permission-secret") + defer stopNode() + + updateForward := func(t *testing.T, ipSpeedID interface{}) *httptest.ResponseRecorder { + t.Helper() + if err := repo.DB().Exec(`UPDATE forward SET ip_speed_id = 10 WHERE id = 30`).Error; err != nil { + t.Fatalf("reset forward ip speed limit: %v", err) + } + body, err := json.Marshal(map[string]interface{}{ + "id": 30, + "name": "ip-speed-update-forward", + "tunnelId": 13, + "remoteAddr": "1.1.1.1:443", + "ipSpeedId": ipSpeedID, + }) + if err != nil { + t.Fatalf("marshal update payload: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/update", bytes.NewReader(body)) + req.Header.Set("Authorization", userToken) + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + return res + } + assertStoredIPSpeedID := func(t *testing.T, want int64) { + t.Helper() + var got sql.NullInt64 + if err := repo.DB().Raw(`SELECT ip_speed_id FROM forward WHERE id = 30`).Scan(&got).Error; err != nil { + t.Fatalf("read forward ip_speed_id: %v", err) + } + if !got.Valid || got.Int64 != want { + t.Fatalf("expected ip_speed_id %d, got valid=%v value=%d", want, got.Valid, got.Int64) + } + } + + t.Run("non-admin cannot change existing ipSpeedId", func(t *testing.T) { + res := updateForward(t, 11) + assertCodeMsg(t, res, -1, "普通用户无法修改每 IP 限速规则") + assertStoredIPSpeedID(t, 10) + }) + + t.Run("non-admin cannot clear existing ipSpeedId", func(t *testing.T) { + res := updateForward(t, nil) + assertCodeMsg(t, res, -1, "普通用户无法修改每 IP 限速规则") + assertStoredIPSpeedID(t, 10) + }) + + t.Run("non-admin can keep existing ipSpeedId", func(t *testing.T) { + res := updateForward(t, 10) + assertCode(t, res, 0) + assertStoredIPSpeedID(t, 10) + }) +} + func TestNonAdminCannotSetSpeedIdOrPort(t *testing.T) { secret := "contract-jwt-secret-perm" router, repo := setupContractRouter(t, secret) From 46394388b1e959b9645192085d6353da7ab8b72a Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 22:32:35 +0800 Subject: [PATCH 06/11] feat: sync per-IP runtime limiters --- .../internal/http/handler/control_plane.go | 149 ++++++++++++---- .../http/handler/control_plane_test.go | 54 +++++- .../handler/forward_proxy_protocol_test.go | 2 +- .../contract/max_conn_limit_contract_test.go | 9 +- .../per_ip_speed_limit_contract_test.go | 168 ++++++++++++++++++ 5 files changed, 337 insertions(+), 45 deletions(-) create mode 100644 go-backend/tests/contract/per_ip_speed_limit_contract_test.go diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 8fd2c33..3bb7b25 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -27,6 +27,16 @@ type nodeRecord = model.NodeRecord type chainNodeRecord = model.ChainNodeRecord +type forwardRuntimeLimiters struct { + TrafficLimiter string + ConnLimiter string +} + +type forwardLimiterConfig struct { + Name string + Limits []string +} + type diagnosisTarget struct { Address string IP string @@ -264,6 +274,13 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method speed = utSpeed } + var ipSpeed *int + if forward.IPSpeedID.Valid && forward.IPSpeedID.Int64 > 0 { + if speedVal, err := h.repo.GetSpeedLimitSpeed(forward.IPSpeedID.Int64); err == nil && speedVal > 0 { + ipSpeed = &speedVal + } + } + serviceBase := buildForwardServiceBaseWithResolvedUserTunnel(forward.ID, forward.UserID, userTunnelID) user, err := h.repo.GetUserByID(forward.UserID) @@ -271,19 +288,31 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method return nil, err } - var cLimiterName string - var maxConnToSet int - - if forward.MaxConn > 0 { - maxConnToSet = forward.MaxConn - cLimiterName = fmt.Sprintf("rule_conn_limit_%d", forward.ID) - } else if user != nil && user.MaxConn > 0 { - maxConnToSet = user.MaxConn - cLimiterName = fmt.Sprintf("user_conn_limit_%d", user.ID) + userMaxConn := 0 + if user != nil && user.MaxConn > 0 { + userMaxConn = user.MaxConn } + connLimiterConfig := buildConnLimiterConfig(forward, userMaxConn) for _, fp := range ports { - if limiterID != nil && speed != nil { + runtimeLimiters := forwardRuntimeLimiters{ConnLimiter: connLimiterConfig.Name} + if ipSpeed != nil { + runtimeLimiters.TrafficLimiter = fmt.Sprintf("rule_traffic_limit_%d", forward.ID) + if err := h.ensureTrafficLimiterOnNode(fp.NodeID, runtimeLimiters.TrafficLimiter, speed, ipSpeed); err != nil { + // If the limiter push fails because the node is offline, skip it with a warning + if isNodeOfflineOrTimeoutError(err) { + node, _ := h.getNodeRecord(fp.NodeID) + nodeName := fmt.Sprintf("%d", fp.NodeID) + if node != nil && strings.TrimSpace(node.Name) != "" { + nodeName = strings.TrimSpace(node.Name) + } + warnings = append(warnings, fmt.Sprintf("节点 %s 不在线,已跳过下发", nodeName)) + continue + } + return nil, err + } + } else if limiterID != nil && speed != nil { + runtimeLimiters.TrafficLimiter = strconv.FormatInt(*limiterID, 10) if err := h.ensureLimiterOnNode(fp.NodeID, *limiterID, *speed); err != nil { // If the limiter push fails because the node is offline, skip it with a warning if isNodeOfflineOrTimeoutError(err) { @@ -299,8 +328,8 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method } } - if cLimiterName != "" { - if err := h.ensureConnLimiterOnNode(fp.NodeID, cLimiterName, maxConnToSet); err != nil { + if connLimiterConfig.Name != "" { + if err := h.ensureConnLimiterOnNode(fp.NodeID, connLimiterConfig); err != nil { warnings = append(warnings, fmt.Sprintf("节点 %d 连接限制器下发失败: %v", fp.NodeID, err)) } } @@ -309,7 +338,7 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method if err != nil { return nil, err } - services := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, strings.TrimSpace(fp.InIP), limiterID, cLimiterName) + services := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, strings.TrimSpace(fp.InIP), runtimeLimiters) _, err = h.sendNodeCommand(node.ID, method, services, true, false) if err != nil && allowFallbackAdd && method == "UpdateService" { if isNotFoundError(err) { @@ -324,7 +353,7 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method } if err != nil && strings.EqualFold(strings.TrimSpace(method), "UpdateService") && isCannotAssignRequestedAddressError(err) { var warning string - warning, err = h.fallbackForwardPortToDefaultBind(forward, tunnel, node, fp, serviceBase, limiterID, cLimiterName) + warning, err = h.fallbackForwardPortToDefaultBind(forward, tunnel, node, fp, serviceBase, runtimeLimiters) if err == nil && warning != "" { warnings = append(warnings, warning) } @@ -350,7 +379,7 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method return warnings, nil } -func (h *Handler) fallbackForwardPortToDefaultBind(forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, fp forwardPortRecord, serviceBase string, limiterID *int64, cLimiterName string) (string, error) { +func (h *Handler) fallbackForwardPortToDefaultBind(forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, fp forwardPortRecord, serviceBase string, runtimeLimiters forwardRuntimeLimiters) (string, error) { if h == nil || forward == nil || tunnel == nil || node == nil { return "", errors.New("invalid bind fallback context") } @@ -367,7 +396,7 @@ func (h *Handler) fallbackForwardPortToDefaultBind(forward *forwardRecord, tunne } time.Sleep(150 * time.Millisecond) - defaultServices := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, "", limiterID, cLimiterName) + defaultServices := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, "", runtimeLimiters) if _, err := h.sendNodeCommand(node.ID, "AddService", defaultServices, true, false); err != nil { return "", err } @@ -1659,7 +1688,7 @@ func compactErrorMessage(msg string) string { return strings.Join(strings.Fields(strings.ToLower(msg)), "") } -func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, bindIP string, limiterID *int64, cLimiterName string) []map[string]interface{} { +func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, bindIP string, runtimeLimiters forwardRuntimeLimiters) []map[string]interface{} { protocols := []string{"tcp", "udp"} services := make([]map[string]interface{}, 0, 2) targets := splitRemoteTargets(forward.RemoteAddr) @@ -1702,8 +1731,11 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel }, }, } - if cLimiterName != "" { - service["climiter"] = cLimiterName + if runtimeLimiters.ConnLimiter != "" { + service["climiter"] = runtimeLimiters.ConnLimiter + } + if runtimeLimiters.TrafficLimiter != "" { + service["limiter"] = runtimeLimiters.TrafficLimiter } if forward.ProxyProtocol > 0 { handlerConfig := service["handler"].(map[string]interface{}) @@ -1728,9 +1760,6 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel } service["metadata"].(map[string]interface{})["interface"] = node.InterfaceName } - if limiterID != nil && *limiterID > 0 { - service["limiter"] = strconv.FormatInt(*limiterID, 10) - } services = append(services, service) } @@ -1830,22 +1859,16 @@ func (h *Handler) ensureLimiterOnNode(nodeID int64, limiterID int64, speed int) return nil } -func (h *Handler) ensureConnLimiterOnNode(nodeID int64, limiterName string, maxConn int) error { - limitStr := fmt.Sprintf("$ %d", maxConn) - - payload := map[string]interface{}{ - "name": limiterName, - "limits": []string{limitStr}, +func (h *Handler) ensureConnLimiterOnNode(nodeID int64, cfg forwardLimiterConfig) error { + if cfg.Name == "" || len(cfg.Limits) == 0 { + return nil } - + payload := map[string]interface{}{"name": cfg.Name, "limits": cfg.Limits} if _, err := h.sendNodeCommand(nodeID, "AddCLimiters", payload, false, false); err != nil { if !isAlreadyExistsMessage(err.Error()) { return fmt.Errorf("连接限制器下发失败: %w", err) } - updatePayload := map[string]interface{}{ - "limiter": limiterName, - "data": payload, - } + updatePayload := map[string]interface{}{"limiter": cfg.Name, "data": payload} if _, updateErr := h.sendNodeCommand(nodeID, "UpdateCLimiters", updatePayload, false, false); updateErr != nil { return fmt.Errorf("连接限制器更新失败: %w", updateErr) } @@ -1853,14 +1876,51 @@ func (h *Handler) ensureConnLimiterOnNode(nodeID int64, limiterName string, maxC return nil } -func buildLimiterAddPayload(limiterID int64, speed int) (string, map[string]interface{}) { +func buildConnLimiterConfig(forward *forwardRecord, userMaxConn int) forwardLimiterConfig { + if forward == nil { + return forwardLimiterConfig{} + } + limits := make([]string, 0, 2) + if forward.MaxConn > 0 { + limits = append(limits, fmt.Sprintf("$ %d", forward.MaxConn)) + } else if userMaxConn > 0 { + limits = append(limits, fmt.Sprintf("$ %d", userMaxConn)) + } + if forward.IPMaxConn > 0 { + limits = append(limits, fmt.Sprintf("$$ %d", forward.IPMaxConn)) + } + if len(limits) == 0 { + return forwardLimiterConfig{} + } + name := fmt.Sprintf("user_conn_limit_%d", forward.UserID) + if forward.MaxConn > 0 || forward.IPMaxConn > 0 { + name = fmt.Sprintf("rule_conn_limit_%d", forward.ID) + } + return forwardLimiterConfig{Name: name, Limits: limits} +} + +func speedToLimitLine(key string, speed int) string { rate := float64(speed) / 8.0 - limitStr := fmt.Sprintf("$ %.1fMB %.1fMB", rate, rate) + return fmt.Sprintf("%s %.1fMB %.1fMB", key, rate, rate) +} + +func buildTrafficLimiterPayload(name string, totalSpeed *int, ipSpeed *int) map[string]interface{} { + limits := make([]string, 0, 3) + if totalSpeed != nil && *totalSpeed > 0 { + limits = append(limits, speedToLimitLine("$", *totalSpeed)) + } + if ipSpeed != nil && *ipSpeed > 0 { + limits = append(limits, speedToLimitLine("0.0.0.0/0", *ipSpeed), speedToLimitLine("::/0", *ipSpeed)) + } + return map[string]interface{}{"name": name, "limits": limits} +} + +func buildLimiterAddPayload(limiterID int64, speed int) (string, map[string]interface{}) { name := strconv.FormatInt(limiterID, 10) return name, map[string]interface{}{ "name": name, - "limits": []string{limitStr}, + "limits": []string{speedToLimitLine("$", speed)}, } } @@ -1888,3 +1948,20 @@ func (h *Handler) upsertLimiterOnNode(nodeID int64, limiterID int64, speed int) return nil } + +func (h *Handler) ensureTrafficLimiterOnNode(nodeID int64, name string, totalSpeed *int, ipSpeed *int) error { + payload := buildTrafficLimiterPayload(name, totalSpeed, ipSpeed) + limits, _ := payload["limits"].([]string) + if name == "" || len(limits) == 0 { + return nil + } + if _, err := h.sendNodeCommand(nodeID, "AddLimiters", payload, false, false); err != nil { + if !isAlreadyExistsMessage(err.Error()) { + return fmt.Errorf("限速规则下发失败: %w", err) + } + if _, updateErr := h.sendNodeCommand(nodeID, "UpdateLimiters", buildLimiterUpdatePayload(name, payload), false, false); updateErr != nil { + return fmt.Errorf("限速规则更新失败: %w", updateErr) + } + } + return nil +} diff --git a/go-backend/internal/http/handler/control_plane_test.go b/go-backend/internal/http/handler/control_plane_test.go index 97bd80a..c3fb79a 100644 --- a/go-backend/internal/http/handler/control_plane_test.go +++ b/go-backend/internal/http/handler/control_plane_test.go @@ -378,7 +378,7 @@ func TestRetryTunnelServiceAddWithCleanupReturnsCleanupError(t *testing.T) { func TestBuildForwardServiceConfigs_UsesBindIPForListen(t *testing.T) { forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7} node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"} - services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22000, "10.9.8.7", nil, "") + services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22000, "10.9.8.7", forwardRuntimeLimiters{}) if len(services) != 2 { t.Fatalf("expected 2 services, got %d", len(services)) } @@ -393,7 +393,7 @@ func TestBuildForwardServiceConfigs_UsesBindIPForListen(t *testing.T) { func TestBuildForwardServiceConfigs_DefaultListenAddrWhenBindIPEmpty(t *testing.T) { forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7} node := &nodeRecord{TCPListenAddr: "0.0.0.0", UDPListenAddr: "[::]"} - services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22001, "", nil, "") + services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22001, "", forwardRuntimeLimiters{}) if len(services) != 2 { t.Fatalf("expected 2 services, got %d", len(services)) } @@ -409,7 +409,7 @@ func TestBuildForwardServiceConfigs_DefaultListenAddrWhenBindIPEmpty(t *testing. func TestBuildForwardServiceConfigs_BindIPAlreadyContainsPort(t *testing.T) { forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7} node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"} - services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 55555, "3.3.3.3:12345", nil, "") + services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 55555, "3.3.3.3:12345", forwardRuntimeLimiters{}) if len(services) != 2 { t.Fatalf("expected 2 services, got %d", len(services)) } @@ -464,7 +464,7 @@ func TestBuildForwardServiceConfigs_IPv6BindIP(t *testing.T) { t.Run(tt.name, func(t *testing.T) { forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7} node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"} - services := buildForwardServiceConfigs("1_2_0", forward, nil, node, tt.port, tt.bindIP, nil, "") + services := buildForwardServiceConfigs("1_2_0", forward, nil, node, tt.port, tt.bindIP, forwardRuntimeLimiters{}) if len(services) != 2 { t.Fatalf("expected 2 services, got %d", len(services)) } @@ -478,6 +478,52 @@ func TestBuildForwardServiceConfigs_IPv6BindIP(t *testing.T) { } } +func TestBuildConnLimiterConfigCombinesTotalAndPerIP(t *testing.T) { + cfg := buildConnLimiterConfig(&forwardRecord{ID: 42, UserID: 9, MaxConn: 100, IPMaxConn: 5}, 37) + want := forwardLimiterConfig{Name: "rule_conn_limit_42", Limits: []string{"$ 100", "$$ 5"}} + if !reflect.DeepEqual(cfg, want) { + t.Fatalf("expected %+v, got %+v", want, cfg) + } +} + +func TestBuildConnLimiterConfigUsesUserTotalWithRulePerIP(t *testing.T) { + cfg := buildConnLimiterConfig(&forwardRecord{ID: 42, UserID: 9, IPMaxConn: 5}, 37) + want := forwardLimiterConfig{Name: "rule_conn_limit_42", Limits: []string{"$ 37", "$$ 5"}} + if !reflect.DeepEqual(cfg, want) { + t.Fatalf("expected %+v, got %+v", want, cfg) + } +} + +func TestBuildTrafficLimiterPayloadCombinesTotalAndPerIP(t *testing.T) { + payload := buildTrafficLimiterPayload("rule_traffic_limit_42", intPtr(80), intPtr(40)) + wantLimits := []string{"$ 10.0MB 10.0MB", "0.0.0.0/0 5.0MB 5.0MB", "::/0 5.0MB 5.0MB"} + if payload["name"] != "rule_traffic_limit_42" { + t.Fatalf("expected name rule_traffic_limit_42, got %v", payload["name"]) + } + if !reflect.DeepEqual(payload["limits"], wantLimits) { + t.Fatalf("expected limits %v, got %v", wantLimits, payload["limits"]) + } +} + +func TestBuildForwardServiceConfigsUsesRuntimeLimiterNames(t *testing.T) { + forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7} + node := &nodeRecord{TCPListenAddr: "0.0.0.0", UDPListenAddr: "[::]"} + services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22001, "", forwardRuntimeLimiters{TrafficLimiter: "rule_traffic_limit_42", ConnLimiter: "rule_conn_limit_42"}) + if len(services) != 2 { + t.Fatalf("expected 2 services, got %d", len(services)) + } + for _, service := range services { + if service["limiter"] != "rule_traffic_limit_42" { + t.Fatalf("expected traffic limiter rule_traffic_limit_42, got %v", service["limiter"]) + } + if service["climiter"] != "rule_conn_limit_42" { + t.Fatalf("expected conn limiter rule_conn_limit_42, got %v", service["climiter"]) + } + } +} + +func intPtr(v int) *int { return &v } + func TestProcessServerAddress_StripsURLSchemeAndPath(t *testing.T) { tests := []struct { name string diff --git a/go-backend/internal/http/handler/forward_proxy_protocol_test.go b/go-backend/internal/http/handler/forward_proxy_protocol_test.go index c5fc9e9..b54df23 100644 --- a/go-backend/internal/http/handler/forward_proxy_protocol_test.go +++ b/go-backend/internal/http/handler/forward_proxy_protocol_test.go @@ -25,7 +25,7 @@ func TestBuildForwardServiceConfigsSendsProxyProtocolToForwardHandler(t *testing UDPListenAddr: "0.0.0.0", } - services := buildForwardServiceConfigs("1_2_3", forward, tunnel, node, 4001, "", nil, "") + services := buildForwardServiceConfigs("1_2_3", forward, tunnel, node, 4001, "", forwardRuntimeLimiters{}) if len(services) != 2 { t.Fatalf("expected 2 services, got %d", len(services)) } diff --git a/go-backend/tests/contract/max_conn_limit_contract_test.go b/go-backend/tests/contract/max_conn_limit_contract_test.go index c55e7c5..b356be1 100644 --- a/go-backend/tests/contract/max_conn_limit_contract_test.go +++ b/go-backend/tests/contract/max_conn_limit_contract_test.go @@ -97,6 +97,7 @@ func TestMaxConnLimit(t *testing.T) { "remoteAddr": "1.1.1.1:443", "strategy": "fifo", "maxConn": 42, + "ipMaxConn": 7, "proxyProtocol": 2, } body, err := json.Marshal(payload) @@ -195,8 +196,8 @@ func TestMaxConnLimit(t *testing.T) { t.Fatalf("expected limiter name %s, got %v", expectedName, addData["name"]) } if limits, ok := addData["limits"].([]interface{}); ok { - if len(limits) != 1 || limits[0] != "$ 42" { - t.Fatalf("expected limits to contain '$ 42', got %v", limits) + if len(limits) != 2 || limits[0] != "$ 42" || limits[1] != "$$ 7" { + t.Fatalf("expected limits to contain '$ 42' and '$$ 7', got %v", limits) } } else { t.Fatalf("invalid limits type in AddCLimiters data: %v", addData) @@ -218,8 +219,8 @@ func TestMaxConnLimit(t *testing.T) { t.Fatalf("expected nested name %s, got %v", expectedName, nestedData["name"]) } if nestedLimits, ok := nestedData["limits"].([]interface{}); ok { - if len(nestedLimits) != 1 || nestedLimits[0] != "$ 42" { - t.Fatalf("expected nested limits to contain '$ 42', got %v", nestedLimits) + if len(nestedLimits) != 2 || nestedLimits[0] != "$ 42" || nestedLimits[1] != "$$ 7" { + t.Fatalf("expected nested limits to contain '$ 42' and '$$ 7', got %v", nestedLimits) } } else { t.Fatalf("invalid limits type in UpdateCLimiters nested data: %v", nestedData) diff --git a/go-backend/tests/contract/per_ip_speed_limit_contract_test.go b/go-backend/tests/contract/per_ip_speed_limit_contract_test.go new file mode 100644 index 0000000..faaa1ec --- /dev/null +++ b/go-backend/tests/contract/per_ip_speed_limit_contract_test.go @@ -0,0 +1,168 @@ +package contract_test + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "sync" + "testing" + "time" + + "go-backend/internal/auth" + "go-backend/internal/http/response" +) + +func TestPerIPSpeedLimitRuntimePayload(t *testing.T) { + secret := "contract-jwt-secret" + router, r := setupContractRouter(t, secret) + server := httptest.NewServer(router) + defer server.Close() + + adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret) + if err != nil { + t.Fatalf("generate admin token: %v", err) + } + + now := time.Now().UnixMilli() + if err := r.DB().Exec(` + INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, "per-ip-speed-tunnel", 1.0, 1, "tls", 99999, now, now, 1, nil, 0).Error; err != nil { + t.Fatalf("insert tunnel: %v", err) + } + var tunnelID int64 + if err := r.DB().Raw("SELECT id FROM tunnel WHERE name = ?", "per-ip-speed-tunnel").Scan(&tunnelID).Error; err != nil { + t.Fatalf("get tunnel ID: %v", err) + } + + if err := r.DB().Exec(` + INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, "per-ip-speed-node", "per-ip-speed-secret", "10.22.0.1", "10.22.0.1", "", "32200-32210", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0).Error; err != nil { + t.Fatalf("insert node: %v", err) + } + var nodeID int64 + if err := r.DB().Raw("SELECT id FROM node WHERE name = ?", "per-ip-speed-node").Scan(&nodeID).Error; err != nil { + t.Fatalf("get node ID: %v", err) + } + + if err := r.DB().Exec(` + INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) + VALUES(?, 1, ?, 32201, 'round', 1, 'tls') + `, tunnelID, nodeID).Error; err != nil { + t.Fatalf("insert chain_tunnel: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO user_tunnel(user_id, tunnel_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) + VALUES(1, ?, 10, 99999, 0, 0, 1, ?, 1) + `, tunnelID, now+365*24*3600*1000).Error; err != nil { + t.Fatalf("insert user_tunnel: %v", err) + } + + totalSpeedID, err := r.CreateSpeedLimit("per-ip-total-speed", 80, now, 1) + if err != nil { + t.Fatalf("create total speed limit: %v", err) + } + ipSpeedID, err := r.CreateSpeedLimit("per-ip-client-speed", 40, now, 1) + if err != nil { + t.Fatalf("create per-ip speed limit: %v", err) + } + + var commandMu sync.Mutex + receivedCommands := make([]string, 0) + var addLimitersData json.RawMessage + var updateServiceData json.RawMessage + + stopNode := startMockSessionForMaxConn(t, server.URL, "per-ip-speed-secret", func(cmdType string, data json.RawMessage) (bool, string) { + commandMu.Lock() + defer commandMu.Unlock() + receivedCommands = append(receivedCommands, cmdType) + if cmdType == "AddLimiters" { + addLimitersData = append([]byte(nil), data...) + } + if cmdType == "UpdateService" { + updateServiceData = append([]byte(nil), data...) + } + return false, "" + }) + defer stopNode() + + waitNodeStatus(t, r, nodeID, 1) + + payload := map[string]interface{}{ + "name": "per-ip-speed-forward", + "tunnelId": tunnelID, + "remoteAddr": "1.1.1.1:443", + "strategy": "fifo", + "speedId": totalSpeedID, + "ipSpeedId": ipSpeedID, + } + body, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal payload: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(body)) + req.Header.Set("Authorization", adminToken) + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + + var out response.R + if err := json.NewDecoder(res.Body).Decode(&out); err != nil { + t.Fatalf("decode response: %v", err) + } + if out.Code != 0 { + t.Fatalf("expected create success, got code=%d msg=%s", out.Code, out.Msg) + } + + var forwardID int64 + if err := r.DB().Raw("SELECT id FROM forward WHERE name = ?", "per-ip-speed-forward").Scan(&forwardID).Error; err != nil { + t.Fatalf("get forward ID: %v", err) + } + expectedName := fmt.Sprintf("rule_traffic_limit_%d", forwardID) + expectedLimits := []string{"$ 10.0MB 10.0MB", "0.0.0.0/0 5.0MB 5.0MB", "::/0 5.0MB 5.0MB"} + + commandMu.Lock() + defer commandMu.Unlock() + if addLimitersData == nil { + t.Fatalf("expected AddLimiters to be sent. Received: %v", receivedCommands) + } + if updateServiceData == nil { + t.Fatalf("expected UpdateService to be sent. Received: %v", receivedCommands) + } + + var addData map[string]interface{} + if err := json.Unmarshal(addLimitersData, &addData); err != nil { + t.Fatalf("unmarshal AddLimiters data: %v", err) + } + if addData["name"] != expectedName { + t.Fatalf("expected limiter name %s, got %v", expectedName, addData["name"]) + } + limits, ok := addData["limits"].([]interface{}) + if !ok { + t.Fatalf("expected limits array, got %T", addData["limits"]) + } + gotLimits := make([]string, 0, len(limits)) + for _, limit := range limits { + gotLimits = append(gotLimits, fmt.Sprint(limit)) + } + if !reflect.DeepEqual(gotLimits, expectedLimits) { + t.Fatalf("expected limits %v, got %v", expectedLimits, gotLimits) + } + + var services []map[string]interface{} + if err := json.Unmarshal(updateServiceData, &services); err != nil { + t.Fatalf("unmarshal UpdateService data: %v", err) + } + if len(services) == 0 { + t.Fatalf("expected services in UpdateService") + } + for _, service := range services { + if service["limiter"] != expectedName { + t.Fatalf("expected service limiter %s, got %v", expectedName, service["limiter"]) + } + } +} From 7a9ba8bd817560b840c0aeb8d2c992abc6c814ee Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 23:05:31 +0800 Subject: [PATCH 07/11] fix: apply per-client limits to udp listener --- go-gost/x/limiter/conn/conn_test.go | 30 +++++++ go-gost/x/limiter/traffic/traffic_test.go | 28 +++++++ go-gost/x/listener/udp/listener.go | 43 +++++++++- go-gost/x/listener/udp/listener_test.go | 95 +++++++++++++++++++++++ 4 files changed, 194 insertions(+), 2 deletions(-) create mode 100644 go-gost/x/limiter/conn/conn_test.go create mode 100644 go-gost/x/limiter/traffic/traffic_test.go create mode 100644 go-gost/x/listener/udp/listener_test.go diff --git a/go-gost/x/limiter/conn/conn_test.go b/go-gost/x/limiter/conn/conn_test.go new file mode 100644 index 0000000..063a4e9 --- /dev/null +++ b/go-gost/x/limiter/conn/conn_test.go @@ -0,0 +1,30 @@ +package conn + +import ( + "io" + "testing" + + corelogger "github.com/go-gost/core/logger" + xlogger "github.com/go-gost/x/logger" +) + +func TestIPLimitKeyCreatesIndependentLimiters(t *testing.T) { + limiter := NewConnLimiter( + LimitsOption("$$ 1"), + LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel))), + ) + first := limiter.Limiter("192.0.2.1") + second := limiter.Limiter("192.0.2.2") + if first == nil || second == nil { + t.Fatalf("expected non-nil per-IP limiters") + } + if !first.Allow(1) { + t.Fatalf("expected first IP first connection to be allowed") + } + if first.Allow(1) { + t.Fatalf("expected first IP second connection to be rejected") + } + if !second.Allow(1) { + t.Fatalf("expected second IP first connection to be allowed independently") + } +} diff --git a/go-gost/x/limiter/traffic/traffic_test.go b/go-gost/x/limiter/traffic/traffic_test.go new file mode 100644 index 0000000..702cc76 --- /dev/null +++ b/go-gost/x/limiter/traffic/traffic_test.go @@ -0,0 +1,28 @@ +package traffic + +import ( + "context" + "io" + "testing" + + corelogger "github.com/go-gost/core/logger" + xlogger "github.com/go-gost/x/logger" +) + +func TestCIDRLimitCreatesIndependentClientLimiters(t *testing.T) { + limiter := NewTrafficLimiter( + LimitsOption("0.0.0.0/0 2B 2B"), + LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel))), + ) + first := limiter.In(context.Background(), "192.0.2.1:1000") + second := limiter.In(context.Background(), "192.0.2.2:1000") + if first == nil || second == nil { + t.Fatalf("expected non-nil CIDR client limiters") + } + if first == second { + t.Fatalf("expected different clients to receive independent limiter instances") + } + if first.Limit() != 2 || second.Limit() != 2 { + t.Fatalf("expected both limits to be 2, got %d and %d", first.Limit(), second.Limit()) + } +} diff --git a/go-gost/x/listener/udp/listener.go b/go-gost/x/listener/udp/listener.go index 6f21f7c..23aff31 100644 --- a/go-gost/x/listener/udp/listener.go +++ b/go-gost/x/listener/udp/listener.go @@ -10,6 +10,7 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/udp" + climiter "github.com/go-gost/x/limiter/conn/wrapper" traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -70,7 +71,7 @@ func (l *udpListener) Init(md md.Metadata) (err error) { limiter.NetworkOption(conn.LocalAddr().Network()), ) - l.ln = udp.NewListener(conn, &udp.ListenConfig{ + ln := udp.NewListener(conn, &udp.ListenConfig{ Backlog: l.md.backlog, ReadQueueSize: l.md.readQueueSize, ReadBufferSize: l.md.readBufferSize, @@ -78,11 +79,49 @@ func (l *udpListener) Init(md md.Metadata) (err error) { TTL: l.md.ttl, Logger: l.logger, }) + l.ln = ln return } func (l *udpListener) Accept() (conn net.Conn, err error) { - return l.ln.Accept() + conn, err = l.ln.Accept() + if err != nil { + return + } + + if l.options.ConnLimiter != nil { + host, _, _ := net.SplitHostPort(conn.RemoteAddr().String()) + if lim := l.options.ConnLimiter.Limiter(host); lim != nil { + if !lim.Allow(1) { + _ = conn.Close() + return closedConn{Conn: conn}, nil + } + conn = climiter.WrapConn(lim, conn) + } + } + + conn = limiter_wrapper.WrapConn( + conn, + l.options.TrafficLimiter, + conn.RemoteAddr().String(), + limiter.ScopeOption(limiter.ScopeConn), + limiter.ServiceOption(l.options.Service), + limiter.NetworkOption(conn.LocalAddr().Network()), + limiter.SrcOption(conn.RemoteAddr().String()), + ) + return +} + +type closedConn struct { + net.Conn +} + +func (c closedConn) Read([]byte) (int, error) { + return 0, net.ErrClosed +} + +func (c closedConn) Write([]byte) (int, error) { + return 0, net.ErrClosed } func (l *udpListener) Addr() net.Addr { diff --git a/go-gost/x/listener/udp/listener_test.go b/go-gost/x/listener/udp/listener_test.go new file mode 100644 index 0000000..9b56847 --- /dev/null +++ b/go-gost/x/listener/udp/listener_test.go @@ -0,0 +1,95 @@ +package udp + +import ( + "io" + "net" + "testing" + "time" + + corelistener "github.com/go-gost/core/listener" + corelogger "github.com/go-gost/core/logger" + xconn "github.com/go-gost/x/limiter/conn" + xlogger "github.com/go-gost/x/logger" +) + +func TestAcceptAppliesConnLimiterAndReleasesOnClose(t *testing.T) { + ln := NewListener( + corelistener.AddrOption("127.0.0.1:0"), + corelistener.ConnLimiterOption(xconn.NewConnLimiter( + xconn.LimitsOption("$$ 1"), + xconn.LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel))), + )), + corelistener.LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel))), + ) + if err := ln.Init(nil); err != nil { + t.Fatalf("init listener: %v", err) + } + defer ln.Close() + + addr := ln.Addr().String() + client, err := net.Dial("udp", addr) + if err != nil { + t.Fatalf("dial udp listener: %v", err) + } + defer client.Close() + if _, err := client.Write([]byte("first")); err != nil { + t.Fatalf("write first packet: %v", err) + } + first, err := acceptWithTimeout(t, ln, time.Second) + if err != nil { + t.Fatalf("accept first conn: %v", err) + } + + blockedClient, err := net.Dial("udp", addr) + if err != nil { + t.Fatalf("dial blocked udp client: %v", err) + } + defer blockedClient.Close() + if _, err := blockedClient.Write([]byte("blocked")); err != nil { + t.Fatalf("write blocked packet: %v", err) + } + blocked, err := acceptWithTimeout(t, ln, time.Second) + if err != nil { + t.Fatalf("expected blocked same-IP pseudo-connection to be returned closed: %v", err) + } + buf := make([]byte, 16) + if _, err := blocked.Read(buf); err == nil { + _ = blocked.Close() + t.Fatalf("expected blocked same-IP pseudo-connection to be closed") + } + _ = blocked.Close() + _ = first.Close() + + reopenedClient, err := net.Dial("udp", addr) + if err != nil { + t.Fatalf("dial reopened udp client: %v", err) + } + defer reopenedClient.Close() + if _, err := reopenedClient.Write([]byte("after-close")); err != nil { + t.Fatalf("write after close packet: %v", err) + } + reopened, err := acceptWithTimeout(t, ln, time.Second) + if err != nil { + t.Fatalf("expected same client to be accepted after close: %v", err) + } + _ = reopened.Close() +} + +func acceptWithTimeout(t *testing.T, ln corelistener.Listener, timeout time.Duration) (net.Conn, error) { + t.Helper() + type result struct { + conn net.Conn + err error + } + ch := make(chan result, 1) + go func() { + conn, err := ln.Accept() + ch <- result{conn: conn, err: err} + }() + select { + case res := <-ch: + return res.conn, res.err + case <-time.After(timeout): + return nil, net.ErrClosed + } +} From edfe2a2372d70fc14b0fc06a601bcb19522f2725 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 23:13:16 +0800 Subject: [PATCH 08/11] fix: preserve udp packet semantics with limiters --- go-gost/x/listener/udp/listener.go | 105 +++++++++++++++++++++--- go-gost/x/listener/udp/listener_test.go | 65 +++++++++++++++ 2 files changed, 158 insertions(+), 12 deletions(-) diff --git a/go-gost/x/listener/udp/listener.go b/go-gost/x/listener/udp/listener.go index 23aff31..3ac3119 100644 --- a/go-gost/x/listener/udp/listener.go +++ b/go-gost/x/listener/udp/listener.go @@ -2,15 +2,17 @@ package udp import ( "net" + "sync" + "time" "github.com/go-gost/core/limiter" + conn_limiter "github.com/go-gost/core/limiter/conn" "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/udp" - climiter "github.com/go-gost/x/limiter/conn/wrapper" traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -94,26 +96,77 @@ func (l *udpListener) Accept() (conn net.Conn, err error) { if lim := l.options.ConnLimiter.Limiter(host); lim != nil { if !lim.Allow(1) { _ = conn.Close() - return closedConn{Conn: conn}, nil + return newClosedConn(conn), nil } - conn = climiter.WrapConn(lim, conn) + conn = wrapConnLimiter(lim, conn) } } - conn = limiter_wrapper.WrapConn( - conn, - l.options.TrafficLimiter, - conn.RemoteAddr().String(), - limiter.ScopeOption(limiter.ScopeConn), - limiter.ServiceOption(l.options.Service), - limiter.NetworkOption(conn.LocalAddr().Network()), - limiter.SrcOption(conn.RemoteAddr().String()), - ) + if pc, ok := conn.(net.PacketConn); ok { + conn = limiter_wrapper.WrapUDPConn( + pc, + l.options.TrafficLimiter, + conn.RemoteAddr().String(), + limiter.ScopeOption(limiter.ScopeConn), + limiter.ServiceOption(l.options.Service), + limiter.NetworkOption(conn.LocalAddr().Network()), + limiter.SrcOption(conn.RemoteAddr().String()), + ) + } return } +type connLimiterConn struct { + net.Conn + net.PacketConn + limiter conn_limiter.Limiter + once sync.Once +} + +func wrapConnLimiter(limiter conn_limiter.Limiter, conn net.Conn) net.Conn { + pc, ok := conn.(net.PacketConn) + if !ok { + return conn + } + return &connLimiterConn{ + Conn: conn, + PacketConn: pc, + limiter: limiter, + } +} + +func (c *connLimiterConn) Close() (err error) { + c.once.Do(func() { + c.limiter.Allow(-1) + err = c.Conn.Close() + }) + return +} + +func (c *connLimiterConn) LocalAddr() net.Addr { + return c.Conn.LocalAddr() +} + +func (c *connLimiterConn) SetDeadline(t time.Time) error { + return c.Conn.SetDeadline(t) +} + +func (c *connLimiterConn) SetReadDeadline(t time.Time) error { + return c.Conn.SetReadDeadline(t) +} + +func (c *connLimiterConn) SetWriteDeadline(t time.Time) error { + return c.Conn.SetWriteDeadline(t) +} + type closedConn struct { net.Conn + net.PacketConn +} + +func newClosedConn(conn net.Conn) net.Conn { + pc, _ := conn.(net.PacketConn) + return closedConn{Conn: conn, PacketConn: pc} } func (c closedConn) Read([]byte) (int, error) { @@ -124,6 +177,34 @@ func (c closedConn) Write([]byte) (int, error) { return 0, net.ErrClosed } +func (c closedConn) ReadFrom([]byte) (int, net.Addr, error) { + return 0, nil, net.ErrClosed +} + +func (c closedConn) WriteTo([]byte, net.Addr) (int, error) { + return 0, net.ErrClosed +} + +func (c closedConn) Close() error { + return c.Conn.Close() +} + +func (c closedConn) LocalAddr() net.Addr { + return c.Conn.LocalAddr() +} + +func (c closedConn) SetDeadline(t time.Time) error { + return c.Conn.SetDeadline(t) +} + +func (c closedConn) SetReadDeadline(t time.Time) error { + return c.Conn.SetReadDeadline(t) +} + +func (c closedConn) SetWriteDeadline(t time.Time) error { + return c.Conn.SetWriteDeadline(t) +} + func (l *udpListener) Addr() net.Addr { return l.ln.Addr() } diff --git a/go-gost/x/listener/udp/listener_test.go b/go-gost/x/listener/udp/listener_test.go index 9b56847..504042a 100644 --- a/go-gost/x/listener/udp/listener_test.go +++ b/go-gost/x/listener/udp/listener_test.go @@ -9,9 +9,61 @@ import ( corelistener "github.com/go-gost/core/listener" corelogger "github.com/go-gost/core/logger" xconn "github.com/go-gost/x/limiter/conn" + xtraffic "github.com/go-gost/x/limiter/traffic" xlogger "github.com/go-gost/x/logger" ) +func TestAcceptWithLimitersPreservesPacketConn(t *testing.T) { + ln := NewListener( + corelistener.AddrOption("127.0.0.1:0"), + corelistener.ConnLimiterOption(xconn.NewConnLimiter( + xconn.LimitsOption("$$ 1"), + xconn.LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel))), + )), + corelistener.TrafficLimiterOption(xtraffic.NewTrafficLimiter( + xtraffic.LimitsOption("$$ 1024B 1024B"), + xtraffic.LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel))), + )), + corelistener.LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel))), + ) + if err := ln.Init(nil); err != nil { + t.Fatalf("init listener: %v", err) + } + defer ln.Close() + + client, err := net.Dial("udp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial udp listener: %v", err) + } + defer client.Close() + if _, err := client.Write([]byte("packet")); err != nil { + t.Fatalf("write packet: %v", err) + } + + conn, err := acceptWithTimeout(t, ln, time.Second) + if err != nil { + t.Fatalf("accept conn: %v", err) + } + defer conn.Close() + + packetConn, ok := conn.(net.PacketConn) + if !ok { + t.Fatalf("expected accepted UDP conn with limiters to implement net.PacketConn, got %T", conn) + } + + buf := make([]byte, 16) + n, addr, err := packetConn.ReadFrom(buf) + if err != nil { + t.Fatalf("read packet: %v", err) + } + if string(buf[:n]) != "packet" { + t.Fatalf("expected original datagram, got %q", string(buf[:n])) + } + if addr == nil || addr.String() != client.LocalAddr().String() { + t.Fatalf("expected client addr %v, got %v", client.LocalAddr(), addr) + } +} + func TestAcceptAppliesConnLimiterAndReleasesOnClose(t *testing.T) { ln := NewListener( corelistener.AddrOption("127.0.0.1:0"), @@ -57,6 +109,19 @@ func TestAcceptAppliesConnLimiterAndReleasesOnClose(t *testing.T) { _ = blocked.Close() t.Fatalf("expected blocked same-IP pseudo-connection to be closed") } + packetConn, ok := blocked.(net.PacketConn) + if !ok { + _ = blocked.Close() + t.Fatalf("expected blocked same-IP pseudo-connection to preserve net.PacketConn, got %T", blocked) + } + if _, _, err := packetConn.ReadFrom(buf); err == nil { + _ = blocked.Close() + t.Fatalf("expected blocked same-IP packet connection to be closed") + } + if _, err := packetConn.WriteTo([]byte("blocked"), client.LocalAddr()); err == nil { + _ = blocked.Close() + t.Fatalf("expected blocked same-IP packet write to be closed") + } _ = blocked.Close() _ = first.Close() From a5a500bc0fac9ab6b7fe2fb2ffddeef5b0685b80 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 23:17:02 +0800 Subject: [PATCH 09/11] feat: add per-IP limit controls --- vite-frontend/src/api/types.ts | 5 ++ vite-frontend/src/pages/forward.tsx | 89 +++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+) diff --git a/vite-frontend/src/api/types.ts b/vite-frontend/src/api/types.ts index cd3c5f0..1ff16cc 100644 --- a/vite-frontend/src/api/types.ts +++ b/vite-frontend/src/api/types.ts @@ -71,6 +71,9 @@ export interface ForwardApiItem { userId?: number; tunnelId?: number; speedId?: number | null; + ipMaxConn?: number; + ipSpeedId?: number | null; + ipSpeedLimitName?: string; maxConn?: number; proxyProtocol?: number; inx?: number; @@ -372,6 +375,8 @@ export interface ForwardMutationPayload { remoteAddr?: string; strategy?: string; speedId?: number | null; + ipMaxConn?: number; + ipSpeedId?: number | null; maxConn?: number; proxyProtocol?: number; } diff --git a/vite-frontend/src/pages/forward.tsx b/vite-frontend/src/pages/forward.tsx index 2e86b78..3d23627 100644 --- a/vite-frontend/src/pages/forward.tsx +++ b/vite-frontend/src/pages/forward.tsx @@ -125,6 +125,9 @@ interface Forward { userId?: number; inx?: number; speedId?: number | null; + ipMaxConn?: number; + ipSpeedId?: number | null; + ipSpeedLimitName?: string; proxyProtocol?: number; } @@ -160,6 +163,8 @@ interface ForwardForm { interfaceName?: string; strategy: string; speedId: number | null; + ipMaxConn?: number; + ipSpeedId: number | null; maxConn?: number; proxyProtocol?: number; } @@ -578,6 +583,16 @@ const mapForwardApiItems = (items: ForwardApiItem[]): Forward[] => { typeof forward.speedId === "number" || forward.speedId === null ? forward.speedId : undefined, + ipMaxConn: + typeof forward.ipMaxConn === "number" ? forward.ipMaxConn : undefined, + ipSpeedId: + typeof forward.ipSpeedId === "number" || forward.ipSpeedId === null + ? forward.ipSpeedId + : undefined, + ipSpeedLimitName: + typeof forward.ipSpeedLimitName === "string" + ? forward.ipSpeedLimitName + : undefined, maxConn: typeof forward.maxConn === "number" ? forward.maxConn : undefined, proxyProtocol: typeof forward.proxyProtocol === "number" @@ -1316,6 +1331,8 @@ export default function ForwardPage() { interfaceName: "", strategy: "fifo", speedId: null, + ipMaxConn: 0, + ipSpeedId: null, maxConn: 0, proxyProtocol: 0, }); @@ -2030,6 +2047,7 @@ export default function ForwardPage() { }; const selectedSpeedId = normalizeSpeedId(form.speedId); + const selectedIPSpeedId = normalizeSpeedId(form.ipSpeedId); const validateForm = (): boolean => { const newErrors: { [key: string]: string } = {}; @@ -2105,6 +2123,8 @@ export default function ForwardPage() { interfaceName: "", strategy: "fifo", speedId: null, + ipMaxConn: 0, + ipSpeedId: null, proxyProtocol: 0, }); setErrors({}); @@ -2126,6 +2146,8 @@ export default function ForwardPage() { interfaceName: forward.interfaceName || "", strategy: forward.strategy || "fifo", speedId: normalizeSpeedId(forward.speedId), + ipMaxConn: forward.ipMaxConn ?? 0, + ipSpeedId: normalizeSpeedId(forward.ipSpeedId), maxConn: forward.maxConn ?? 0, proxyProtocol: forward.proxyProtocol ?? 0, }); @@ -2245,6 +2267,8 @@ export default function ForwardPage() { let res: { code: number; msg: string }; const normalizedSpeedId = normalizeSpeedId(form.speedId); const speedLimitAutoCleared = isMissingSpeedLimit(form.speedId); + const normalizedIPSpeedId = normalizeSpeedId(form.ipSpeedId); + const ipSpeedLimitAutoCleared = isMissingSpeedLimit(form.ipSpeedId); if (isEdit) { const updateData = { @@ -2256,6 +2280,8 @@ export default function ForwardPage() { remoteAddr: processedRemoteAddr, strategy: addressCount > 1 ? form.strategy : "fifo", speedId: normalizedSpeedId, + ipMaxConn: form.ipMaxConn, + ipSpeedId: normalizedIPSpeedId, maxConn: form.maxConn, proxyProtocol: form.proxyProtocol, }; @@ -2270,6 +2296,8 @@ export default function ForwardPage() { remoteAddr: processedRemoteAddr, strategy: addressCount > 1 ? form.strategy : "fifo", speedId: normalizedSpeedId, + ipMaxConn: form.ipMaxConn, + ipSpeedId: normalizedIPSpeedId, maxConn: form.maxConn, proxyProtocol: form.proxyProtocol, }; @@ -2297,6 +2325,12 @@ export default function ForwardPage() { duration: 5000, }); } + if (ipSpeedLimitAutoCleared) { + toast("所选每 IP 限速规则不存在,已自动清除为不限速", { + icon: "⚠️", + duration: 5000, + }); + } toast.success(isEdit ? "修改成功" : "创建成功"); setModalOpen(false); await refreshForwardList(false); @@ -4911,6 +4945,27 @@ export default function ForwardPage() { setForm((prev) => ({ ...prev, maxConn: value })); }} /> + { + const value = Math.max( + Number(e.target.value) || 0, + 0, + ); + + setForm((prev) => ({ ...prev, ipMaxConn: value })); + }} + /> { + const selectedKey = Array.from(keys)[0] as + | string + | undefined; + + setForm((prev) => ({ + ...prev, + ipSpeedId: selectedKey + ? Number(selectedKey) + : null, + })); + }} + > + {availableSpeedLimits.map((speedLimit) => ( + + {speedLimit.name} + + ))} + + )} From bd27b949090dbd3992a30795c069d8b95beed3d2 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Mon, 27 Apr 2026 23:34:35 +0800 Subject: [PATCH 10/11] fix: omit per-IP speed payload for users --- vite-frontend/src/pages/forward.tsx | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/vite-frontend/src/pages/forward.tsx b/vite-frontend/src/pages/forward.tsx index 3d23627..3842f63 100644 --- a/vite-frontend/src/pages/forward.tsx +++ b/vite-frontend/src/pages/forward.tsx @@ -2281,7 +2281,7 @@ export default function ForwardPage() { strategy: addressCount > 1 ? form.strategy : "fifo", speedId: normalizedSpeedId, ipMaxConn: form.ipMaxConn, - ipSpeedId: normalizedIPSpeedId, + ...(isAdmin ? { ipSpeedId: normalizedIPSpeedId } : {}), maxConn: form.maxConn, proxyProtocol: form.proxyProtocol, }; @@ -2297,7 +2297,7 @@ export default function ForwardPage() { strategy: addressCount > 1 ? form.strategy : "fifo", speedId: normalizedSpeedId, ipMaxConn: form.ipMaxConn, - ipSpeedId: normalizedIPSpeedId, + ...(isAdmin ? { ipSpeedId: normalizedIPSpeedId } : {}), maxConn: form.maxConn, proxyProtocol: form.proxyProtocol, }; @@ -2325,7 +2325,7 @@ export default function ForwardPage() { duration: 5000, }); } - if (ipSpeedLimitAutoCleared) { + if (isAdmin && ipSpeedLimitAutoCleared) { toast("所选每 IP 限速规则不存在,已自动清除为不限速", { icon: "⚠️", duration: 5000, From 3373e5ade9bbff27e0902c07ee27fb67fc9198c9 Mon Sep 17 00:00:00 2001 From: sagitchu Date: Tue, 28 Apr 2026 00:18:32 +0800 Subject: [PATCH 11/11] fix: preserve shared limiters with per-IP rules --- .../internal/http/handler/control_plane.go | 81 ++++--- .../http/handler/control_plane_test.go | 28 ++- .../contract/max_conn_limit_contract_test.go | 138 ++++++++++++ .../per_ip_speed_limit_contract_test.go | 50 +++-- .../parsing/service/composite_limiter.go | 199 ++++++++++++++++++ .../parsing/service/composite_limiter_test.go | 79 +++++++ go-gost/x/config/parsing/service/parse.go | 19 +- 7 files changed, 511 insertions(+), 83 deletions(-) create mode 100644 go-gost/x/config/parsing/service/composite_limiter.go create mode 100644 go-gost/x/config/parsing/service/composite_limiter_test.go diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 3bb7b25..75af395 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -292,27 +292,13 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method if user != nil && user.MaxConn > 0 { userMaxConn = user.MaxConn } - connLimiterConfig := buildConnLimiterConfig(forward, userMaxConn) + connLimiterConfigs := buildConnLimiterConfigs(forward, userMaxConn) for _, fp := range ports { - runtimeLimiters := forwardRuntimeLimiters{ConnLimiter: connLimiterConfig.Name} - if ipSpeed != nil { - runtimeLimiters.TrafficLimiter = fmt.Sprintf("rule_traffic_limit_%d", forward.ID) - if err := h.ensureTrafficLimiterOnNode(fp.NodeID, runtimeLimiters.TrafficLimiter, speed, ipSpeed); err != nil { - // If the limiter push fails because the node is offline, skip it with a warning - if isNodeOfflineOrTimeoutError(err) { - node, _ := h.getNodeRecord(fp.NodeID) - nodeName := fmt.Sprintf("%d", fp.NodeID) - if node != nil && strings.TrimSpace(node.Name) != "" { - nodeName = strings.TrimSpace(node.Name) - } - warnings = append(warnings, fmt.Sprintf("节点 %s 不在线,已跳过下发", nodeName)) - continue - } - return nil, err - } - } else if limiterID != nil && speed != nil { - runtimeLimiters.TrafficLimiter = strconv.FormatInt(*limiterID, 10) + runtimeLimiters := forwardRuntimeLimiters{ConnLimiter: joinLimiterNames(connLimiterConfigs)} + trafficLimiterNames := make([]string, 0, 2) + if limiterID != nil && speed != nil { + totalLimiterName := strconv.FormatInt(*limiterID, 10) if err := h.ensureLimiterOnNode(fp.NodeID, *limiterID, *speed); err != nil { // If the limiter push fails because the node is offline, skip it with a warning if isNodeOfflineOrTimeoutError(err) { @@ -326,9 +312,28 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method } return nil, err } + trafficLimiterNames = append(trafficLimiterNames, totalLimiterName) } + if ipSpeed != nil { + ruleLimiterName := fmt.Sprintf("rule_traffic_limit_%d", forward.ID) + if err := h.ensureTrafficLimiterOnNode(fp.NodeID, ruleLimiterName, nil, ipSpeed); err != nil { + // If the limiter push fails because the node is offline, skip it with a warning + if isNodeOfflineOrTimeoutError(err) { + node, _ := h.getNodeRecord(fp.NodeID) + nodeName := fmt.Sprintf("%d", fp.NodeID) + if node != nil && strings.TrimSpace(node.Name) != "" { + nodeName = strings.TrimSpace(node.Name) + } + warnings = append(warnings, fmt.Sprintf("节点 %s 不在线,已跳过下发", nodeName)) + continue + } + return nil, err + } + trafficLimiterNames = append(trafficLimiterNames, ruleLimiterName) + } + runtimeLimiters.TrafficLimiter = strings.Join(trafficLimiterNames, ",") - if connLimiterConfig.Name != "" { + for _, connLimiterConfig := range connLimiterConfigs { if err := h.ensureConnLimiterOnNode(fp.NodeID, connLimiterConfig); err != nil { warnings = append(warnings, fmt.Sprintf("节点 %d 连接限制器下发失败: %v", fp.NodeID, err)) } @@ -1876,27 +1881,35 @@ func (h *Handler) ensureConnLimiterOnNode(nodeID int64, cfg forwardLimiterConfig return nil } -func buildConnLimiterConfig(forward *forwardRecord, userMaxConn int) forwardLimiterConfig { +func buildConnLimiterConfigs(forward *forwardRecord, userMaxConn int) []forwardLimiterConfig { if forward == nil { - return forwardLimiterConfig{} + return nil } - limits := make([]string, 0, 2) if forward.MaxConn > 0 { - limits = append(limits, fmt.Sprintf("$ %d", forward.MaxConn)) - } else if userMaxConn > 0 { - limits = append(limits, fmt.Sprintf("$ %d", userMaxConn)) + limits := []string{fmt.Sprintf("$ %d", forward.MaxConn)} + if forward.IPMaxConn > 0 { + limits = append(limits, fmt.Sprintf("$$ %d", forward.IPMaxConn)) + } + return []forwardLimiterConfig{{Name: fmt.Sprintf("rule_conn_limit_%d", forward.ID), Limits: limits}} + } + configs := make([]forwardLimiterConfig, 0, 2) + if userMaxConn > 0 { + configs = append(configs, forwardLimiterConfig{Name: fmt.Sprintf("user_conn_limit_%d", forward.UserID), Limits: []string{fmt.Sprintf("$ %d", userMaxConn)}}) } if forward.IPMaxConn > 0 { - limits = append(limits, fmt.Sprintf("$$ %d", forward.IPMaxConn)) + configs = append(configs, forwardLimiterConfig{Name: fmt.Sprintf("rule_conn_limit_%d", forward.ID), Limits: []string{fmt.Sprintf("$$ %d", forward.IPMaxConn)}}) } - if len(limits) == 0 { - return forwardLimiterConfig{} + return configs +} + +func joinLimiterNames(configs []forwardLimiterConfig) string { + names := make([]string, 0, len(configs)) + for _, cfg := range configs { + if cfg.Name != "" { + names = append(names, cfg.Name) + } } - name := fmt.Sprintf("user_conn_limit_%d", forward.UserID) - if forward.MaxConn > 0 || forward.IPMaxConn > 0 { - name = fmt.Sprintf("rule_conn_limit_%d", forward.ID) - } - return forwardLimiterConfig{Name: name, Limits: limits} + return strings.Join(names, ",") } func speedToLimitLine(key string, speed int) string { diff --git a/go-backend/internal/http/handler/control_plane_test.go b/go-backend/internal/http/handler/control_plane_test.go index c3fb79a..86c6433 100644 --- a/go-backend/internal/http/handler/control_plane_test.go +++ b/go-backend/internal/http/handler/control_plane_test.go @@ -479,24 +479,30 @@ func TestBuildForwardServiceConfigs_IPv6BindIP(t *testing.T) { } func TestBuildConnLimiterConfigCombinesTotalAndPerIP(t *testing.T) { - cfg := buildConnLimiterConfig(&forwardRecord{ID: 42, UserID: 9, MaxConn: 100, IPMaxConn: 5}, 37) - want := forwardLimiterConfig{Name: "rule_conn_limit_42", Limits: []string{"$ 100", "$$ 5"}} - if !reflect.DeepEqual(cfg, want) { - t.Fatalf("expected %+v, got %+v", want, cfg) + cfgs := buildConnLimiterConfigs(&forwardRecord{ID: 42, UserID: 9, MaxConn: 100, IPMaxConn: 5}, 37) + want := []forwardLimiterConfig{{Name: "rule_conn_limit_42", Limits: []string{"$ 100", "$$ 5"}}} + if !reflect.DeepEqual(cfgs, want) { + t.Fatalf("expected %+v, got %+v", want, cfgs) } } func TestBuildConnLimiterConfigUsesUserTotalWithRulePerIP(t *testing.T) { - cfg := buildConnLimiterConfig(&forwardRecord{ID: 42, UserID: 9, IPMaxConn: 5}, 37) - want := forwardLimiterConfig{Name: "rule_conn_limit_42", Limits: []string{"$ 37", "$$ 5"}} - if !reflect.DeepEqual(cfg, want) { - t.Fatalf("expected %+v, got %+v", want, cfg) + cfgs := buildConnLimiterConfigs(&forwardRecord{ID: 42, UserID: 9, IPMaxConn: 5}, 37) + want := []forwardLimiterConfig{ + {Name: "user_conn_limit_9", Limits: []string{"$ 37"}}, + {Name: "rule_conn_limit_42", Limits: []string{"$$ 5"}}, + } + if !reflect.DeepEqual(cfgs, want) { + t.Fatalf("expected %+v, got %+v", want, cfgs) + } + if got := joinLimiterNames(cfgs); got != "user_conn_limit_9,rule_conn_limit_42" { + t.Fatalf("expected composite limiter names, got %q", got) } } -func TestBuildTrafficLimiterPayloadCombinesTotalAndPerIP(t *testing.T) { - payload := buildTrafficLimiterPayload("rule_traffic_limit_42", intPtr(80), intPtr(40)) - wantLimits := []string{"$ 10.0MB 10.0MB", "0.0.0.0/0 5.0MB 5.0MB", "::/0 5.0MB 5.0MB"} +func TestBuildTrafficLimiterPayloadUsesOnlyPerIPRulesWhenTotalIsSeparate(t *testing.T) { + payload := buildTrafficLimiterPayload("rule_traffic_limit_42", nil, intPtr(40)) + wantLimits := []string{"0.0.0.0/0 5.0MB 5.0MB", "::/0 5.0MB 5.0MB"} if payload["name"] != "rule_traffic_limit_42" { t.Fatalf("expected name rule_traffic_limit_42, got %v", payload["name"]) } diff --git a/go-backend/tests/contract/max_conn_limit_contract_test.go b/go-backend/tests/contract/max_conn_limit_contract_test.go index b356be1..5e1176a 100644 --- a/go-backend/tests/contract/max_conn_limit_contract_test.go +++ b/go-backend/tests/contract/max_conn_limit_contract_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "reflect" "strings" "sync" "testing" @@ -366,6 +367,143 @@ func TestUserMaxConnUpdateResyncsExistingForwards(t *testing.T) { } } +func TestUserMaxConnWithPerIPRuleSplitsRuntimeLimiters(t *testing.T) { + secret := "contract-jwt-secret" + router, r := setupContractRouter(t, secret) + server := httptest.NewServer(router) + defer server.Close() + + userToken, err := auth.GenerateToken(3, "per_ip_user", 1, secret) + if err != nil { + t.Fatalf("generate user token: %v", err) + } + + now := time.Now().UnixMilli() + if err := r.DB().Exec(` + INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, max_conn, created_time, updated_time, status) + VALUES(3, 'per_ip_user', 'pwd', 1, ?, 99999, 0, 0, 1, 10, 37, ?, ?, 1) + `, now+365*24*3600*1000, now, now).Error; err != nil { + t.Fatalf("insert user: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) + VALUES(11, 'user-per-ip-conn-tunnel', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 0) + `, now, now).Error; err != nil { + t.Fatalf("insert tunnel: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO node(id, name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx) + VALUES(21, 'user-per-ip-conn-node', 'user-per-ip-conn-secret', '10.23.0.1', '10.23.0.1', '', '32300-32310', '', 'v1', 1, 1, 1, ?, ?, 1, '[::]', '[::]', 0) + `, now, now).Error; err != nil { + t.Fatalf("insert node: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) + VALUES(11, 1, 21, 32301, 'round', 1, 'tls') + `).Error; err != nil { + t.Fatalf("insert chain_tunnel: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO user_tunnel(id, user_id, tunnel_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) + VALUES(31, 3, 11, 10, 99999, 0, 0, 1, ?, 1) + `, now+365*24*3600*1000).Error; err != nil { + t.Fatalf("insert user_tunnel: %v", err) + } + + var commandMu sync.Mutex + receivedCommands := make([]string, 0) + addCLimitersData := make([]json.RawMessage, 0) + var updateServiceData json.RawMessage + + stopNode := startMockSessionForMaxConn(t, server.URL, "user-per-ip-conn-secret", func(cmdType string, data json.RawMessage) (bool, string) { + commandMu.Lock() + defer commandMu.Unlock() + receivedCommands = append(receivedCommands, cmdType) + if cmdType == "AddCLimiters" { + addCLimitersData = append(addCLimitersData, append([]byte(nil), data...)) + } + if cmdType == "UpdateService" { + updateServiceData = append([]byte(nil), data...) + } + return false, "" + }) + defer stopNode() + + waitNodeStatus(t, r, 21, 1) + + payload := map[string]interface{}{ + "name": "user-per-ip-conn-forward", + "tunnelId": int64(11), + "remoteAddr": "1.1.1.1:443", + "strategy": "fifo", + "ipMaxConn": 7, + } + body, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal payload: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(body)) + req.Header.Set("Authorization", userToken) + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + + var out response.R + if err := json.NewDecoder(res.Body).Decode(&out); err != nil { + t.Fatalf("decode response: %v", err) + } + if out.Code != 0 { + t.Fatalf("expected create success, got code=%d msg=%s", out.Code, out.Msg) + } + + var forwardID int64 + if err := r.DB().Raw("SELECT id FROM forward WHERE name = ?", "user-per-ip-conn-forward").Scan(&forwardID).Error; err != nil { + t.Fatalf("get forward ID: %v", err) + } + expectedRuleName := fmt.Sprintf("rule_conn_limit_%d", forwardID) + + commandMu.Lock() + defer commandMu.Unlock() + if len(addCLimitersData) != 2 { + t.Fatalf("expected two AddCLimiters commands. Received: %v", receivedCommands) + } + if updateServiceData == nil { + t.Fatalf("expected UpdateService. Received: %v", receivedCommands) + } + + gotLimits := make(map[string][]string) + for _, raw := range addCLimitersData { + var data map[string]interface{} + if err := json.Unmarshal(raw, &data); err != nil { + t.Fatalf("unmarshal AddCLimiters data: %v", err) + } + limits, ok := data["limits"].([]interface{}) + if !ok { + t.Fatalf("expected limits array, got %T", data["limits"]) + } + for _, limit := range limits { + gotLimits[fmt.Sprint(data["name"])] = append(gotLimits[fmt.Sprint(data["name"])], fmt.Sprint(limit)) + } + } + if !reflect.DeepEqual(gotLimits["user_conn_limit_3"], []string{"$ 37"}) { + t.Fatalf("expected user max limiter payload, got %v", gotLimits["user_conn_limit_3"]) + } + if !reflect.DeepEqual(gotLimits[expectedRuleName], []string{"$$ 7"}) { + t.Fatalf("expected rule per-IP limiter payload, got %v", gotLimits[expectedRuleName]) + } + + var services []map[string]interface{} + if err := json.Unmarshal(updateServiceData, &services); err != nil { + t.Fatalf("unmarshal UpdateService data: %v", err) + } + expectedCLimiter := "user_conn_limit_3," + expectedRuleName + for _, service := range services { + if service["climiter"] != expectedCLimiter { + t.Fatalf("expected service climiter %s, got %v", expectedCLimiter, service["climiter"]) + } + } +} + func startMockSessionForMaxConn(t *testing.T, baseURL string, nodeSecret string, onCommand func(cmdType string, data json.RawMessage) (bool, string)) func() { t.Helper() diff --git a/go-backend/tests/contract/per_ip_speed_limit_contract_test.go b/go-backend/tests/contract/per_ip_speed_limit_contract_test.go index faaa1ec..04f2314 100644 --- a/go-backend/tests/contract/per_ip_speed_limit_contract_test.go +++ b/go-backend/tests/contract/per_ip_speed_limit_contract_test.go @@ -73,7 +73,7 @@ func TestPerIPSpeedLimitRuntimePayload(t *testing.T) { var commandMu sync.Mutex receivedCommands := make([]string, 0) - var addLimitersData json.RawMessage + addLimitersData := make([]json.RawMessage, 0) var updateServiceData json.RawMessage stopNode := startMockSessionForMaxConn(t, server.URL, "per-ip-speed-secret", func(cmdType string, data json.RawMessage) (bool, string) { @@ -81,7 +81,7 @@ func TestPerIPSpeedLimitRuntimePayload(t *testing.T) { defer commandMu.Unlock() receivedCommands = append(receivedCommands, cmdType) if cmdType == "AddLimiters" { - addLimitersData = append([]byte(nil), data...) + addLimitersData = append(addLimitersData, append([]byte(nil), data...)) } if cmdType == "UpdateService" { updateServiceData = append([]byte(nil), data...) @@ -123,34 +123,41 @@ func TestPerIPSpeedLimitRuntimePayload(t *testing.T) { t.Fatalf("get forward ID: %v", err) } expectedName := fmt.Sprintf("rule_traffic_limit_%d", forwardID) - expectedLimits := []string{"$ 10.0MB 10.0MB", "0.0.0.0/0 5.0MB 5.0MB", "::/0 5.0MB 5.0MB"} + expectedTotalName := fmt.Sprint(totalSpeedID) + expectedRuleLimits := []string{"0.0.0.0/0 5.0MB 5.0MB", "::/0 5.0MB 5.0MB"} + expectedTotalLimits := []string{"$ 10.0MB 10.0MB"} commandMu.Lock() defer commandMu.Unlock() - if addLimitersData == nil { + if len(addLimitersData) != 2 { t.Fatalf("expected AddLimiters to be sent. Received: %v", receivedCommands) } if updateServiceData == nil { t.Fatalf("expected UpdateService to be sent. Received: %v", receivedCommands) } - var addData map[string]interface{} - if err := json.Unmarshal(addLimitersData, &addData); err != nil { - t.Fatalf("unmarshal AddLimiters data: %v", err) + gotLimiterLimits := make(map[string][]string) + for _, raw := range addLimitersData { + var addData map[string]interface{} + if err := json.Unmarshal(raw, &addData); err != nil { + t.Fatalf("unmarshal AddLimiters data: %v", err) + } + name := fmt.Sprint(addData["name"]) + limits, ok := addData["limits"].([]interface{}) + if !ok { + t.Fatalf("expected limits array, got %T", addData["limits"]) + } + gotLimits := make([]string, 0, len(limits)) + for _, limit := range limits { + gotLimits = append(gotLimits, fmt.Sprint(limit)) + } + gotLimiterLimits[name] = gotLimits } - if addData["name"] != expectedName { - t.Fatalf("expected limiter name %s, got %v", expectedName, addData["name"]) + if !reflect.DeepEqual(gotLimiterLimits[expectedTotalName], expectedTotalLimits) { + t.Fatalf("expected total limits %v, got %v", expectedTotalLimits, gotLimiterLimits[expectedTotalName]) } - limits, ok := addData["limits"].([]interface{}) - if !ok { - t.Fatalf("expected limits array, got %T", addData["limits"]) - } - gotLimits := make([]string, 0, len(limits)) - for _, limit := range limits { - gotLimits = append(gotLimits, fmt.Sprint(limit)) - } - if !reflect.DeepEqual(gotLimits, expectedLimits) { - t.Fatalf("expected limits %v, got %v", expectedLimits, gotLimits) + if !reflect.DeepEqual(gotLimiterLimits[expectedName], expectedRuleLimits) { + t.Fatalf("expected rule limits %v, got %v", expectedRuleLimits, gotLimiterLimits[expectedName]) } var services []map[string]interface{} @@ -161,8 +168,9 @@ func TestPerIPSpeedLimitRuntimePayload(t *testing.T) { t.Fatalf("expected services in UpdateService") } for _, service := range services { - if service["limiter"] != expectedName { - t.Fatalf("expected service limiter %s, got %v", expectedName, service["limiter"]) + expectedLimiter := expectedTotalName + "," + expectedName + if service["limiter"] != expectedLimiter { + t.Fatalf("expected service limiter %s, got %v", expectedLimiter, service["limiter"]) } } } diff --git a/go-gost/x/config/parsing/service/composite_limiter.go b/go-gost/x/config/parsing/service/composite_limiter.go new file mode 100644 index 0000000..e1304b1 --- /dev/null +++ b/go-gost/x/config/parsing/service/composite_limiter.go @@ -0,0 +1,199 @@ +package service + +import ( + "context" + "fmt" + "sort" + "strconv" + "strings" + + corelimiter "github.com/go-gost/core/limiter" + connlimiter "github.com/go-gost/core/limiter/conn" + trafficlimiter "github.com/go-gost/core/limiter/traffic" + xtraffic "github.com/go-gost/x/limiter/traffic" + "github.com/go-gost/x/registry" +) + +func resolveTrafficLimiter(names string) trafficlimiter.TrafficLimiter { + parts := splitLimiterNames(names) + if len(parts) == 0 { + return nil + } + if len(parts) == 1 { + return resolveSingleTrafficLimiter(parts[0]) + } + limiters := make([]trafficlimiter.TrafficLimiter, 0, len(parts)) + for _, part := range parts { + if lim := resolveSingleTrafficLimiter(part); lim != nil { + limiters = append(limiters, lim) + } + } + if len(limiters) == 0 { + return nil + } + if len(limiters) == 1 { + return limiters[0] + } + return &compositeTrafficLimiter{limiters: limiters} +} + +func resolveSingleTrafficLimiter(name string) trafficlimiter.TrafficLimiter { + lim := registry.TrafficLimiterRegistry().Get(name) + if lim != nil { + return lim + } + if val, err := strconv.Atoi(name); err == nil && val > 0 { + return xtraffic.NewTrafficLimiter( + xtraffic.LimitsOption(fmt.Sprintf("%s %dB %dB", xtraffic.ServiceLimitKey, val, val)), + ) + } + return xtraffic.NewTrafficLimiter( + xtraffic.LimitsOption(fmt.Sprintf("%s %s %s", xtraffic.ServiceLimitKey, name, name)), + ) +} + +func resolveConnLimiter(names string) connlimiter.ConnLimiter { + parts := splitLimiterNames(names) + if len(parts) == 0 { + return nil + } + if len(parts) == 1 { + return registry.ConnLimiterRegistry().Get(parts[0]) + } + limiters := make([]connlimiter.ConnLimiter, 0, len(parts)) + for _, part := range parts { + if lim := registry.ConnLimiterRegistry().Get(part); lim != nil { + limiters = append(limiters, lim) + } + } + if len(limiters) == 0 { + return nil + } + if len(limiters) == 1 { + return limiters[0] + } + return &compositeConnLimiter{limiters: limiters} +} + +func splitLimiterNames(names string) []string { + parts := strings.Split(names, ",") + out := make([]string, 0, len(parts)) + for _, part := range parts { + if part = strings.TrimSpace(part); part != "" { + out = append(out, part) + } + } + return out +} + +type compositeTrafficLimiter struct { + limiters []trafficlimiter.TrafficLimiter +} + +func (l *compositeTrafficLimiter) In(ctx context.Context, key string, opts ...corelimiter.Option) trafficlimiter.Limiter { + limiters := make([]trafficlimiter.Limiter, 0, len(l.limiters)) + for _, child := range l.limiters { + if lim := child.In(ctx, key, opts...); lim != nil { + limiters = append(limiters, lim) + } + } + return newCompositeTrafficChildLimiter(limiters) +} + +func (l *compositeTrafficLimiter) Out(ctx context.Context, key string, opts ...corelimiter.Option) trafficlimiter.Limiter { + limiters := make([]trafficlimiter.Limiter, 0, len(l.limiters)) + for _, child := range l.limiters { + if lim := child.Out(ctx, key, opts...); lim != nil { + limiters = append(limiters, lim) + } + } + return newCompositeTrafficChildLimiter(limiters) +} + +type compositeTrafficChildLimiter struct { + limiters []trafficlimiter.Limiter +} + +func newCompositeTrafficChildLimiter(limiters []trafficlimiter.Limiter) trafficlimiter.Limiter { + if len(limiters) == 0 { + return nil + } + if len(limiters) == 1 { + return limiters[0] + } + sort.Slice(limiters, func(i, j int) bool { + return limiters[i].Limit() < limiters[j].Limit() + }) + return &compositeTrafficChildLimiter{limiters: limiters} +} + +func (l *compositeTrafficChildLimiter) Wait(ctx context.Context, n int) int { + for _, lim := range l.limiters { + if v := lim.Wait(ctx, n); v < n { + n = v + } + } + return n +} + +func (l *compositeTrafficChildLimiter) Limit() int { + if len(l.limiters) == 0 { + return 0 + } + return l.limiters[0].Limit() +} + +func (l *compositeTrafficChildLimiter) Set(n int) {} + +type compositeConnLimiter struct { + limiters []connlimiter.ConnLimiter +} + +func (l *compositeConnLimiter) Limiter(key string) connlimiter.Limiter { + limiters := make([]connlimiter.Limiter, 0, len(l.limiters)) + for _, child := range l.limiters { + if lim := child.Limiter(key); lim != nil { + limiters = append(limiters, lim) + } + } + return newCompositeConnChildLimiter(limiters) +} + +type compositeConnChildLimiter struct { + limiters []connlimiter.Limiter +} + +func newCompositeConnChildLimiter(limiters []connlimiter.Limiter) connlimiter.Limiter { + if len(limiters) == 0 { + return nil + } + if len(limiters) == 1 { + return limiters[0] + } + sort.Slice(limiters, func(i, j int) bool { + return limiters[i].Limit() < limiters[j].Limit() + }) + return &compositeConnChildLimiter{limiters: limiters} +} + +func (l *compositeConnChildLimiter) Allow(n int) (allowed bool) { + var i int + for i = range l.limiters { + if allowed = l.limiters[i].Allow(n); !allowed { + break + } + } + if !allowed && i > 0 && n > 0 { + for _, lim := range l.limiters[:i] { + lim.Allow(-n) + } + } + return allowed +} + +func (l *compositeConnChildLimiter) Limit() int { + if len(l.limiters) == 0 { + return 0 + } + return l.limiters[0].Limit() +} diff --git a/go-gost/x/config/parsing/service/composite_limiter_test.go b/go-gost/x/config/parsing/service/composite_limiter_test.go new file mode 100644 index 0000000..2f7a2d4 --- /dev/null +++ b/go-gost/x/config/parsing/service/composite_limiter_test.go @@ -0,0 +1,79 @@ +package service + +import ( + "context" + "io" + "testing" + + corelimiter "github.com/go-gost/core/limiter" + corelogger "github.com/go-gost/core/logger" + xconn "github.com/go-gost/x/limiter/conn" + xtraffic "github.com/go-gost/x/limiter/traffic" + xlogger "github.com/go-gost/x/logger" + "github.com/go-gost/x/registry" +) + +func TestResolveTrafficLimiterComposesCommaSeparatedNames(t *testing.T) { + const totalName = "test_total_speed_composite" + const ruleName = "test_rule_speed_composite" + registry.TrafficLimiterRegistry().Unregister(totalName) + registry.TrafficLimiterRegistry().Unregister(ruleName) + defer registry.TrafficLimiterRegistry().Unregister(totalName) + defer registry.TrafficLimiterRegistry().Unregister(ruleName) + + logger := xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel)) + if err := registry.TrafficLimiterRegistry().Register(totalName, xtraffic.NewTrafficLimiter(xtraffic.LimitsOption("$ 10B 10B"), xtraffic.LoggerOption(logger))); err != nil { + t.Fatalf("register total limiter: %v", err) + } + if err := registry.TrafficLimiterRegistry().Register(ruleName, xtraffic.NewTrafficLimiter(xtraffic.LimitsOption("0.0.0.0/0 3B 3B"), xtraffic.LoggerOption(logger))); err != nil { + t.Fatalf("register rule limiter: %v", err) + } + + lim := resolveTrafficLimiter(totalName + "," + ruleName) + if lim == nil { + t.Fatalf("expected composite traffic limiter") + } + serviceLimiter := lim.In(context.Background(), "192.0.2.1:1000", corelimiter.ScopeOption(corelimiter.ScopeService)) + if serviceLimiter == nil || serviceLimiter.Limit() != 10 { + t.Fatalf("expected service-scope total limiter 10, got %#v", serviceLimiter) + } + connLimiter := lim.In(context.Background(), "192.0.2.1:1000", corelimiter.ScopeOption(corelimiter.ScopeConn)) + if connLimiter == nil || connLimiter.Limit() != 3 { + t.Fatalf("expected conn-scope per-IP limiter 3, got %#v", connLimiter) + } +} + +func TestResolveConnLimiterComposesCommaSeparatedNames(t *testing.T) { + const totalName = "test_total_conn_composite" + const ruleName = "test_rule_conn_composite" + registry.ConnLimiterRegistry().Unregister(totalName) + registry.ConnLimiterRegistry().Unregister(ruleName) + defer registry.ConnLimiterRegistry().Unregister(totalName) + defer registry.ConnLimiterRegistry().Unregister(ruleName) + + logger := xlogger.NewLogger(xlogger.OutputOption(io.Discard), xlogger.LevelOption(corelogger.ErrorLevel)) + if err := registry.ConnLimiterRegistry().Register(totalName, xconn.NewConnLimiter(xconn.LimitsOption("$ 2"), xconn.LoggerOption(logger))); err != nil { + t.Fatalf("register total conn limiter: %v", err) + } + if err := registry.ConnLimiterRegistry().Register(ruleName, xconn.NewConnLimiter(xconn.LimitsOption("$$ 1"), xconn.LoggerOption(logger))); err != nil { + t.Fatalf("register rule conn limiter: %v", err) + } + + lim := resolveConnLimiter(totalName + "," + ruleName) + if lim == nil { + t.Fatalf("expected composite conn limiter") + } + clientLimiter := lim.Limiter("192.0.2.1") + if clientLimiter == nil || clientLimiter.Limit() != 1 { + t.Fatalf("expected composite client limiter with strictest limit 1, got %#v", clientLimiter) + } + if !clientLimiter.Allow(1) { + t.Fatalf("expected first connection to be allowed") + } + if clientLimiter.Allow(1) { + t.Fatalf("expected per-IP rule limiter to reject second connection") + } + if !lim.Limiter("192.0.2.2").Allow(1) { + t.Fatalf("expected another client to share total limiter but have independent per-IP capacity") + } +} diff --git a/go-gost/x/config/parsing/service/parse.go b/go-gost/x/config/parsing/service/parse.go index 636419c..584398a 100644 --- a/go-gost/x/config/parsing/service/parse.go +++ b/go-gost/x/config/parsing/service/parse.go @@ -3,7 +3,6 @@ package service import ( "fmt" "runtime" - "strconv" "strings" "time" @@ -31,7 +30,6 @@ import ( logger_parser "github.com/go-gost/x/config/parsing/logger" selector_parser "github.com/go-gost/x/config/parsing/selector" tls_util "github.com/go-gost/x/internal/util/tls" - xtraffic "github.com/go-gost/x/limiter/traffic" cache_limiter "github.com/go-gost/x/limiter/traffic/cache" "github.com/go-gost/x/metadata" mdutil "github.com/go-gost/x/metadata/util" @@ -185,20 +183,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { var trafficLimiter listener.Option if cfg.Limiter != "" { - lim := registry.TrafficLimiterRegistry().Get(cfg.Limiter) - if lim == nil { - // Try to parse as simple number (bandwidth in bytes/sec) - if val, err := strconv.Atoi(cfg.Limiter); err == nil && val > 0 { - lim = xtraffic.NewTrafficLimiter( - xtraffic.LimitsOption(fmt.Sprintf("%s %dB %dB", xtraffic.ServiceLimitKey, val, val)), - ) - } - if lim == nil { - lim = xtraffic.NewTrafficLimiter( - xtraffic.LimitsOption(fmt.Sprintf("%s %s %s", xtraffic.ServiceLimitKey, cfg.Limiter, cfg.Limiter)), - ) - } - } + lim := resolveTrafficLimiter(cfg.Limiter) trafficLimiter = listener.TrafficLimiterOption( cache_limiter.NewCachedTrafficLimiter( lim, @@ -216,7 +201,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { listener.AuthOption(auth_parser.Info(cfg.Listener.Auth)), listener.TLSConfigOption(tlsConfig), listener.AdmissionOption(xadmission.AdmissionGroup(admissions...)), - listener.ConnLimiterOption(registry.ConnLimiterRegistry().Get(cfg.CLimiter)), + listener.ConnLimiterOption(resolveConnLimiter(cfg.CLimiter)), listener.ServiceOption(cfg.Name), listener.ProxyProtocolOption(ppv), listener.StatsOption(pStats),