feat: add remote control SSH logging and session management over MQTT

README.md

@@ -344,6 +344,7 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_MQTT_URI`                            | An MQTT broker endpoint that is used for bi-directional communication (live view, onvif, etc)   | "tcp://mqtt.kerberos.io:1883"  |
 | `AGENT_MQTT_USERNAME`                       | Username of the MQTT broker.                                                                    | ""                             |
 | `AGENT_MQTT_PASSWORD`                       | Password of the MQTT broker.                                                                    | ""                             |
+| `AGENT_REMOTE_ACCESS_ENABLED`               | Allow encrypted Hub MQTT sessions to stream Agent logs and open an interactive shell. Enable only for trusted deployments. | "false"                        |
 | `AGENT_REALTIME_PROCESSING`                 | If `AGENT_REALTIME_PROCESSING` set to `true`, the agent will send key frames to the topic       | ""                             |
 | `AGENT_REALTIME_PROCESSING_TOPIC`           | The topic to which keyframes will be sent in base64 encoded format.                             | ""                             |
 | `AGENT_STUN_URI`                            | When using WebRTC, you'll need to provide a STUN server.                                        | "stun:turn-fra1.kerberos.io:3478"|
@@ -379,6 +380,13 @@ See [RTSPS and TLS certificates](README-RTSPS-TLS.md) for the complete Bosch UI,
 | `AGENT_SIGNING`                             | Enable 'true' or disable 'false' for signing recordings.                                        | "true"                         |
 | `AGENT_SIGNING_PRIVATE_KEY`                 | The private key (RSA) to sign the recordings fingerprint to validate origin.                    | "" - uses default one if empty |
 
+Remote console access is disabled unless `AGENT_REMOTE_ACCESS_ENABLED=true`.
+The Agent also rejects remote session messages unless Hub encryption or
+end-to-end MQTT encryption is configured and used. A remote shell runs inside
+the Agent process environment as the Agent operating-system user; it is not an
+SSH server and does not expose a new network port. Keep the feature disabled on
+deployments where Hub owners should not have operating-system access.
+
 ### Resumable upload chunk size
 
 Hub and Vault resumable uploads use `AGENT_TUS_CHUNK_SIZE_BYTES` as the maximum

machinery/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/bluenviron/gortsplib/v5 v5.6.3
 	github.com/bluenviron/mediacommon v1.14.0
 	github.com/cedricve/go-onvif v0.0.0-20200222191200-567e8ce298f6
+	github.com/creack/pty v1.1.24
 	github.com/dromara/carbon/v2 v2.6.8
 	github.com/dropbox/dropbox-sdk-go-unofficial/v6 v6.0.5
 	github.com/eclipse/paho.mqtt.golang v1.5.0

machinery/go.sum

@@ -456,6 +456,8 @@ github.com/cncf/xds/go v0.0.0-20240905190251-b4127c9b8d78/go.mod h1:W+zGtBO5Y1Ig
 github.com/cncf/xds/go v0.0.0-20241223141626-cff3c89139a3/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
 github.com/cncf/xds/go v0.0.0-20250121191232-2f005788dc42/go.mod h1:W+zGtBO5Y1IgJhy4+A9GOqVhqLpfZi+vwmdNXUehLA8=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s=
+github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

machinery/src/models/mqtt.go

@@ -324,3 +324,23 @@ type TriggerRelay struct {
 	DeviceId  string `json:"device_id"` // device id
 	Token     string `json:"token"`     // token
 }
+
+// RemoteSessionPayload controls an interactive shell or log stream over MQTT.
+// Data is base64 encoded so terminal control bytes remain valid JSON.
+type RemoteSessionPayload struct {
+	Timestamp int64  `json:"timestamp"`
+	SessionID string `json:"session_id"`
+	Kind      string `json:"kind,omitempty"`
+	Data      string `json:"data,omitempty"`
+	Rows      uint16 `json:"rows,omitempty"`
+	Columns   uint16 `json:"columns,omitempty"`
+	Tail      int    `json:"tail,omitempty"`
+}
+
+type RemoteSessionStatus struct {
+	Timestamp int64  `json:"timestamp"`
+	SessionID string `json:"session_id"`
+	Kind      string `json:"kind"`
+	State     string `json:"state"`
+	Error     string `json:"error,omitempty"`
+}

machinery/src/routers/mqtt/main.go

@@ -59,6 +59,7 @@ func HasMQTTClientModified(configuration *models.Configuration) bool {
 // - kerberos/{hubkey}/device/{devicekey}/motion: a motion signal
 
 func ConfigureMQTT(configDirectory string, configuration *models.Configuration, communication *models.Communication) mqtt.Client {
+	installRemoteAccessHook()
 
 	config := configuration.Config
 
@@ -273,6 +274,7 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 			// We will receive all messages from our hub, so we'll need to filter to the relevant device.
 			if message.Mid != "" && message.Timestamp != 0 && message.DeviceId == configuration.Config.Key {
 				var payload models.Payload
+				remoteAuthenticated := false
 
 				// Messages might be hidden, if so we'll need to decrypt them using the Kerberos Hub private key.
 				if message.Hidden && configuration.Config.HubEncryption == "true" {
@@ -289,8 +291,10 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 								log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message: " + err.Error())
 								return
 							}
-							json.Unmarshal(visibleValue, &payload)
-							message.Payload = payload
+							if err := json.Unmarshal(visibleValue, &payload); err == nil {
+								message.Payload = payload
+								remoteAuthenticated = true
+							}
 						} else {
 							log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message, no private key provided.")
 						}
@@ -338,7 +342,9 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 											log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message: " + err.Error())
 											return
 										}
-										json.Unmarshal(decryptedValue, &payload)
+										if err := json.Unmarshal(decryptedValue, &payload); err == nil {
+											remoteAuthenticated = true
+										}
 									} else {
 										log.Error("routers.mqtt.main.MQTTListenerHandler(): error decrypting message, assymetric keys do not match.")
 										return
@@ -396,6 +402,14 @@ func MQTTListenerHandler(mqttClient mqtt.Client, hubKey string, configDirectory
 					go HandleReceiveHDCandidates(mqttClient, hubKey, payload, configuration, communication)
 				case "trigger-relay":
 					go HandleTriggerRelay(mqttClient, hubKey, payload, configuration, communication)
+				case "remote-session-open":
+					go HandleRemoteSessionOpen(mqttClient, hubKey, payload, remoteAuthenticated, configuration)
+				case "remote-session-input":
+					go HandleRemoteSessionInput(mqttClient, hubKey, payload, remoteAuthenticated, configuration)
+				case "remote-session-resize":
+					go HandleRemoteSessionResize(payload, remoteAuthenticated)
+				case "remote-session-close":
+					go HandleRemoteSessionClose(payload, remoteAuthenticated)
 				}
 
 			}

machinery/src/routers/mqtt/main_test.go

@@ -1,6 +1,8 @@
 package mqtt
 
 import (
+	"errors"
+	"os"
 	"testing"
 	"time"
 
@@ -53,3 +55,73 @@ func TestEnqueueLatestAudioDoesNotBlockNilChannel(t *testing.T) {
 		t.Fatal("enqueueLatestAudio() blocked on a nil channel")
 	}
 }
+
+func TestRemoteAccessRequiresExplicitOptIn(t *testing.T) {
+	previous, present := os.LookupEnv(remoteAccessEnvironment)
+	t.Cleanup(func() {
+		if present {
+			_ = os.Setenv(remoteAccessEnvironment, previous)
+		} else {
+			_ = os.Unsetenv(remoteAccessEnvironment)
+		}
+	})
+
+	_ = os.Unsetenv(remoteAccessEnvironment)
+	if remoteAccessEnabled() {
+		t.Fatal("remoteAccessEnabled() = true without opt-in")
+	}
+	_ = os.Setenv(remoteAccessEnvironment, "true")
+	if !remoteAccessEnabled() {
+		t.Fatal("remoteAccessEnabled() = false after opt-in")
+	}
+}
+
+func TestNormalizeTerminalSize(t *testing.T) {
+	rows, columns := normalizeTerminalSize(0, 0)
+	if rows != 24 || columns != 80 {
+		t.Fatalf("normalizeTerminalSize(0, 0) = (%d, %d), want (24, 80)", rows, columns)
+	}
+
+	rows, columns = normalizeTerminalSize(500, 500)
+	if rows != 200 || columns != 400 {
+		t.Fatalf("normalizeTerminalSize(500, 500) = (%d, %d), want (200, 400)", rows, columns)
+	}
+}
+
+func TestDecodeRemotePayloadRejectsMissingSession(t *testing.T) {
+	_, err := decodeRemotePayload(models.Payload{Value: map[string]interface{}{
+		"kind": "shell",
+	}})
+	if err == nil {
+		t.Fatal("decodeRemotePayload() accepted a missing session id")
+	}
+}
+
+func TestRemoteSessionOpenRejectsUnprovenEncryption(t *testing.T) {
+	previous, present := os.LookupEnv(remoteAccessEnvironment)
+	t.Cleanup(func() {
+		if present {
+			_ = os.Setenv(remoteAccessEnvironment, previous)
+		} else {
+			_ = os.Unsetenv(remoteAccessEnvironment)
+		}
+	})
+	_ = os.Setenv(remoteAccessEnvironment, "true")
+
+	// The listener passes false when an envelope merely claims to be hidden but
+	// no ciphertext was successfully decrypted. The remote handler must reject it.
+	if err := validateRemoteAccess(false); !errors.Is(err, errRemoteUnauthenticated) {
+		t.Fatalf("validateRemoteAccess(false) error = %v, want %v", err, errRemoteUnauthenticated)
+	}
+}
+
+func TestRemoteSessionReservationIsIdempotent(t *testing.T) {
+	manager := newRemoteAccessManager()
+	session := &remoteSession{id: "session-1", kind: "logs"}
+	if _, err := manager.reserve(session); err != nil {
+		t.Fatalf("first reserve() failed: %v", err)
+	}
+	if _, err := manager.reserve(session); !errors.Is(err, errRemoteSessionExists) {
+		t.Fatalf("duplicate reserve() error = %v, want %v", err, errRemoteSessionExists)
+	}
+}

machinery/src/routers/mqtt/remote_access.go

@@ -0,0 +1,397 @@
+package mqtt
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/creack/pty"
+	paho "github.com/eclipse/paho.mqtt.golang"
+	"github.com/kerberos-io/agent/machinery/src/models"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	remoteAccessEnvironment = "AGENT_REMOTE_ACCESS_ENABLED"
+	remoteHistoryLimit      = 500
+	remoteSessionLimit      = 5
+	remoteOutputChunkSize   = 4096
+	remoteInputLimit        = 64 * 1024
+	remoteSessionLifetime   = time.Hour
+)
+
+type remoteSession struct {
+	id     string
+	kind   string
+	pty    *os.File
+	cancel context.CancelFunc
+	logs   chan string
+	done   chan struct{}
+	timer  *time.Timer
+}
+
+type remoteAccessManager struct {
+	mu       sync.Mutex
+	sessions map[string]*remoteSession
+	history  []string
+}
+
+var (
+	remoteAccess             = newRemoteAccessManager()
+	remoteHookOnce           sync.Once
+	errRemoteDisabled        = errors.New("remote access is disabled on this agent")
+	errRemoteUnauthenticated = errors.New("remote access requires encrypted MQTT")
+	errRemoteSessionExists   = errors.New("remote session already exists")
+	errRemoteSessionLimit    = errors.New("remote session limit reached")
+)
+
+func newRemoteAccessManager() *remoteAccessManager {
+	return &remoteAccessManager{sessions: make(map[string]*remoteSession)}
+}
+
+func installRemoteAccessHook() {
+	remoteHookOnce.Do(func() {
+		log.AddHook(remoteAccess)
+	})
+}
+
+func (manager *remoteAccessManager) Levels() []log.Level {
+	return log.AllLevels
+}
+
+func (manager *remoteAccessManager) Fire(entry *log.Entry) error {
+	line, err := json.Marshal(map[string]interface{}{
+		"timestamp": entry.Time.Format(time.RFC3339Nano),
+		"level":     entry.Level.String(),
+		"message":   entry.Message,
+		"fields":    entry.Data,
+	})
+	if err != nil {
+		return nil
+	}
+	encoded := base64.StdEncoding.EncodeToString(append(line, '\n'))
+
+	manager.mu.Lock()
+	manager.history = append(manager.history, encoded)
+	if len(manager.history) > remoteHistoryLimit {
+		manager.history = manager.history[len(manager.history)-remoteHistoryLimit:]
+	}
+	for _, session := range manager.sessions {
+		if session.kind != "logs" {
+			continue
+		}
+		select {
+		case session.logs <- encoded:
+		default:
+		}
+	}
+	manager.mu.Unlock()
+	return nil
+}
+
+func remoteAccessEnabled() bool {
+	enabled, err := strconv.ParseBool(strings.TrimSpace(os.Getenv(remoteAccessEnvironment)))
+	return err == nil && enabled
+}
+
+func validateRemoteAccess(authenticated bool) error {
+	if !authenticated {
+		return errRemoteUnauthenticated
+	}
+	if !remoteAccessEnabled() {
+		return errRemoteDisabled
+	}
+	return nil
+}
+
+func decodeRemotePayload(payload models.Payload) (models.RemoteSessionPayload, error) {
+	data, err := json.Marshal(payload.Value)
+	if err != nil {
+		return models.RemoteSessionPayload{}, err
+	}
+	var request models.RemoteSessionPayload
+	if err := json.Unmarshal(data, &request); err != nil {
+		return models.RemoteSessionPayload{}, err
+	}
+	if request.SessionID == "" || len(request.SessionID) > 128 {
+		return models.RemoteSessionPayload{}, errors.New("invalid remote session id")
+	}
+	return request, nil
+}
+
+func normalizeTerminalSize(rows uint16, columns uint16) (uint16, uint16) {
+	if rows < 5 {
+		rows = 24
+	}
+	if rows > 200 {
+		rows = 200
+	}
+	if columns < 20 {
+		columns = 80
+	}
+	if columns > 400 {
+		columns = 400
+	}
+	return rows, columns
+}
+
+func HandleRemoteSessionOpen(client paho.Client, hubKey string, payload models.Payload, authenticated bool, configuration *models.Configuration) {
+	request, err := decodeRemotePayload(payload)
+	if err != nil {
+		return
+	}
+	if accessErr := validateRemoteAccess(authenticated); accessErr != nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "error", accessErr.Error())
+		return
+	}
+
+	switch request.Kind {
+	case "logs":
+		err = remoteAccess.openLogs(client, hubKey, configuration, request)
+	case "shell":
+		err = remoteAccess.openShell(client, hubKey, configuration, request)
+	default:
+		err = errors.New("unsupported remote session kind")
+	}
+	if err != nil {
+		if errors.Is(err, errRemoteSessionExists) {
+			publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "opened", "")
+			return
+		}
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, request.Kind, "error", err.Error())
+	}
+}
+
+func HandleRemoteSessionInput(client paho.Client, hubKey string, payload models.Payload, authenticated bool, configuration *models.Configuration) {
+	if validateRemoteAccess(authenticated) != nil {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err != nil || len(request.Data) > remoteInputLimit*2 {
+		return
+	}
+	data, err := base64.StdEncoding.DecodeString(request.Data)
+	if err != nil || len(data) > remoteInputLimit {
+		return
+	}
+	remoteAccess.mu.Lock()
+	session := remoteAccess.sessions[request.SessionID]
+	remoteAccess.mu.Unlock()
+	if session == nil || session.kind != "shell" || session.pty == nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, "shell", "error", "remote session is not open")
+		return
+	}
+	if _, err := session.pty.Write(data); err != nil {
+		publishRemoteStatus(client, hubKey, configuration, request.SessionID, "shell", "error", "failed to write terminal input")
+	}
+}
+
+func HandleRemoteSessionResize(payload models.Payload, authenticated bool) {
+	if validateRemoteAccess(authenticated) != nil {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err != nil {
+		return
+	}
+	rows, columns := normalizeTerminalSize(request.Rows, request.Columns)
+	remoteAccess.mu.Lock()
+	session := remoteAccess.sessions[request.SessionID]
+	remoteAccess.mu.Unlock()
+	if session != nil && session.kind == "shell" && session.pty != nil {
+		_ = pty.Setsize(session.pty, &pty.Winsize{Rows: rows, Cols: columns})
+	}
+}
+
+func HandleRemoteSessionClose(payload models.Payload, authenticated bool) {
+	if !authenticated {
+		return
+	}
+	request, err := decodeRemotePayload(payload)
+	if err == nil {
+		remoteAccess.close(request.SessionID)
+	}
+}
+
+func (manager *remoteAccessManager) reserve(session *remoteSession) ([]string, error) {
+	manager.mu.Lock()
+	defer manager.mu.Unlock()
+	if _, exists := manager.sessions[session.id]; exists {
+		return nil, errRemoteSessionExists
+	}
+	if len(manager.sessions) >= remoteSessionLimit {
+		return nil, errRemoteSessionLimit
+	}
+	manager.sessions[session.id] = session
+	history := append([]string(nil), manager.history...)
+	return history, nil
+}
+
+func (manager *remoteAccessManager) expire(client paho.Client, hubKey string, configuration *models.Configuration, session *remoteSession) {
+	session.timer = time.AfterFunc(remoteSessionLifetime, func() {
+		manager.close(session.id)
+		if session.kind == "logs" {
+			publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "closed", "session lifetime reached")
+		}
+	})
+}
+
+func (manager *remoteAccessManager) openLogs(client paho.Client, hubKey string, configuration *models.Configuration, request models.RemoteSessionPayload) error {
+	session := &remoteSession{
+		id:   request.SessionID,
+		kind: "logs",
+		logs: make(chan string, 256),
+		done: make(chan struct{}),
+	}
+	history, err := manager.reserve(session)
+	if err != nil {
+		return err
+	}
+	manager.expire(client, hubKey, configuration, session)
+	tail := request.Tail
+	if tail <= 0 || tail > remoteHistoryLimit {
+		tail = 200
+	}
+	if len(history) > tail {
+		history = history[len(history)-tail:]
+	}
+
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "opened", "")
+	go func() {
+		for _, line := range history {
+			publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, line)
+		}
+		for {
+			select {
+			case line := <-session.logs:
+				publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, line)
+			case <-session.done:
+				return
+			}
+		}
+	}()
+	return nil
+}
+
+func (manager *remoteAccessManager) openShell(client paho.Client, hubKey string, configuration *models.Configuration, request models.RemoteSessionPayload) error {
+	rows, columns := normalizeTerminalSize(request.Rows, request.Columns)
+	ctx, cancel := context.WithCancel(context.Background())
+	session := &remoteSession{
+		id:     request.SessionID,
+		kind:   "shell",
+		cancel: cancel,
+		done:   make(chan struct{}),
+	}
+	if _, err := manager.reserve(session); err != nil {
+		cancel()
+		return err
+	}
+	manager.expire(client, hubKey, configuration, session)
+	command := exec.CommandContext(ctx, "/bin/sh")
+	command.Env = append(os.Environ(), "TERM=xterm-256color", "HISTFILE=/dev/null")
+	terminal, err := pty.StartWithSize(command, &pty.Winsize{Rows: rows, Cols: columns})
+	if err != nil {
+		manager.remove(session.id)
+		cancel()
+		return err
+	}
+	session.pty = terminal
+
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "opened", "")
+	go manager.forwardShell(client, hubKey, configuration, session, command)
+	return nil
+}
+
+func (manager *remoteAccessManager) forwardShell(client paho.Client, hubKey string, configuration *models.Configuration, session *remoteSession, command *exec.Cmd) {
+	buffer := make([]byte, remoteOutputChunkSize)
+	for {
+		count, err := session.pty.Read(buffer)
+		if count > 0 {
+			publishRemoteOutput(client, hubKey, configuration, session.id, session.kind, base64.StdEncoding.EncodeToString(buffer[:count]))
+		}
+		if err != nil {
+			if !errors.Is(err, io.EOF) && !errors.Is(err, os.ErrClosed) {
+				publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "error", "terminal stream closed unexpectedly")
+			}
+			break
+		}
+	}
+	_ = command.Wait()
+	manager.remove(session.id)
+	publishRemoteStatus(client, hubKey, configuration, session.id, session.kind, "closed", "")
+}
+
+func (manager *remoteAccessManager) close(sessionID string) {
+	manager.mu.Lock()
+	session := manager.sessions[sessionID]
+	delete(manager.sessions, sessionID)
+	manager.mu.Unlock()
+	if session == nil {
+		return
+	}
+	if session.cancel != nil {
+		session.cancel()
+	}
+	if session.pty != nil {
+		_ = session.pty.Close()
+	}
+	if session.logs != nil {
+		close(session.done)
+	}
+	if session.timer != nil {
+		session.timer.Stop()
+	}
+}
+
+func (manager *remoteAccessManager) remove(sessionID string) {
+	manager.mu.Lock()
+	session := manager.sessions[sessionID]
+	delete(manager.sessions, sessionID)
+	manager.mu.Unlock()
+	if session != nil && session.timer != nil {
+		session.timer.Stop()
+	}
+}
+
+func publishRemoteStatus(client paho.Client, hubKey string, configuration *models.Configuration, sessionID string, kind string, state string, errorMessage string) {
+	status := models.RemoteSessionStatus{
+		Timestamp: time.Now().Unix(),
+		SessionID: sessionID,
+		Kind:      kind,
+		State:     state,
+		Error:     errorMessage,
+	}
+	value, _ := json.Marshal(status)
+	var statusValue map[string]interface{}
+	_ = json.Unmarshal(value, &statusValue)
+	publishRemote(client, hubKey, configuration, "remote-session-status", statusValue, 1)
+}
+
+func publishRemoteOutput(client paho.Client, hubKey string, configuration *models.Configuration, sessionID string, kind string, data string) {
+	publishRemote(client, hubKey, configuration, "remote-session-output", map[string]interface{}{
+		"timestamp":  time.Now().Unix(),
+		"session_id": sessionID,
+		"kind":       kind,
+		"data":       data,
+	}, 0)
+}
+
+func publishRemote(client paho.Client, hubKey string, configuration *models.Configuration, action string, value map[string]interface{}, qos byte) {
+	message := models.Message{Payload: models.Payload{
+		Action:   action,
+		DeviceId: configuration.Config.Key,
+		Value:    value,
+	}}
+	payload, err := models.PackageMQTTMessage(configuration, message)
+	if err == nil {
+		client.Publish("kerberos/hub/"+hubKey, qos, false, payload)
+	}
+}