dsh/meshtastic_mqtt_server
1
0 Fork 0
forked from kevin/meshtastic_mqtt_server
Code Pull Requests Activity

安全加固：MQTT broker 新增可选连接认证(bcrypt 用户+匿名开关+按 IP 失败限速，配置明文密码首启自动转哈希，默认关闭零影响)，py 迁移脚本数据库口令改环境变量(新增 db_config.example.py 模板)，新增 doc/SECURITY_FIX_TODO.md 安全修复清单，后端 v1.3.0

Browse Source
This commit is contained in:
kevin
2026-08-20 16:51:33 +08:00
parent fbd105bd6c
commit fe0cf036cf
11 changed files with 925 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+28
View File
@@ -144,6 +144,10 @@ export MESH_ADMIN_SESSION_SECRET='replace-with-a-long-random-string'
mqtt: mqtt:
host: 0.0.0.0 host: 0.0.0.0
port: 1883 port: 1883
auth:
enabled: false
allow_anonymous: false
users: []
tls: tls:
enabled: false enabled: false
cert_file: "" cert_file: ""
@@ -208,6 +212,30 @@ mkdir -p /srv/mesh_mqtt_go
-tls-key /path/to/server.key -tls-key /path/to/server.key
``` ```
## 启用 MQTT 连接认证
默认 `mqtt.auth.enabled: false`,所有客户端均可连接(与历史版本一致)。broker 暴露公网时建议开启认证：
```yaml
mqtt:
auth:
enabled: true
allow_anonymous: false # true 时允许空用户名/密码连接
users:
- username: mesh
password_hash: "$2y$10$..." # bcrypt,见下方生成方法
```
- `password` 支持直接写明文：首次载入时会自动转为 `password_hash` 并从配置文件中剔除明文；
- 也可以直接写 `password_hash`(bcrypt)。生成方法：
```bash
htpasswd -bnBC 10 "" '你的密码' | tr -d ':\n'
```
- 同一来源 IP 连续认证失败达到阈值(默认 5 次/分钟)会被临时封禁 5 分钟，防止在线爆破；
- 开启认证后，Meshtastic 节点/客户端需在 MQTT 上行配置中填写相同的用户名密码。
## 访问服务 ## 访问服务
启动后可访问： 启动后可访问：
doc/SECURITY_FIX_TODO.md
+111
View File
@@ -0,0 +1,111 @@
# 安全修复 TODO
依据 2026-08-20 安全审计(源码 + meshmap.lmve.net 线上实测)整理,按优先级排列。
完成一项勾选一项;每项含位置、修复方案、验收标准。
---
## P0 - 立即处理(已被暴露/可直接利用)
- [x] **T1 数据库口令治理(2026-08-20 完成,含一处误报修正)**
- ~~位置:`py/db_config.py` 两套 MySQL 明文口令已入库~~
- **修正:该文件从未进入 git 历史**(`py/.gitignore` 已忽略,`git log --all -- py/db_config.py` 为空),无需 filter-repo 清洗
- 已完成:`py/db_config.py` 与可入库模板 `py/db_config.example.py` 全部改为环境变量读取(`MESH_SOURCE_DB_PASSWORD` / `MESH_TARGET_DB_PASSWORD`),口令不再落盘
- **剩余(需手动)**:建议仍轮换两套 MySQL 口令(本地明文留存过,且出现在审计输出中):`ALTER USER ... IDENTIFIED BY '新口令';`
- 验收:口令文件不存在明文;脚本从环境变量读取 ✓
- [x] **T2 MQTT broker 认证(2026-08-20 代码完成,待部署启用)**
- 位置:`main.go`(原 AllowHook 零认证,1883 公网可达)
- 已完成:
1. 新增 `internal/mqttauth`:可配置用户(bcrypt)+ 匿名开关 + 按 IP 失败限速(默认 5 次/分钟,封禁 5 分钟)+ 未知用户名 dummy bcrypt 防时间侧信道
2. 配置段 `mqtt.auth`(enabled/allow_anonymous/users);配置中明文 `password` 首次载入自动转 `password_hash` 并从文件剔除
3. `auth.enabled: false` 默认,升级零影响;install.sh 模板与 README 已更新
4. 测试:单元 + 真实 broker/paho 端到端(有效凭据放行、错误拒绝、匿名拒绝、超限封禁)全部通过
- **剩余(需手动,服务器上执行)**:
1. `/etc/mesh_mqtt_go/config.yaml` 增加:`mqtt.auth.enabled: true` 与用户(哈希生成:`htpasswd -bnBC 10 "" '密码' | tr -d ':\n'`,或直接写明文 password 由首启转哈希)
2. 重启服务;Meshtastic 节点/客户端 MQTT 上行配置填入相同账号
3. 仍建议防火墙限制 1883 来源网关 IP(认证之外的纵深防御)
## P1 - 高优先(1-2 周内)
- [ ] **T3 install.sh 不再生成 admin/admin 默认口令**
- 位置:`install.sh:90-92`;`internal/config/config.go:184-199`
- 方案:首启随机生成 16 字节密码打印到终端(仅一次);或 web 绑定非 loopback 且口令为默认值时拒绝启动;`README.md:219-224` 删除默认口令描述
- 验收:新部署无法用 admin/admin 登录
- [ ] **T4 登录防爆破与用户名枚举**
- 位置:`internal/web/web.go:227-248`
- 方案:
1. 按 IP + 用户名维度限速(如 5 次/分钟,锁定 10 分钟,内存或 login_log 实现均可)
2. 用户不存在时也执行一次 `bcrypt.CompareHashAndPassword`(dummy hash)消除时间侧信道
- 验收:连续错误登录返回 429;存在/不存在用户名的响应耗时一致
- [ ] **T5 瓦片代理 SSRF 加固**
- 位置:`internal/web/map_tile_proxy_routes.go:35,79-87`
- 方案:
1. 自定义 `http.Transport.DialContext`:解析后拒绝 loopback/RFC1918/169.254.0.0/16/CGNAT/组播/IPv6 ULA 地址(防 DNS rebinding,必须在 connect 时按 IP 校验)
2. `CheckRedirect`:每跳重新校验目标,最多 2 跳
3. `writeMapTile`(198-202 行)Content-Type 白名单:仅 `image/*` 通过,否则强制 `application/octet-stream`(堵同源 HTML XSS)
- 验收:模板指向 `http://169.254.169.254/...` 及外网 302->内网均失败;上游返回 HTML 时浏览器下载而非渲染
- [ ] **T6 `/api/discard-details` 去敏**
- 位置:`internal/web/web.go:154-161,632-634`(线上实测匿名可读 MQTT 客户端 ID/IP/端口/raw_base64)
- 方案:公开响应剔除 `mqtt_remote_addr/host/port`、`raw_base64`;完整数据仅 `RequireAdmin` 分组提供(或直接整体移入 admin)
- 验收:匿名请求响应中无 IP 与原始报文字段
- [ ] **T7 高德地图 key 不再下发到前端**
- 位置:`internal/mapsource/admin_map_source_routes.go:36-47`(线上 `/api/map-source/enabled` 已泄露 `key=35206f...`)
- 方案:外部模板一律走服务端代理(存 hash 形式);`enabled` 接口只返回代理 URL 不返回原始 url_template
- 验收:`/api/map-source/enabled` 响应中无任何 `key=`/外部域名
## P2 - 中优先(迭代内)
- [ ] **T8 LLM 会话按 (bot, peer) 隔离**
- 位置:`internal/conversation/store.go:87-98`(`peerNodeID` 参数被忽略,所有 DM 对端共享上下文)
- 方案:`GetOrCreateForBot` 以 `(botID, peerNodeID)` 为键;历史消息数量设上限(如最近 50 条)
- 验收:不同 peer DM 得到独立会话;A 的注入不会影响 B 的回复
- [ ] **T9 LLM 入队限流/白名单**
- 位置:`internal/store/llm_store.go:545-578`;`internal/autoreply/service.go:29,178`
- 方案:按 from 节点维度限流(现仅有 bot 级 10 msg/5s);可选 allowlist 只响应已登记节点
- 验收:单一来源高频消息只消耗有限 LLM 调用
- [ ] **T10 瓦片磁盘缓存设上限**
- 位置:`internal/web/map_tile_proxy_routes.go:174-196`
- 方案:按 sourceHash 限制总字节数/文件数,超限 LRU 淘汰;每 IP 瓦片请求限速
- 验收:遍历坐标脚本无法使缓存目录超过配额
- [ ] **T11 sign 强制触发的污染治理**
- 位置:`internal/toolrouter/loop.go:116-161,221-283`
- 方案:同一 from 节点每日限 1 条(后端已有?核实);`/api/signs` 增加频率限制与管理员删除
- 验收:伪造大量 node_num 刷签到无法批量入库公开墙
- [ ] **T12 敏感数据落盘加密**
- 位置:`internal/store/db.go:216-224`(forwarder 密码)、`:263`(bot 私钥)、`:488-498`(LLM api_key)
- 方案:AES-GCM 加密存储,主密钥来自环境变量 `MESH_SECRET_KEY`;API 层维持现有脱敏
- 验收:直接读 SQLite 文件无法得到可用明文密钥
- [ ] **T13 admin 密码修改需验证自身当前密码 + session 可撤销**
- 位置:`internal/web/web.go:372-393`;`internal/auth/auth.go:89-149`
- 方案:改他人密码前要求请求方验证自己的密码;claims 增加 `pwd_ver`(密码 hash 版本号),改密后旧 cookie 全部失效
- 验收:改密后所有已登录会话返回 401
## P3 - 低优先(择机)
- [ ] **T14 解密私聊默认不打控制台日志** - `internal/config/config.go:204-210` 将 `console_log.meshtastic` 默认改 false,或至少对 `text_message` 且 DM 来源脱敏(`main.go:239-241`)
- [ ] **T15 session cookie `Secure: true`** - 生产部署 HTTPS 下设置 `session_secure: true`(`install.sh:94` / config 默认值);确认 nginx 强制 HTTP->HTTPS 跳转
- [ ] **T16 config.yaml 回写权限** - `internal/config/config.go:573-582` `Write` 改 0600,避免明文密码 0644 可读
- [ ] **T17 公开接口错误信息脱敏** - `/api/health` 等公开路由将 `err.Error()` 映射为固定文案(`webutil.go:178,191`)
- [ ] **T18 bot PSK 不回显** - `internal/bot/admin_bot_routes.go:308` 改为 `psk_set` 布尔,与 forwarder 路由风格一致
- [ ] **T19 前端 help 页防御性消毒** - `meshmap_frontend/src/components/HelpPage.vue:38` 的 `v-html` 前增加 DOMPurify(纵深防御,当前依赖服务端 bluemonday)
---
## 已确认无需修复(复核时勿重复排查)
- **nodeinfo 公钥替换影响 bot DM 加密**:Meshtastic 协议特性,公钥本就经由频道 PSK 加密的 nodeinfo 分发、无签名认证,官方 broker 与官方客户端(仅本地 key pinning + 变更告警)行为一致,不做服务端拦截
- SQL 注入:store 层全部参数化,无 LIKE/ORDER BY 拼接
- 前端 XSS:消息与节点名均经 Vue 转义,Leaflet popup 手工转义完整
- 静态服务/瓦片缓存路径穿越:不存在
- protobuf/PKI 解析 panic:未发现;calculator 工具为 AST 白名单,无 exec/文件访问
- 私钥/API key/forwarder 密码的 API DTO 脱敏与日志:现状正确
install.sh
+10
View File
@@ -69,6 +69,16 @@ if [[ ! -f "${CONFIG_DIR}/config.yaml" ]]; then
mqtt: mqtt:
host: 0.0.0.0 host: 0.0.0.0
port: 1883 port: 1883
# MQTT 连接认证:enabled 改为 true 后,客户端必须携带 users 中的账号连接
# (或 allow_anonymous: true 放行匿名)。生成哈希:
# htpasswd -bnBC 10 "" '你的密码' | tr -d ':\n'
auth:
enabled: false
allow_anonymous: false
users: []
# users:
# - username: mesh
# password_hash: "\$2y\$10\$..."
tls: tls:
enabled: false enabled: false
cert_file: "" cert_file: ""
internal/config/config.go
+144 -6
View File
@@ -6,7 +6,10 @@ import (
"os" "os"
"path/filepath" "path/filepath"
"runtime" "runtime"
"strconv"
"strings"
"golang.org/x/crypto/bcrypt"
"gopkg.in/yaml.v3" "gopkg.in/yaml.v3"
) )
@@ -28,9 +31,27 @@ type Config struct {
} }
type MQTTConfig struct { type MQTTConfig struct {
Host string `yaml:"host"` Host string `yaml:"host"`
Port int `yaml:"port"` Port int `yaml:"port"`
TLS TLSConfig `yaml:"tls"` TLS TLSConfig `yaml:"tls"`
Auth MQTTAuthConfig `yaml:"auth"`
}
// MQTTAuthConfig 控制 MQTT broker 的 CONNECT 认证。
// Enabled=false 时行为与历史版本一致(全部放行)。
type MQTTAuthConfig struct {
Enabled bool `yaml:"enabled"`
AllowAnonymous bool `yaml:"allow_anonymous"`
Users []MQTTAuthUser `yaml:"users"`
}
// MQTTAuthUser 是一个可连接 broker 的账号。
// Password 仅支持写在配置里由首次载入时自动转为 PasswordHash,
// 序列化写回时剔除明文;也可直接提供 password_hash(bcrypt)。
type MQTTAuthUser struct {
Username string `yaml:"username"`
Password string `yaml:"-"`
PasswordHash string `yaml:"password_hash,omitempty"`
} }
type TLSConfig struct { type TLSConfig struct {
@@ -113,9 +134,22 @@ type rawAIConfig struct {
} }
type rawMQTTConfig struct { type rawMQTTConfig struct {
Host *string `yaml:"host"` Host *string `yaml:"host"`
Port *int `yaml:"port"` Port *int `yaml:"port"`
TLS *rawTLSConfig `yaml:"tls"` TLS *rawTLSConfig `yaml:"tls"`
Auth *rawMQTTAuthConfig `yaml:"auth"`
}
type rawMQTTAuthConfig struct {
Enabled *bool `yaml:"enabled"`
AllowAnonymous *bool `yaml:"allow_anonymous"`
Users *[]rawMQTTAuthUser `yaml:"users"`
}
type rawMQTTAuthUser struct {
Username *string `yaml:"username"`
Password *string `yaml:"password"`
PasswordHash *string `yaml:"password_hash"`
} }
type rawTLSConfig struct { type rawTLSConfig struct {
@@ -172,6 +206,11 @@ func Default() *Config {
CertFile: "", CertFile: "",
KeyFile: "", KeyFile: "",
}, },
Auth: MQTTAuthConfig{
Enabled: false,
AllowAnonymous: false,
Users: []MQTTAuthUser{},
},
}, },
Meshtastic: MeshtasticConfig{ Meshtastic: MeshtasticConfig{
PSK: "AQ==", PSK: "AQ==",
@@ -377,6 +416,45 @@ func normalize(raw rawConfig) (*Config, bool) {
cfg.MQTT.TLS.KeyFile = *raw.MQTT.TLS.KeyFile cfg.MQTT.TLS.KeyFile = *raw.MQTT.TLS.KeyFile
} }
} }
if raw.MQTT.Auth == nil {
changed = true
} else {
if raw.MQTT.Auth.Enabled == nil {
changed = true
} else {
cfg.MQTT.Auth.Enabled = *raw.MQTT.Auth.Enabled
}
if raw.MQTT.Auth.AllowAnonymous == nil {
changed = true
} else {
cfg.MQTT.Auth.AllowAnonymous = *raw.MQTT.Auth.AllowAnonymous
}
if raw.MQTT.Auth.Users == nil {
changed = true
} else {
users := make([]MQTTAuthUser, 0, len(*raw.MQTT.Auth.Users))
for _, ru := range *raw.MQTT.Auth.Users {
u := MQTTAuthUser{}
if ru.Username == nil {
changed = true
} else {
u.Username = *ru.Username
}
if ru.Password == nil {
changed = true
} else {
u.Password = *ru.Password
}
if ru.PasswordHash == nil {
changed = true
} else {
u.PasswordHash = *ru.PasswordHash
}
users = append(users, u)
}
cfg.MQTT.Auth.Users = users
}
}
} }
if raw.Meshtastic == nil { if raw.Meshtastic == nil {
@@ -525,6 +603,18 @@ func normalize(raw rawConfig) (*Config, bool) {
} }
} }
// 明文 password 自动转为 bcrypt 哈希,并标记 changed 以便写回时剔除明文。
for i := range cfg.MQTT.Auth.Users {
if cfg.MQTT.Auth.Users[i].Password != "" {
if hashed, err := bcrypt.GenerateFromPassword([]byte(cfg.MQTT.Auth.Users[i].Password), bcrypt.DefaultCost); err == nil {
cfg.MQTT.Auth.Users[i].PasswordHash = string(hashed)
cfg.MQTT.Auth.Users[i].Password = ""
changed = true
}
// 散列失败(如密码超过 72 字节)时保留明文,交给 Validate 报错。
}
}
return cfg, changed return cfg, changed
} }
@@ -532,6 +622,9 @@ func Validate(cfg *Config) error {
if cfg.MQTT.Port <= 0 || cfg.MQTT.Port > 65535 { if cfg.MQTT.Port <= 0 || cfg.MQTT.Port > 65535 {
return fmt.Errorf("invalid mqtt port %d: must be 1-65535", cfg.MQTT.Port) return fmt.Errorf("invalid mqtt port %d: must be 1-65535", cfg.MQTT.Port)
} }
if err := validateMQTTAuth(cfg.MQTT.Auth); err != nil {
return err
}
switch cfg.Database.Driver { switch cfg.Database.Driver {
case DriverSQLite: case DriverSQLite:
if cfg.Database.SQLite.Path == "" { if cfg.Database.SQLite.Path == "" {
@@ -570,6 +663,51 @@ func Validate(cfg *Config) error {
return nil return nil
} }
func validateMQTTAuth(auth MQTTAuthConfig) error {
seen := make(map[string]bool, len(auth.Users))
for _, u := range auth.Users {
if u.Username == "" {
return fmt.Errorf("mqtt.auth.users[].username is required")
}
if seen[u.Username] {
return fmt.Errorf("mqtt.auth.users: duplicate username %q", u.Username)
}
seen[u.Username] = true
if u.Password != "" {
return fmt.Errorf("mqtt.auth.users[%s]: password 无法转为哈希(长度须 <= 72 字节),或直接改用 password_hash", u.Username)
}
if u.PasswordHash != "" && !isBcryptHash(u.PasswordHash) {
return fmt.Errorf("mqtt.auth.users[%s]: password_hash 不是合法的 bcrypt 散列($2a$/$2b$/$2y$ 开头),可用 htpasswd -bnBC 10 \"\" '密码' 生成", u.Username)
}
}
if auth.Enabled {
if !auth.AllowAnonymous && len(auth.Users) == 0 {
return fmt.Errorf("mqtt.auth.enabled 为 true 时必须配置至少一个用户,或设置 allow_anonymous: true")
}
for _, u := range auth.Users {
if u.PasswordHash == "" {
return fmt.Errorf("mqtt.auth.users[%s]: 启用认证时必须提供 password 或 password_hash", u.Username)
}
}
}
return nil
}
// isBcryptHash 校验 $2a$/$2b$/$2y$<cost>$<53位散列> 的 bcrypt 格式。
func isBcryptHash(s string) bool {
parts := strings.Split(s, "$")
if len(parts) != 4 || parts[0] != "" {
return false
}
if parts[1] != "2a" && parts[1] != "2b" && parts[1] != "2y" {
return false
}
if _, err := strconv.Atoi(parts[2]); err != nil {
return false
}
return len(parts[3]) == 53
}
func Write(path string, cfg *Config) error { func Write(path string, cfg *Config) error {
data, err := yaml.Marshal(cfg) data, err := yaml.Marshal(cfg)
if err != nil { if err != nil {
internal/config/config_test.go
+84
View File
@@ -0,0 +1,84 @@
package config
import (
"strings"
"testing"
"gopkg.in/yaml.v3"
)
func loadRaw(t *testing.T, data string) rawConfig {
t.Helper()
var raw rawConfig
if err := yaml.Unmarshal([]byte(data), &raw); err != nil {
t.Fatalf("yaml: %v", err)
}
return raw
}
func TestNormalizePlaintextPasswordBecomesHash(t *testing.T) {
raw := loadRaw(t, `
mqtt:
auth:
enabled: true
users:
- username: mesh
password: secret
`)
cfg, changed := normalize(raw)
if !changed {
t.Fatal("plaintext password must mark config changed")
}
if err := Validate(cfg); err != nil {
t.Fatalf("validate: %v", err)
}
u := cfg.MQTT.Auth.Users[0]
if u.Password != "" {
t.Error("plaintext must be cleared after hashing")
}
if !strings.HasPrefix(u.PasswordHash, "$2") {
t.Errorf("expected bcrypt hash, got %q", u.PasswordHash)
}
out, err := yaml.Marshal(cfg)
if err != nil {
t.Fatalf("marshal: %v", err)
}
if strings.Contains(string(out), "password: secret") {
t.Error("serialized config must not contain the plaintext password")
}
if !strings.Contains(string(out), "password_hash") {
t.Error("serialized config must contain password_hash")
}
}
func TestValidateAuthErrors(t *testing.T) {
cases := []struct {
name string
yaml string
want string
}{
{"启用但无用户", "mqtt:\n auth:\n enabled: true\n", "至少一个用户"},
{"坏哈希", "mqtt:\n auth:\n enabled: true\n users:\n - username: a\n password_hash: not-bcrypt\n", "bcrypt"},
{"缺哈希", "mqtt:\n auth:\n enabled: true\n users:\n - username: a\n", "password_hash"},
{"重复用户", "mqtt:\n auth:\n enabled: true\n users:\n - username: a\n password_hash: " + fakeHash + "\n - username: a\n password_hash: " + fakeHash + "\n", "duplicate"},
}
for _, tc := range cases {
cfg, _ := normalize(loadRaw(t, tc.yaml))
err := Validate(cfg)
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Errorf("%s: got %v, want error containing %q", tc.name, err, tc.want)
}
}
}
func TestValidateAuthDisabledOk(t *testing.T) {
cfg, _ := normalize(loadRaw(t, "mqtt:\n"))
if err := Validate(cfg); err != nil {
t.Fatalf("disabled auth must pass: %v", err)
}
if cfg.MQTT.Auth.Enabled {
t.Error("auth must default to disabled")
}
}
const fakeHash = "$2a$10$N9qo8uLOickgx2ZMRZoMyeIjZAgcfl7p92ldGxad68LJZdL17lhWy"
internal/mqttauth/integration_test.go
+69
View File
@@ -0,0 +1,69 @@
package mqttauth
import (
"testing"
"time"
paho "github.com/eclipse/paho.mqtt.golang"
mqtt "github.com/mochi-mqtt/server/v2"
"github.com/mochi-mqtt/server/v2/listeners"
)
// TestBrokerIntegration 用真实 TCP broker + paho 客户端验证认证全链路。
func TestBrokerIntegration(t *testing.T) {
hook := NewHook(Config{
Enabled: true,
Users: []User{{Username: "mesh", PasswordHash: hash(t, "secret")}},
MaxFailures: 3,
})
t.Cleanup(func() { _ = hook.Stop() })
server := mqtt.New(&mqtt.Options{InlineClient: true})
if err := server.AddHook(hook, nil); err != nil {
t.Fatalf("add hook: %v", err)
}
addr := "127.0.0.1:18883"
if err := server.AddListener(listeners.NewTCP(listeners.Config{ID: "t", Address: addr})); err != nil {
t.Fatalf("listen: %v", err)
}
if err := server.Serve(); err != nil {
t.Fatalf("serve: %v", err)
}
t.Cleanup(func() { _ = server.Close() })
time.Sleep(200 * time.Millisecond)
connect := func(user, pass string) bool {
opts := paho.NewClientOptions().
AddBroker("tcp://" + addr).
SetClientID("smoke-" + user + time.Now().Format("150405.000")).
SetUsername(user).SetPassword(pass).
SetConnectTimeout(3 * time.Second)
client := paho.NewClient(opts)
token := client.Connect()
token.WaitTimeout(5 * time.Second)
ok := token.Error() == nil
if ok {
client.Disconnect(50)
}
return ok
}
if !connect("mesh", "secret") {
t.Error("valid credentials should connect")
}
if connect("mesh", "wrong") {
t.Error("wrong password should be rejected")
}
if connect("", "") {
t.Error("anonymous should be rejected when disabled")
}
// 连续失败触发封禁后,正确凭据也应被拒绝(默认阈值 3,本例 MaxFailures=3)。
// 前面已失败 2 次(mesh/wrong 与匿名),再失败 1 次触发封禁。
if connect("mesh", "wrong") {
t.Fatal("wrong password should be rejected")
}
if connect("mesh", "secret") {
t.Error("blocked ip must reject even valid credentials")
}
}
internal/mqttauth/mqttauth.go
+265
View File
@@ -0,0 +1,265 @@
// Package mqttauth 为 mochi-mqtt broker 提供基于配置用户的 CONNECT 认证。
//
// 设计要点:
// - Enabled=false 时全部放行,行为与原 mqttauth.AllowHook 完全一致(平滑升级);
// - Enabled=true 时按用户名 + bcrypt 哈希校验,可选允许匿名连接;
// - 未知用户名也执行一次 dummy bcrypt 比较,消除用户名枚举时间侧信道;
// - 同一来源 IP 认证失败达到阈值后临时封禁,防止在线爆破;
// - OnACLCheck 恒返回 true:mochi 在无任何 ACL provider 时默认拒绝,
// 本 hook 不做主题级权限,保持原有全 topic 可读写行为。
package mqttauth
import (
"bytes"
"fmt"
"net"
"sync"
"time"
mqtt "github.com/mochi-mqtt/server/v2"
"github.com/mochi-mqtt/server/v2/packets"
"golang.org/x/crypto/bcrypt"
)
// User 是一个可连接 broker 的账号,PasswordHash 为 bcrypt 散列。
type User struct {
Username string
PasswordHash string
}
// Config 控制 hook 行为。
type Config struct {
Enabled bool
AllowAnonymous bool
Users []User
// MaxFailures 为单个 IP 在 Window 时间窗内允许的连续认证失败次数,
// 达到后封锁该 IP BlockFor 时长。零值使用默认。
MaxFailures int
Window time.Duration
BlockFor time.Duration
// LogEvent 用于输出结构化事件(传入 main 包的 printJSON),可为 nil。
LogEvent func(record map[string]any)
}
const (
defaultMaxFailures = 5
defaultWindow = time.Minute
defaultBlockFor = 5 * time.Minute
// dummyBcryptHash 是固定散列,用于未知用户名的耗时对齐。
dummyBcryptHash = "$2a$10$N9qo8uLOickgx2ZMRZoMyeIjZAgcfl7p92ldGxad68LJZdL17lhWy"
)
// Hook 实现 mochi 认证钩子。
type Hook struct {
mqtt.HookBase
cfg Config
users map[string]string
mu sync.Mutex
fails map[string]*failState
now func() time.Time
stopped chan struct{}
stopOnce sync.Once
}
type failState struct {
count int
windowStart time.Time
blockedUntil time.Time
}
// NewHook 按配置构造 hook 并启动后台清理协程,返回的 hook 交给 server.AddHook。
// 服务退出时应调用 Stop 结束清理协程。
func NewHook(cfg Config) *Hook {
if cfg.MaxFailures <= 0 {
cfg.MaxFailures = defaultMaxFailures
}
if cfg.Window <= 0 {
cfg.Window = defaultWindow
}
if cfg.BlockFor <= 0 {
cfg.BlockFor = defaultBlockFor
}
users := make(map[string]string, len(cfg.Users))
for _, u := range cfg.Users {
users[u.Username] = u.PasswordHash
}
h := &Hook{
cfg: cfg,
users: users,
fails: make(map[string]*failState),
now: time.Now,
stopped: make(chan struct{}),
}
go h.cleanupLoop()
return h
}
// Stop 结束后台清理协程;实现 mochi Hook 接口的 Stop。
func (h *Hook) Stop() error {
h.stopOnce.Do(func() { close(h.stopped) })
return nil
}
func (h *Hook) cleanupLoop() {
ticker := time.NewTicker(time.Minute)
defer ticker.Stop()
for {
select {
case <-h.stopped:
return
case <-ticker.C:
h.purgeExpired()
}
}
}
func (h *Hook) purgeExpired() {
now := h.now()
h.mu.Lock()
defer h.mu.Unlock()
for ip, st := range h.fails {
if now.After(st.blockedUntil) && now.Sub(st.windowStart) > h.cfg.Window {
delete(h.fails, ip)
}
}
}
// ID 返回 hook 标识。
func (h *Hook) ID() string { return "mqttauth" }
// Provides 声明处理认证与 ACL 检查。
func (h *Hook) Provides(b byte) bool {
return bytes.Contains([]byte{
mqtt.OnConnectAuthenticate,
mqtt.OnACLCheck,
}, []byte{b})
}
// OnACLCheck 恒允许:本 hook 只做连接级认证,不做主题级权限。
func (h *Hook) OnACLCheck(cl *mqtt.Client, topic string, write bool) bool {
return true
}
// OnConnectAuthenticate 校验 CONNECT 报文中的用户名/密码。
func (h *Hook) OnConnectAuthenticate(cl *mqtt.Client, pk packets.Packet) bool {
if !h.cfg.Enabled {
return true
}
ip := remoteHost(cl)
if reason, ok := h.checkBlocked(ip); !ok {
h.logEvent(map[string]any{
"event": "mqtt_auth_rejected", "reason": reason,
"client_id": cl.ID, "username": string(pk.Connect.Username), "remote_addr": cl.Net.Remote,
})
return false
}
username := string(pk.Connect.Username)
password := string(pk.Connect.Password)
allowed := h.authenticate(username, password)
if allowed {
h.resetFailures(ip)
return true
}
blockedNow := h.recordFailure(ip, username)
event := map[string]any{
"event": "mqtt_auth_rejected", "reason": "invalid credentials",
"client_id": cl.ID, "username": username, "remote_addr": cl.Net.Remote,
}
if blockedNow {
event["reason"] = "invalid credentials; ip now blocked"
event["blocked_for"] = h.cfg.BlockFor.String()
}
h.logEvent(event)
return false
}
// authenticate 给出凭据判定;抽离以便单元测试。
// 未知用户名也执行 dummy bcrypt,使两种失败路径耗时一致。
func (h *Hook) authenticate(username, password string) bool {
if !h.cfg.Enabled {
return true
}
if username == "" && password == "" {
return h.cfg.AllowAnonymous
}
if hash, found := h.users[username]; found {
return bcrypt.CompareHashAndPassword([]byte(hash), []byte(password)) == nil
}
_ = bcrypt.CompareHashAndPassword([]byte(dummyBcryptHash), []byte(password))
return false
}
func (h *Hook) checkBlocked(ip string) (string, bool) {
now := h.now()
h.mu.Lock()
defer h.mu.Unlock()
st, ok := h.fails[ip]
if ok && now.Before(st.blockedUntil) {
return fmt.Sprintf("ip blocked, retry after %s", time.Until(st.blockedUntil).Round(time.Second)), false
}
return "", true
}
// recordFailure 记录一次失败;达到阈值时返回 true 表示本次触发封禁。
func (h *Hook) recordFailure(ip, username string) bool {
now := h.now()
h.mu.Lock()
defer h.mu.Unlock()
st, ok := h.fails[ip]
if !ok || now.Sub(st.windowStart) > h.cfg.Window {
st = &failState{windowStart: now}
h.fails[ip] = st
if len(h.fails) > 4096 {
h.purgeLocked(now)
}
}
st.count++
if st.count >= h.cfg.MaxFailures {
st.blockedUntil = now.Add(h.cfg.BlockFor)
st.count = 0
st.windowStart = now
return true
}
return false
}
func (h *Hook) resetFailures(ip string) {
h.mu.Lock()
defer h.mu.Unlock()
delete(h.fails, ip)
}
func (h *Hook) purgeLocked(now time.Time) {
for ip, st := range h.fails {
if now.After(st.blockedUntil) && now.Sub(st.windowStart) > h.cfg.Window {
delete(h.fails, ip)
}
}
}
func (h *Hook) logEvent(record map[string]any) {
if h.cfg.LogEvent != nil {
h.cfg.LogEvent(record)
}
}
// remoteHost 提取客户端 IP,供失败限流使用。
func remoteHost(cl *mqtt.Client) string {
if cl == nil {
return "unknown"
}
remote := cl.Net.Remote
if remote == "" && cl.Net.Conn != nil && cl.Net.Conn.RemoteAddr() != nil {
remote = cl.Net.Conn.RemoteAddr().String()
}
if remote == "" {
return "unknown"
}
if host, _, err := net.SplitHostPort(remote); err == nil {
return host
}
return remote
}
internal/mqttauth/mqttauth_test.go
+147
View File
@@ -0,0 +1,147 @@
package mqttauth
import (
"testing"
"time"
mqtt "github.com/mochi-mqtt/server/v2"
"github.com/mochi-mqtt/server/v2/packets"
"golang.org/x/crypto/bcrypt"
)
func hash(t *testing.T, password string) string {
t.Helper()
h, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.MinCost)
if err != nil {
t.Fatalf("bcrypt: %v", err)
}
return string(h)
}
func newTestHook(t *testing.T, cfg Config) *Hook {
t.Helper()
if cfg.MaxFailures == 0 {
cfg.MaxFailures = 3
}
if cfg.Window == 0 {
cfg.Window = time.Minute
}
if cfg.BlockFor == 0 {
cfg.BlockFor = 5 * time.Minute
}
h := NewHook(cfg)
t.Cleanup(func() { _ = h.Stop() })
return h
}
func TestAuthenticate(t *testing.T) {
h := newTestHook(t, Config{
Enabled: true,
AllowAnonymous: false,
Users: []User{{Username: "mesh", PasswordHash: hash(t, "secret")}},
})
cases := []struct {
name string
user string
pass string
allowed bool
}{
{"正确凭据", "mesh", "secret", true},
{"错误密码", "mesh", "wrong", false},
{"未知用户", "nobody", "secret", false},
{"匿名未启用", "", "", false},
{"仅用户名", "mesh", "", false},
{"仅密码", "", "secret", false},
}
for _, tc := range cases {
if got := h.authenticate(tc.user, tc.pass); got != tc.allowed {
t.Errorf("%s: authenticate(%q,%q)=%v want %v", tc.name, tc.user, tc.pass, got, tc.allowed)
}
}
}
func TestAuthenticateDisabledAllowsAll(t *testing.T) {
h := newTestHook(t, Config{Enabled: false})
for _, tc := range [][2]string{{"", ""}, {"any", "thing"}} {
if !h.authenticate(tc[0], tc[1]) {
t.Errorf("disabled hook must allow %v", tc)
}
}
}
func TestAuthenticateAnonymousAllowed(t *testing.T) {
h := newTestHook(t, Config{Enabled: true, AllowAnonymous: true})
if !h.authenticate("", "") {
t.Error("anonymous should be allowed")
}
if h.authenticate("mesh", "x") {
t.Error("unknown user must still be rejected")
}
}
func TestOnConnectAuthenticateDisabled(t *testing.T) {
h := newTestHook(t, Config{Enabled: false})
if !h.OnConnectAuthenticate(&mqtt.Client{}, packets.Packet{}) {
t.Error("disabled hook must return true")
}
}
func TestFailureLimiterBlocks(t *testing.T) {
h := newTestHook(t, Config{Enabled: true, Users: []User{{Username: "mesh", PasswordHash: hash(t, "secret")}}})
now := time.Unix(1700000000, 0)
h.now = func() time.Time { return now }
for i := 0; i < 3; i++ {
if h.OnConnectAuthenticate(&mqtt.Client{}, connectPacket("mesh", "wrong")) {
t.Fatalf("attempt %d should fail", i)
}
}
// 第 3 次失败触发封禁;之后即使凭据正确也应被拒。
now = now.Add(time.Second)
if h.OnConnectAuthenticate(&mqtt.Client{}, connectPacket("mesh", "secret")) {
t.Fatal("blocked ip must be rejected even with valid credentials")
}
// 封禁到期后恢复。
now = now.Add(6 * time.Minute)
if !h.OnConnectAuthenticate(&mqtt.Client{}, connectPacket("mesh", "secret")) {
t.Fatal("credentials should work after block expires")
}
}
func TestSuccessResetsFailures(t *testing.T) {
h := newTestHook(t, Config{Enabled: true, Users: []User{{Username: "mesh", PasswordHash: hash(t, "secret")}}})
now := time.Unix(1700000000, 0)
h.now = func() time.Time { return now }
for i := 0; i < 2; i++ {
h.OnConnectAuthenticate(&mqtt.Client{}, connectPacket("mesh", "wrong"))
}
if !h.OnConnectAuthenticate(&mqtt.Client{}, connectPacket("mesh", "secret")) {
t.Fatal("valid login should succeed")
}
// 成功后计数清零:再失败 2 次应有计数条目但未被封禁(阈值为 3)。
now = now.Add(2 * time.Second)
h.OnConnectAuthenticate(&mqtt.Client{}, connectPacket("mesh", "wrong"))
h.OnConnectAuthenticate(&mqtt.Client{}, connectPacket("mesh", "wrong"))
h.mu.Lock()
st := h.fails["unknown"]
h.mu.Unlock()
if st != nil && !st.blockedUntil.IsZero() {
t.Fatal("failures should have been reset by successful login")
}
}
func TestRemoteHost(t *testing.T) {
cl := &mqtt.Client{}
if got := remoteHost(cl); got == "" {
t.Error("remoteHost should never return empty string")
}
}
func connectPacket(username, password string) packets.Packet {
pk := packets.Packet{}
pk.Connect.Username = []byte(username)
pk.Connect.Password = []byte(password)
return pk
}
internal/web/web.go
+1 -1
View File
@@ -84,7 +84,7 @@ func NewRouter(cfg configpkg.WebConfig, consoleLog bool, store *storepkg.Store,
return r return r
} }
const BackendVersion = "1.2.1" const BackendVersion = "1.3.0"
var CommitVersion = "dev" var CommitVersion = "dev"
main.go
+30 -11
View File
@@ -17,7 +17,6 @@ import (
"time" "time"
mqtt "github.com/mochi-mqtt/server/v2" mqtt "github.com/mochi-mqtt/server/v2"
mqttauth "github.com/mochi-mqtt/server/v2/hooks/auth"
"github.com/mochi-mqtt/server/v2/listeners" "github.com/mochi-mqtt/server/v2/listeners"
"github.com/mochi-mqtt/server/v2/packets" "github.com/mochi-mqtt/server/v2/packets"
@@ -28,6 +27,7 @@ import (
botpkg "meshtastic_mqtt_server/internal/bot" botpkg "meshtastic_mqtt_server/internal/bot"
configpkg "meshtastic_mqtt_server/internal/config" configpkg "meshtastic_mqtt_server/internal/config"
"meshtastic_mqtt_server/internal/llm" "meshtastic_mqtt_server/internal/llm"
"meshtastic_mqtt_server/internal/mqttauth"
"meshtastic_mqtt_server/internal/mqtpp" "meshtastic_mqtt_server/internal/mqtpp"
mqttforwardpkg "meshtastic_mqtt_server/internal/mqttforward" mqttforwardpkg "meshtastic_mqtt_server/internal/mqttforward"
rspkg "meshtastic_mqtt_server/internal/runtimesettings" rspkg "meshtastic_mqtt_server/internal/runtimesettings"
@@ -381,10 +381,11 @@ func run(cfg *configpkg.Config) error {
messageStats := &mqttforwardpkg.Stats{} messageStats := &mqttforwardpkg.Stats{}
clientStats := mqttforwardpkg.NewClientStats() clientStats := mqttforwardpkg.NewClientStats()
server, mqttHook, mqttAddr, err := startMQTTServer(cfg, store, dbQueue, messageStats, clientStats, blocking, settings) server, mqttHook, authHook, mqttAddr, err := startMQTTServer(cfg, store, dbQueue, messageStats, clientStats, blocking, settings)
if err != nil { if err != nil {
return err return err
} }
defer authHook.Stop()
botSender := botpkg.NewService(store, server, cfg.Key) botSender := botpkg.NewService(store, server, cfg.Key)
mqttHook.autoAcker = botSender.MaybeAutoAck mqttHook.autoAcker = botSender.MaybeAutoAck
botCtx, stopBotBroadcaster := context.WithCancel(context.Background()) botCtx, stopBotBroadcaster := context.WithCancel(context.Background())
@@ -527,10 +528,21 @@ func run(cfg *configpkg.Config) error {
return runErr return runErr
} }
func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *storepkg.WriteQueue, stats *mqttforwardpkg.Stats, clientStats *mqttforwardpkg.ClientStats, blocking *blockingpkg.Cache, settings *rspkg.Cache) (*mqtt.Server, *meshtasticFilterHook, string, error) { func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *storepkg.WriteQueue, stats *mqttforwardpkg.Stats, clientStats *mqttforwardpkg.ClientStats, blocking *blockingpkg.Cache, settings *rspkg.Cache) (*mqtt.Server, *meshtasticFilterHook, *mqttauth.Hook, string, error) {
server := mqtt.New(&mqtt.Options{InlineClient: true}) server := mqtt.New(&mqtt.Options{InlineClient: true})
if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil { authUsers := make([]mqttauth.User, 0, len(cfg.MQTT.Auth.Users))
return nil, nil, "", err for _, u := range cfg.MQTT.Auth.Users {
authUsers = append(authUsers, mqttauth.User{Username: u.Username, PasswordHash: u.PasswordHash})
}
authHook := mqttauth.NewHook(mqttauth.Config{
Enabled: cfg.MQTT.Auth.Enabled,
AllowAnonymous: cfg.MQTT.Auth.AllowAnonymous,
Users: authUsers,
LogEvent: printJSON,
})
if err := server.AddHook(authHook, nil); err != nil {
authHook.Stop()
return nil, nil, nil, "", err
} }
dedupQueue := mqttforwardpkg.NewDedupQueue() dedupQueue := mqttforwardpkg.NewDedupQueue()
dedupQueue.Start() dedupQueue.Start()
@@ -548,23 +560,30 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
dedupQueue: dedupQueue, dedupQueue: dedupQueue,
} }
if err := server.AddHook(hook, nil); err != nil { if err := server.AddHook(hook, nil); err != nil {
return nil, nil, "", err authHook.Stop()
return nil, nil, nil, "", err
} }
addr := net.JoinHostPort(cfg.MQTT.Host, strconv.Itoa(cfg.MQTT.Port)) addr := net.JoinHostPort(cfg.MQTT.Host, strconv.Itoa(cfg.MQTT.Port))
tlsConfig, err := configpkg.BuildTLS(cfg.MQTT.TLS) tlsConfig, err := configpkg.BuildTLS(cfg.MQTT.TLS)
if err != nil { if err != nil {
return nil, nil, "", err authHook.Stop()
return nil, nil, nil, "", err
} }
listener := listeners.NewTCP(listeners.Config{ID: "tcp", Address: addr, TLSConfig: tlsConfig}) listener := listeners.NewTCP(listeners.Config{ID: "tcp", Address: addr, TLSConfig: tlsConfig})
if err := server.AddListener(listener); err != nil { if err := server.AddListener(listener); err != nil {
return nil, nil, "", err authHook.Stop()
return nil, nil, nil, "", err
} }
if err := server.Serve(); err != nil { if err := server.Serve(); err != nil {
return nil, nil, "", err authHook.Stop()
return nil, nil, nil, "", err
} }
printJSON(map[string]any{"event": "broker_started", "address": addr, "tls": cfg.MQTT.TLS.Enabled}) printJSON(map[string]any{
return server, hook, addr, nil "event": "broker_started", "address": addr, "tls": cfg.MQTT.TLS.Enabled,
"auth_enabled": cfg.MQTT.Auth.Enabled, "auth_users": len(authUsers), "auth_allow_anonymous": cfg.MQTT.Auth.AllowAnonymous,
})
return server, hook, authHook, addr, nil
} }
// printJSON 将记录编码为 JSON 后按数据包类型着色输出。 // printJSON 将记录编码为 JSON 后按数据包类型着色输出。
py/db_config.example.py
+36
View File
@@ -0,0 +1,36 @@
"""db_config 模板:复制为 db_config.py(已被 py/.gitignore 忽略)并按需修改。
口令一律从环境变量读取,严禁写进文件:
export MESH_SOURCE_DB_PASSWORD='...'
export MESH_TARGET_DB_PASSWORD='...'
"""
from __future__ import annotations
import os
def _require(name: str) -> str:
value = os.environ.get(name, "")
if not value:
raise RuntimeError(f"环境变量 {name} 未设置。请先 export 口令再运行。")
return value
SOURCE_CONFIG = {
"host": os.environ.get("MESH_SOURCE_DB_HOST", "source-db.example.com"),
"port": int(os.environ.get("MESH_SOURCE_DB_PORT", "3306")),
"user": os.environ.get("MESH_SOURCE_DB_USER", "meshuser"),
"password": _require("MESH_SOURCE_DB_PASSWORD"),
"database": os.environ.get("MESH_SOURCE_DB_NAME", "meshtastic_db"),
"charset": "utf8mb4",
}
TARGET_CONFIG = {
"host": os.environ.get("MESH_TARGET_DB_HOST", "target-db.example.com"),
"port": int(os.environ.get("MESH_TARGET_DB_PORT", "3306")),
"user": os.environ.get("MESH_TARGET_DB_USER", "meshtastic_db"),
"password": _require("MESH_TARGET_DB_PASSWORD"),
"database": os.environ.get("MESH_TARGET_DB_NAME", "meshtastic_db"),
"charset": "utf8mb4",
}