From 594ab2d9c7c6ac805a6689f2f232cb7826c1100d Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 23:50:30 +0800 Subject: [PATCH] feat: support yutong mqtt tls config --- deploy/portainer/docker-compose-go.yml | 10 ++ go/vehicle-gateway/cmd/gateway/main.go | 27 ++-- .../internal/gateway/mqtt_client.go | 116 ++++++++++++++---- .../internal/gateway/mqtt_client_test.go | 116 ++++++++++++++++++ 4 files changed, 236 insertions(+), 33 deletions(-) diff --git a/deploy/portainer/docker-compose-go.yml b/deploy/portainer/docker-compose-go.yml index 9abbef86..32bbadd5 100644 --- a/deploy/portainer/docker-compose-go.yml +++ b/deploy/portainer/docker-compose-go.yml @@ -44,11 +44,21 @@ services: YUTONG_MQTT_CLIENT_ID: ${YUTONG_MQTT_CLIENT_ID:-lingniu-go-yutong-mqtt} YUTONG_MQTT_USERNAME: ${YUTONG_MQTT_USERNAME:-} YUTONG_MQTT_PASSWORD: ${YUTONG_MQTT_PASSWORD:-} + YUTONG_MQTT_CLEAN_SESSION: ${YUTONG_MQTT_CLEAN_SESSION:-false} + YUTONG_MQTT_KEEP_ALIVE_SECONDS: ${YUTONG_MQTT_KEEP_ALIVE_SECONDS:-20} + YUTONG_MQTT_CONNECTION_TIMEOUT_SECONDS: ${YUTONG_MQTT_CONNECTION_TIMEOUT_SECONDS:-10} + YUTONG_MQTT_TLS_CA_PEM: ${YUTONG_MQTT_TLS_CA_PEM:-} + YUTONG_MQTT_TLS_CLIENT_PEM: ${YUTONG_MQTT_TLS_CLIENT_PEM:-} + YUTONG_MQTT_TLS_CLIENT_KEY: ${YUTONG_MQTT_TLS_CLIENT_KEY:-} + YUTONG_MQTT_TLS_HOSTNAME_VERIFICATION_ENABLED: ${YUTONG_MQTT_TLS_HOSTNAME_VERIFICATION_ENABLED:-true} IDENTITY_MYSQL_DSN: ${IDENTITY_MYSQL_DSN:-} VEHICLE_IDENTITY_TABLE: ${VEHICLE_IDENTITY_TABLE:-vehicle_identity_binding} ports: - "${GO_GB32960_TCP_PORT:-32960}:32960" - "${GO_JT808_TCP_PORT:-808}:808" + volumes: + - "${KAFKA_SPOOL_HOST_DIR:-/opt/lingniu-go/spool/gateway}:${KAFKA_SPOOL_DIR:-/data/spool/gateway}" + - "${YUTONG_MQTT_CERT_HOST_DIR:-/opt/lingniuServices/certificate/yutong/vehicledatareception}:${YUTONG_MQTT_CERT_CONTAINER_DIR:-/opt/lingniuServices/certificate/yutong/vehicledatareception}:ro" go-history-writer: <<: *restart-policy diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index 9e70e021..f9a9dd3d 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -81,16 +81,23 @@ func main() { } if envBool("YUTONG_MQTT_ENABLED", false) { client, err := gateway.NewMQTTClient(gateway.MQTTClientConfig{ - EndpointName: env("YUTONG_MQTT_ENDPOINT", "yutong"), - Broker: env("YUTONG_MQTT_URI", ""), - ClientID: env("YUTONG_MQTT_CLIENT_ID", "lingniu-go-yutong-mqtt"), - Username: env("YUTONG_MQTT_USERNAME", ""), - Password: env("YUTONG_MQTT_PASSWORD", ""), - Topics: splitCSV(env("YUTONG_MQTT_TOPICS", env("YUTONG_MQTT_TOPIC", "/ytforward/shln/+"))), - QoS: byte(envInt("YUTONG_MQTT_QOS", 2)), - Sink: sink, - Resolver: resolver, - Logger: logger, + EndpointName: env("YUTONG_MQTT_ENDPOINT", env("YUTONG_MQTT_ENDPOINT_NAME", "yutong")), + Broker: env("YUTONG_MQTT_URI", ""), + ClientID: env("YUTONG_MQTT_CLIENT_ID", "lingniu-go-yutong-mqtt"), + Username: env("YUTONG_MQTT_USERNAME", ""), + Password: env("YUTONG_MQTT_PASSWORD", ""), + Topics: splitCSV(env("YUTONG_MQTT_TOPICS", env("YUTONG_MQTT_TOPIC", "/ytforward/shln/+"))), + QoS: byte(envInt("YUTONG_MQTT_QOS", 2)), + CleanSession: envBool("YUTONG_MQTT_CLEAN_SESSION", false), + KeepAlive: time.Duration(envInt("YUTONG_MQTT_KEEP_ALIVE_SECONDS", 20)) * time.Second, + ConnectTimeout: time.Duration(envInt("YUTONG_MQTT_CONNECTION_TIMEOUT_SECONDS", 10)) * time.Second, + TLSCACertPath: env("YUTONG_MQTT_TLS_CA_PEM", ""), + TLSClientCertPath: env("YUTONG_MQTT_TLS_CLIENT_PEM", ""), + TLSClientKeyPath: env("YUTONG_MQTT_TLS_CLIENT_KEY", ""), + TLSHostnameVerification: envBool("YUTONG_MQTT_TLS_HOSTNAME_VERIFICATION_ENABLED", true), + Sink: sink, + Resolver: resolver, + Logger: logger, }) if err != nil { logger.Error("build yutong mqtt client failed", "error", err) diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go index 4c798d72..40289aba 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go @@ -2,9 +2,13 @@ package gateway import ( "context" + "crypto/tls" + "crypto/x509" "encoding/hex" "errors" + "fmt" "log/slog" + "os" "strings" "time" @@ -17,16 +21,23 @@ import ( ) type MQTTClientConfig struct { - EndpointName string - Broker string - ClientID string - Username string - Password string - Topics []string - QoS byte - Sink eventbus.Sink - Resolver identity.Resolver - Logger *slog.Logger + EndpointName string + Broker string + ClientID string + Username string + Password string + Topics []string + QoS byte + CleanSession bool + KeepAlive time.Duration + ConnectTimeout time.Duration + TLSCACertPath string + TLSClientCertPath string + TLSClientKeyPath string + TLSHostnameVerification bool + Sink eventbus.Sink + Resolver identity.Resolver + Logger *slog.Logger } type MQTTClient struct { @@ -60,17 +71,45 @@ func NewMQTTClient(cfg MQTTClientConfig) (*MQTTClient, error) { } func (c *MQTTClient) Start(ctx context.Context) error { + opts, err := c.buildOptions(ctx) + if err != nil { + return err + } + + c.client = mqtt.NewClient(opts) + token := c.client.Connect() + if token.Wait() && token.Error() != nil { + return token.Error() + } + go func() { + <-ctx.Done() + if c.client != nil && c.client.IsConnected() { + c.client.Disconnect(250) + } + }() + return nil +} + +func (c *MQTTClient) buildOptions(ctx context.Context) (*mqtt.ClientOptions, error) { + keepAlive := c.cfg.KeepAlive + if keepAlive <= 0 { + keepAlive = 20 * time.Second + } + connectTimeout := c.cfg.ConnectTimeout + if connectTimeout <= 0 { + connectTimeout = 10 * time.Second + } opts := mqtt.NewClientOptions(). AddBroker(c.cfg.Broker). SetClientID(c.cfg.ClientID). SetUsername(c.cfg.Username). SetPassword(c.cfg.Password). - SetCleanSession(false). + SetCleanSession(c.cfg.CleanSession). SetAutoReconnect(true). SetConnectRetry(true). SetConnectRetryInterval(5 * time.Second). - SetKeepAlive(20 * time.Second). - SetConnectTimeout(10 * time.Second) + SetKeepAlive(keepAlive). + SetConnectTimeout(connectTimeout) opts.SetDefaultPublishHandler(func(_ mqtt.Client, message mqtt.Message) { c.handleMessage(ctx, message.Topic(), message.Payload()) @@ -89,18 +128,49 @@ func (c *MQTTClient) Start(ctx context.Context) error { c.cfg.Logger.Warn("mqtt connection lost", "broker", c.cfg.Broker, "error", err) } - c.client = mqtt.NewClient(opts) - token := c.client.Connect() - if token.Wait() && token.Error() != nil { - return token.Error() + tlsConfig, err := c.buildTLSConfig() + if err != nil { + return nil, err } - go func() { - <-ctx.Done() - if c.client != nil && c.client.IsConnected() { - c.client.Disconnect(250) + if tlsConfig != nil { + opts.SetTLSConfig(tlsConfig) + } + return opts, nil +} + +func (c *MQTTClient) buildTLSConfig() (*tls.Config, error) { + caPath := strings.TrimSpace(c.cfg.TLSCACertPath) + certPath := strings.TrimSpace(c.cfg.TLSClientCertPath) + keyPath := strings.TrimSpace(c.cfg.TLSClientKeyPath) + if caPath == "" && certPath == "" && keyPath == "" { + return nil, nil + } + config := &tls.Config{ + MinVersion: tls.VersionTLS12, + InsecureSkipVerify: !c.cfg.TLSHostnameVerification, + } + if caPath != "" { + caPEM, err := os.ReadFile(caPath) + if err != nil { + return nil, fmt.Errorf("read mqtt ca certificate: %w", err) } - }() - return nil + roots := x509.NewCertPool() + if !roots.AppendCertsFromPEM(caPEM) { + return nil, fmt.Errorf("parse mqtt ca certificate %s", caPath) + } + config.RootCAs = roots + } + if certPath != "" || keyPath != "" { + if certPath == "" || keyPath == "" { + return nil, errors.New("mqtt client certificate and key must be configured together") + } + cert, err := tls.LoadX509KeyPair(certPath, keyPath) + if err != nil { + return nil, fmt.Errorf("load mqtt client certificate: %w", err) + } + config.Certificates = []tls.Certificate{cert} + } + return config, nil } func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []byte) { diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go index 3d92da11..1f21fd26 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client_test.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client_test.go @@ -2,8 +2,17 @@ package gateway import ( "context" + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" "log/slog" + "math/big" + "os" + "path/filepath" "testing" + "time" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" ) @@ -59,3 +68,110 @@ func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) { t.Fatalf("parse status = %q", sink.raw[0].ParseStatus) } } + +func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) { + dir := t.TempDir() + caPath, certPath, keyPath := writeTestTLSMaterial(t, dir) + client, err := NewMQTTClient(MQTTClientConfig{ + EndpointName: "endpoint-a", + Broker: "ssl://mqtt.example.test:8883", + ClientID: "test-client", + Topics: []string{"/ytforward/shln/+"}, + QoS: 1, + Sink: &recordingSink{}, + Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)), + TLSCACertPath: caPath, + TLSClientCertPath: certPath, + TLSClientKeyPath: keyPath, + TLSHostnameVerification: false, + CleanSession: true, + KeepAlive: 20 * time.Second, + ConnectTimeout: 10 * time.Second, + }) + if err != nil { + t.Fatalf("NewMQTTClient() error = %v", err) + } + + opts, err := client.buildOptions(context.Background()) + if err != nil { + t.Fatalf("buildOptions() error = %v", err) + } + if opts.TLSConfig == nil { + t.Fatal("TLSConfig is nil") + } + if opts.TLSConfig.RootCAs == nil { + t.Fatal("RootCAs is nil") + } + if len(opts.TLSConfig.Certificates) != 1 { + t.Fatalf("client certificates = %d, want 1", len(opts.TLSConfig.Certificates)) + } + if !opts.TLSConfig.InsecureSkipVerify { + t.Fatal("InsecureSkipVerify should be true when hostname verification is disabled") + } + if !opts.CleanSession { + t.Fatal("CleanSession should be true") + } + if got := opts.KeepAlive; got != 20 { + t.Fatalf("KeepAlive = %d, want 20", got) + } + if got := opts.ConnectTimeout; got != 10*time.Second { + t.Fatalf("ConnectTimeout = %v, want 10s", got) + } +} + +func writeTestTLSMaterial(t *testing.T, dir string) (string, string, string) { + t.Helper() + caKey, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatalf("generate ca key: %v", err) + } + caTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "test-ca"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign, + BasicConstraintsValid: true, + IsCA: true, + } + caDER, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &caKey.PublicKey, caKey) + if err != nil { + t.Fatalf("create ca cert: %v", err) + } + clientKey, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatalf("generate client key: %v", err) + } + clientTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(2), + Subject: pkix.Name{CommonName: "test-client"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, + } + clientDER, err := x509.CreateCertificate(rand.Reader, clientTemplate, caTemplate, &clientKey.PublicKey, caKey) + if err != nil { + t.Fatalf("create client cert: %v", err) + } + caPath := filepath.Join(dir, "ca.pem") + certPath := filepath.Join(dir, "client.pem") + keyPath := filepath.Join(dir, "client-key.pem") + writePEM(t, caPath, "CERTIFICATE", caDER) + writePEM(t, certPath, "CERTIFICATE", clientDER) + keyDER := x509.MarshalPKCS1PrivateKey(clientKey) + writePEM(t, keyPath, "RSA PRIVATE KEY", keyDER) + return caPath, certPath, keyPath +} + +func writePEM(t *testing.T, path, typ string, der []byte) { + t.Helper() + file, err := os.Create(path) + if err != nil { + t.Fatalf("create %s: %v", path, err) + } + defer file.Close() + if err := pem.Encode(file, &pem.Block{Type: typ, Bytes: der}); err != nil { + t.Fatalf("write pem %s: %v", path, err) + } +}