feat(security): Add SSH output limits and improve host key management

Addresses two security vulnerabilities:

1. SSH Output Size Limits:
   - Prevents memory exhaustion from malicious remote nodes
   - Configurable max_ssh_output_bytes (default 1MB)
   - Stream with io.LimitReader to cap output size
   - New metric: pulse_proxy_ssh_output_oversized_total{node}
   - WARN logging for oversized outputs

2. Improved Host Key Management:
   - Seed host keys from Proxmox cluster store (/etc/pve/priv/known_hosts)
   - Falls back to ssh-keyscan only if Proxmox unavailable (with WARN)
   - Fingerprint change detection with ERROR logging
   - require_proxmox_hostkeys option for strict mode
   - New metric: pulse_proxy_hostkey_changes_total{node}
   - Reduces MITM attack surface significantly

Known hosts manager now normalizes entries, reuses existing fingerprints,
and raises typed HostKeyChangeError when fingerprints differ.

Related to security audit 2025-11-07.

Co-authored-by: Codex <codex@openai.com>

cmd/pulse-sensor-proxy/ssh.go

@@ -1,16 +1,21 @@
 package main
 
 import (
+	"bufio"
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
+	"github.com/rcourtman/pulse-go-rewrite/internal/ssh/knownhosts"
 	"github.com/rs/zerolog/log"
 )
 
@@ -42,12 +47,204 @@ exit 1
 `
 )
 
+const proxmoxClusterKnownHostsPath = "/etc/pve/priv/known_hosts"
+
 // execCommand executes a shell command and returns output
 func execCommand(cmd string) (string, error) {
 	out, err := exec.Command("sh", "-c", cmd).CombinedOutput()
 	return string(out), err
 }
 
+func execCommandWithLimits(cmd string, stdoutLimit, stderrLimit int64) (string, string, bool, bool, error) {
+	command := exec.Command("sh", "-c", cmd)
+
+	stdoutPipe, err := command.StdoutPipe()
+	if err != nil {
+		return "", "", false, false, fmt.Errorf("stdout pipe: %w", err)
+	}
+	stderrPipe, err := command.StderrPipe()
+	if err != nil {
+		return "", "", false, false, fmt.Errorf("stderr pipe: %w", err)
+	}
+
+	if err := command.Start(); err != nil {
+		return "", "", false, false, err
+	}
+
+	type pipeResult struct {
+		data     []byte
+		exceeded bool
+		err      error
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	stdoutCh := make(chan pipeResult, 1)
+	stderrCh := make(chan pipeResult, 1)
+
+	go func() {
+		defer wg.Done()
+		data, exceeded, readErr := readAllWithLimit(stdoutPipe, stdoutLimit)
+		stdoutCh <- pipeResult{data: data, exceeded: exceeded, err: readErr}
+	}()
+
+	go func() {
+		defer wg.Done()
+		data, exceeded, readErr := readAllWithLimit(stderrPipe, stderrLimit)
+		stderrCh <- pipeResult{data: data, exceeded: exceeded, err: readErr}
+	}()
+
+	var stdoutRes, stderrRes pipeResult
+	wgDone := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(wgDone)
+	}()
+
+	<-wgDone
+	stdoutRes = <-stdoutCh
+	stderrRes = <-stderrCh
+
+	waitErr := command.Wait()
+
+	if stdoutRes.err != nil {
+		return "", "", stdoutRes.exceeded, stderrRes.exceeded, fmt.Errorf("stdout read: %w", stdoutRes.err)
+	}
+	if stderrRes.err != nil {
+		return "", "", stdoutRes.exceeded, stderrRes.exceeded, fmt.Errorf("stderr read: %w", stderrRes.err)
+	}
+
+	if waitErr != nil {
+		return string(stdoutRes.data), string(stderrRes.data), stdoutRes.exceeded, stderrRes.exceeded, waitErr
+	}
+
+	return string(stdoutRes.data), string(stderrRes.data), stdoutRes.exceeded, stderrRes.exceeded, nil
+}
+
+func readAllWithLimit(r io.Reader, limit int64) ([]byte, bool, error) {
+	if limit <= 0 {
+		data, err := io.ReadAll(r)
+		return data, false, err
+	}
+
+	const chunkSize = 32 * 1024
+	buf := make([]byte, chunkSize)
+	var out bytes.Buffer
+	var total int64
+	exceeded := false
+
+	for {
+		n, err := r.Read(buf)
+		if n > 0 {
+			if total < limit {
+				remaining := limit - total
+				toWrite := n
+				if int64(n) > remaining {
+					toWrite = int(remaining)
+					exceeded = true
+				}
+				if toWrite > 0 {
+					if _, writeErr := out.Write(buf[:toWrite]); writeErr != nil {
+						return nil, exceeded, writeErr
+					}
+				}
+				if int64(n) > remaining {
+					exceeded = true
+				}
+			} else {
+				exceeded = true
+			}
+			total += int64(n)
+		}
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return nil, exceeded, err
+		}
+	}
+
+	return out.Bytes(), exceeded, nil
+}
+
+func (p *Proxy) ensureHostKeyFromProxmox(ctx context.Context, node string) error {
+	if !isProxmoxHost() {
+		return fmt.Errorf("not running on Proxmox host")
+	}
+
+	entries, err := loadProxmoxHostKeys(node)
+	if err != nil {
+		return err
+	}
+
+	if err := p.knownHosts.EnsureWithEntries(ctx, node, 22, entries); err != nil {
+		return p.handleHostKeyEnsureError(node, err)
+	}
+
+	log.Debug().
+		Str("node", node).
+		Msg("Loaded host key from Proxmox cluster store")
+	return nil
+}
+
+func (p *Proxy) handleHostKeyEnsureError(node string, err error) error {
+	var changeErr *knownhosts.HostKeyChangeError
+	if errors.As(err, &changeErr) {
+		log.Error().
+			Str("node", node).
+			Str("host_spec", changeErr.Host).
+			Msg("Detected SSH host key change")
+		if p.metrics != nil {
+			p.metrics.recordHostKeyChange(node)
+		}
+	}
+	return err
+}
+
+func loadProxmoxHostKeys(host string) ([][]byte, error) {
+	file, err := os.Open(proxmoxClusterKnownHostsPath)
+	if err != nil {
+		return nil, err
+	}
+	defer file.Close()
+
+	var entries [][]byte
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" || strings.HasPrefix(line, "#") {
+			continue
+		}
+		fields := strings.Fields(line)
+		if len(fields) < 3 {
+			continue
+		}
+		if !knownhosts.HostFieldMatches(host, fields[0]) {
+			continue
+		}
+
+		comment := ""
+		if len(fields) > 3 {
+			comment = strings.Join(fields[3:], " ")
+		}
+		var entry string
+		if comment != "" {
+			entry = fmt.Sprintf("%s %s %s %s", host, fields[1], fields[2], comment)
+		} else {
+			entry = fmt.Sprintf("%s %s %s", host, fields[1], fields[2])
+		}
+		entries = append(entries, []byte(entry))
+	}
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+	if len(entries) == 0 {
+		return nil, fmt.Errorf("no Proxmox host keys found for %s in %s", host, proxmoxClusterKnownHostsPath)
+	}
+	return entries, nil
+}
+
 // getPublicKey reads the SSH public key from the default directory
 func (p *Proxy) getPublicKey() (string, error) {
 	return p.getPublicKeyFrom(p.sshKeyPath)
@@ -87,7 +284,28 @@ func (p *Proxy) ensureHostKey(node string) error {
 	if p.knownHosts == nil {
 		return fmt.Errorf("host key manager not configured")
 	}
-	return p.knownHosts.Ensure(context.Background(), node)
+
+	ctx := context.Background()
+	if isProxmoxHost() {
+		if err := p.ensureHostKeyFromProxmox(ctx, node); err == nil {
+			return nil
+		} else {
+			if p.config.RequireProxmoxHostkeys {
+				return err
+			}
+			log.Warn().
+				Str("node", node).
+				Err(err).
+				Msg("Failed to load host key from Proxmox; falling back to ssh-keyscan")
+		}
+	} else if p.config.RequireProxmoxHostkeys {
+		return fmt.Errorf("require_proxmox_hostkeys enabled but not running on Proxmox host")
+	}
+
+	if err := p.knownHosts.Ensure(ctx, node); err != nil {
+		return p.handleHostKeyEnsureError(node, err)
+	}
+	return nil
 }
 
 func (p *Proxy) sshCommonOptions() string {
@@ -253,11 +471,27 @@ func (p *Proxy) testSSHConnection(nodeHost string) error {
 		nodeHost,
 	)
 
-	output, err := execCommand(cmd)
+	_, stderr, stdoutExceeded, stderrExceeded, err := execCommandWithLimits(cmd, p.maxSSHOutputBytes, p.maxSSHOutputBytes)
+	if stdoutExceeded {
+		log.Warn().Str("node", nodeHost).Int64("limit_bytes", p.maxSSHOutputBytes).Msg("SSH test output exceeded limit")
+		if p.metrics != nil {
+			p.metrics.recordSSHOutputOversized(nodeHost)
+		}
+		return fmt.Errorf("ssh test output exceeded %d bytes", p.maxSSHOutputBytes)
+	}
+
+	if stderrExceeded {
+		log.Warn().Str("node", nodeHost).Int64("limit_bytes", p.maxSSHOutputBytes).Msg("SSH test stderr exceeded limit")
+	}
+
 	if err != nil {
 		p.metrics.sshRequests.WithLabelValues(nodeLabel, "error").Inc()
 		p.metrics.sshLatency.WithLabelValues(nodeLabel).Observe(time.Since(startTime).Seconds())
-		return fmt.Errorf("SSH test failed: %w (output: %s)", err, output)
+		stderrMsg := strings.TrimSpace(stderr)
+		if stderrMsg != "" {
+			return fmt.Errorf("SSH test failed: %w (stderr: %s)", err, stderrMsg)
+		}
+		return fmt.Errorf("SSH test failed: %w", err)
 	}
 
 	// The forced command will run "sensors -j" instead of "echo test"
@@ -291,16 +525,34 @@ func (p *Proxy) getTemperatureViaSSH(nodeHost string) (string, error) {
 		nodeHost,
 	)
 
-	output, err := execCommand(cmd)
+	stdout, stderr, stdoutExceeded, stderrExceeded, err := execCommandWithLimits(cmd, p.maxSSHOutputBytes, p.maxSSHOutputBytes)
+	if stdoutExceeded {
+		log.Warn().Str("node", nodeHost).Int64("limit_bytes", p.maxSSHOutputBytes).Msg("SSH temperature output exceeded limit")
+		if p.metrics != nil {
+			p.metrics.recordSSHOutputOversized(nodeHost)
+		}
+		p.metrics.sshRequests.WithLabelValues(nodeLabel, "error").Inc()
+		p.metrics.sshLatency.WithLabelValues(nodeLabel).Observe(time.Since(startTime).Seconds())
+		return "", fmt.Errorf("ssh output exceeded %d bytes", p.maxSSHOutputBytes)
+	}
+
+	if stderrExceeded {
+		log.Warn().Str("node", nodeHost).Int64("limit_bytes", p.maxSSHOutputBytes).Msg("SSH temperature stderr exceeded limit")
+	}
+
 	if err != nil {
 		p.metrics.sshRequests.WithLabelValues(nodeLabel, "error").Inc()
 		p.metrics.sshLatency.WithLabelValues(nodeLabel).Observe(time.Since(startTime).Seconds())
+		stderrMsg := strings.TrimSpace(stderr)
+		if stderrMsg != "" {
+			return "", fmt.Errorf("failed to fetch temperatures: %w (stderr: %s)", err, stderrMsg)
+		}
 		return "", fmt.Errorf("failed to fetch temperatures: %w", err)
 	}
 
 	p.metrics.sshRequests.WithLabelValues(nodeLabel, "success").Inc()
 	p.metrics.sshLatency.WithLabelValues(nodeLabel).Observe(time.Since(startTime).Seconds())
-	return output, nil
+	return stdout, nil
 }
 
 // discoverClusterNodes discovers all nodes in the Proxmox cluster

cmd/pulse-sensor-proxy/ssh_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -136,3 +137,38 @@ func TestTempWrapperPrefersSensorsOutput(t *testing.T) {
 		t.Fatalf("expected wrapper to return sensors output %s, got %s", jsonOutput, trimmed)
 	}
 }
+
+func TestReadAllWithLimit(t *testing.T) {
+	reader := bytes.NewBufferString("abcdefg")
+	data, exceeded, err := readAllWithLimit(reader, 4)
+	if err != nil {
+		t.Fatalf("readAllWithLimit returned error: %v", err)
+	}
+	if string(data) != "abcd" {
+		t.Fatalf("expected truncated output 'abcd', got %q", string(data))
+	}
+	if !exceeded {
+		t.Fatalf("expected exceeded flag when data exceeds limit")
+	}
+
+	reader2 := bytes.NewBufferString("xyz")
+	data, exceeded, err = readAllWithLimit(reader2, 10)
+	if err != nil {
+		t.Fatalf("readAllWithLimit returned error: %v", err)
+	}
+	if string(data) != "xyz" {
+		t.Fatalf("expected full output 'xyz', got %q", string(data))
+	}
+	if exceeded {
+		t.Fatalf("did not expect exceeded flag for small output")
+	}
+
+	reader3 := bytes.NewBufferString("12345")
+	data, exceeded, err = readAllWithLimit(reader3, 0)
+	if err != nil {
+		t.Fatalf("readAllWithLimit returned error: %v", err)
+	}
+	if string(data) != "12345" || exceeded {
+		t.Fatalf("expected unlimited read to return full data without exceeding")
+	}
+}

internal/ssh/knownhosts/manager.go

@@ -23,16 +23,18 @@ type Manager interface {
 	// EnsureWithPort guarantees that the host key for the provided host:port exists
 	// in the managed known_hosts file.
 	EnsureWithPort(ctx context.Context, host string, port int) error
+	// EnsureWithEntries installs provided host key entries for the given host/port.
+	EnsureWithEntries(ctx context.Context, host string, port int, entries [][]byte) error
 	// Path returns the absolute path to the managed known_hosts file.
 	Path() string
 }
 
 type manager struct {
-	path            string
+	path           string
-	cache           map[string]struct{}
+	cache          map[string]struct{}
-	mu              sync.Mutex
+	mu             sync.Mutex
-	keyscanFn       keyscanFunc
+	keyscanFn      keyscanFunc
-	keyscanTimeout  time.Duration
+	keyscanTimeout time.Duration
 }
 
 type keyscanFunc func(ctx context.Context, host string, port int, timeout time.Duration) ([]byte, error)
@@ -44,8 +46,25 @@ const (
 var (
 	// ErrNoHostKeys is returned when ssh-keyscan yields no usable entries.
 	ErrNoHostKeys = errors.New("knownhosts: no host keys discovered")
+	// ErrHostKeyChanged signals that a host key already exists with a different fingerprint.
+	ErrHostKeyChanged = errors.New("knownhosts: host key changed")
 )
 
+// HostKeyChangeError describes a detected host key mismatch.
+type HostKeyChangeError struct {
+	Host     string
+	Existing string
+	Provided string
+}
+
+func (e *HostKeyChangeError) Error() string {
+	return fmt.Sprintf("knownhosts: host key for %s changed", e.Host)
+}
+
+func (e *HostKeyChangeError) Unwrap() error {
+	return ErrHostKeyChanged
+}
+
 // Option allows customizing Manager construction.
 type Option func(*manager)
 
@@ -101,31 +120,16 @@ func (m *manager) EnsureWithPort(ctx context.Context, host string, port int) err
 		port = 22 // Default to standard SSH port
 	}
 
-	cacheKey := fmt.Sprintf("%s:%d", host, port)
-
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if _, ok := m.cache[cacheKey]; ok {
-		return nil
-	}
-
-	if err := m.ensureKnownHostsFile(); err != nil {
-		return err
-	}
-
-	// For non-standard ports, check for [host]:port format
 	hostSpec := host
 	if port != 22 {
 		hostSpec = fmt.Sprintf("[%s]:%d", host, port)
 	}
 
-	exists, err := hostKeyExists(m.path, hostSpec)
+	cacheKey := fmt.Sprintf("%s:%d", host, port)
-	if err != nil {
+	m.mu.Lock()
-		return err
+	_, cached := m.cache[cacheKey]
-	}
+	m.mu.Unlock()
-	if exists {
+	if cached {
-		m.cache[cacheKey] = struct{}{}
 		return nil
 	}
 
@@ -139,10 +143,66 @@ func (m *manager) EnsureWithPort(ctx context.Context, host string, port int) err
 		return fmt.Errorf("%w for %s:%d", ErrNoHostKeys, host, port)
 	}
 
-	if err := appendHostKey(m.path, entries); err != nil {
+	return m.EnsureWithEntries(ctx, host, port, entries)
+}
+
+// EnsureWithEntries installs the provided host key entries for host:port.
+func (m *manager) EnsureWithEntries(ctx context.Context, host string, port int, entries [][]byte) error {
+	if strings.TrimSpace(host) == "" {
+		return fmt.Errorf("knownhosts: missing host")
+	}
+	if port <= 0 {
+		port = 22
+	}
+	if len(entries) == 0 {
+		return fmt.Errorf("knownhosts: no host key entries provided for %s", host)
+	}
+
+	cacheKey := fmt.Sprintf("%s:%d", host, port)
+	hostSpec := host
+	if port != 22 {
+		hostSpec = fmt.Sprintf("[%s]:%d", host, port)
+	}
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if err := m.ensureKnownHostsFile(); err != nil {
 		return err
 	}
 
+	var toAppend [][]byte
+	for _, entry := range entries {
+		normalized, keyType, err := normalizeHostEntry(hostSpec, entry)
+		if err != nil {
+			return err
+		}
+
+		existing, err := findHostKeyLine(m.path, hostSpec, keyType)
+		if err != nil {
+			return err
+		}
+
+		if existing != "" {
+			if existing != string(normalized) {
+				return &HostKeyChangeError{
+					Host:     hostSpec,
+					Existing: existing,
+					Provided: string(normalized),
+				}
+			}
+			continue
+		}
+
+		toAppend = append(toAppend, normalized)
+	}
+
+	if len(toAppend) > 0 {
+		if err := appendHostKey(m.path, toAppend); err != nil {
+			return err
+		}
+	}
+
 	m.cache[cacheKey] = struct{}{}
 	return nil
 }
@@ -225,6 +285,58 @@ func sanitizeKeyscanOutput(host string, raw []byte) [][]byte {
 	return entries
 }
 
+func normalizeHostEntry(host string, entry []byte) ([]byte, string, error) {
+	trimmed := strings.TrimSpace(string(entry))
+	fields := strings.Fields(trimmed)
+	if len(fields) < 3 {
+		return nil, "", fmt.Errorf("knownhosts: invalid host key entry for %s", host)
+	}
+
+	keyType := fields[1]
+	keyData := fields[2]
+	var comment string
+	if len(fields) > 3 {
+		comment = strings.Join(fields[3:], " ")
+	}
+
+	if comment != "" {
+		return []byte(fmt.Sprintf("%s %s %s %s", host, keyType, keyData, comment)), keyType, nil
+	}
+	return []byte(fmt.Sprintf("%s %s %s", host, keyType, keyData)), keyType, nil
+}
+
+func findHostKeyLine(path, host, keyType string) (string, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return "", nil
+		}
+		return "", err
+	}
+	defer f.Close()
+
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		line := scanner.Text()
+		if !hostLineMatches(host, line) {
+			continue
+		}
+
+		fields := strings.Fields(line)
+		if len(fields) < 3 {
+			continue
+		}
+		if keyType != "" && fields[1] != keyType {
+			continue
+		}
+		return strings.TrimSpace(line), nil
+	}
+	if err := scanner.Err(); err != nil {
+		return "", err
+	}
+	return "", nil
+}
+
 func hostLineMatches(host, line string) bool {
 	trimmed := strings.TrimSpace(line)
 	if trimmed == "" || strings.HasPrefix(trimmed, "#") {
@@ -253,6 +365,11 @@ func hostFieldMatches(host, field string) bool {
 	return false
 }
 
+// HostFieldMatches reports whether a known_hosts host field matches the provided host.
+func HostFieldMatches(host, field string) bool {
+	return hostFieldMatches(host, field)
+}
+
 func hostCandidates(part string) []string {
 	part = strings.TrimSpace(part)
 	if part == "" {

internal/ssh/knownhosts/manager_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 )
@@ -65,7 +66,7 @@ other.com ssh-ed25519 CCCC
 	if err != nil {
 		t.Fatalf("ReadFile: %v", err)
 	}
-	if want := "example.com ssh-ed25519 AAAA\nexample.com,192.0.2.10 ssh-rsa BBBB\n"; string(data) != want {
+	if want := "example.com ssh-ed25519 AAAA\nexample.com ssh-rsa BBBB\n"; string(data) != want {
 		t.Fatalf("unexpected known_hosts contents\nwant:\n%s\ngot:\n%s", want, data)
 	}
 }
@@ -131,3 +132,58 @@ func TestHostCandidates(t *testing.T) {
 		}
 	}
 }
+
+func TestEnsureWithEntriesDetectsChange(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "known_hosts")
+
+	mgr, err := NewManager(path)
+	if err != nil {
+		t.Fatalf("NewManager: %v", err)
+	}
+
+	entry := []byte("example.com ssh-ed25519 AAAA")
+	if err := mgr.EnsureWithEntries(context.Background(), "example.com", 22, [][]byte{entry}); err != nil {
+		t.Fatalf("EnsureWithEntries: %v", err)
+	}
+
+	// Same entry should be a no-op
+	if err := mgr.EnsureWithEntries(context.Background(), "example.com", 22, [][]byte{entry}); err != nil {
+		t.Fatalf("EnsureWithEntries repeat: %v", err)
+	}
+
+	// Different key should trigger change error
+	changeEntry := []byte("example.com ssh-ed25519 BBBB")
+	err = mgr.EnsureWithEntries(context.Background(), "example.com", 22, [][]byte{changeEntry})
+	var changeErr *HostKeyChangeError
+	if !errors.As(err, &changeErr) {
+		t.Fatalf("expected HostKeyChangeError, got %v", err)
+	}
+}
+
+func TestEnsureWithEntriesAppendsNewKeyTypes(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "known_hosts")
+
+	mgr, err := NewManager(path)
+	if err != nil {
+		t.Fatalf("NewManager: %v", err)
+	}
+
+	ctx := context.Background()
+	if err := mgr.EnsureWithEntries(ctx, "example.com", 22, [][]byte{[]byte("example.com ssh-ed25519 AAAA")}); err != nil {
+		t.Fatalf("EnsureWithEntries ed25519: %v", err)
+	}
+	if err := mgr.EnsureWithEntries(ctx, "example.com", 22, [][]byte{[]byte("example.com ssh-rsa BBBB")}); err != nil {
+		t.Fatalf("EnsureWithEntries rsa: %v", err)
+	}
+
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("ReadFile: %v", err)
+	}
+	got := string(data)
+	if !strings.Contains(got, "ssh-ed25519 AAAA") || !strings.Contains(got, "ssh-rsa BBBB") {
+		t.Fatalf("expected both key types, got %s", got)
+	}
+}