From b2e65f7b3e781a3403fe619ab935114398eaf795 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 7 Nov 2025 17:09:02 +0000 Subject: [PATCH] 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 --- cmd/pulse-sensor-proxy/ssh.go | 262 +++++++++++++++++++++++- cmd/pulse-sensor-proxy/ssh_test.go | 36 ++++ internal/ssh/knownhosts/manager.go | 169 ++++++++++++--- internal/ssh/knownhosts/manager_test.go | 58 +++++- 4 files changed, 493 insertions(+), 32 deletions(-) diff --git a/cmd/pulse-sensor-proxy/ssh.go b/cmd/pulse-sensor-proxy/ssh.go index c39ff21..76dfe34 100644 --- a/cmd/pulse-sensor-proxy/ssh.go +++ b/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 diff --git a/cmd/pulse-sensor-proxy/ssh_test.go b/cmd/pulse-sensor-proxy/ssh_test.go index 4113b28..c6a2227 100644 --- a/cmd/pulse-sensor-proxy/ssh_test.go +++ b/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") + } +} diff --git a/internal/ssh/knownhosts/manager.go b/internal/ssh/knownhosts/manager.go index 6a4cfb4..e53954d 100644 --- a/internal/ssh/knownhosts/manager.go +++ b/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 - cache map[string]struct{} - mu sync.Mutex - keyscanFn keyscanFunc - keyscanTimeout time.Duration + path string + cache map[string]struct{} + mu sync.Mutex + keyscanFn keyscanFunc + 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) - if err != nil { - return err - } - if exists { - m.cache[cacheKey] = struct{}{} + cacheKey := fmt.Sprintf("%s:%d", host, port) + m.mu.Lock() + _, cached := m.cache[cacheKey] + m.mu.Unlock() + if cached { 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 == "" { diff --git a/internal/ssh/knownhosts/manager_test.go b/internal/ssh/knownhosts/manager_test.go index d29ce63..0fbf204 100644 --- a/internal/ssh/knownhosts/manager_test.go +++ b/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) + } +}