mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
fix: verify cluster CA on peer and predastore readiness probes
Peer /health and predastore readiness probes skipped TLS verification. Both now trust the cluster CA (nats.cacert) that issues every node's server cert. A node without the CA stops peer probing, leaving peers reachable, and reports predastore not ready.
This commit is contained in:
@@ -92,7 +92,10 @@ func TestWaitForClusterReady_PredastoreBecomesReady(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
|
||||
d := &Daemon{
|
||||
config: &config.Config{Predastore: config.PredastoreConfig{Host: closedPort(t)}},
|
||||
config: &config.Config{
|
||||
Predastore: config.PredastoreConfig{Host: closedPort(t)},
|
||||
NATS: config.NATSConfig{CACert: writeServerCA(t, srv)},
|
||||
},
|
||||
natsConn: nc,
|
||||
}
|
||||
|
||||
|
||||
@@ -2574,15 +2574,6 @@ func (d *Daemon) checkViperblockReady() bool {
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// predastoreReadinessClient is hoisted to package scope so the 2s readiness
|
||||
// poll loop doesn't build a fresh transport every call. Verification is off
|
||||
// because the probe authenticates nothing and reads no data.
|
||||
var predastoreReadinessClient = &http.Client{
|
||||
Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // readiness probe only, no data exchanged
|
||||
},
|
||||
}
|
||||
|
||||
// checkPredastoreReady reports whether predastore is serving HTTPS, not just
|
||||
// listening. Any response counts as ready, including the 401/403 an S3
|
||||
// endpoint returns for this deliberately unsigned request.
|
||||
@@ -2595,11 +2586,18 @@ func (d *Daemon) checkPredastoreReady() bool {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
rootCAs, err := loadClusterTrustRoot(d)
|
||||
if err != nil {
|
||||
slog.Warn("predastore readiness probe: could not load cluster CA", "err", err)
|
||||
return false
|
||||
}
|
||||
client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: rootCAs}}}
|
||||
defer client.CloseIdleConnections()
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+host+"/", nil)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
resp, err := predastoreReadinessClient.Do(req)
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -165,7 +165,7 @@ func computePredastoreHealth(ctx context.Context, d *Daemon) string {
|
||||
return probeLocalGate(ctx, d, cfg, hostID)
|
||||
}
|
||||
|
||||
rootCAs, err := loadPredastoreTrustRoot(d)
|
||||
rootCAs, err := loadClusterTrustRoot(d)
|
||||
if err != nil {
|
||||
slog.Warn("predastore health probe: could not load cluster CA", "err", err)
|
||||
return predastoreHealthUnreachable
|
||||
@@ -332,11 +332,11 @@ func predastoreConfigPath(daemonConfigPath string) string {
|
||||
return filepath.Join(filepath.Dir(daemonConfigPath), "predastore", "predastore.toml")
|
||||
}
|
||||
|
||||
// loadPredastoreTrustRoot returns the cluster CA pool predastore's meta
|
||||
// replicas are issued from — the same CA (d.config.NATS.CACert, conventionally
|
||||
// /etc/spinifex/ca.pem) the daemon already trusts NATS peers with.
|
||||
func loadPredastoreTrustRoot(d *Daemon) (*x509.CertPool, error) {
|
||||
if d.config.NATS.CACert == "" {
|
||||
// loadClusterTrustRoot returns the cluster CA pool (d.config.NATS.CACert,
|
||||
// conventionally /etc/spinifex/ca.pem) that issues every node's server cert:
|
||||
// daemon peers, predastore replicas and NATS alike.
|
||||
func loadClusterTrustRoot(d *Daemon) (*x509.CertPool, error) {
|
||||
if d.config == nil || d.config.NATS.CACert == "" {
|
||||
return nil, fmt.Errorf("cluster CA not configured (nats.cacert)")
|
||||
}
|
||||
return utils.LoadCertPool(d.config.NATS.CACert)
|
||||
|
||||
@@ -55,10 +55,17 @@ func (d *Daemon) monitorPeerReachability() {
|
||||
return
|
||||
}
|
||||
|
||||
// Without the CA every probe would fail, flipping the node to partitioned;
|
||||
// leave peersReachable at its initial true instead.
|
||||
rootCAs, err := loadClusterTrustRoot(d)
|
||||
if err != nil {
|
||||
slog.Error("peer reachability probe disabled: could not load cluster CA", "err", err)
|
||||
return
|
||||
}
|
||||
client := &http.Client{
|
||||
Timeout: peerProbeTimeout,
|
||||
Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // self-signed per-node certs (matches harness DaemonClient)
|
||||
TLSClientConfig: &tls.Config{RootCAs: rootCAs},
|
||||
},
|
||||
}
|
||||
defer client.CloseIdleConnections()
|
||||
|
||||
@@ -2,11 +2,20 @@ package daemon
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/ecdsa"
|
||||
"crypto/elliptic"
|
||||
"crypto/rand"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"crypto/x509/pkix"
|
||||
"encoding/pem"
|
||||
"math/big"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -20,8 +29,9 @@ import (
|
||||
// /health responses, returning the cluster config the daemon-under-test should
|
||||
// see plus a handle to flip each peer up or down at runtime.
|
||||
type peerHealthFixture struct {
|
||||
peers []*peerStub
|
||||
cfg *config.ClusterConfig
|
||||
peers []*peerStub
|
||||
cfg *config.ClusterConfig
|
||||
caPath string
|
||||
}
|
||||
|
||||
type peerStub struct {
|
||||
@@ -61,6 +71,7 @@ func newPeerHealthFixture(t *testing.T, peerCount int) *peerHealthFixture {
|
||||
Daemon: config.DaemonConfig{Host: "0.0.0.0:" + port},
|
||||
}
|
||||
f.peers = append(f.peers, stub)
|
||||
f.caPath = writeServerCA(t, stub.srv)
|
||||
}
|
||||
f.cfg = &config.ClusterConfig{Node: "node-self", Nodes: nodes}
|
||||
return f
|
||||
@@ -76,6 +87,40 @@ func daemonForPeerHealth(t *testing.T, cfg *config.ClusterConfig) (*Daemon, cont
|
||||
return d, cancel
|
||||
}
|
||||
|
||||
// writeServerCA writes srv's self-signed cert as a PEM CA file, standing in for
|
||||
// the cluster CA. Every httptest TLS server shares that one cert.
|
||||
func writeServerCA(t *testing.T, srv *httptest.Server) string {
|
||||
t.Helper()
|
||||
return writePEMCert(t, srv.Certificate().Raw)
|
||||
}
|
||||
|
||||
// writeUntrustedCA writes a freshly generated self-signed CA that signed none
|
||||
// of the httptest servers' certs.
|
||||
func writeUntrustedCA(t *testing.T) string {
|
||||
t.Helper()
|
||||
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
require.NoError(t, err)
|
||||
tmpl := &x509.Certificate{
|
||||
SerialNumber: big.NewInt(1),
|
||||
Subject: pkix.Name{CommonName: "untrusted test CA"},
|
||||
NotBefore: time.Now().Add(-time.Hour),
|
||||
NotAfter: time.Now().Add(time.Hour),
|
||||
IsCA: true,
|
||||
BasicConstraintsValid: true,
|
||||
KeyUsage: x509.KeyUsageCertSign,
|
||||
}
|
||||
der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
|
||||
require.NoError(t, err)
|
||||
return writePEMCert(t, der)
|
||||
}
|
||||
|
||||
func writePEMCert(t *testing.T, der []byte) string {
|
||||
t.Helper()
|
||||
path := filepath.Join(t.TempDir(), "ca.pem")
|
||||
require.NoError(t, os.WriteFile(path, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600))
|
||||
return path
|
||||
}
|
||||
|
||||
func TestPeerCount_SingleNode(t *testing.T) {
|
||||
cfg := &config.ClusterConfig{
|
||||
Node: "only",
|
||||
@@ -194,6 +239,7 @@ func TestMonitorPeerReachability_EndToEnd(t *testing.T) {
|
||||
}
|
||||
f := newPeerHealthFixture(t, 2)
|
||||
d, cancel := daemonForPeerHealth(t, f.cfg)
|
||||
d.config = &config.Config{NATS: config.NATSConfig{CACert: f.caPath}}
|
||||
|
||||
for _, p := range f.peers {
|
||||
p.setHealthy(false)
|
||||
@@ -211,3 +257,47 @@ func TestMonitorPeerReachability_EndToEnd(t *testing.T) {
|
||||
peerProbeInterval+peerProbeTimeout+time.Second, 50*time.Millisecond,
|
||||
"peersReachable should flip true when any peer comes back")
|
||||
}
|
||||
|
||||
// TestMonitorPeerReachability_UntrustedPeerCert asserts a peer whose cert the
|
||||
// cluster CA did not sign counts as unreachable, even while serving 200.
|
||||
func TestMonitorPeerReachability_UntrustedPeerCert(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping ticker-driven peer probe test in short mode")
|
||||
}
|
||||
f := newPeerHealthFixture(t, 1)
|
||||
d, cancel := daemonForPeerHealth(t, f.cfg)
|
||||
d.config = &config.Config{NATS: config.NATSConfig{CACert: writeUntrustedCA(t)}}
|
||||
d.peersReachable.Store(true)
|
||||
|
||||
go d.monitorPeerReachability()
|
||||
defer cancel()
|
||||
|
||||
require.Eventually(t, func() bool { return !d.peersReachable.Load() },
|
||||
peerProbeTimeout+time.Second, 50*time.Millisecond,
|
||||
"a peer presenting an untrusted cert must not count as reachable")
|
||||
}
|
||||
|
||||
// TestMonitorPeerReachability_NoCAKeepsPeersReachable asserts a node that
|
||||
// cannot load the cluster CA stops probing rather than declaring every peer
|
||||
// unreachable and dropping out of cluster mode.
|
||||
func TestMonitorPeerReachability_NoCAKeepsPeersReachable(t *testing.T) {
|
||||
f := newPeerHealthFixture(t, 2)
|
||||
for _, p := range f.peers {
|
||||
p.setHealthy(false)
|
||||
}
|
||||
d, _ := daemonForPeerHealth(t, f.cfg)
|
||||
d.config = &config.Config{}
|
||||
d.peersReachable.Store(true)
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
d.monitorPeerReachability()
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("monitorPeerReachability kept running without a cluster CA")
|
||||
}
|
||||
assert.True(t, d.peersReachable.Load())
|
||||
}
|
||||
|
||||
@@ -26,10 +26,46 @@ func TestCheckPredastoreReady_HTTPSServing(t *testing.T) {
|
||||
u, err := url.Parse(srv.URL)
|
||||
require.NoError(t, err)
|
||||
|
||||
d := &Daemon{config: &config.Config{Predastore: config.PredastoreConfig{Host: u.Host}}}
|
||||
d := &Daemon{config: &config.Config{
|
||||
Predastore: config.PredastoreConfig{Host: u.Host},
|
||||
NATS: config.NATSConfig{CACert: writeServerCA(t, srv)},
|
||||
}}
|
||||
assert.True(t, d.checkPredastoreReady())
|
||||
}
|
||||
|
||||
// TestCheckPredastoreReady_UntrustedCert asserts an endpoint whose cert the
|
||||
// cluster CA did not sign is not ready, even though it answers HTTPS.
|
||||
func TestCheckPredastoreReady_UntrustedCert(t *testing.T) {
|
||||
srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
u, err := url.Parse(srv.URL)
|
||||
require.NoError(t, err)
|
||||
|
||||
d := &Daemon{config: &config.Config{
|
||||
Predastore: config.PredastoreConfig{Host: u.Host},
|
||||
NATS: config.NATSConfig{CACert: writeUntrustedCA(t)},
|
||||
}}
|
||||
assert.False(t, d.checkPredastoreReady())
|
||||
}
|
||||
|
||||
// TestCheckPredastoreReady_NoClusterCA asserts a node without a configured
|
||||
// cluster CA fails closed instead of probing unverified.
|
||||
func TestCheckPredastoreReady_NoClusterCA(t *testing.T) {
|
||||
srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
u, err := url.Parse(srv.URL)
|
||||
require.NoError(t, err)
|
||||
|
||||
d := &Daemon{config: &config.Config{Predastore: config.PredastoreConfig{Host: u.Host}}}
|
||||
assert.False(t, d.checkPredastoreReady())
|
||||
}
|
||||
|
||||
// TestCheckPredastoreReady_ClosedPort asserts a port nothing is listening on
|
||||
// is reported not-ready, rather than panicking or failing open.
|
||||
func TestCheckPredastoreReady_ClosedPort(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user