diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..52eb4fe --- /dev/null +++ b/CHANGELOG.md @@ -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. diff --git a/telemetry.go b/telemetry.go index 08651d6..309f22f 100644 --- a/telemetry.go +++ b/telemetry.go @@ -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) diff --git a/theta-agent-linux-amd64 b/theta-agent-linux-amd64 index 7faf828..cb9c281 100755 Binary files a/theta-agent-linux-amd64 and b/theta-agent-linux-amd64 differ diff --git a/websocket.go b/websocket.go index fc7b3e6..23cb365 100644 --- a/websocket.go +++ b/websocket.go @@ -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 +} diff --git a/websocket_test.go b/websocket_test.go index a829b4f..d55d597 100644 --- a/websocket_test.go +++ b/websocket_test.go @@ -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))