将mqtt客户端改成mqtt服务端

README.md

@@ -1,5 +1,17 @@
 # Meshtastic MQTT Server
 
+本程序启动一个本地 MQTT broker，并在转发客户端发布的消息前校验 Meshtastic MQTT payload。
+
+每条传入的 `PUBLISH` 都会先进入：
+
+```go
+valid, _, record := mqtpp.MQTTPP(topic, payload, key)
+```
+
+- `valid == true`：保留原始 topic、payload、QoS、retain 等字段，正常转发给订阅匹配 topic 的客户端
+- `valid == false`：丢弃该消息，不转发给订阅客户端
+
+当前不桥接到 `mqtt.meshtastic.org` 等上游 broker。
 
 ## 运行
 
@@ -7,39 +19,58 @@
 go run .
 ```
 
-默认连接：
+默认监听：
 
-- broker：`mqtt.meshtastic.org:1883`
+- host：`0.0.0.0`
-- username：`meshdev`
+- port：`1883`
-- password：`large4cats`
-- topic：`msh/US/#`
 - PSK：`AQ==`
 
-也可以指定 topic：
+也可以指定监听地址和 PSK：
 
 ```bash
-go run . --topic 'msh/US/#'
+go run . --host 127.0.0.1 --port 1883 --psk AQ==
-```
-
-多个 topic 可重复传入：
-
-```bash
-go run . --topic 'msh/US/#' --topic 'msh/EU_868/#'
 ```
 
 ## 参数
 
 ```text
---host       MQTT broker hostname
+--host  MQTT broker listen host
---port       MQTT broker port
+--port  MQTT broker listen port
---username   MQTT username
---password   MQTT password
 --psk   Base64 channel PSK used to try decrypting encrypted packets
---topic      Topic to subscribe; may be repeated
---qos        MQTT subscription QoS: 0, 1, or 2
---client-id  MQTT client id
 ```
 
+## 转发规则
+
+程序监听所有传入 publish。payload 能被 `mqtpp.MQTTPP` 解析时，认为 `valid == true`，broker 会继续把原始 MQTT 消息转发给订阅者；解析失败时，认为 `valid == false`，broker 会拒绝并丢弃该 publish。
+
+`empty_packet` 仍然属于 `valid == true`，会被转发；只是控制台默认不显示它。
+
+无法解密但能解析的加密包通常会输出为 `encrypted_packet`，仍然属于 `valid == true`，因此会被转发。
+
+## 本地验证
+
+一个终端启动 broker：
+
+```bash
+go run . --host 127.0.0.1 --port 1883 --psk AQ==
+```
+
+另一个终端订阅：
+
+```bash
+mosquitto_sub -h 127.0.0.1 -p 1883 -t '#'
+```
+
+发布非法 payload：
+
+```bash
+mosquitto_pub -h 127.0.0.1 -p 1883 -t 'msh/US/test' -m 'not protobuf'
+```
+
+订阅端应该收不到该消息。
+
+要验证 valid 消息转发，请使用真实 Meshtastic MQTT payload 发布到本 broker；订阅匹配 topic 的客户端应收到原始消息，broker 控制台会打印解析后的 `record`。
+
 ## 控制台颜色说明
 
 程序会按数据包类型使用不同背景色，方便快速区分消息类型。

go.mod

@@ -3,12 +3,12 @@ module meshtastic_mqtt_server
 go 1.23
 
 require (
-	github.com/eclipse/paho.mqtt.golang v1.5.0
+	github.com/mochi-mqtt/server/v2 v2.7.9
 	google.golang.org/protobuf v1.36.11
 )
 
 require (
 	github.com/gorilla/websocket v1.5.3 // indirect
-	golang.org/x/net v0.27.0 // indirect
+	github.com/rs/xid v1.4.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,10 +1,20 @@
-github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
-github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-golang.org/x/net v0.27.0 h1:5K3Njcw06/l2y9vpGCSdcxWOYHOUk3dVNGDXN+FvAys=
+github.com/jinzhu/copier v0.3.5 h1:GlvfUwHk62RokgqVNvYsku0TATCF7bAHVwEXoBh3iJg=
-golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE=
+github.com/jinzhu/copier v0.3.5/go.mod h1:DfbEm0FYsaqBcKcFuvmOZb218JkPGtvSHsKg8S8hyyg=
-golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
+github.com/mochi-mqtt/server/v2 v2.7.9 h1:y0g4vrSLAag7T07l2oCzOa/+nKVLoazKEWAArwqBNYI=
-golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+github.com/mochi-mqtt/server/v2 v2.7.9/go.mod h1:lZD3j35AVNqJL5cezlnSkuG05c0FCHSsfAKSPBOSbqc=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rs/xid v1.4.0 h1:qd7wPTDkN6KQx2VmMBLrpHkiyQwgFXRnkOLacUiaSNY=
+github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

main.go

@@ -1,25 +1,25 @@
 package main
 
 import (
-	"encoding/json"
 	"flag"
 	"fmt"
+	"net"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
-	"time"
 
 	"meshtastic_mqtt_server/mqtpp"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
+	mqtt "github.com/mochi-mqtt/server/v2"
+	"github.com/mochi-mqtt/server/v2/hooks/auth"
+	"github.com/mochi-mqtt/server/v2/listeners"
+	"github.com/mochi-mqtt/server/v2/packets"
 )
 
 const (
-	defaultHost     = "mqtt.meshtastic.org"
+	defaultHost = "0.0.0.0"
-	defaultUsername = "meshdev"
-	defaultPassword = "large4cats"
 	defaultPSK  = "AQ=="
-	defaultTopic    = "msh/US/#"
 
 	ansiGreenBGWhiteText  = "\033[42;37m"
 	ansiBlueBGWhiteText   = "\033[44;37m"
@@ -34,33 +34,39 @@ const (
 type config struct {
 	host string
 	port int
-	username string
-	password string
 	psk  string
-	topics   topicsFlag
-	qos      int
-	clientID string
 	key  []byte
 }
 
-type topicsFlag []string
+type meshtasticFilterHook struct {
+	mqtt.HookBase
+	key []byte
+}
 
-// String 将已配置的 topic 列表转换为字符串，供 flag 包显示默认值或帮助信息。
+// ID 返回用于识别 Meshtastic payload 过滤器的 hook 名称。
-func (t *topicsFlag) String() string {
+func (h *meshtasticFilterHook) ID() string {
-	if t == nil {
+	return "meshtastic-filter"
-		return ""
+}
+
+// Provides 声明该 hook 只处理客户端发布消息。
+func (h *meshtasticFilterHook) Provides(b byte) bool {
+	return b == mqtt.OnPublish
+}
+
+// OnPublish 在 broker 转发消息前校验 payload；无效消息会被拒绝并丢弃。
+func (h *meshtasticFilterHook) OnPublish(_ *mqtt.Client, pk packets.Packet) (packets.Packet, error) {
+	valid, _, record := mqtpp.MQTTPP(pk.TopicName, pk.Payload, h.key)
+	if !valid {
+		return pk, packets.ErrRejectPacket
 	}
-	b, _ := json.Marshal([]string(*t))
+
-	return string(b)
+	if record["type"] != "empty_packet" {
+		printJSON(record)
+	}
+	return pk, nil
 }
 
-// Set 追加一个 --topic 参数值，支持命令行重复传入多个订阅主题。
+// main 是程序入口，负责解析参数并启动 MQTT broker。
-func (t *topicsFlag) Set(value string) error {
-	*t = append(*t, value)
-	return nil
-}
-
-// main 是程序入口，负责解析参数并启动 MQTT 订阅流程。
 func main() {
 	cfg, err := parseArgs()
 	if err != nil {
@@ -76,22 +82,11 @@ func main() {
 // parseArgs 解析命令行参数，并展开 Meshtastic channel PSK。
 func parseArgs() (*config, error) {
 	cfg := &config{}
-	flag.StringVar(&cfg.host, "host", defaultHost, "MQTT broker hostname")
+	flag.StringVar(&cfg.host, "host", defaultHost, "MQTT broker listen host")
-	flag.IntVar(&cfg.port, "port", 1883, "MQTT broker port")
+	flag.IntVar(&cfg.port, "port", 1883, "MQTT broker listen port")
-	flag.StringVar(&cfg.username, "username", defaultUsername, "MQTT username")
-	flag.StringVar(&cfg.password, "password", defaultPassword, "MQTT password")
 	flag.StringVar(&cfg.psk, "psk", defaultPSK, "Base64 channel PSK used to try decrypting encrypted packets")
-	flag.Var(&cfg.topics, "topic", "Topic to subscribe; may be repeated. Defaults to msh/US/#")
-	flag.IntVar(&cfg.qos, "qos", 0, "MQTT subscription QoS (0, 1, or 2)")
-	flag.StringVar(&cfg.clientID, "client-id", "meshtastic-nodeinfo-subscriber", "MQTT client id")
 	flag.Parse()
 
-	if len(cfg.topics) == 0 {
-		cfg.topics = topicsFlag{defaultTopic}
-	}
-	if cfg.qos < 0 || cfg.qos > 2 {
-		return nil, fmt.Errorf("invalid qos %d: must be 0, 1, or 2", cfg.qos)
-	}
 	key, err := mqtpp.ExpandPSK(cfg.psk)
 	if err != nil {
 		return nil, err
@@ -100,62 +95,30 @@ func parseArgs() (*config, error) {
 	return cfg, nil
 }
 
-// run 创建 MQTT 客户端，连接 broker，订阅 topic，并阻塞等待退出信号。
+// run 创建 MQTT broker，监听传入 publish，并阻塞等待退出信号。
 func run(cfg *config) error {
-	opts := mqtt.NewClientOptions()
+	server := mqtt.New(nil)
-	opts.AddBroker(fmt.Sprintf("tcp://%s:%d", cfg.host, cfg.port))
+	if err := server.AddHook(new(auth.AllowHook), nil); err != nil {
-	opts.SetClientID(cfg.clientID)
-	opts.SetKeepAlive(60 * time.Second)
-	if cfg.username != "" {
-		opts.SetUsername(cfg.username)
-		opts.SetPassword(cfg.password)
-	}
-
-	opts.OnConnect = func(client mqtt.Client) {
-		printJSON(map[string]any{"event": "connected", "reason_code": "0"})
-		for _, topic := range cfg.topics {
-			token := client.Subscribe(topic, byte(cfg.qos), handleMessage(cfg.key))
-			token.Wait()
-			if err := token.Error(); err != nil {
-				printJSON(map[string]any{"event": "subscribe_error", "topic": topic, "qos": cfg.qos, "error": err.Error()})
-				continue
-			}
-			printJSON(map[string]any{"event": "subscribed", "topic": topic, "qos": cfg.qos})
-		}
-	}
-
-	client := mqtt.NewClient(opts)
-	token := client.Connect()
-	token.Wait()
-	if err := token.Error(); err != nil {
 		return err
 	}
+	if err := server.AddHook(&meshtasticFilterHook{key: cfg.key}, nil); err != nil {
+		return err
+	}
+
+	addr := net.JoinHostPort(cfg.host, strconv.Itoa(cfg.port))
+	listener := listeners.NewTCP(listeners.Config{ID: "tcp", Address: addr})
+	if err := server.AddListener(listener); err != nil {
+		return err
+	}
+	if err := server.Serve(); err != nil {
+		return err
+	}
+	printJSON(map[string]any{"event": "broker_started", "address": addr})
 
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
 	<-sigCh
-	client.Disconnect(250)
+	return server.Close()
-	return nil
-}
-
-// handleMessage 返回 MQTT 消息回调，把原始 payload 交给 MQTTPP 处理后按类型输出。
-func handleMessage(key []byte) mqtt.MessageHandler {
-	return func(_ mqtt.Client, msg mqtt.Message) {
-		valid, _, decodedJSON := mqtpp.MQTTPP(msg.Topic(), msg.Payload(), key)
-		if !valid || len(decodedJSON) == 0 {
-			return
-		}
-
-		var record map[string]any
-		if err := json.Unmarshal(decodedJSON, &record); err != nil {
-			printJSON(map[string]any{"topic": msg.Topic(), "error": "json decode failed: " + err.Error(), "payload_len": len(msg.Payload())})
-			return
-		}
-		if record["type"] == "empty_packet" {
-			return
-		}
-		printJSONBytes(record, decodedJSON)
-	}
 }
 
 // printJSON 将记录编码为 JSON 后按数据包类型着色输出。

mqtpp/mqtpp.go

@@ -113,22 +113,21 @@ type telemetryInfo struct {
 	Metrics map[string]any
 }
 
-// MQTTPP 处理一个 MQTT 原始 payload，返回合规状态、原始数据和解码后的 JSON。
+// MQTTPP 处理一个 MQTT 原始 payload，返回合规状态、原始数据和解码后的记录。
-// 第一个返回值表示数据是否合规；第二个返回值在不合规时为 nil；第三个返回值是解码结果 JSON。
+// 第一个返回值表示数据是否合规；第二个返回值在不合规时为 nil；第三个返回值是解码结果记录。
-func MQTTPP(topic string, raw []byte, key []byte) (bool, []byte, []byte) {
+func MQTTPP(topic string, raw []byte, key []byte) (bool, []byte, map[string]any) {
-	if !isCompliantMQTTPacket(raw) {
-		return false, nil, nil
-	}
 
 	env, err := parseServiceEnvelope(raw)
 	if err != nil {
-		return true, raw, MustJSON(map[string]any{"topic": topic, "error": "protobuf decode failed: " + err.Error(), "payload_len": len(raw)})
+		//解包失败
+		return false, nil, map[string]any{"topic": topic, "error": "protobuf decode failed: " + err.Error(), "payload_len": len(raw)}
 	}
 	record, err := describePacket(topic, env, key)
 	if err != nil {
-		return true, raw, MustJSON(map[string]any{"topic": topic, "error": err.Error(), "payload_len": len(raw)})
+		//解码失败
+		return false, nil, map[string]any{"topic": topic, "error": err.Error(), "payload_len": len(raw)}
 	}
-	return true, raw, MustJSON(record)
+	return true, raw, record
 }
 
 // ExpandPSK 展开 Base64 PSK，兼容 Meshtastic 默认索引 PSK 和短 key 补零规则。