release: v1.2.0 - Pure Go update engine, enhanced journal log fetcher, and protocol concurrency fixes

2026-08-03 02:10:22 -04:00
parent e2ec7e7234
commit af78d4683e
5 changed files with 183 additions and 47 deletions
@@ -0,0 +1,25 @@
+# Changelog
+
+All notable changes to the `theta-agent` daemon will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [v1.2.0] - 2026-08-03
+
+### Added
+- **Protocol v1.1.0 Compliance**: Full alignment with `PROTOCOL.md` (v1.1.0) specification.
+- **Ed25519 Cryptographic Verification**: Verification of Ed25519 Base64 signatures for high-risk C2 commands (`reboot`, `service_restart`, `configure_ldap`, `arbitrary_bash`, `update_binary`).
+- **Enhanced Journal Log Fetcher (`fetch_logs`)**: Support for querying service-specific systemd logs (`service` parameter) with configurable line count (`lines` parameter).
+- **Pure Go Self-Update Engine (`update_binary`)**: Replaced shell script execution with pure Go HTTP client fetching, SHA256 verification, atomic file replacement, and clean daemon restart.
+
+### Fixed
+- **Goroutine Leak Prevention**: Added `stopCh` lifecycle management to terminate background telemetry and heartbeat tickers upon WebSocket disconnection.
+- **Dynamic Config Rerenders**: Resolved data races when toggling capabilities or reloading `agent.yml` via `reload_config`.
+- **Config Path Uniformity**: Standardized canonical configuration file path across codebase, installer, and documentation to `/etc/theta42/agent.yml`.
+
+## [v0.1.0] - 2026-08-01
+
+### Added
+- Initial release of `theta-agent` Go daemon replacing legacy bash metric scripts.
+- Persistent outbound WebSocket telemetry and local capability matrix enforcement.
@@ -114,38 +114,57 @@ func collectGPUUsage(exec Executor) float64 {
 }

 // StartTelemetryLoop manages the initial discovery push and the periodic telemetry stream.
-func StartTelemetryLoop(c *websocket.Conn, cfg *Config, exec Executor) {
+func StartTelemetryLoop(c MessageWriter, cm *ConfigManager, exec Executor, stopCh <-chan struct{}) {
+	cfg := cm.Get()
+
 	// 1. Immediate Discovery Push
 	pushDiscovery(c, cfg)

+	// If telemetry capability is disabled in agent.yml, return early after discovery
+	if !cfg.Capabilities.Telemetry {
+		log.Println("Telemetry capability is disabled in agent.yml; skipping telemetry stream.")
+		return
+	}
+
 	// 2. Periodic Telemetry Stream
 	ticker := time.NewTicker(30 * time.Second)
 	go func() {
+		defer ticker.Stop()
 		var lastIPs []string
-		for range ticker.C {
-			// Network Change Detection
-			currentIPs := collectIPs()
-			if !equalSlices(lastIPs, currentIPs) {
-				log.Println("Network change detected. Pushing discovery update...")
-				pushDiscovery(c, cfg)
-				lastIPs = currentIPs
-			}
-
-			telemetry := CollectTelemetryData(exec)
-			payload, _ := json.Marshal(WSMessage{
-				Type: "telemetry",
-				Payload: map[string]interface{}{
-					"cpu_usage_percent":  telemetry.CPUUsagePercent,
-					"ram_usage_percent":  telemetry.RAMUsagePercent,
-					"disk_usage_percent": telemetry.DiskUsagePercent,
-					"zfs_health":         telemetry.ZFSHealth,
-					"gpu_usage_percent":   telemetry.GPUUsage,
-					"timestamp":          telemetry.Timestamp,
-				},
-			})
-			if err := c.WriteMessage(websocket.TextMessage, payload); err != nil {
-				log.Printf("Failed to stream telemetry: %v", err)
+		for {
+			select {
+			case <-stopCh:
 				return
+			case <-ticker.C:
+				currentCFG := cm.Get()
+				if !currentCFG.Capabilities.Telemetry {
+					continue
+				}
+
+				// Network Change Detection
+				currentIPs := collectIPs()
+				if !equalSlices(lastIPs, currentIPs) {
+					log.Println("Network change detected. Pushing discovery update...")
+					pushDiscovery(c, currentCFG)
+					lastIPs = currentIPs
+				}
+
+				telemetry := CollectTelemetryData(exec)
+				payload, _ := json.Marshal(WSMessage{
+					Type: "telemetry",
+					Payload: map[string]interface{}{
+						"cpu_usage_percent":  telemetry.CPUUsagePercent,
+						"ram_usage_percent":  telemetry.RAMUsagePercent,
+						"disk_usage_percent": telemetry.DiskUsagePercent,
+						"zfs_health":         telemetry.ZFSHealth,
+						"gpu_usage_percent":   telemetry.GPUUsage,
+						"timestamp":          telemetry.Timestamp,
+					},
+				})
+				if err := c.WriteMessage(websocket.TextMessage, payload); err != nil {
+					log.Printf("Failed to stream telemetry: %v", err)
+					return
+				}
 			}
 		}
 	}()
@@ -176,7 +195,7 @@ func equalSlices(a, b []string) bool {
 	return true
 }

-func pushDiscovery(c *websocket.Conn, cfg *Config) {
+func pushDiscovery(c MessageWriter, cfg *Config) {
 	discovery := CollectDiscoveryData(cfg)
 	discoveryPayload, _ := json.Marshal(discovery)

@@ -2,11 +2,16 @@ package main

 import (
 	"crypto/ed25519"
+	"crypto/sha256"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log"
+	"net/http"
 	"net/url"
+	"os"
+	"path/filepath"
 	"strings"
 	"time"

@@ -83,17 +88,25 @@ func connectWebSocket(cm *ConfigManager, exec Executor) {

 		log.Println("Successfully connected to SSO Manager.")

-		// Start telemetry and discovery
-		StartTelemetryLoop(c, cfg, exec)
+		stopCh := make(chan struct{})
+
+		// Start telemetry and discovery with stopCh lifecycle control
+		StartTelemetryLoop(c, cm, exec, stopCh)

 		// Heartbeat loop
 		go func() {
 			ticker := time.NewTicker(60 * time.Second)
-			for range ticker.C {
-				hb := WSMessage{Type: "heartbeat", Payload: map[string]interface{}{"timestamp": time.Now().Format(time.RFC3339)}}
-				payload, _ := json.Marshal(hb)
-				if err := c.WriteMessage(websocket.TextMessage, payload); err != nil {
+			defer ticker.Stop()
+			for {
+				select {
+				case <-stopCh:
 					return
+				case <-ticker.C:
+					hb := WSMessage{Type: "heartbeat", Payload: map[string]interface{}{"timestamp": time.Now().Format(time.RFC3339)}}
+					payload, _ := json.Marshal(hb)
+					if err := c.WriteMessage(websocket.TextMessage, payload); err != nil {
+						return
+					}
 				}
 			}
 		}()
@@ -116,6 +129,7 @@ func connectWebSocket(cm *ConfigManager, exec Executor) {
 		}

 		// Cleanup on disconnect
+		close(stopCh)
 		c.Close()

 		log.Println("WebSocket disconnected. Reconnecting in 5 seconds...")
@@ -142,15 +156,33 @@ func handleCommand(cm *ConfigManager, msg WSMessage, c MessageWriter, exec Execu
 			sendResponse("ok", "configuration reloaded")
 		}
 	case "fetch_logs":
-		out, err := exec.Execute("journalctl", "-u", "theta-agent", "-n", "100")
+		serviceName, _ := msg.Payload["service"].(string)
+		if serviceName == "" {
+			serviceName = "theta-agent"
+		}
+
+		if serviceName != "theta-agent" && !cfg.Capabilities.CanManageService(serviceName) {
+			log.Printf("Fetch logs rejected for '%s': not in allowed service list", serviceName)
+			sendResponse("error", "service log fetch rejected")
+			return
+		}
+
+		linesCount := 100
+		if l, ok := msg.Payload["lines"].(float64); ok && l > 0 {
+			linesCount = int(l)
+		}
+
+		log.Printf("Fetching logs for service %s (%d lines)...", serviceName, linesCount)
+		out, err := exec.Execute("journalctl", "-u", serviceName, "-n", fmt.Sprintf("%d", linesCount), "--no-pager")
 		if err != nil {
 			log.Printf("Log fetch failed: %v", err)
 			sendResponse("error", "failed to fetch logs")
 			return
 		}
-		resp := map[string]string{
-			"status": "ok",
-			"logs":   string(out),
+		resp := map[string]interface{}{
+			"status":  "ok",
+			"service": serviceName,
+			"logs":    string(out),
 		}
 		respPayload, _ := json.Marshal(resp)
 		c.WriteMessage(websocket.TextMessage, respPayload)
@@ -160,28 +192,25 @@ func handleCommand(cm *ConfigManager, msg WSMessage, c MessageWriter, exec Execu
 			sendResponse("error", "signature verification failed")
 			return
 		}
-		if !cfg.Capabilities.ArbitraryBash { // Use Bash as a proxy for "dangerous update" capability
+		if !cfg.Capabilities.ArbitraryBash {
 			sendResponse("error", "update capability disabled")
 			return
 		}

-		url, _ := msg.Payload["url"].(string)
+		urlStr, _ := msg.Payload["url"].(string)
 		checksum, _ := msg.Payload["sha256"].(string)
-		if url == "" || checksum == "" {
-			sendResponse("error", "missing url or checksum")
+		if urlStr == "" || checksum == "" {
+			sendResponse("error", "missing url or sha256 checksum")
 			return
 		}

-		log.Printf("Updating binary from %s...", url)
-		// implementation of download and replace
-		// ... (simplified for now, using a shell command via executor for brevity in this turn)
-		script := fmt.Sprintf("curl -fsSL %s -o /tmp/theta-agent.new && sha256sum -c <(echo '%s  /tmp/theta-agent.new') && mv /tmp/theta-agent.new $(readlink -f /proc/self/exe)", url, checksum)
-		if _, err := exec.Execute("bash", "-c", script); err != nil {
+		log.Printf("Updating binary from %s...", urlStr)
+		if err := downloadAndUpdateBinary(urlStr, checksum); err != nil {
 			log.Printf("Update failed: %v", err)
-			sendResponse("error", "update failed")
+			sendResponse("error", fmt.Sprintf("update failed: %v", err))
 			return
 		}
-		sendResponse("ok", "update applied. restarting agent...")
+		sendResponse("ok", "update applied successfully; restarting agent...")
 		os.Exit(0)
 	case "config":
 		log.Printf("Received config payload: %v", msg.Payload)
@@ -287,3 +316,56 @@ func handleCommand(cm *ConfigManager, msg WSMessage, c MessageWriter, exec Execu
 		sendResponse("error", "unknown command type")
 	}
 }
+
+func downloadAndUpdateBinary(downloadURL string, expectedSHA256 string) error {
+	resp, err := http.Get(downloadURL)
+	if err != nil {
+		return fmt.Errorf("http fetch failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("unexpected http status: %s", resp.Status)
+	}
+
+	tmpFile, err := os.CreateTemp("", "theta-agent-update-*")
+	if err != nil {
+		return fmt.Errorf("failed to create temp file: %w", err)
+	}
+	tmpPath := tmpFile.Name()
+	defer os.Remove(tmpPath)
+
+	hasher := sha256.New()
+	writer := io.MultiWriter(tmpFile, hasher)
+
+	if _, err := io.Copy(writer, resp.Body); err != nil {
+		tmpFile.Close()
+		return fmt.Errorf("failed to save binary: %w", err)
+	}
+	tmpFile.Close()
+
+	actualSHA256 := fmt.Sprintf("%x", hasher.Sum(nil))
+	if !strings.EqualFold(actualSHA256, strings.TrimSpace(expectedSHA256)) {
+		return fmt.Errorf("sha256 mismatch: expected %s, got %s", expectedSHA256, actualSHA256)
+	}
+
+	if err := os.Chmod(tmpPath, 0755); err != nil {
+		return fmt.Errorf("failed to set executable permissions: %w", err)
+	}
+
+	selfPath, err := os.Executable()
+	if err != nil {
+		return fmt.Errorf("failed to resolve current binary path: %w", err)
+	}
+
+	resolvedPath, err := filepath.EvalSymlinks(selfPath)
+	if err == nil {
+		selfPath = resolvedPath
+	}
+
+	if err := os.Rename(tmpPath, selfPath); err != nil {
+		return fmt.Errorf("failed to replace binary: %w", err)
+	}
+
+	return nil
+}
@@ -33,6 +33,15 @@ func (m *MockExecutor) WriteFile(path string, data []byte, perm os.FileMode) err
 	return nil
 }

+func (m *MockExecutor) ReadFile(path string) ([]byte, error) {
+	if m.WrittenFiles != nil {
+		if data, ok := m.WrittenFiles[path]; ok {
+			return data, nil
+		}
+	}
+	return []byte("mock file content"), nil
+}
+
 func TestHandleCommand(t *testing.T) {
 	tests := []struct {
 		name            string
@@ -182,7 +191,8 @@ func TestHandleCommand(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			mockConn := &MockConn{}
 			mockExec := &MockExecutor{}
-			handleCommand(tc.cfg, tc.msg, mockConn, mockExec)
+			cm := &ConfigManager{current: tc.cfg}
+			handleCommand(cm, tc.msg, mockConn, mockExec)

 			if len(mockConn.Messages) != 1 {
 				t.Fatalf("expected 1 response message, got %d", len(mockConn.Messages))