301 lines
10 KiB
Go
301 lines
10 KiB
Go
// Package service 内嵌 MQTT Broker，并在内部 HTTP 边界接收待下发命令。
|
||
package service
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/subtle"
|
||
"crypto/tls"
|
||
"crypto/x509"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"os"
|
||
"strings"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"git.apinb.com/heqiapp/platforms/backend/iot-server/internal/config"
|
||
"git.apinb.com/heqiapp/platforms/backend/iot-server/internal/protocol"
|
||
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"
|
||
)
|
||
|
||
type Service struct {
|
||
cfg config.Config
|
||
keys protocol.Keyring
|
||
broker *mqtt.Server
|
||
packet atomic.Uint32
|
||
http *http.Server
|
||
ready atomic.Bool
|
||
client *http.Client
|
||
}
|
||
type Command struct {
|
||
Identity string `json:"identity"`
|
||
IdempotencyKey string `json:"idempotency_key"`
|
||
DeviceID string `json:"device_id"`
|
||
Action string `json:"action"`
|
||
ExpiresAt time.Time `json:"expires_at"`
|
||
KeyID byte `json:"key_id"`
|
||
DeviceKind byte `json:"device_kind"`
|
||
DeviceType byte `json:"device_type"`
|
||
DeviceModel [3]byte `json:"device_model"`
|
||
Controller byte `json:"controller"`
|
||
Loop byte `json:"loop"`
|
||
Component byte `json:"component"`
|
||
}
|
||
type Envelope struct {
|
||
Type, Topic, DeviceID, ReceivedAt string
|
||
PayloadHex string
|
||
Frame *DecodedFrame `json:"frame,omitempty"`
|
||
}
|
||
type DecodedFrame struct {
|
||
KeyID, Version, Control, MainID byte
|
||
PacketNumber uint16
|
||
Sequence byte
|
||
Final bool
|
||
DeviceTime string
|
||
PayloadHex string
|
||
Realtime *protocol.RealtimeData `json:"realtime,omitempty"`
|
||
}
|
||
|
||
func New(cfg config.Config) (*Service, error) {
|
||
keys, err := cfg.Keys()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
broker := mqtt.New(&mqtt.Options{InlineClient: true})
|
||
ledger := &auth.Ledger{}
|
||
for _, device := range cfg.MQTT.Devices {
|
||
ledger.Auth = append(ledger.Auth, auth.AuthRule{
|
||
Client: auth.RString(device.ClientID), Username: auth.RString(device.Username),
|
||
Password: auth.RString(device.Password), Allow: true,
|
||
})
|
||
ledger.ACL = append(ledger.ACL, auth.ACLRule{
|
||
Client: auth.RString(device.ClientID), Username: auth.RString(device.Username),
|
||
Filters: auth.Filters{
|
||
auth.RString(strings.ReplaceAll(cfg.MQTT.UpTopic, "+", device.DeviceID)): auth.WriteOnly,
|
||
auth.RString(strings.ReplaceAll(cfg.MQTT.AckTopic, "+", device.DeviceID)): auth.WriteOnly,
|
||
auth.RString(strings.ReplaceAll(cfg.MQTT.DownTopic, "{deviceId}", device.DeviceID)): auth.ReadOnly,
|
||
},
|
||
})
|
||
}
|
||
if err = broker.AddHook(new(auth.Hook), &auth.Options{Ledger: ledger}); err != nil {
|
||
return nil, fmt.Errorf("配置 MQTT 认证: %w", err)
|
||
}
|
||
listenerConfig := listeners.Config{ID: cfg.MQTT.ListenerID, Address: cfg.MQTT.Address}
|
||
if cfg.MQTT.TLS {
|
||
tlsConfig, tlsErr := makeTLSConfig(cfg)
|
||
if tlsErr != nil {
|
||
return nil, tlsErr
|
||
}
|
||
listenerConfig.TLSConfig = tlsConfig
|
||
}
|
||
if err = broker.AddListener(listeners.NewTCP(listenerConfig)); err != nil {
|
||
return nil, fmt.Errorf("配置 MQTT 监听器: %w", err)
|
||
}
|
||
return &Service{cfg: cfg, keys: keys, broker: broker, client: &http.Client{Timeout: 12 * time.Second}}, nil
|
||
}
|
||
|
||
func makeTLSConfig(cfg config.Config) (*tls.Config, error) {
|
||
roots, err := x509.SystemCertPool()
|
||
if err != nil {
|
||
roots = x509.NewCertPool()
|
||
}
|
||
if cfg.MQTT.CAFile != "" {
|
||
data, readErr := os.ReadFile(cfg.MQTT.CAFile)
|
||
if readErr != nil {
|
||
return nil, readErr
|
||
}
|
||
if !roots.AppendCertsFromPEM(data) {
|
||
return nil, fmt.Errorf("MQTT CA 证书无效")
|
||
}
|
||
}
|
||
if cfg.MQTT.CertificateFile == "" || cfg.MQTT.PrivateKeyFile == "" {
|
||
return nil, fmt.Errorf("启用 MQTT TLS 时 CertificateFile 和 PrivateKeyFile 必填")
|
||
}
|
||
certificate, err := tls.LoadX509KeyPair(cfg.MQTT.CertificateFile, cfg.MQTT.PrivateKeyFile)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
result := &tls.Config{MinVersion: tls.VersionTLS12, Certificates: []tls.Certificate{certificate}}
|
||
if cfg.MQTT.CAFile != "" {
|
||
result.ClientCAs = roots
|
||
result.ClientAuth = tls.RequireAndVerifyClientCert
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
func (s *Service) Run(ctx context.Context) error {
|
||
if err := s.broker.Subscribe(s.cfg.MQTT.UpTopic, 1, s.onMessage); err != nil {
|
||
return fmt.Errorf("订阅设备上行 Topic: %w", err)
|
||
}
|
||
if err := s.broker.Subscribe(s.cfg.MQTT.AckTopic, 2, s.onMessage); err != nil {
|
||
return fmt.Errorf("订阅设备回执 Topic: %w", err)
|
||
}
|
||
if err := s.broker.Serve(); err != nil {
|
||
return fmt.Errorf("启动 MQTT Broker: %w", err)
|
||
}
|
||
s.ready.Store(true)
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("/health", s.health)
|
||
mux.HandleFunc("/internal/v1/commands", s.command)
|
||
s.http = &http.Server{Addr: s.cfg.HTTP.Address, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
|
||
go func() {
|
||
<-ctx.Done()
|
||
shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer cancel()
|
||
s.ready.Store(false)
|
||
_ = s.http.Shutdown(shutdown)
|
||
_ = s.broker.Close()
|
||
}()
|
||
err := s.http.ListenAndServe()
|
||
if err == http.ErrServerClosed {
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
|
||
func (s *Service) health(w http.ResponseWriter, _ *http.Request) {
|
||
if !s.ready.Load() {
|
||
http.Error(w, "mqtt broker unavailable", http.StatusServiceUnavailable)
|
||
return
|
||
}
|
||
w.WriteHeader(http.StatusNoContent)
|
||
}
|
||
|
||
func (s *Service) command(w http.ResponseWriter, r *http.Request) {
|
||
if r.Method != http.MethodPost {
|
||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||
return
|
||
}
|
||
if !secureEqual(r.Header.Get("X-Heqi-Iot-Token"), s.cfg.HTTP.InternalToken) {
|
||
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
||
return
|
||
}
|
||
var cmd Command
|
||
if err := json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&cmd); err != nil {
|
||
http.Error(w, "invalid json", http.StatusBadRequest)
|
||
return
|
||
}
|
||
if cmd.Identity == "" || cmd.IdempotencyKey == "" || cmd.DeviceID == "" || time.Now().After(cmd.ExpiresAt) {
|
||
http.Error(w, "invalid or expired command", http.StatusUnprocessableEntity)
|
||
return
|
||
}
|
||
deviceID, err := decodeDeviceID(cmd.DeviceID)
|
||
if err != nil {
|
||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||
return
|
||
}
|
||
var payload []byte
|
||
switch cmd.Action {
|
||
case "open_valve":
|
||
payload, err = protocol.ValveCommand(cmd.Controller, cmd.Loop, cmd.Component, true)
|
||
case "close_valve":
|
||
payload, err = protocol.ValveCommand(cmd.Controller, cmd.Loop, cmd.Component, false)
|
||
default:
|
||
err = fmt.Errorf("unsupported action")
|
||
}
|
||
if err != nil {
|
||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||
return
|
||
}
|
||
n := s.packet.Add(1)
|
||
frame := protocol.Frame{KeyID: cmd.KeyID, Version: 1, Control: 0x10, DeviceKind: cmd.DeviceKind, DeviceType: cmd.DeviceType, DeviceModel: cmd.DeviceModel, DeviceID: deviceID, PacketNumber: uint16(n%65535 + 1), Sequence: 1, Final: true, DeviceTime: time.Now(), Payload: payload}
|
||
raw, err := protocol.Encode(frame, s.keys)
|
||
if err != nil {
|
||
http.Error(w, err.Error(), http.StatusUnprocessableEntity)
|
||
return
|
||
}
|
||
topic := strings.ReplaceAll(s.cfg.MQTT.DownTopic, "{deviceId}", cmd.DeviceID)
|
||
if err := s.broker.Publish(topic, raw, false, s.cfg.MQTT.QoS); err != nil {
|
||
http.Error(w, "mqtt publish failed", http.StatusBadGateway)
|
||
return
|
||
}
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusAccepted)
|
||
_ = json.NewEncoder(w).Encode(map[string]any{"command_identity": cmd.Identity, "packet_number": frame.PacketNumber, "status": "dispatched"})
|
||
}
|
||
|
||
func (s *Service) onMessage(_ *mqtt.Client, _ packets.Subscription, message packets.Packet) {
|
||
raw := append([]byte(nil), message.Payload...)
|
||
expectedDeviceID, topicOK := deviceIDFromTopic(s.cfg, message.TopicName)
|
||
envelope := Envelope{Type: "device_message", Topic: message.TopicName, DeviceID: expectedDeviceID, ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano), PayloadHex: hex.EncodeToString(raw)}
|
||
if frame, err := protocol.Decode(raw, s.keys); err == nil {
|
||
deviceID := hex.EncodeToString(frame.DeviceID[:])
|
||
if !topicOK || deviceID != expectedDeviceID {
|
||
s.forwardEnvelope(envelope)
|
||
return
|
||
}
|
||
main := byte(0)
|
||
if len(frame.Payload) > 0 {
|
||
main = frame.Payload[0]
|
||
}
|
||
decoded := &DecodedFrame{KeyID: frame.KeyID, Version: frame.Version, Control: frame.Control, MainID: main, PacketNumber: frame.PacketNumber, Sequence: frame.Sequence, Final: frame.Final, DeviceTime: frame.DeviceTime.Format(time.RFC3339), PayloadHex: hex.EncodeToString(frame.Payload)}
|
||
if main == protocol.MainRealtime {
|
||
if realtime, parseErr := protocol.DecodeRealtime(frame.Payload); parseErr == nil {
|
||
decoded.Realtime = &realtime
|
||
}
|
||
}
|
||
envelope.Frame = decoded
|
||
}
|
||
s.forwardEnvelope(envelope)
|
||
}
|
||
|
||
func (s *Service) forwardEnvelope(envelope Envelope) {
|
||
data, _ := json.Marshal(envelope)
|
||
request, err := http.NewRequest(http.MethodPost, s.cfg.HTTP.CallbackURL, bytes.NewReader(data))
|
||
if err != nil {
|
||
return
|
||
}
|
||
request.Header.Set("Content-Type", "application/json")
|
||
request.Header.Set("X-Heqi-Iot-Token", s.cfg.HTTP.InternalToken)
|
||
response, err := s.client.Do(request)
|
||
if err == nil {
|
||
_ = response.Body.Close()
|
||
}
|
||
}
|
||
|
||
func deviceIDFromTopic(cfg config.Config, topic string) (string, bool) {
|
||
for _, pattern := range []string{cfg.MQTT.UpTopic, cfg.MQTT.AckTopic} {
|
||
parts := strings.Split(pattern, "/")
|
||
values := strings.Split(topic, "/")
|
||
if len(parts) != len(values) {
|
||
continue
|
||
}
|
||
deviceID := ""
|
||
matched := true
|
||
for index := range parts {
|
||
if parts[index] == "+" {
|
||
deviceID = values[index]
|
||
continue
|
||
}
|
||
if parts[index] != values[index] {
|
||
matched = false
|
||
break
|
||
}
|
||
}
|
||
if matched && len(deviceID) == 16 {
|
||
return strings.ToLower(deviceID), true
|
||
}
|
||
}
|
||
return "", false
|
||
}
|
||
|
||
func decodeDeviceID(value string) ([8]byte, error) {
|
||
var result [8]byte
|
||
decoded, err := hex.DecodeString(value)
|
||
if err != nil || len(decoded) != 8 {
|
||
return result, fmt.Errorf("device_id 必须是 16 位 BCD/十六进制字符串")
|
||
}
|
||
copy(result[:], decoded)
|
||
return result, nil
|
||
}
|
||
func secureEqual(left, right string) bool {
|
||
return len(left) == len(right) && subtle.ConstantTimeCompare([]byte(left), []byte(right)) == 1
|
||
}
|