change how plugins are sent to clients (https is better than whatever tf I can do)

2026-05-03 22:37:08 -05:00
parent d7c05af1a3
commit 8a7217cda2
7 changed files with 256 additions and 166 deletions
@@ -575,45 +575,19 @@ func HandleCommand(ctx context.Context, env *runtime.Env, envelope map[string]in
 		}
 		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "loaded"})
 		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: true})
-	case "plugin_load_init":
+	case "plugin_load_http":
 		payload, _ := envelope["payload"].(map[string]interface{})
-		if payload == nil || env.Plugins == nil {
-			return nil
+		if payload == nil {
+			return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: "missing payload"})
 		}
 		manifestRaw, _ := payload["manifest"].(map[string]interface{})
-		totalSize := toInt(payload["size"])
-		totalChunks := toInt(payload["chunks"])
 		manifest, err := plugins.ManifestFromMap(manifestRaw)
 		if err != nil {
-			return nil
+			return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: err.Error()})
 		}
-		_ = env.Plugins.StartBundle(manifest, totalSize, totalChunks)
-		return nil
-	case "plugin_load_chunk":
-		payload, _ := envelope["payload"].(map[string]interface{})
-		if payload == nil || env.Plugins == nil {
-			return nil
-		}
-		pluginId, _ := payload["pluginId"].(string)
-		index := toInt(payload["index"])
-		data, _ := payload["data"].([]byte)
-		_ = env.Plugins.AddChunk(pluginId, index, data)
-		return nil
-	case "plugin_load_finish":
-		payload, _ := envelope["payload"].(map[string]interface{})
-		if payload == nil || env.Plugins == nil {
-			return nil
-		}
-		pluginId, _ := payload["pluginId"].(string)
-		if pluginId == "" {
-			return nil
-		}
-		if err := env.Plugins.FinalizeBundle(ctx, pluginId); err != nil {
-			_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: pluginId, Event: "error", Error: err.Error()})
-			return nil
-		}
-		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: pluginId, Event: "loaded"})
-		return nil
+		pullURL, _ := payload["url"].(string)
+		expectedSize := int64(toInt(payload["size"]))
+		return HandlePluginLoadHTTP(ctx, env, cmdID, manifest, pullURL, expectedSize)
 	case "plugin_unload":
 		payload, _ := envelope["payload"].(map[string]interface{})
 		if payload == nil || env.Plugins == nil {
@@ -0,0 +1,142 @@
+package handlers
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+
+	"overlord-client/cmd/agent/plugins"
+	agentRuntime "overlord-client/cmd/agent/runtime"
+	"overlord-client/cmd/agent/wire"
+)
+
+const maxPluginBinaryBytes = 200 * 1024 * 1024
+
+func HandlePluginLoadHTTP(ctx context.Context, env *agentRuntime.Env, cmdID string, manifest plugins.PluginManifest, pullURL string, expectedSize int64) error {
+	if env.Plugins == nil {
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: "plugin manager not ready"})
+	}
+
+	resolved, err := resolvePluginPullURL(env, pullURL)
+	if err != nil {
+		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "error", Error: err.Error()})
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: err.Error()})
+	}
+
+	tlsConfig := &tls.Config{InsecureSkipVerify: env.Cfg.TLSInsecureSkipVerify, MinVersion: tls.VersionTLS12}
+	if caPath := strings.TrimSpace(env.Cfg.TLSCAPath); caPath != "" {
+		caBytes, err := os.ReadFile(caPath)
+		if err != nil {
+			return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: fmt.Sprintf("failed to read TLS CA: %v", err)})
+		}
+		pool := x509.NewCertPool()
+		if !pool.AppendCertsFromPEM(caBytes) {
+			return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: "failed to parse TLS CA"})
+		}
+		tlsConfig.RootCAs = pool
+	}
+
+	client := &http.Client{Transport: &http.Transport{TLSClientConfig: tlsConfig}}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, resolved, nil)
+	if err != nil {
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: err.Error()})
+	}
+	if token := strings.TrimSpace(env.Cfg.AgentToken); token != "" {
+		req.Header.Set("x-agent-token", token)
+	}
+	if id := strings.TrimSpace(env.Cfg.ID); id != "" {
+		req.Header.Set("x-overlord-client-id", id)
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "error", Error: err.Error()})
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: err.Error()})
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		msg := fmt.Sprintf("plugin pull failed: status %d", resp.StatusCode)
+		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "error", Error: msg})
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: msg})
+	}
+
+	limit := int64(maxPluginBinaryBytes)
+	if expectedSize > 0 && expectedSize < limit {
+		limit = expectedSize
+	}
+	binary, err := io.ReadAll(io.LimitReader(resp.Body, limit+1))
+	if err != nil {
+		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "error", Error: err.Error()})
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: err.Error()})
+	}
+	if int64(len(binary)) > limit {
+		msg := "plugin pull exceeded size limit"
+		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "error", Error: msg})
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: msg})
+	}
+	if expectedSize > 0 && int64(len(binary)) != expectedSize {
+		msg := "plugin pull size mismatch"
+		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "error", Error: msg})
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: msg})
+	}
+
+	if err := env.Plugins.Load(ctx, manifest, binary); err != nil {
+		_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "error", Error: err.Error()})
+		return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: false, Message: err.Error()})
+	}
+	_ = wire.WriteMsg(ctx, env.Conn, wire.PluginEvent{Type: "plugin_event", PluginID: manifest.ID, Event: "loaded"})
+	return wire.WriteMsg(ctx, env.Conn, wire.CommandResult{Type: "command_result", CommandID: cmdID, OK: true})
+}
+
+func resolvePluginPullURL(env *agentRuntime.Env, raw string) (string, error) {
+	trimmed := strings.TrimSpace(raw)
+	if trimmed == "" {
+		return "", errors.New("missing plugin pull url")
+	}
+
+	parsed, err := url.Parse(trimmed)
+	if err != nil {
+		return "", fmt.Errorf("invalid plugin pull url: %w", err)
+	}
+
+	if parsed.IsAbs() {
+		scheme := strings.ToLower(parsed.Scheme)
+		if scheme != "http" && scheme != "https" {
+			return "", fmt.Errorf("unsupported plugin pull scheme: %s", parsed.Scheme)
+		}
+		return parsed.String(), nil
+	}
+
+	if len(env.Cfg.ServerURLs) == 0 {
+		return "", errors.New("no server url configured for plugin pull")
+	}
+	idx := env.Cfg.ServerIndex
+	if idx < 0 || idx >= len(env.Cfg.ServerURLs) {
+		idx = 0
+	}
+	server, err := url.Parse(env.Cfg.ServerURLs[idx])
+	if err != nil {
+		return "", fmt.Errorf("invalid agent server url: %w", err)
+	}
+	switch strings.ToLower(server.Scheme) {
+	case "wss":
+		server.Scheme = "https"
+	case "ws":
+		server.Scheme = "http"
+	case "https", "http":
+	default:
+		return "", fmt.Errorf("unsupported agent server scheme: %s", server.Scheme)
+	}
+	server.Path = parsed.Path
+	server.RawQuery = parsed.RawQuery
+	server.Fragment = ""
+	return server.String(), nil
+}
@@ -13,7 +13,6 @@ import (
 type Manager struct {
 	mu      sync.Mutex
 	plugins map[string]*pluginInstance
-	pending map[string]*pendingBundle
 	writer  wire.Writer
 	host    HostInfo
 }
@@ -24,103 +23,14 @@ type pluginInstance struct {
 	native   NativePlugin
 }

-type pendingBundle struct {
-	manifest    PluginManifest
-	totalSize   int
-	totalChunks int
-	chunks      map[int][]byte
-	received    int
-	receivedSz  int
-}
-
 func NewManager(writer wire.Writer, host HostInfo) *Manager {
 	return &Manager{
 		plugins: make(map[string]*pluginInstance),
-		pending: make(map[string]*pendingBundle),
 		writer:  writer,
 		host:    host,
 	}
 }

-func (m *Manager) StartBundle(manifest PluginManifest, totalSize int, totalChunks int) error {
-	if manifest.ID == "" {
-		return errors.New("missing plugin id")
-	}
-	if totalChunks <= 0 {
-		return errors.New("invalid total chunks")
-	}
-	if totalSize <= 0 {
-		return errors.New("invalid total size")
-	}
-	if totalChunks > 10000 {
-		return errors.New("too many chunks")
-	}
-
-	m.mu.Lock()
-	m.pending[manifest.ID] = &pendingBundle{
-		manifest:    manifest,
-		totalSize:   totalSize,
-		totalChunks: totalChunks,
-		chunks:      make(map[int][]byte),
-	}
-	m.mu.Unlock()
-	return nil
-}
-
-func (m *Manager) AddChunk(pluginId string, index int, data []byte) error {
-	if pluginId == "" {
-		return errors.New("missing plugin id")
-	}
-	if index < 0 {
-		return errors.New("invalid chunk index")
-	}
-
-	m.mu.Lock()
-	b := m.pending[pluginId]
-	if b == nil {
-		m.mu.Unlock()
-		return errors.New("bundle not initialized")
-	}
-	if _, exists := b.chunks[index]; !exists {
-		b.chunks[index] = data
-		b.received++
-		b.receivedSz += len(data)
-	}
-	m.mu.Unlock()
-	return nil
-}
-
-func (m *Manager) FinalizeBundle(ctx context.Context, pluginId string) error {
-	m.mu.Lock()
-	b := m.pending[pluginId]
-	if b == nil {
-		m.mu.Unlock()
-		return errors.New("bundle not initialized")
-	}
-	if b.received < b.totalChunks {
-		m.mu.Unlock()
-		return errors.New("bundle incomplete")
-	}
-
-	chunks := make([][]byte, b.totalChunks)
-	for i := 0; i < b.totalChunks; i++ {
-		chunks[i] = b.chunks[i]
-		if chunks[i] == nil {
-			m.mu.Unlock()
-			return errors.New("missing chunk")
-		}
-	}
-	manifest := b.manifest
-	delete(m.pending, pluginId)
-	m.mu.Unlock()
-
-	combined := make([]byte, 0, b.totalSize)
-	for _, part := range chunks {
-		combined = append(combined, part...)
-	}
-	return m.Load(ctx, manifest, combined)
-}
-
 func (m *Manager) Load(ctx context.Context, manifest PluginManifest, binary []byte) error {
 	//garble:controlflow block_splits=max junk_jumps=max flatten_passes=max
 	if len(binary) == 0 {