diff --git a/README.md b/README.md index b8c090e..b510f25 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,10 @@ export MESH_ADMIN_SESSION_SECRET='replace-with-a-long-random-string' mqtt: host: 0.0.0.0 port: 1883 + auth: + enabled: false + allow_anonymous: false + users: [] tls: enabled: false cert_file: "" @@ -208,6 +212,30 @@ mkdir -p /srv/mesh_mqtt_go -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 上行配置中填写相同的用户名密码。 + ## 访问服务 启动后可访问： diff --git a/doc/SECURITY_FIX_TODO.md b/doc/SECURITY_FIX_TODO.md new file mode 100644 index 0000000..a708a0f --- /dev/null +++ b/doc/SECURITY_FIX_TODO.md @@ -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 脱敏与日志:现状正确 diff --git a/install.sh b/install.sh index 6e2e24d..abbf8ff 100644 --- a/install.sh +++ b/install.sh @@ -69,6 +69,16 @@ if [[ ! -f "${CONFIG_DIR}/config.yaml" ]]; then mqtt: host: 0.0.0.0 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: enabled: false cert_file: "" diff --git a/internal/config/config.go b/internal/config/config.go index de5cd64..5145431 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,7 +6,10 @@ import ( "os" "path/filepath" "runtime" + "strconv" + "strings" + "golang.org/x/crypto/bcrypt" "gopkg.in/yaml.v3" ) @@ -28,9 +31,27 @@ type Config struct { } type MQTTConfig struct { - Host string `yaml:"host"` - Port int `yaml:"port"` - TLS TLSConfig `yaml:"tls"` + Host string `yaml:"host"` + Port int `yaml:"port"` + 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 { @@ -113,9 +134,22 @@ type rawAIConfig struct { } type rawMQTTConfig struct { - Host *string `yaml:"host"` - Port *int `yaml:"port"` - TLS *rawTLSConfig `yaml:"tls"` + Host *string `yaml:"host"` + Port *int `yaml:"port"` + 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 { @@ -172,6 +206,11 @@ func Default() *Config { CertFile: "", KeyFile: "", }, + Auth: MQTTAuthConfig{ + Enabled: false, + AllowAnonymous: false, + Users: []MQTTAuthUser{}, + }, }, Meshtastic: MeshtasticConfig{ PSK: "AQ==", @@ -377,6 +416,45 @@ func normalize(raw rawConfig) (*Config, bool) { 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 { @@ -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 } @@ -532,6 +622,9 @@ func Validate(cfg *Config) error { if cfg.MQTT.Port <= 0 || cfg.MQTT.Port > 65535 { 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 { case DriverSQLite: if cfg.Database.SQLite.Path == "" { @@ -570,6 +663,51 @@ func Validate(cfg *Config) error { 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$$<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 { data, err := yaml.Marshal(cfg) if err != nil { diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..b4a43d6 --- /dev/null +++ b/internal/config/config_test.go @@ -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" diff --git a/internal/mqttauth/integration_test.go b/internal/mqttauth/integration_test.go new file mode 100644 index 0000000..70d1e68 --- /dev/null +++ b/internal/mqttauth/integration_test.go @@ -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") + } +} diff --git a/internal/mqttauth/mqttauth.go b/internal/mqttauth/mqttauth.go new file mode 100644 index 0000000..8a48084 --- /dev/null +++ b/internal/mqttauth/mqttauth.go @@ -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 +} diff --git a/internal/mqttauth/mqttauth_test.go b/internal/mqttauth/mqttauth_test.go new file mode 100644 index 0000000..5a8c36e --- /dev/null +++ b/internal/mqttauth/mqttauth_test.go @@ -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 +} diff --git a/internal/web/web.go b/internal/web/web.go index 70c9c81..1be0c28 100644 --- a/internal/web/web.go +++ b/internal/web/web.go @@ -84,7 +84,7 @@ func NewRouter(cfg configpkg.WebConfig, consoleLog bool, store *storepkg.Store, return r } -const BackendVersion = "1.2.1" +const BackendVersion = "1.3.0" var CommitVersion = "dev" diff --git a/main.go b/main.go index 2dbb0c1..1ebaffa 100644 --- a/main.go +++ b/main.go @@ -17,7 +17,6 @@ import ( "time" 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/packets" @@ -28,6 +27,7 @@ import ( botpkg "meshtastic_mqtt_server/internal/bot" configpkg "meshtastic_mqtt_server/internal/config" "meshtastic_mqtt_server/internal/llm" + "meshtastic_mqtt_server/internal/mqttauth" "meshtastic_mqtt_server/internal/mqtpp" mqttforwardpkg "meshtastic_mqtt_server/internal/mqttforward" rspkg "meshtastic_mqtt_server/internal/runtimesettings" @@ -381,10 +381,11 @@ func run(cfg *configpkg.Config) error { messageStats := &mqttforwardpkg.Stats{} 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 { return err } + defer authHook.Stop() botSender := botpkg.NewService(store, server, cfg.Key) mqttHook.autoAcker = botSender.MaybeAutoAck botCtx, stopBotBroadcaster := context.WithCancel(context.Background()) @@ -527,10 +528,21 @@ func run(cfg *configpkg.Config) error { 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}) - if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil { - return nil, nil, "", err + authUsers := make([]mqttauth.User, 0, len(cfg.MQTT.Auth.Users)) + 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.Start() @@ -548,23 +560,30 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor dedupQueue: dedupQueue, } 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)) tlsConfig, err := configpkg.BuildTLS(cfg.MQTT.TLS) 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}) if err := server.AddListener(listener); err != nil { - return nil, nil, "", err + authHook.Stop() + return nil, nil, nil, "", err } 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}) - return server, hook, addr, nil + printJSON(map[string]any{ + "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 后按数据包类型着色输出。 diff --git a/py/db_config.example.py b/py/db_config.example.py new file mode 100644 index 0000000..5a2f787 --- /dev/null +++ b/py/db_config.example.py @@ -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", +}