From ebc70f17ebda68bc38faba487bc281b4740a5bee Mon Sep 17 00:00:00 2001 From: cynfo3000 Date: Sat, 28 Feb 2026 09:01:04 +0100 Subject: [PATCH] Update-Commands: update, major_update, update_check, reboot Agent: - update: opnsense-update + pkg upgrade -y, optionaler Reboot - major_update: opnsense-update -r , optionaler Reboot - update_check: Prueft Core- und Paket-Updates, liefert Details - reboot: Shutdown mit konfigurierbarer Verzoegerung Backend: - Synchrone API-Endpoints (warten auf Agent-Response) - Response-Waiter-System im Hub (Command-ID basiert) - Konfigurierbare Timeouts pro Command-Typ - POST /agents/{id}/update[-check] und /major-update und /reboot README mit Update-Beispielen und API-Doku aktualisiert --- README.md | 53 +++++- agent/ws/handler.go | 356 ++++++++++++++++++++++++++++++++---- backend/api/tunnel_proxy.go | 115 ++++++++++++ backend/ws/hub.go | 65 ++++++- 4 files changed, 544 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index bfbd54b..5cc1df9 100644 --- a/README.md +++ b/README.md @@ -124,6 +124,15 @@ daemon -f -p /var/run/rmm-agent.pid -o /usr/local/rmm/rmm-agent.log \ | GET | `/api/v1/agent/ws?agent_id=X&api_key=X` | WebSocket-Verbindung (Agent) | | GET | `/api/v1/hub/status` | Hub-Status (connected agents, aktive Tunnel) | +### Updates & Reboot + +| Methode | Endpoint | Beschreibung | +|---------|----------|-------------| +| POST | `/api/v1/agents/{id}/update-check` | Verfuegbare Updates pruefen | +| POST | `/api/v1/agents/{id}/update` | Normales Update (Core + Packages) | +| POST | `/api/v1/agents/{id}/major-update` | Major-Upgrade auf Zielversion | +| POST | `/api/v1/agents/{id}/reboot` | Reboot ausfuehren | + ### Tunnel-Management | Methode | Endpoint | Beschreibung | @@ -158,17 +167,55 @@ curl -sk -H "X-API-Key: KEY" -X POST \ # Im Browser: https://backend-ip:10001/ ``` +### Update-Beispiele + +```bash +# Verfuegbare Updates pruefen +curl -sk -H "X-API-Key: KEY" -X POST \ + https://backend:8443/api/v1/agents/AGENT_ID/update-check + +# Normales Update (ohne Reboot) +curl -sk -H "X-API-Key: KEY" -X POST \ + https://backend:8443/api/v1/agents/AGENT_ID/update + +# Normales Update mit Reboot +curl -sk -H "X-API-Key: KEY" -X POST \ + https://backend:8443/api/v1/agents/AGENT_ID/update \ + -d '{"reboot":true}' + +# Major-Upgrade auf Version 26.7 +curl -sk -H "X-API-Key: KEY" -X POST \ + https://backend:8443/api/v1/agents/AGENT_ID/major-update \ + -d '{"version":"26.7","reboot":true}' + +# Reboot (5s Verzoegerung default) +curl -sk -H "X-API-Key: KEY" -X POST \ + https://backend:8443/api/v1/agents/AGENT_ID/reboot +``` + ### WebSocket Commands (Backend → Agent) ```json // Remote-Befehl ausfuehren {"type":"command","id":"cmd1","command":"exec","params":{"command":"uptime","timeout":30}} -// Tunnel-Session oeffnen (wird automatisch vom Backend gesendet) -{"type":"command","id":"cmd2","command":"tunnel_connect","params":{"session_id":"xxx","tunnel_id":"xxx","target_host":"127.0.0.1","target_port":22}} +// Update-Check +{"type":"command","id":"cmd2","command":"update_check"} + +// Normales Update +{"type":"command","id":"cmd3","command":"update","params":{"reboot":false}} + +// Major-Update +{"type":"command","id":"cmd4","command":"major_update","params":{"version":"26.7","reboot":true}} + +// Reboot +{"type":"command","id":"cmd5","command":"reboot","params":{"delay":5}} + +// Tunnel-Session oeffnen (automatisch vom Backend gesendet) +{"type":"command","id":"cmd6","command":"tunnel_connect","params":{"session_id":"xxx","tunnel_id":"xxx","target_host":"127.0.0.1","target_port":22}} // Tunnel-Session schliessen -{"type":"command","id":"cmd3","command":"tunnel_disconnect","params":{"session_id":"xxx"}} +{"type":"command","id":"cmd7","command":"tunnel_disconnect","params":{"session_id":"xxx"}} ``` ### Binary WebSocket Messages (Tunnel-Daten) diff --git a/agent/ws/handler.go b/agent/ws/handler.go index 7c415b8..a5dc6f9 100644 --- a/agent/ws/handler.go +++ b/agent/ws/handler.go @@ -6,6 +6,7 @@ import ( "log" "os/exec" "strconv" + "strings" "time" ) @@ -27,17 +28,23 @@ func (h *Handler) HandleCommand(msg Message) { switch msg.Command { case "exec": response = h.handleExec(msg) - case "tunnel_open": - // Legacy: wird nicht mehr verwendet, aber fuer Kompatibilitaet - response.Status = "ok" - response.Data = map[string]interface{}{"message": "tunnel_open deprecated, use tunnel_connect"} + case "update": + response = h.handleUpdate(msg) + case "major_update": + response = h.handleMajorUpdate(msg) + case "update_check": + response = h.handleUpdateCheck(msg) + case "reboot": + response = h.handleReboot(msg) case "tunnel_connect": response = h.handleTunnelConnect(msg) case "tunnel_disconnect": h.handleTunnelDisconnect(msg) - return // Keine Response noetig + return + case "tunnel_open": + response.Status = "ok" + response.Data = map[string]interface{}{"message": "tunnel_open deprecated, use tunnel_connect"} case "tunnel_close": - // Legacy: alle Sessions fuer diesen Tunnel schliessen response.Status = "ok" default: response.Status = "error" @@ -49,11 +56,9 @@ func (h *Handler) HandleCommand(msg Message) { } } +// handleExec fuehrt einen beliebigen Befehl aus func (h *Handler) handleExec(msg Message) Message { - response := Message{ - Type: "response", - ID: msg.ID, - } + response := Message{Type: "response", ID: msg.ID} command, ok := msg.Params["command"].(string) if !ok || command == "" { @@ -73,34 +78,269 @@ func (h *Handler) handleExec(msg Message) Message { } } - ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutSec)*time.Second) - defer cancel() - - cmd := exec.CommandContext(ctx, "/bin/sh", "-c", command) - output, err := cmd.CombinedOutput() + output, err := h.runCommand(command, timeoutSec) if err != nil { response.Status = "error" response.Error = err.Error() - response.Data = map[string]interface{}{ - "output": string(output), - } + response.Data = map[string]interface{}{"output": output} } else { response.Status = "ok" - response.Data = map[string]interface{}{ - "output": string(output), - } + response.Data = map[string]interface{}{"output": output} } return response } -func (h *Handler) handleTunnelConnect(msg Message) Message { - response := Message{ - Type: "response", - ID: msg.ID, +// handleUpdateCheck prueft ob Updates verfuegbar sind +func (h *Handler) handleUpdateCheck(msg Message) Message { + response := Message{Type: "response", ID: msg.ID} + log.Println("Update-Check gestartet...") + + result := map[string]interface{}{} + + // 1. OPNsense Core Update pruefen + coreOutput, coreErr := h.runCommand("/usr/local/sbin/opnsense-update -c", 60) + if coreErr != nil { + // Exit 1 = Update verfuegbar + result["core_update_available"] = true + result["core_update_info"] = strings.TrimSpace(coreOutput) + } else { + result["core_update_available"] = false + result["core_update_info"] = "Kein Core-Update verfuegbar" } + // 2. Paket-Updates pruefen + pkgOutput, _ := h.runCommand("/usr/sbin/pkg upgrade -n", 60) + pendingPackages := h.parsePkgUpgrade(pkgOutput) + result["package_update_available"] = len(pendingPackages) > 0 + result["pending_packages"] = pendingPackages + result["pending_count"] = len(pendingPackages) + + response.Status = "ok" + response.Data = result + log.Printf("Update-Check abgeschlossen: Core=%v, Packages=%d", + result["core_update_available"], len(pendingPackages)) + return response +} + +// handleUpdate fuehrt ein normales Update durch (Core + Packages) +func (h *Handler) handleUpdate(msg Message) Message { + response := Message{Type: "response", ID: msg.ID} + log.Println("Normales Update gestartet...") + + reboot := false + if r, ok := msg.Params["reboot"].(bool); ok { + reboot = r + } + + result := map[string]interface{}{ + "steps": []map[string]interface{}{}, + } + var steps []map[string]interface{} + + // Schritt 1: opnsense-update (Core + Firmware) + log.Println("Update Schritt 1: opnsense-update") + coreOutput, coreErr := h.runCommand("/usr/local/sbin/opnsense-update", 600) + step1 := map[string]interface{}{ + "step": "opnsense-update", + "output": coreOutput, + "ok": coreErr == nil, + } + if coreErr != nil { + step1["error"] = coreErr.Error() + } + steps = append(steps, step1) + + // Schritt 2: pkg upgrade -y (Paket-Updates) + log.Println("Update Schritt 2: pkg upgrade -y") + pkgOutput, pkgErr := h.runCommand("/usr/sbin/pkg upgrade -y", 600) + step2 := map[string]interface{}{ + "step": "pkg upgrade", + "output": pkgOutput, + "ok": pkgErr == nil, + } + if pkgErr != nil { + step2["error"] = pkgErr.Error() + } + steps = append(steps, step2) + + // Schritt 3: opnsense-update -c (Nachpruefung) + log.Println("Update Schritt 3: Nachpruefung") + checkOutput, checkErr := h.runCommand("/usr/local/sbin/opnsense-update -c", 60) + stillPending := checkErr != nil + step3 := map[string]interface{}{ + "step": "verify", + "output": checkOutput, + "updates_pending": stillPending, + } + steps = append(steps, step3) + + result["steps"] = steps + result["reboot_requested"] = reboot + + // Gesamtstatus + if coreErr == nil && pkgErr == nil { + response.Status = "ok" + result["message"] = "Update erfolgreich abgeschlossen" + } else { + response.Status = "ok" // Nicht "error" — teilweise Updates sind OK + result["message"] = "Update mit Warnungen abgeschlossen" + } + + // Reboot wenn gewuenscht + if reboot { + log.Println("Reboot nach Update angefordert...") + result["reboot"] = "Reboot in 5 Sekunden..." + response.Data = result + + // Response erst senden, dann rebooten + go func() { + time.Sleep(5 * time.Second) + log.Println("Reboot wird ausgefuehrt...") + exec.Command("/sbin/shutdown", "-r", "now").Run() + }() + + return response + } + + response.Data = result + log.Println("Update abgeschlossen") + return response +} + +// handleMajorUpdate fuehrt ein Major-Upgrade durch +func (h *Handler) handleMajorUpdate(msg Message) Message { + response := Message{Type: "response", ID: msg.ID} + + targetVersion, ok := msg.Params["version"].(string) + if !ok || targetVersion == "" { + response.Status = "error" + response.Error = "Parameter 'version' erforderlich (z.B. '26.7')" + return response + } + + reboot := false + if r, ok := msg.Params["reboot"].(bool); ok { + reboot = r + } + + log.Printf("Major-Update auf Version %s gestartet...", targetVersion) + + result := map[string]interface{}{ + "target_version": targetVersion, + } + var steps []map[string]interface{} + + // Schritt 1: opnsense-update -r (Release wechseln) + log.Printf("Major-Update Schritt 1: opnsense-update -r %s", targetVersion) + releaseCmd := fmt.Sprintf("/usr/local/sbin/opnsense-update -r %s", targetVersion) + releaseOutput, releaseErr := h.runCommand(releaseCmd, 300) + step1 := map[string]interface{}{ + "step": fmt.Sprintf("opnsense-update -r %s", targetVersion), + "output": releaseOutput, + "ok": releaseErr == nil, + } + if releaseErr != nil { + step1["error"] = releaseErr.Error() + steps = append(steps, step1) + + // Bei Fehler im Release-Wechsel abbrechen + result["steps"] = steps + result["message"] = "Release-Wechsel fehlgeschlagen" + response.Status = "error" + response.Error = "Release-Wechsel fehlgeschlagen" + response.Data = result + return response + } + steps = append(steps, step1) + + // Schritt 2: opnsense-update (eigentliches Update laden) + log.Println("Major-Update Schritt 2: opnsense-update") + updateOutput, updateErr := h.runCommand("/usr/local/sbin/opnsense-update", 900) + step2 := map[string]interface{}{ + "step": "opnsense-update", + "output": updateOutput, + "ok": updateErr == nil, + } + if updateErr != nil { + step2["error"] = updateErr.Error() + } + steps = append(steps, step2) + + // Schritt 3: pkg upgrade -y + log.Println("Major-Update Schritt 3: pkg upgrade -y") + pkgOutput, pkgErr := h.runCommand("/usr/sbin/pkg upgrade -y", 600) + step3 := map[string]interface{}{ + "step": "pkg upgrade", + "output": pkgOutput, + "ok": pkgErr == nil, + } + if pkgErr != nil { + step3["error"] = pkgErr.Error() + } + steps = append(steps, step3) + + result["steps"] = steps + result["reboot_requested"] = reboot + + if updateErr == nil { + result["message"] = fmt.Sprintf("Major-Update auf %s abgeschlossen", targetVersion) + } else { + result["message"] = fmt.Sprintf("Major-Update auf %s mit Warnungen", targetVersion) + } + + response.Status = "ok" + + if reboot { + log.Println("Reboot nach Major-Update angefordert...") + result["reboot"] = "Reboot in 5 Sekunden..." + response.Data = result + + go func() { + time.Sleep(5 * time.Second) + log.Println("Reboot wird ausgefuehrt...") + exec.Command("/sbin/shutdown", "-r", "now").Run() + }() + + return response + } + + response.Data = result + log.Printf("Major-Update auf %s abgeschlossen", targetVersion) + return response +} + +// handleReboot fuehrt einen Reboot durch +func (h *Handler) handleReboot(msg Message) Message { + response := Message{Type: "response", ID: msg.ID} + + delaySec := 5 + if d, ok := msg.Params["delay"].(float64); ok && d > 0 { + delaySec = int(d) + } + + log.Printf("Reboot angefordert (Verzoegerung: %ds)", delaySec) + + response.Status = "ok" + response.Data = map[string]interface{}{ + "message": fmt.Sprintf("Reboot in %d Sekunden", delaySec), + } + + go func() { + time.Sleep(time.Duration(delaySec) * time.Second) + log.Println("Reboot wird ausgefuehrt...") + exec.Command("/sbin/shutdown", "-r", "now").Run() + }() + + return response +} + +// --- Tunnel Handlers --- + +func (h *Handler) handleTunnelConnect(msg Message) Message { + response := Message{Type: "response", ID: msg.ID} + sessionID, _ := msg.Params["session_id"].(string) tunnelID, _ := msg.Params["tunnel_id"].(string) targetHost, _ := msg.Params["target_host"].(string) @@ -120,23 +360,71 @@ func (h *Handler) handleTunnelConnect(msg Message) Message { } response.Status = "ok" - response.Data = map[string]interface{}{ - "session_id": sessionID, - } - + response.Data = map[string]interface{}{"session_id": sessionID} return response } func (h *Handler) handleTunnelDisconnect(msg Message) { sessionID, _ := msg.Params["session_id"].(string) - if sessionID == "" { - return + if sessionID != "" { + h.client.tunnelManager.DisconnectSession(sessionID) } - - h.client.tunnelManager.DisconnectSession(sessionID) } -// SendSessionData schickt Daten als Binary-Message zurueck zum Backend +// --- Helper --- + +// runCommand fuehrt einen Shell-Befehl mit Timeout aus +func (h *Handler) runCommand(command string, timeoutSec int) (string, error) { + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutSec)*time.Second) + defer cancel() + + cmd := exec.CommandContext(ctx, "/bin/sh", "-c", command) + output, err := cmd.CombinedOutput() + return string(output), err +} + +// parsePkgUpgrade parst die Ausgabe von "pkg upgrade -n" +func (h *Handler) parsePkgUpgrade(output string) []map[string]string { + var packages []map[string]string + + inUpgradeBlock := false + for _, line := range strings.Split(output, "\n") { + trimmed := strings.TrimSpace(line) + + if strings.Contains(trimmed, "packages to be UPGRADED") || + strings.Contains(trimmed, "will be affected") { + inUpgradeBlock = true + continue + } + + if inUpgradeBlock && strings.Contains(trimmed, "->") { + parts := strings.SplitN(trimmed, ":", 2) + if len(parts) == 2 { + pkg := strings.TrimSpace(parts[0]) + versions := strings.TrimSpace(parts[1]) + + arrowParts := strings.SplitN(versions, "->", 2) + if len(arrowParts) == 2 { + packages = append(packages, map[string]string{ + "package": pkg, + "current_version": strings.TrimSpace(arrowParts[0]), + "new_version": strings.TrimSpace(arrowParts[1]), + }) + } + } + } + + if inUpgradeBlock && (trimmed == "" || strings.HasPrefix(trimmed, "Number of")) { + if !strings.HasPrefix(trimmed, "Number of") { + inUpgradeBlock = false + } + } + } + + return packages +} + +// SendSessionData schickt Tunnel-Daten als Binary-Message zurueck zum Backend func (h *Handler) SendSessionData(sessionID string, data []byte) error { idBytes := []byte(sessionID) if len(idBytes) > 255 { diff --git a/backend/api/tunnel_proxy.go b/backend/api/tunnel_proxy.go index 0c8997d..1e73723 100644 --- a/backend/api/tunnel_proxy.go +++ b/backend/api/tunnel_proxy.go @@ -12,6 +12,10 @@ import ( func (h *Handler) setupTunnelRoutes(mux *http.ServeMux, hub *ws.Hub) { // Command-Endpoints mux.HandleFunc("POST /api/v1/agents/{id}/exec", h.executeCommand(hub)) + mux.HandleFunc("POST /api/v1/agents/{id}/update", h.agentUpdate(hub)) + mux.HandleFunc("POST /api/v1/agents/{id}/major-update", h.agentMajorUpdate(hub)) + mux.HandleFunc("POST /api/v1/agents/{id}/update-check", h.agentUpdateCheck(hub)) + mux.HandleFunc("POST /api/v1/agents/{id}/reboot", h.agentReboot(hub)) mux.HandleFunc("POST /api/v1/agents/{id}/tunnel", h.openTunnel(hub)) mux.HandleFunc("DELETE /api/v1/agents/{id}/tunnel/{tunnel_id}", h.closeTunnel(hub)) mux.HandleFunc("GET /api/v1/agents/{id}/tunnels", h.listTunnels(hub)) @@ -178,6 +182,117 @@ func (h *Handler) httpProxy(hub *ws.Hub) http.HandlerFunc { } } +// --- Update Commands --- + +func (h *Handler) agentUpdateCheck(hub *ws.Hub) http.HandlerFunc { + return h.sendAgentCommand(hub, "update_check", nil, 120) +} + +func (h *Handler) agentUpdate(hub *ws.Hub) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + agentID := r.PathValue("id") + + var req struct { + Reboot bool `json:"reboot"` + } + json.NewDecoder(r.Body).Decode(&req) + + params := map[string]interface{}{ + "reboot": req.Reboot, + } + + h.sendCommandAndWait(hub, agentID, "update", params, 900, w) + } +} + +func (h *Handler) agentMajorUpdate(hub *ws.Hub) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + agentID := r.PathValue("id") + + var req struct { + Version string `json:"version"` + Reboot bool `json:"reboot"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Version == "" { + writeError(w, http.StatusBadRequest, "Parameter 'version' erforderlich (z.B. '26.7')") + return + } + + params := map[string]interface{}{ + "version": req.Version, + "reboot": req.Reboot, + } + + h.sendCommandAndWait(hub, agentID, "major_update", params, 1200, w) + } +} + +func (h *Handler) agentReboot(hub *ws.Hub) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + agentID := r.PathValue("id") + + var req struct { + Delay int `json:"delay"` + } + json.NewDecoder(r.Body).Decode(&req) + + params := map[string]interface{}{} + if req.Delay > 0 { + params["delay"] = req.Delay + } + + h.sendCommandAndWait(hub, agentID, "reboot", params, 30, w) + } +} + +// sendAgentCommand erstellt einen einfachen Command-Handler ohne Request-Body +func (h *Handler) sendAgentCommand(hub *ws.Hub, command string, params map[string]interface{}, timeoutSec int) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + agentID := r.PathValue("id") + h.sendCommandAndWait(hub, agentID, command, params, timeoutSec, w) + } +} + +// sendCommandAndWait schickt ein Command an den Agent und wartet auf Response +func (h *Handler) sendCommandAndWait(hub *ws.Hub, agentID, command string, params map[string]interface{}, timeoutSec int, w http.ResponseWriter) { + if !hub.IsAgentConnected(agentID) { + writeError(w, http.StatusServiceUnavailable, "Agent nicht verbunden") + return + } + + cmdID := generateID() + + cmd := map[string]interface{}{ + "type": "command", + "id": cmdID, + "command": command, + } + if params != nil { + cmd["params"] = params + } + + // Response-Channel registrieren + responseCh := hub.RegisterResponseWaiter(cmdID) + defer hub.UnregisterResponseWaiter(cmdID) + + cmdJSON, _ := json.Marshal(cmd) + if err := hub.SendToAgent(agentID, cmdJSON); err != nil { + log.Printf("Command '%s' senden fehlgeschlagen: %v", command, err) + writeError(w, http.StatusInternalServerError, "Command senden fehlgeschlagen") + return + } + + log.Printf("Command '%s' (ID: %s) an Agent %s gesendet, warte auf Response...", command, cmdID, agentID) + + // Auf Response warten + select { + case resp := <-responseCh: + writeJSON(w, http.StatusOK, resp) + case <-time.After(time.Duration(timeoutSec) * time.Second): + writeError(w, http.StatusGatewayTimeout, "Timeout: Agent hat nicht rechtzeitig geantwortet") + } +} + func (h *Handler) hubStatus(hub *ws.Hub) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { stats := hub.GetAgentStats() diff --git a/backend/ws/hub.go b/backend/ws/hub.go index 67b9c7a..db7a840 100644 --- a/backend/ws/hub.go +++ b/backend/ws/hub.go @@ -22,14 +22,19 @@ type Hub struct { broadcast chan []byte tunnelManager *TunnelManager + + // Response-Waiters: Command-ID -> Response-Channel + responseWaiters map[string]chan map[string]interface{} + waitersMux sync.RWMutex } func NewHub() *Hub { hub := &Hub{ - agents: make(map[string]*AgentConnection), - register: make(chan *AgentConnection), - unregister: make(chan *AgentConnection), - broadcast: make(chan []byte, 256), + agents: make(map[string]*AgentConnection), + register: make(chan *AgentConnection), + unregister: make(chan *AgentConnection), + broadcast: make(chan []byte, 256), + responseWaiters: make(map[string]chan map[string]interface{}), } hub.tunnelManager = NewTunnelManager(hub) @@ -193,6 +198,22 @@ func (h *Hub) cleanupInactiveAgents() { } } +// RegisterResponseWaiter registriert einen Channel fuer eine Command-Response +func (h *Hub) RegisterResponseWaiter(cmdID string) chan map[string]interface{} { + ch := make(chan map[string]interface{}, 1) + h.waitersMux.Lock() + h.responseWaiters[cmdID] = ch + h.waitersMux.Unlock() + return ch +} + +// UnregisterResponseWaiter entfernt einen Response-Waiter +func (h *Hub) UnregisterResponseWaiter(cmdID string) { + h.waitersMux.Lock() + delete(h.responseWaiters, cmdID) + h.waitersMux.Unlock() +} + // GetTunnelManager gibt den TunnelManager zurück (für API-Handler) func (h *Hub) GetTunnelManager() *TunnelManager { return h.tunnelManager @@ -245,10 +266,38 @@ func (h *Hub) ProcessAgentMessage(agentID string, messageType int, data []byte) return nil // Ignorieren wenn nicht parsebar } - // Tunnel-Connect Responses verarbeiten - if msg.Type == "response" && msg.Status == "ok" && msg.Data != nil { - if sessionID, ok := msg.Data["session_id"].(string); ok { - h.tunnelManager.ConfirmSession(sessionID) + if msg.Type == "response" { + // Tunnel-Connect Responses verarbeiten + if msg.Status == "ok" && msg.Data != nil { + if sessionID, ok := msg.Data["session_id"].(string); ok { + h.tunnelManager.ConfirmSession(sessionID) + } + } + + // Response an wartende API-Handler weiterleiten + if msg.ID != "" { + h.waitersMux.RLock() + ch, exists := h.responseWaiters[msg.ID] + h.waitersMux.RUnlock() + + if exists { + resp := map[string]interface{}{ + "status": msg.Status, + "data": msg.Data, + } + if msg.Status == "error" { + var errMsg struct { + Error string `json:"error"` + } + json.Unmarshal(data, &errMsg) + resp["error"] = errMsg.Error + } + + select { + case ch <- resp: + default: + } + } } }