Configure the ProxyAdminService listeners and peer discovery

config/config.go

@@ -10,6 +10,7 @@ import (
 	"gopkg.in/yaml.v3"
 
 	"github.com/temporalio/s2s-proxy/collect"
+	"github.com/temporalio/s2s-proxy/encryption"
 )
 
 const (
@@ -47,11 +48,75 @@ type (
 	S2SProxyConfig struct {
 		Metrics            *MetricsConfig           `yaml:"metrics"`
 		ProfilingConfig    *ProfilingConfig         `yaml:"profiling"`
+		ProxyAdmin         ProxyAdminConfig         `yaml:"proxyAdmin"`
 		Logging            LoggingConfig            `yaml:"logging"`
 		LogConfigs         map[string]LoggingConfig `yaml:"logConfigs"`
 		ClusterConnections []ClusterConnConfig      `yaml:"clusterConnections"`
 	}
 
+	ProxyAdminConfig struct {
+		// ListenAddress serves ProxyAdminService for local operator queries.
+		// Empty or absent means no server runs.
+		// This listener has no TLS and no authorization.
+		// Validate rejects anything but a loopback address.
+		ListenAddress string `yaml:"listenAddress"`
+
+		// Peer serves ProxyAdminService to the other pods of this proxy deployment.
+		// One pod can then answer for all of them.
+		// Absent means this pod only ever describes itself.
+		Peer *ProxyAdminPeerConfig `yaml:"peer"`
+	}
+
+	ProxyAdminPeerConfig struct {
+		// ListenAddress must be reachable from sibling pods.
+		// Unlike the operator listener it is normally not loopback.
+		ListenAddress string `yaml:"listenAddress"`
+
+		// AllowInsecure permits a non-loopback ListenAddress with no TLS.
+		// Without it that combination is a startup error.
+		// It publishes an unauthenticated view of the deployment's topology to anything that can reach the pod network.
+		AllowInsecure bool `yaml:"allowInsecure"`
+
+		// TLS secures peer traffic.
+		// Unlike the mux listener this verifies the client chain.
+		// CAServerName must name a SAN that every pod's certificate carries.
+		// Peers are dialed by IP.
+		// A per-pod SAN scheme fails every handshake.
+		TLS *encryption.TLSConfig `yaml:"tls"`
+
+		Discovery DiscoveryConfig `yaml:"discovery"`
+	}
+
+	// DiscoveryConfig selects one provider by name.
+	// Each provider reads only its own block.
+	// A block left behind by an unselected provider is inert.
+	//
+	// Every layered configuration tool in this stack deep-merges and cannot delete keys.
+	// Switching Provider from "dns" to "static" through a Helm values override cannot remove the dns block.
+	// Strict decoding would reject the config if the two shared a namespace.
+	DiscoveryConfig struct {
+		// Provider is "dns", "static", or "none".
+		// Empty means "none".
+		Provider string `yaml:"provider"`
+
+		DNS    DNSDiscoveryConfig    `yaml:"dns"`
+		Static StaticDiscoveryConfig `yaml:"static"`
+	}
+
+	DNSDiscoveryConfig struct {
+		// Name resolves to one address per sibling pod, as a Kubernetes headless Service does.
+		// Such a Service publishes ready endpoints only.
+		// A crash-looping pod is invisible to discovery rather than reported as unreachable.
+		Name string `yaml:"name"`
+
+		// Port defaults to the port in the peer ListenAddress.
+		Port int `yaml:"port"`
+	}
+
+	StaticDiscoveryConfig struct {
+		Addresses []string `yaml:"addresses"`
+	}
+
 	SATranslationConfig struct {
 		NamespaceMappings []SANamespaceMapping `yaml:"namespaceMappings"`
 		cachedBiMap       SearchAttributeTranslation
@@ -93,6 +158,25 @@ type (
 
 	AllowedMethods struct {
 		AdminService []string `yaml:"adminService"`
+
+		// ProxyAdmin narrows what the remote cluster may call on this proxy's own admin API.
+		// Short method names, like AdminService above.
+		//
+		// Absent means the built-in ceiling applies.
+		// That ceiling is DescribeClusterConnections and nothing else.
+		// A present-but-empty list means serve nothing.
+		// An operator declines to answer this counterparty at all by writing the empty list.
+		// A name the counterparty could never be served is a startup error.
+		// Configuration can only narrow.
+		//
+		// AdminService differs: its empty list means allow everything.
+		// That asymmetry is deliberate.
+		// The replication ACL's fail-open default is a compatibility promise.
+		// Repeating it here would make the natural spelling of "off" the widest possible setting.
+		//
+		// Writing "proxyAdmin:" with no value is absent, not empty.
+		// Write "proxyAdmin: []" to serve nothing.
+		ProxyAdmin []string `yaml:"proxyAdmin"`
 	}
 
 	ACLPolicy struct {
@@ -366,5 +450,6 @@ func (c *S2SProxyConfig) Validate() error {
 	return validation.Validate(
 		"",
 		validation.Children("clusterConnections", c.ClusterConnections, (*ClusterConnConfig).Validate),
+		validation.Nested("proxyAdmin", &c.ProxyAdmin),
 	)
 }

config/config_test.go

@@ -322,3 +322,83 @@ func TestExampleChart(t *testing.T) {
 	require.Equal(t, ConnectionType("mux-client"), cc.Remote.ConnectionType)
 	require.Equal(t, "s2s-proxy-sample.example.tmprl.cloud:8233", cc.Remote.MuxAddressInfo.ConnectionString)
 }
+
+// TestProxyAdminConfig covers decoding only.
+// The validation rules are TestProxyAdminValidate, in validate_test.go.
+func TestProxyAdminConfig(t *testing.T) {
+	load := func(t *testing.T, body string) S2SProxyConfig {
+		t.Helper()
+		cfg, err := LoadConfig[S2SProxyConfig](writeYAML(t, body))
+		require.NoError(t, err)
+		return cfg
+	}
+
+	const clusterConnections = `
+clusterConnections:
+  - name: only
+`
+
+	t.Run("absent means disabled", func(t *testing.T) {
+		cfg := load(t, clusterConnections)
+		require.Empty(t, cfg.ProxyAdmin.ListenAddress)
+		require.Nil(t, cfg.ProxyAdmin.Peer)
+	})
+
+	t.Run("listen address round-trips", func(t *testing.T) {
+		cfg := load(t, clusterConnections+"proxyAdmin:\n  listenAddress: \"localhost:6061\"\n")
+		require.Equal(t, "localhost:6061", cfg.ProxyAdmin.ListenAddress)
+	})
+
+	// KnownFields(true) makes an unrecognized key fatal.
+	// The Go field has to exist before any YAML can set it.
+	t.Run("unknown keys are rejected", func(t *testing.T) {
+		for name, body := range map[string]string{
+			"under proxyAdmin": "proxyAdmin:\n  nope: 1\n",
+			"under discovery": `
+proxyAdmin:
+  peer:
+    listenAddress: "127.0.0.1:9234"
+    discovery:
+      provider: dns
+      nmae: typo
+`,
+		} {
+			t.Run(name, func(t *testing.T) {
+				_, err := LoadConfig[S2SProxyConfig](writeYAML(t, clusterConnections+body))
+				require.Error(t, err)
+			})
+		}
+	})
+
+	// Every layered configuration tool deep-merges and cannot delete keys.
+	// Switching provider leaves the previous provider's block behind.
+	// Strict decoding has to accept it, or there is no way to change provider through an override.
+	t.Run("an unselected provider's block still decodes", func(t *testing.T) {
+		cfg := load(t, clusterConnections+`
+proxyAdmin:
+  peer:
+    listenAddress: "127.0.0.1:9234"
+    discovery:
+      provider: static
+      dns:
+        name: leftover.svc.cluster.local
+      static:
+        addresses: ["a:9234", "b:9234"]
+`)
+		require.Equal(t, DiscoveryStatic, cfg.ProxyAdmin.Peer.Discovery.Provider)
+		require.Equal(t, "leftover.svc.cluster.local", cfg.ProxyAdmin.Peer.Discovery.DNS.Name)
+	})
+
+	t.Run("peer port defaults the dns port", func(t *testing.T) {
+		cfg := load(t, clusterConnections+`
+proxyAdmin:
+  peer:
+    listenAddress: "127.0.0.1:9234"
+    discovery:
+      provider: dns
+      dns:
+        name: peers.svc.cluster.local
+`)
+		require.Equal(t, 9234, cfg.ProxyAdmin.Peer.PeerPort())
+	})
+}

config/proxyadmin.go

@@ -0,0 +1,197 @@
+package config
+
+import (
+	"errors"
+	"fmt"
+	"net"
+	"slices"
+	"strconv"
+
+	"github.com/temporalio/temporal-proxy/pkg/validation"
+)
+
+// Discovery provider names. Empty is equivalent to DiscoveryNone.
+const (
+	DiscoveryNone   = "none"
+	DiscoveryDNS    = "dns"
+	DiscoveryStatic = "static"
+)
+
+// DiscoveryProviders lists every provider name Validate accepts, for error messages.
+var DiscoveryProviders = []string{DiscoveryNone, DiscoveryDNS, DiscoveryStatic}
+
+// Validate reports configuration that can never work.
+// A typo fails startup rather than leaving a listener that silently never starts.
+//
+// Failures that depend on the environment rather than the config are handled at runtime instead.
+// A port already in use and a name that does not resolve are both of that kind.
+func (c *ProxyAdminConfig) Validate() error {
+	return validation.Validate(
+		"",
+		validation.Field("listenAddress", c.ListenAddress,
+			validation.When(isSet,
+				validation.IsHostPort(),
+				validation.When(parsesAsHostPort, isLoopback(
+					"Bind it to loopback, or serve siblings through proxyAdmin.peer, which authenticates its callers.")),
+			)),
+		validation.WhenNested(func() bool { return c.Peer != nil }, "peer", c.Peer),
+	)
+}
+
+func (p *ProxyAdminPeerConfig) Validate() error {
+	// TLSConfig.IsEnabled is true when only caServerName is set.
+	// The fields below are therefore required individually rather than taken as a group.
+	tlsEnabled := p.TLS != nil && p.TLS.IsEnabled()
+
+	rules := []validation.Rule{
+		validation.Field("listenAddress", p.ListenAddress,
+			validation.Required[string](),
+			validation.When(isSet,
+				validation.IsHostPort(),
+				validation.When(parsesAsHostPort,
+					validation.WhenFn(func() bool { return !tlsEnabled && !p.AllowInsecure }, isLoopback(
+						"Configure proxyAdmin.peer.tls, or set proxyAdmin.peer.allowInsecure to accept it."))),
+			)),
+	}
+	if tlsEnabled {
+		rules = append(rules, p.tlsRules()...)
+	}
+	rules = append(rules,
+		validation.Field("discovery.dns.port", p.Discovery.DNS.Port,
+			validation.WhenFn(p.discoveryPortUnresolvable, validation.Required[int]())),
+		validation.Nested("discovery", &p.Discovery),
+	)
+
+	return validation.Validate("", rules...)
+}
+
+// tlsRules validates the peer TLS block.
+// That block serves the peer listener and dials every sibling.
+func (p *ProxyAdminPeerConfig) tlsRules() []validation.Rule {
+	return []validation.Rule{
+		validation.Field("tls.certificatePath", p.TLS.CertificatePath, validation.Required[string]()),
+		validation.Field("tls.keyPath", p.TLS.KeyPath, validation.Required[string]()),
+		// The peer listener verifies its callers, unlike the mux.
+		// It needs the CA to verify them against.
+		// Without this the listener fails to build and is skipped at startup.
+		// The only symptom is one log line plus every sibling reporting this pod unreachable.
+		validation.Field("tls.remoteCAPath", p.TLS.RemoteCAPath, validation.Required[string]()),
+		// Peers are dialed by IP.
+		// This must name a SAN that every pod's certificate carries.
+		validation.Field("tls.caServerName", p.TLS.CAServerName, validation.Required[string]()),
+		validation.Field("tls.skipCAVerification", p.TLS.SkipCAVerification, isFalse(
+			"disables verification of every sibling this pod dials, which defeats the peer TLS it is set alongside")),
+	}
+}
+
+// discoveryPortUnresolvable reports whether the dns provider has no port to dial siblings on.
+// Neither discovery.dns.port nor the peer listen address supplies one in that case.
+func (p *ProxyAdminPeerConfig) discoveryPortUnresolvable() bool {
+	return p.Discovery.Provider == DiscoveryDNS && p.PeerPort() == 0
+}
+
+func (d *DiscoveryConfig) Validate() error {
+	return validation.Validate(
+		"",
+		validation.Field("provider", d.Provider, knownDiscoveryProvider()),
+		// Blocks belonging to an unselected provider are deliberately not validated.
+		// Layered configuration deep-merges and cannot delete keys.
+		// Switching provider leaves the previous provider's block behind.
+		// It must stay inert.
+		validation.WhenRules(func() bool { return d.Provider == DiscoveryDNS },
+			validation.Field("dns.name", d.DNS.Name, validation.Required[string]()),
+		),
+		validation.WhenRules(func() bool { return d.Provider == DiscoveryStatic },
+			validation.Field("static.addresses", d.Static.Addresses, nonEmpty[string]()),
+		),
+	)
+}
+
+func isSet(s string) bool { return s != "" }
+
+// parsesAsHostPort gates the loopback check.
+// A malformed address already reports its own error.
+// Every unparseable host reads as non-loopback.
+// Running both checks would report "localhost" as exposed when the real problem is the missing port.
+func parsesAsHostPort(listenAddress string) bool {
+	_, _, err := net.SplitHostPort(listenAddress)
+	return err == nil
+}
+
+func nonEmpty[V any]() validation.Check[[]V] {
+	return func(vs []V) error {
+		if len(vs) == 0 {
+			return errors.New("is required")
+		}
+		return nil
+	}
+}
+
+func knownDiscoveryProvider() validation.Check[string] {
+	return func(provider string) error {
+		if provider == "" || slices.Contains(DiscoveryProviders, provider) {
+			return nil
+		}
+		return fmt.Errorf("is %q, want one of %v", provider, DiscoveryProviders)
+	}
+}
+
+// isLoopback rejects an address reachable from outside this host.
+// An unparseable or name-based host is rejected too.
+// The check errs toward making the operator state their intent.
+func isLoopback(remedy string) validation.Check[string] {
+	return func(listenAddress string) error {
+		if loopbackListenAddress(listenAddress) {
+			return nil
+		}
+		return fmt.Errorf("is %q, which is not loopback: this publishes an unauthenticated view "+
+			"of the deployment topology to anything that can reach it. %s", listenAddress, remedy)
+	}
+}
+
+func isFalse(because string) validation.Check[bool] {
+	return func(v bool) error {
+		if !v {
+			return nil
+		}
+		return errors.New(because)
+	}
+}
+
+// peerPort returns the port of a host:port listen address, or 0 when the address binds an arbitrary port.
+func peerPort(listenAddress string) (int, error) {
+	_, portStr, err := net.SplitHostPort(listenAddress)
+	if err != nil {
+		return 0, err
+	}
+	if portStr == "" || portStr == "0" {
+		return 0, nil
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		return 0, fmt.Errorf("invalid port %q: %w", portStr, err)
+	}
+	return port, nil
+}
+
+// loopbackListenAddress reports whether an address only accepts connections from this host.
+func loopbackListenAddress(listenAddress string) bool {
+	host, _, err := net.SplitHostPort(listenAddress)
+	if err != nil {
+		return false
+	}
+	if host == "localhost" {
+		return true
+	}
+	ip := net.ParseIP(host)
+	return ip != nil && ip.IsLoopback()
+}
+
+// PeerPort returns the port the peer listener binds, for defaulting discovery ports.
+func (p ProxyAdminPeerConfig) PeerPort() int {
+	port, err := peerPort(p.ListenAddress)
+	if err != nil {
+		return 0
+	}
+	return port
+}