Add optional dedicated agent-ingest listener (PULSE_AGENT_INGEST_PORT)

Operators can set PULSE_AGENT_INGEST_PORT to serve agent report and
management traffic (/api/agents/*) on a second listener, so that surface
can be placed on its own network or firewall boundary without exposing
the web UI or the rest of the REST API on that port. The dedicated port
serves only the /api/agents/* prefix and 404s everything else.

The option is additive and fail-closed: it is disabled at 0, the main
listener keeps serving agent ingest so existing single-port deployments
and agents are unaffected, and validation rejects out-of-range ports or
collisions with the frontend or HTTP redirect ports.
This commit is contained in:
rcourtman 2026-05-28 22:55:38 +01:00
parent 2f0a5a818f
commit 9be163324d
5 changed files with 191 additions and 4 deletions

View file

@ -737,3 +737,16 @@ That same live auth-env reload boundary also owns watcher lifecycle cleanup:
its fsnotify or polling goroutine can still read debounce or callback state.
Stopping the watcher is the synchronization point that lets tests and runtime
teardown restore auth/config state without racing a background reload.
That same server-bind config boundary now also owns optional agent-ingest
network isolation. `internal/config/config.go` may accept
`PULSE_AGENT_INGEST_PORT` as a dedicated listener for agent report and
management traffic so operators can place `/api/agents/*` on its own network or
firewall boundary, but the option must fail closed at validation: the agent
ingest port stays disabled at `0`, must be a valid `1`-`65535` port, and must
differ from both the frontend port and any HTTP redirect port. When that
listener is active the runtime must serve only the `/api/agents/*` surface on
it and must never expose the web UI or the rest of the REST API through that
port, so a port reachable from an untrusted agent network cannot widen into the
operator console. Enabling the dedicated port is additive: the main listener
keeps serving agent ingest too, so the default single-port deployment and
existing agents are unaffected.

View file

@ -141,9 +141,10 @@ type Config struct {
FrontendPort int `envconfig:"FRONTEND_PORT" default:"7655"`
ConfigPath string
DataPath string
AppRoot string `json:"-"` // Root directory of the application (where binary lives)
PublicURL string `envconfig:"PULSE_PUBLIC_URL" default:""` // Full URL to access Pulse (e.g., http://198.51.100.100:7655)
AgentConnectURL string `envconfig:"PULSE_AGENT_CONNECT_URL" default:""` // Dedicated direct connect URL for agents (e.g. http://192.0.2.5:7655)
AppRoot string `json:"-"` // Root directory of the application (where binary lives)
PublicURL string `envconfig:"PULSE_PUBLIC_URL" default:""` // Full URL to access Pulse (e.g., http://198.51.100.100:7655)
AgentConnectURL string `envconfig:"PULSE_AGENT_CONNECT_URL" default:""` // Dedicated direct connect URL for agents (e.g. http://192.0.2.5:7655)
AgentIngestPort int `envconfig:"PULSE_AGENT_INGEST_PORT" default:"0"` // When >0, additionally serve agent ingest (/api/agents/*) on this dedicated port so it can be network-isolated from the UI/API. 0 = disabled (single-port, default).
// Proxmox VE connections
PVEInstances []PVEInstance
@ -1314,6 +1315,16 @@ func load(initLogging bool) (*Config, error) {
}
}
if agentIngestPort := utils.GetenvTrim("PULSE_AGENT_INGEST_PORT"); agentIngestPort != "" {
if p, err := parsePortOverride("PULSE_AGENT_INGEST_PORT", agentIngestPort); err == nil {
cfg.AgentIngestPort = p
cfg.EnvOverrides["PULSE_AGENT_INGEST_PORT"] = true
log.Info().Int("port", p).Msg("Serving agent ingest on dedicated port from PULSE_AGENT_INGEST_PORT env var")
} else {
log.Warn().Str("value", agentIngestPort).Msg("Ignoring invalid PULSE_AGENT_INGEST_PORT from environment")
}
}
// Detect deprecated DISABLE_AUTH flag and strip it from the runtime env so downstream
// components behave as if it were never set.
if disableAuthEnv := os.Getenv("DISABLE_AUTH"); disableAuthEnv != "" {
@ -1683,6 +1694,17 @@ func (c *Config) Validate() error {
return fmt.Errorf("HTTP redirect port (%d) must differ from frontend port", c.HTTPRedirectPort)
}
}
if c.AgentIngestPort != 0 {
if c.AgentIngestPort < 1 || c.AgentIngestPort > 65535 {
return fmt.Errorf("invalid agent ingest port: %d (must be between 1 and 65535)", c.AgentIngestPort)
}
if c.AgentIngestPort == c.FrontendPort {
return fmt.Errorf("agent ingest port (%d) must differ from frontend port", c.AgentIngestPort)
}
if c.HTTPRedirectPort != 0 && c.AgentIngestPort == c.HTTPRedirectPort {
return fmt.Errorf("agent ingest port (%d) must differ from HTTP redirect port", c.AgentIngestPort)
}
}
// Validate monitoring settings
if c.PVEPollingInterval < 10*time.Second {

View file

@ -85,6 +85,43 @@ func TestLoad_EnvOverrides(t *testing.T) {
assert.Equal(t, "admin", cfg.AuthUser)
}
func TestLoad_AgentIngestPortDefaultsDisabled(t *testing.T) {
t.Setenv("PULSE_DATA_DIR", t.TempDir())
os.Unsetenv("PULSE_AGENT_INGEST_PORT")
cfg, err := Load()
require.NoError(t, err)
assert.Equal(t, 0, cfg.AgentIngestPort)
}
func TestLoad_AgentIngestPortEnvOverride(t *testing.T) {
t.Setenv("PULSE_DATA_DIR", t.TempDir())
t.Setenv("PULSE_AGENT_INGEST_PORT", "7656")
cfg, err := Load()
require.NoError(t, err)
assert.Equal(t, 7656, cfg.AgentIngestPort)
}
func TestLoad_AgentIngestPortInvalidValueIgnored(t *testing.T) {
t.Setenv("PULSE_DATA_DIR", t.TempDir())
t.Setenv("PULSE_AGENT_INGEST_PORT", "not-a-port")
cfg, err := Load()
require.NoError(t, err)
assert.Equal(t, 0, cfg.AgentIngestPort)
}
func TestLoad_AgentIngestPortRejectsFrontendPortCollision(t *testing.T) {
t.Setenv("PULSE_DATA_DIR", t.TempDir())
t.Setenv("FRONTEND_PORT", "7655")
t.Setenv("PULSE_AGENT_INGEST_PORT", "7655")
_, err := Load()
require.Error(t, err)
assert.Contains(t, err.Error(), "agent ingest port")
}
func TestLoad_ProxmoxGuestDockerDetectionEnvOptIn(t *testing.T) {
t.Setenv("PULSE_DATA_DIR", t.TempDir())

View file

@ -182,6 +182,20 @@ func Run(ctx context.Context, version string) error {
}
defer mainListener.Close()
// Optional dedicated agent-ingest listener. When PULSE_AGENT_INGEST_PORT is
// set, agent report/management traffic (/api/agents/*) is also served on this
// separate port so operators can expose it on its own network/firewall
// boundary without exposing the web UI or the rest of the REST API there.
var agentListener net.Listener
if cfg.AgentIngestPort > 0 {
agentAddr := fmt.Sprintf("%s:%d", cfg.BindAddress, cfg.AgentIngestPort)
agentListener, err = net.Listen("tcp", agentAddr)
if err != nil {
return fmt.Errorf("failed to bind agent ingest server on %s: %w", agentAddr, err)
}
defer agentListener.Close()
}
// Multi-tenant persistence is the canonical way to resolve the base data directory.
// It uses cfg.DataPath, which already includes PULSE_DATA_DIR overrides.
mtPersistence := config.NewMultiTenantPersistence(cfg.DataPath)
@ -566,6 +580,22 @@ func Run(ctx context.Context, version string) error {
},
}
// When a dedicated agent-ingest listener is configured, serve only the
// /api/agents/* surface on it so that port never exposes the web UI or the
// rest of the REST API.
var agentSrv *http.Server
if agentListener != nil {
agentSrv = &http.Server{
Handler: agentIngestHandler(router.Handler()),
ReadHeaderTimeout: 15 * time.Second,
WriteTimeout: 0, // Disabled to support SSE/streaming
IdleTimeout: 120 * time.Second,
TLSConfig: &tls.Config{
MinVersion: tls.VersionTLS12,
},
}
}
// Start config watcher for .env file changes
configWatcher, err := config.NewConfigWatcher(cfg)
if err != nil {
@ -652,7 +682,7 @@ func Run(ctx context.Context, version string) error {
}
// Start server
serverErr := make(chan error, 1)
serverErr := make(chan error, 2)
go func() {
var err error
if cfg.HTTPSEnabled && cfg.TLSCertFile != "" && cfg.TLSKeyFile != "" {
@ -678,6 +708,30 @@ func Run(ctx context.Context, version string) error {
}
}()
if agentSrv != nil {
go func() {
var err error
if cfg.HTTPSEnabled && cfg.TLSCertFile != "" && cfg.TLSKeyFile != "" {
log.Info().
Str("host", cfg.BindAddress).
Int("port", cfg.AgentIngestPort).
Str("protocol", "HTTPS").
Msg("Agent ingest server listening")
err = agentSrv.ServeTLS(agentListener, cfg.TLSCertFile, cfg.TLSKeyFile)
} else {
log.Info().
Str("host", cfg.BindAddress).
Int("port", cfg.AgentIngestPort).
Str("protocol", "HTTP").
Msg("Agent ingest server listening")
err = agentSrv.Serve(agentListener)
}
if err != nil && err != http.ErrServerClosed {
serverErr <- err
}
}()
}
// Setup signal handlers
sigChan := make(chan os.Signal, 1)
reloadChan := make(chan os.Signal, 1)
@ -729,6 +783,11 @@ shutdown:
if err := srv.Shutdown(shutdownCtx); err != nil {
log.Error().Err(err).Msg("Server shutdown error")
}
if agentSrv != nil {
if err := agentSrv.Shutdown(shutdownCtx); err != nil {
log.Error().Err(err).Msg("Agent ingest server shutdown error")
}
}
// Stop license grant refresh loops
router.StopGrantRefresh()
@ -763,6 +822,20 @@ shutdown:
return runErr
}
// agentIngestHandler restricts a handler to the agent-ingest surface
// (/api/agents/*). It backs the optional dedicated agent-ingest listener so
// that port serves only agent report/management endpoints and never the web UI
// or the rest of the REST API. Every other path returns 404.
func agentIngestHandler(inner http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if !strings.HasPrefix(r.URL.Path, "/api/agents/") {
http.NotFound(w, r)
return
}
inner.ServeHTTP(w, r)
})
}
// startMetricsServer starts the Prometheus /metrics endpoint. When metricsToken
// is non-empty, requests must include a matching Authorization: Bearer <token> header.
func startMetricsServer(ctx context.Context, addr string, metricsToken string) {

View file

@ -5,6 +5,7 @@ import (
"fmt"
"net"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
@ -17,6 +18,47 @@ import (
"github.com/rcourtman/pulse-go-rewrite/pkg/metrics"
)
func TestAgentIngestHandler(t *testing.T) {
var innerCalled bool
inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
innerCalled = true
w.WriteHeader(http.StatusOK)
})
h := agentIngestHandler(inner)
cases := []struct {
path string
wantInner bool
wantCode int
}{
{"/api/agents/agent/report", true, http.StatusOK},
{"/api/agents/docker/report", true, http.StatusOK},
{"/api/agents/kubernetes/report", true, http.StatusOK},
{"/api/agents/agent/lookup", true, http.StatusOK},
{"/api/agents/agent/config", true, http.StatusOK},
// Everything outside the agent-ingest surface must be rejected so the
// dedicated port never exposes the web UI or the rest of the REST API.
{"/", false, http.StatusNotFound},
{"/index.html", false, http.StatusNotFound},
{"/api/health", false, http.StatusNotFound},
{"/api/state", false, http.StatusNotFound},
{"/api/security/status", false, http.StatusNotFound},
{"/api/agents", false, http.StatusNotFound},
}
for _, tc := range cases {
innerCalled = false
req := httptest.NewRequest(http.MethodPost, tc.path, nil)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if innerCalled != tc.wantInner {
t.Errorf("path %q: innerCalled=%v, want %v", tc.path, innerCalled, tc.wantInner)
}
if rec.Code != tc.wantCode {
t.Errorf("path %q: status=%d, want %d", tc.path, rec.Code, tc.wantCode)
}
}
}
func TestBusinessHooks(t *testing.T) {
called := false
hook := func(store *metrics.Store) {