From 6e725f1cfc769cd1cc66d72d0ff273645cd77525 Mon Sep 17 00:00:00 2001 From: Guillaume Belanger Date: Wed, 26 Aug 2026 08:26:10 -0400 Subject: [PATCH 1/5] fix(ha): redeem join tokens atomically Signed-off-by: Guillaume Belanger --- internal/api/server/cluster_pki.go | 32 ++-- internal/cluster/pkiagent/agent.go | 98 ++++++++-- internal/cluster/pkiagent/agent_test.go | 72 +++++++ internal/cluster/pkiissuer/service.go | 148 +++++++-------- internal/cluster/pkiissuer/service_test.go | 176 +++++++++++++++--- internal/db/cluster_pki.go | 121 ++++++++++++ internal/db/cluster_pki_test.go | 107 +++++++++++ internal/db/db.go | 4 +- internal/db/error.go | 2 + internal/db/flow_reports.go | 1 - internal/db/ip_leases.go | 6 +- internal/db/operations_register.go | 1 + .../db/operations_registry_internal_test.go | 152 +++++++++++++++ internal/pki/pki.go | 24 --- internal/pki/tokens.go | 29 ++- internal/pki/tokens_test.go | 20 ++ internal/raft/manager.go | 4 +- pkg/runtime/pki.go | 7 + pkg/runtime/pki_leader.go | 19 +- pkg/runtime/runtime.go | 12 +- 20 files changed, 848 insertions(+), 187 deletions(-) create mode 100644 internal/db/operations_registry_internal_test.go diff --git a/internal/api/server/cluster_pki.go b/internal/api/server/cluster_pki.go index a83f166c0..febf4a262 100644 --- a/internal/api/server/cluster_pki.go +++ b/internal/api/server/cluster_pki.go @@ -14,6 +14,7 @@ import ( "bufio" "bytes" "encoding/json" + "errors" "io" "net" "net/http" @@ -22,6 +23,7 @@ import ( "github.com/ellanetworks/core/internal/cluster/listener" "github.com/ellanetworks/core/internal/cluster/pkiagent" "github.com/ellanetworks/core/internal/cluster/pkiissuer" + "github.com/ellanetworks/core/internal/db" "github.com/ellanetworks/core/internal/logger" "github.com/ellanetworks/core/internal/pki" "go.uber.org/zap" @@ -54,6 +56,11 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler { peerNodeID, hasPeer := peerNodeIDFromContext(r.Context()) + var ( + fp string + pins []db.ClusterNodeCert + ) + switch { case hasPeer: // mTLS path: the cert's owner is the only caller who @@ -65,6 +72,12 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler { return } + fp, pins, err = svc.RegisterCert(r.Context(), req.NodeID, []byte(req.CertPEM)) + if err != nil { + writeError(r.Context(), w, http.StatusBadRequest, "register cert", err, logger.APILog) + return + } + default: if req.Token == "" { writeError(r.Context(), w, http.StatusUnauthorized, @@ -73,26 +86,19 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler { return } - claims, err := svc.VerifyAndConsumeJoinToken(r.Context(), req.Token) + fp, pins, err = svc.RedeemJoinToken(r.Context(), req.Token, req.NodeID, []byte(req.CertPEM)) if err != nil { - writeError(r.Context(), w, http.StatusUnauthorized, "verify join token", err, logger.APILog) - return - } + status := http.StatusUnauthorized + if errors.Is(err, db.ErrMigrationPending) { + status = http.StatusServiceUnavailable + } - if claims.NodeID != req.NodeID { - writeError(r.Context(), w, http.StatusForbidden, - "node-id in body does not match token claims", nil, logger.APILog) + writeError(r.Context(), w, status, "redeem join token", err, logger.APILog) return } } - fp, pins, err := svc.RegisterCert(r.Context(), req.NodeID, []byte(req.CertPEM)) - if err != nil { - writeError(r.Context(), w, http.StatusBadRequest, "register cert", err, logger.APILog) - return - } - records := make([]pkiagent.PinRecord, 0, len(pins)) for _, p := range pins { records = append(records, pkiagent.PinRecord{NodeID: p.NodeID, Fingerprint: p.Fingerprint}) diff --git a/internal/cluster/pkiagent/agent.go b/internal/cluster/pkiagent/agent.go index b21daa553..a872d26c5 100644 --- a/internal/cluster/pkiagent/agent.go +++ b/internal/cluster/pkiagent/agent.go @@ -38,6 +38,8 @@ import ( const ( leafCertFile = "leaf.crt" leafKeyFile = "leaf.key" + joinCertFile = "join.crt" + joinKeyFile = "join.key" peerPinsFile = "peer-pins.json" ) @@ -217,8 +219,9 @@ func (a *Agent) JoinFlow(ctx context.Context, serverAddr, token string) error { return fmt.Errorf("parse join token: %w", err) } - if claims.LeaderCertPin == "" { - return fmt.Errorf("join token has no leader cert pin") + pins := claims.PinSet() + if len(pins) == 0 { + return fmt.Errorf("join token has no cluster cert pins") } if claims.ClusterID == "" { @@ -233,12 +236,12 @@ func (a *Agent) JoinFlow(ctx context.Context, serverAddr, token string) error { a.ClusterID = claims.ClusterID } - certPEM, keyPEM, cert, err := a.prepareNewCert() + certPEM, keyPEM, cert, err := a.ensureJoinCert() if err != nil { return fmt.Errorf("prepare cert: %w", err) } - client, err := bootstrapHTTPClient(claims.LeaderCertPin) + client, err := bootstrapHTTPClient(pins) if err != nil { return err } @@ -249,7 +252,60 @@ func (a *Agent) JoinFlow(ctx context.Context, serverAddr, token string) error { return err } - return a.installCert(certPEM, keyPEM, cert) + if err := a.installCert(certPEM, keyPEM, cert); err != nil { + return err + } + + a.discardJoinCert() + + return nil +} + +// ensureJoinCert returns the node's join identity, generating and +// persisting it on first call and reloading the same keypair on every +// later call. The identity must survive both a retry and a restart: +// the leader pins the fingerprint it was shown when it consumed the +// token, and only a node presenting that same fingerprint can replay +// the redemption idempotently. +func (a *Agent) ensureJoinCert() (certPEM, keyPEM []byte, cert *x509.Certificate, err error) { + certPEM, err = os.ReadFile(a.path(joinCertFile)) // #nosec G304 -- under dataDir + if err == nil { + keyPEM, err = os.ReadFile(a.path(joinKeyFile)) // #nosec G304 -- under dataDir + if err == nil { + cert, err = pki.ParseCertPEM(certPEM) + if err == nil { + return certPEM, keyPEM, cert, nil + } + } + } + + certPEM, keyPEM, cert, err = a.prepareNewCert() + if err != nil { + return nil, nil, nil, err + } + + if err := os.MkdirAll(filepath.Dir(a.path(joinCertFile)), 0o700); err != nil { + return nil, nil, nil, fmt.Errorf("mkdir cluster-tls: %w", err) + } + + if err := atomicWrite(a.path(joinKeyFile), keyPEM, 0o600); err != nil { + return nil, nil, nil, err + } + + if err := atomicWrite(a.path(joinCertFile), certPEM, 0o644); err != nil { + return nil, nil, nil, err + } + + return certPEM, keyPEM, cert, nil +} + +func (a *Agent) discardJoinCert() { + for _, f := range []string{joinCertFile, joinKeyFile} { + if err := os.Remove(a.path(f)); err != nil && !os.IsNotExist(err) { + logger.EllaLog.Warn("failed to remove pending join cert", + zap.String("file", f), zap.Error(err)) + } + } } // Rotate generates a fresh self-signed cert in memory, registers @@ -395,12 +451,24 @@ func atomicWrite(path string, data []byte, mode os.FileMode) error { } // bootstrapHTTPClient returns an HTTP client that dials the bootstrap -// ALPN without a client cert and pins the server cert to -// expectedFingerprint. -func bootstrapHTTPClient(expectedFingerprint string) (*http.Client, error) { - raw, err := pki.ParseFingerprint(expectedFingerprint) - if err != nil { - return nil, err +// ALPN without a client cert and pins the server cert to any +// fingerprint in expectedFingerprints. The join token carries every +// voter's pin, so a joiner is not tied to the node that minted it and +// can reach whichever node is leader when it retries. +func bootstrapHTTPClient(expectedFingerprints []string) (*http.Client, error) { + raws := make([][]byte, 0, len(expectedFingerprints)) + + for _, fp := range expectedFingerprints { + raw, err := pki.ParseFingerprint(fp) + if err != nil { + return nil, err + } + + raws = append(raws, raw) + } + + if len(raws) == 0 { + return nil, fmt.Errorf("no bootstrap pins supplied") } tlsCfg := &tls.Config{ @@ -414,12 +482,14 @@ func bootstrapHTTPClient(expectedFingerprint string) (*http.Client, error) { for _, c := range cs.PeerCertificates { sum := sha256.Sum256(c.Raw) - if subtle.ConstantTimeCompare(sum[:], raw) == 1 { - return nil + for _, raw := range raws { + if subtle.ConstantTimeCompare(sum[:], raw) == 1 { + return nil + } } } - return fmt.Errorf("bootstrap: server cert chain does not contain pinned %s", expectedFingerprint) + return fmt.Errorf("bootstrap: server cert chain matches none of the %d pinned cluster certs", len(raws)) }, } diff --git a/internal/cluster/pkiagent/agent_test.go b/internal/cluster/pkiagent/agent_test.go index 6ac9a5864..9449adea7 100644 --- a/internal/cluster/pkiagent/agent_test.go +++ b/internal/cluster/pkiagent/agent_test.go @@ -174,3 +174,75 @@ func alwaysFailRegisterHandler() listener.ConnHandler { _ = resp.Write(conn) } } + +// A joining node must present the same identity on every attempt: +// the leader pins the fingerprint it saw when it consumed the token, +// and only a node re-presenting that fingerprint can replay the +// redemption instead of finding the token burnt. +func TestAgent_JoinFlow_ReusesIdentityAcrossRetries(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + leader := newAgent(t, 1, "join-cluster") + + pinFn := func(fp string) listener.PinResult { + return listener.PinResult{Found: fp == pki.Fingerprint(leader.Leaf().Leaf), NodeID: leader.NodeID} + } + + _, leaderAddr := startListener(ctx, t, leader, pinFn, func(ln *listener.Listener) { + ln.Register(listener.ALPNPKIBootstrap, alwaysFailRegisterHandler()) + }) + + joiner := pkiagent.NewAgent(2, "", t.TempDir()) + token := mintTestJoinToken(t, 2, "join-cluster", pki.Fingerprint(leader.Leaf().Leaf)) + + joinCert := filepath.Join(joiner.DataDir, "cluster-tls", "join.crt") + + if err := joiner.JoinFlow(ctx, leaderAddr, token); err == nil { + t.Fatal("JoinFlow should have failed; leader returns 500") + } + + first, err := os.ReadFile(joinCert) + if err != nil { + t.Fatalf("join.crt must be persisted before the first POST: %v", err) + } + + if err := joiner.JoinFlow(ctx, leaderAddr, token); err == nil { + t.Fatal("second JoinFlow should have failed too") + } + + second, err := os.ReadFile(joinCert) + if err != nil { + t.Fatalf("read join.crt after retry: %v", err) + } + + if !bytes.Equal(first, second) { + t.Error("retry generated a new identity; the leader's pin would no longer match") + } + + if joiner.HaveLeafOnDisk() { + t.Error("a failed join must not install a live leaf") + } +} + +func mintTestJoinToken(t *testing.T, nodeID int, clusterID, leaderPin string) string { + t.Helper() + + key := bytes.Repeat([]byte{0xAB}, 32) + now := time.Now() + + token, err := pki.MintJoinToken(key, pki.JoinClaims{ + TokenID: "test-token", + NodeID: nodeID, + IssuedAt: now.Unix(), + ExpiresAt: now.Add(time.Hour).Unix(), + LeaderCertPin: leaderPin, + ClusterID: clusterID, + ClusterPins: []string{leaderPin}, + }) + if err != nil { + t.Fatalf("mint join token: %v", err) + } + + return token +} diff --git a/internal/cluster/pkiissuer/service.go b/internal/cluster/pkiissuer/service.go index e8f01147c..7459db93c 100644 --- a/internal/cluster/pkiissuer/service.go +++ b/internal/cluster/pkiissuer/service.go @@ -13,7 +13,7 @@ package pkiissuer import ( "bytes" "context" - "crypto/rand" + "crypto/x509" "encoding/json" "errors" "fmt" @@ -34,14 +34,13 @@ type Store interface { ListClusterNodeCerts(ctx context.Context) ([]db.ClusterNodeCert, error) MintJoinTokenRecord(ctx context.Context, r *db.ClusterJoinToken) error - GetJoinToken(ctx context.Context, id string) (*db.ClusterJoinToken, error) - ConsumeJoinToken(ctx context.Context, id string, nodeID int) error + RedeemJoinToken(ctx context.Context, tokenID string, nodeID int, fingerprint, certPEM string) ([]db.ClusterNodeCert, error) IsLeader() bool } -// Service runs on every voter. Bootstrap, MintJoinToken, and -// RegisterCert require IsLeader; CurrentPins works on followers. +// Service runs on every voter. Bootstrap and MintJoinToken require +// IsLeader; RedeemJoinToken and RegisterCert forward to the leader. type Service struct { store Store } @@ -64,7 +63,7 @@ func (s *Service) Bootstrap(ctx context.Context) error { return fmt.Errorf("get hmac key: %w", err) } - key, err := newHMACKey() + key, err := pki.NewHMACKey() if err != nil { return fmt.Errorf("generate hmac key: %w", err) } @@ -114,7 +113,7 @@ func (s *Service) MintJoinToken(ctx context.Context, nodeID int, ttl time.Durati return "", fmt.Errorf("cluster id not yet populated") } - leaderPin, err := s.leaderPin(ctx, leaderNodeID) + leaderPin, allPins, err := s.pinsForToken(ctx, leaderNodeID) if err != nil { return "", err } @@ -133,6 +132,7 @@ func (s *Service) MintJoinToken(ctx context.Context, nodeID int, ttl time.Durati ExpiresAt: now.Add(ttl).Unix(), LeaderCertPin: leaderPin, ClusterID: op.ClusterID, + ClusterPins: allPins, } tokenStr, err := pki.MintJoinToken(hmacKey, claims) @@ -157,63 +157,69 @@ func (s *Service) MintJoinToken(ctx context.Context, nodeID int, ttl time.Durati return tokenStr, nil } -func (s *Service) leaderPin(ctx context.Context, leaderNodeID int) (string, error) { +func (s *Service) pinsForToken(ctx context.Context, leaderNodeID int) (string, []string, error) { rows, err := s.store.ListClusterNodeCerts(ctx) if err != nil { - return "", fmt.Errorf("list pins: %w", err) + return "", nil, fmt.Errorf("list pins: %w", err) + } + + all := make([]string, 0, len(rows)) + for _, r := range rows { + all = append(all, r.Fingerprint) } if leaderNodeID == 0 { // Standalone or single-node test path: with exactly one // registered pin, that pin belongs to the leader. if len(rows) == 1 { - return rows[0].Fingerprint, nil + return rows[0].Fingerprint, all, nil } - return "", fmt.Errorf("leaderNodeID is zero and registry has %d pins", len(rows)) + return "", nil, fmt.Errorf("leaderNodeID is zero and registry has %d pins", len(rows)) } for _, r := range rows { if r.NodeID == leaderNodeID { - return r.Fingerprint, nil + return r.Fingerprint, all, nil } } - return "", fmt.Errorf("leader node %d has no registered pin", leaderNodeID) + return "", nil, fmt.Errorf("leader node %d has no registered pin", leaderNodeID) } -// VerifyAndConsumeJoinToken authenticates tokenStr, enforces -// expiry, marks the token consumed, and returns its claims. A -// second consumption on any voter returns an error. -func (s *Service) VerifyAndConsumeJoinToken(ctx context.Context, tokenStr string) (*pki.JoinClaims, error) { +// RedeemJoinToken authenticates tokenStr, validates certPEM, and then +// consumes the token and pins the cert in a single replicated apply. +// Consumption can no longer commit without the pin that it authorises, +// so a leadership change mid-join leaves the token usable rather than +// burnt. A repeat redemption by the same node presenting the same cert +// succeeds and returns the current pin set, making the joiner's retry +// loop safe; any other repeat returns db.ErrJoinTokenAlreadyConsumed. +func (s *Service) RedeemJoinToken(ctx context.Context, tokenStr string, nodeID int, certPEM []byte) (string, []db.ClusterNodeCert, error) { hmacKey, err := s.store.GetClusterJoinHMACKey(ctx) if err != nil { - return nil, err + return "", nil, err } claims, err := pki.VerifyJoinToken(hmacKey, time.Now(), tokenStr) if err != nil { - return nil, err + return "", nil, err } - row, err := s.store.GetJoinToken(ctx, claims.TokenID) - if err != nil { - return nil, fmt.Errorf("lookup join token: %w", err) + if claims.NodeID != nodeID { + return "", nil, fmt.Errorf("token is for node %d, not %d", claims.NodeID, nodeID) } - if row.ConsumedAt != 0 { - return nil, fmt.Errorf("token already consumed") + cert, fp, err := s.validateNodeCert(ctx, nodeID, certPEM) + if err != nil { + return "", nil, err } - if err := s.store.ConsumeJoinToken(ctx, claims.TokenID, claims.NodeID); err != nil { - if errors.Is(err, db.ErrJoinTokenAlreadyConsumed) { - return nil, fmt.Errorf("token already consumed") - } - - return nil, fmt.Errorf("consume join token: %w", err) + pins, err := s.store.RedeemJoinToken(ctx, claims.TokenID, nodeID, fp, string(pki.EncodeCertPEM(cert))) + if err != nil { + return "", nil, err } - return claims, nil + return fp, pins, nil } // RegisterCert validates certPEM (SPIFFE URI matches the cluster's @@ -222,45 +228,11 @@ func (s *Service) VerifyAndConsumeJoinToken(ctx context.Context, tokenStr string // Returns the pin fingerprint and the post-commit snapshot of // every registered pin so the caller can seed its local pin map. func (s *Service) RegisterCert(ctx context.Context, nodeID int, certPEM []byte) (string, []db.ClusterNodeCert, error) { - if !s.store.IsLeader() { - return "", nil, fmt.Errorf("not leader") - } - - op, err := s.store.GetOperator(ctx) - if err != nil { - return "", nil, fmt.Errorf("get operator: %w", err) - } - - cert, err := pki.ParseCertPEM(certPEM) - if err != nil { - return "", nil, fmt.Errorf("parse cert: %w", err) - } - - clusterID, certNodeID, err := pki.IdentityFromCert(cert) + cert, fp, err := s.validateNodeCert(ctx, nodeID, certPEM) if err != nil { - return "", nil, fmt.Errorf("invalid cluster cert: %w", err) - } - - if clusterID != op.ClusterID { - return "", nil, fmt.Errorf("cert clusterID %q != operator clusterID %q", clusterID, op.ClusterID) - } - - if certNodeID != nodeID { - return "", nil, fmt.Errorf("cert URI nodeID %d != requested nodeID %d", certNodeID, nodeID) - } - - // Issuer must equal subject; the cluster TLS contract requires - // every node cert to be self-signed. - if !bytes.Equal(cert.RawIssuer, cert.RawSubject) { - return "", nil, fmt.Errorf("cert is not self-signed") - } - - if err := cert.CheckSignature(cert.SignatureAlgorithm, cert.RawTBSCertificate, cert.Signature); err != nil { - return "", nil, fmt.Errorf("self-signature verify: %w", err) + return "", nil, err } - fp := pki.Fingerprint(cert) - row := &db.ClusterNodeCert{ NodeID: nodeID, Fingerprint: fp, @@ -280,29 +252,39 @@ func (s *Service) RegisterCert(ctx context.Context, nodeID int, certPEM []byte) return fp, pins, nil } -// CurrentPins returns the replicated pin set as a fingerprint → -// nodeID map for caching in the listener's PinFunc. -func (s *Service) CurrentPins(ctx context.Context) (map[string]int, error) { - rows, err := s.store.ListClusterNodeCerts(ctx) +func (s *Service) validateNodeCert(ctx context.Context, nodeID int, certPEM []byte) (*x509.Certificate, string, error) { + op, err := s.store.GetOperator(ctx) if err != nil { - return nil, err + return nil, "", fmt.Errorf("get operator: %w", err) } - out := make(map[string]int, len(rows)) - for _, r := range rows { - out[r.Fingerprint] = r.NodeID + cert, err := pki.ParseCertPEM(certPEM) + if err != nil { + return nil, "", fmt.Errorf("parse cert: %w", err) } - return out, nil -} + clusterID, certNodeID, err := pki.IdentityFromCert(cert) + if err != nil { + return nil, "", fmt.Errorf("invalid cluster cert: %w", err) + } + + if clusterID != op.ClusterID { + return nil, "", fmt.Errorf("cert clusterID %q != operator clusterID %q", clusterID, op.ClusterID) + } + + if certNodeID != nodeID { + return nil, "", fmt.Errorf("cert URI nodeID %d != requested nodeID %d", certNodeID, nodeID) + } -func (s *Service) IsLeader() bool { return s.store.IsLeader() } + // Issuer must equal subject; the cluster TLS contract requires + // every node cert to be self-signed. + if !bytes.Equal(cert.RawIssuer, cert.RawSubject) { + return nil, "", fmt.Errorf("cert is not self-signed") + } -func newHMACKey() ([]byte, error) { - b := make([]byte, 32) - if _, err := rand.Read(b); err != nil { - return nil, fmt.Errorf("rand: %w", err) + if err := cert.CheckSignature(cert.SignatureAlgorithm, cert.RawTBSCertificate, cert.Signature); err != nil { + return nil, "", fmt.Errorf("self-signature verify: %w", err) } - return b, nil + return cert, pki.Fingerprint(cert), nil } diff --git a/internal/cluster/pkiissuer/service_test.go b/internal/cluster/pkiissuer/service_test.go index a37067b43..a3ea6077b 100644 --- a/internal/cluster/pkiissuer/service_test.go +++ b/internal/cluster/pkiissuer/service_test.go @@ -26,6 +26,8 @@ type fakeStore struct { tokens map[string]*db.ClusterJoinToken } +const testClusterID = "c" + func newFakeStore(clusterID string) *fakeStore { return &fakeStore{ leader: true, @@ -128,12 +130,57 @@ func (f *fakeStore) ConsumeJoinToken(ctx context.Context, id string, nodeID int) return nil } +func (f *fakeStore) RedeemJoinToken(ctx context.Context, tokenID string, nodeID int, fingerprint, certPEM string) ([]db.ClusterNodeCert, error) { + f.mu.Lock() + + t, ok := f.tokens[tokenID] + if !ok { + f.mu.Unlock() + return nil, db.ErrNotFound + } + + if t.NodeID != nodeID { + f.mu.Unlock() + return nil, db.ErrJoinTokenNodeMismatch + } + + if t.ExpiresAt <= time.Now().Unix() { + f.mu.Unlock() + return nil, db.ErrJoinTokenExpired + } + + if t.ConsumedAt != 0 { + existing, have := f.pins[nodeID] + if t.ConsumedBy != nodeID || !have || existing.Fingerprint != fingerprint { + f.mu.Unlock() + return nil, db.ErrJoinTokenAlreadyConsumed + } + + f.mu.Unlock() + + return f.ListClusterNodeCerts(ctx) + } + + t.ConsumedAt = time.Now().Unix() + t.ConsumedBy = nodeID + f.pins[nodeID] = &db.ClusterNodeCert{ + NodeID: nodeID, + Fingerprint: fingerprint, + CertPEM: certPEM, + AddedAt: time.Now().Unix(), + } + + f.mu.Unlock() + + return f.ListClusterNodeCerts(ctx) +} + // preregisterLeader inserts the leader's pin so MintJoinToken can // embed it in a token's claims. -func preregisterLeader(t *testing.T, store *fakeStore, nodeID int, clusterID string) string { +func preregisterLeader(t *testing.T, store *fakeStore, nodeID int) string { t.Helper() - cert, _, err := pki.GenerateNodeCert(nodeID, clusterID, time.Hour) + cert, _, err := pki.GenerateNodeCert(nodeID, testClusterID, time.Hour) if err != nil { t.Fatalf("generate leader cert: %v", err) } @@ -237,7 +284,7 @@ func TestService_RegisterCert_RejectsNodeIDMismatch(t *testing.T) { func TestService_MintAndVerifyJoinToken_RoundTrip(t *testing.T) { store := newFakeStore("c") - leaderFP := preregisterLeader(t, store, 1, "c") + leaderFP := preregisterLeader(t, store, 1) svc := pkiissuer.New(store) @@ -263,24 +310,27 @@ func TestService_MintAndVerifyJoinToken_RoundTrip(t *testing.T) { t.Fatalf("nodeID mismatch") } - verified, err := svc.VerifyAndConsumeJoinToken(context.Background(), token) + joinerPEM := nodeCertPEM(t, 5) + + fp, pins, err := svc.RedeemJoinToken(context.Background(), token, 5, joinerPEM) if err != nil { - t.Fatalf("verify: %v", err) + t.Fatalf("redeem: %v", err) } - if verified.TokenID != claims.TokenID { - t.Fatal("verify returned different token id") + if fp == "" || len(pins) != 2 { + t.Fatalf("redeem returned fp=%q pins=%d, want non-empty fp and 2 pins", fp, len(pins)) } - // Replay: second consume must fail. - if _, err := svc.VerifyAndConsumeJoinToken(context.Background(), token); err == nil { - t.Fatal("replay should be rejected") + // Replay by a different node must be rejected. + otherPEM := nodeCertPEM(t, 6) + if _, _, err := svc.RedeemJoinToken(context.Background(), token, 6, otherPEM); err == nil { + t.Fatal("replay for a different node should be rejected") } } func TestService_MintJoinToken_RejectsInvalidTTL(t *testing.T) { store := newFakeStore("c") - preregisterLeader(t, store, 1, "c") + preregisterLeader(t, store, 1) svc := pkiissuer.New(store) _ = svc.Bootstrap(context.Background()) @@ -307,30 +357,114 @@ func TestService_NotLeader_RejectsMutations(t *testing.T) { if _, err := svc.MintJoinToken(context.Background(), 5, time.Hour, 1); err == nil { t.Fatal("MintJoinToken should fail on non-leader") } +} - if _, _, err := svc.RegisterCert(context.Background(), 1, []byte("not pem")); err == nil { - t.Fatal("RegisterCert should fail on non-leader") +// RegisterCert no longer gates on leadership: the pin upsert is a +// replicated op that forwards to the leader on its own, so a rotation +// that lands on a follower is completed rather than refused. +func TestService_RegisterCert_WorksOnNonLeader(t *testing.T) { + store := newFakeStore("c") + store.leader = false + + svc := pkiissuer.New(store) + + if _, _, err := svc.RegisterCert(context.Background(), 5, nodeCertPEM(t, 5)); err != nil { + t.Fatalf("RegisterCert on a follower: %v", err) } } -// Smoke test that ErrJoinTokenAlreadyConsumed is preserved through -// the wrapping VerifyAndConsumeJoinToken does. -func TestService_DoubleConsume_PreservesErrPath(t *testing.T) { +func TestService_Redeem_ReplayWithDifferentCertRejected(t *testing.T) { store := newFakeStore("c") - preregisterLeader(t, store, 1, "c") + preregisterLeader(t, store, 1) svc := pkiissuer.New(store) _ = svc.Bootstrap(context.Background()) tok, _ := svc.MintJoinToken(context.Background(), 5, time.Minute*10, 1) - _, err := svc.VerifyAndConsumeJoinToken(context.Background(), tok) + if _, _, err := svc.RedeemJoinToken(context.Background(), tok, 5, nodeCertPEM(t, 5)); err != nil { + t.Fatal(err) + } + + _, _, err := svc.RedeemJoinToken(context.Background(), tok, 5, nodeCertPEM(t, 5)) + if !errors.Is(err, db.ErrJoinTokenAlreadyConsumed) { + t.Fatalf("second redeem with a fresh cert: got %v, want ErrJoinTokenAlreadyConsumed", err) + } +} + +// A joining node that retries after a leadership change presents the +// same cert it persisted before its first attempt, so the redemption +// replays as a no-op instead of reporting the token burnt. +func TestService_Redeem_SameNodeSameCertIsIdempotent(t *testing.T) { + store := newFakeStore("c") + preregisterLeader(t, store, 1) + + svc := pkiissuer.New(store) + _ = svc.Bootstrap(context.Background()) + + tok, _ := svc.MintJoinToken(context.Background(), 5, time.Minute*10, 1) + joinerPEM := nodeCertPEM(t, 5) + + fp1, pins1, err := svc.RedeemJoinToken(context.Background(), tok, 5, joinerPEM) if err != nil { t.Fatal(err) } - _, err = svc.VerifyAndConsumeJoinToken(context.Background(), tok) - if err == nil || (!errors.Is(err, db.ErrJoinTokenAlreadyConsumed) && err.Error() != "token already consumed") { - t.Fatalf("unexpected second-consume error: %v", err) + fp2, pins2, err := svc.RedeemJoinToken(context.Background(), tok, 5, joinerPEM) + if err != nil { + t.Fatalf("retry after a burnt token must succeed: %v", err) + } + + if fp1 != fp2 || len(pins1) != len(pins2) { + t.Fatalf("replay returned a different result: %s/%d vs %s/%d", fp1, len(pins1), fp2, len(pins2)) } } + +func TestService_MintJoinToken_EmbedsEveryVoterPin(t *testing.T) { + store := newFakeStore("c") + leaderFP := preregisterLeader(t, store, 1) + peerFP := preregisterLeader(t, store, 2) + + svc := pkiissuer.New(store) + _ = svc.Bootstrap(context.Background()) + + tok, err := svc.MintJoinToken(context.Background(), 5, time.Minute*10, 1) + if err != nil { + t.Fatal(err) + } + + claims, err := pki.ExtractClaimsUnverified(tok) + if err != nil { + t.Fatal(err) + } + + set := claims.PinSet() + if len(set) != 2 { + t.Fatalf("PinSet has %d entries, want 2", len(set)) + } + + for _, want := range []string{leaderFP, peerFP} { + found := false + + for _, got := range set { + if got == want { + found = true + } + } + + if !found { + t.Fatalf("PinSet is missing %s", want) + } + } +} + +func nodeCertPEM(t *testing.T, nodeID int) []byte { + t.Helper() + + cert, _, err := pki.GenerateNodeCert(nodeID, testClusterID, time.Hour) + if err != nil { + t.Fatalf("generate node cert: %v", err) + } + + return pki.EncodeCertPEM(cert) +} diff --git a/internal/db/cluster_pki.go b/internal/db/cluster_pki.go index c3726cc92..219b4d8ae 100644 --- a/internal/db/cluster_pki.go +++ b/internal/db/cluster_pki.go @@ -42,6 +42,7 @@ const ( const ( listNodeCertsStmtStr = "SELECT &ClusterNodeCert.* FROM %s ORDER BY nodeID ASC" getNodeCertByFPStmtStr = "SELECT &ClusterNodeCert.* FROM %s WHERE fingerprint=$ClusterNodeCert.fingerprint" + getNodeCertByNodeStmtStr = "SELECT &ClusterNodeCert.* FROM %s WHERE nodeID=$ClusterNodeCert.nodeID" upsertNodeCertStmtStr = "INSERT INTO %s (nodeID, fingerprint, certPEM, addedAt) VALUES ($ClusterNodeCert.nodeID, $ClusterNodeCert.fingerprint, $ClusterNodeCert.certPEM, $ClusterNodeCert.addedAt) ON CONFLICT(nodeID) DO UPDATE SET fingerprint=excluded.fingerprint, certPEM=excluded.certPEM, addedAt=excluded.addedAt" deleteNodeCertByNodeStmtStr = "DELETE FROM %s WHERE nodeID=$ClusterNodeCert.nodeID" @@ -133,6 +134,101 @@ func (db *Database) applyConsumeJoinToken(ctx context.Context, r *ClusterJoinTok return nil, nil } +type redeemJoinTokenPayload struct { + TokenID string `json:"token_id"` + NodeID int `json:"node_id"` + Fingerprint string `json:"fingerprint"` + CertPEM string `json:"cert_pem"` + Now int64 `json:"now"` +} + +// RedeemJoinTokenResult is the post-commit pin snapshot returned to a +// joining node so it can seed its local pin map. +type RedeemJoinTokenResult struct { + Pins []ClusterNodeCert `json:"pins"` +} + +func (db *Database) applyRedeemJoinToken(ctx context.Context, p *redeemJoinTokenPayload) (any, error) { + runner := db.runner(ctx) + + token := ClusterJoinToken{ID: p.TokenID} + if err := runner.Query(ctx, db.getJoinTokenStmt, token).Get(&token); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrNotFound + } + + return nil, fmt.Errorf("get join token: %w", err) + } + + if token.NodeID != p.NodeID { + return nil, ErrJoinTokenNodeMismatch + } + + if token.ExpiresAt <= p.Now { + return nil, ErrJoinTokenExpired + } + + if token.ConsumedAt != 0 { + if token.ConsumedBy != p.NodeID { + return nil, ErrJoinTokenAlreadyConsumed + } + + existing := ClusterNodeCert{NodeID: p.NodeID} + if err := runner.Query(ctx, db.getNodeCertByNodeStmt, existing).Get(&existing); err != nil { + return nil, ErrJoinTokenAlreadyConsumed + } + + if existing.Fingerprint != p.Fingerprint { + return nil, ErrJoinTokenAlreadyConsumed + } + + return db.pinSnapshot(ctx, runner) + } + + token.ConsumedAt = p.Now + token.ConsumedBy = p.NodeID + + var outcome sqlair.Outcome + if err := runner.Query(ctx, db.consumeJoinTokenStmt, token).Get(&outcome); err != nil { + return nil, fmt.Errorf("consume join token: %w", err) + } + + rows, err := outcome.Result().RowsAffected() + if err != nil { + return nil, fmt.Errorf("rows affected: %w", err) + } + + if rows == 0 { + return nil, ErrJoinTokenAlreadyConsumed + } + + cert := ClusterNodeCert{ + NodeID: p.NodeID, + Fingerprint: p.Fingerprint, + CertPEM: p.CertPEM, + AddedAt: p.Now, + } + if err := runner.Query(ctx, db.upsertNodeCertStmt, cert).Run(); err != nil { + return nil, fmt.Errorf("upsert node cert: %w", err) + } + + logger.DBLog.Info("redeemed cluster join token", + zap.Int("nodeID", p.NodeID), + zap.String("fingerprint", p.Fingerprint)) + + return db.pinSnapshot(ctx, runner) +} + +func (db *Database) pinSnapshot(ctx context.Context, runner *sqlair.DB) (any, error) { + var pins []ClusterNodeCert + + if err := runner.Query(ctx, db.listNodeCertsStmt).GetAll(&pins); err != nil && !errors.Is(err, sql.ErrNoRows) { + return nil, fmt.Errorf("list pins: %w", err) + } + + return &RedeemJoinTokenResult{Pins: pins}, nil +} + func (db *Database) applyDeleteJoinTokensStale(ctx context.Context, cutoff *ClusterJoinToken) (any, error) { return nil, db.runner(ctx).Query(ctx, db.deleteJoinTokensStaleStmt, cutoff).Run() } @@ -235,6 +331,31 @@ func (db *Database) ConsumeJoinToken(ctx context.Context, id string, nodeID int) return err } +// RedeemJoinToken consumes a join token and pins the joining node's +// cert in one replicated apply, so a leadership change between the two +// can never burn the token without registering the cert. A repeat +// redemption by the same node presenting the same fingerprint is a +// no-op that returns the current pin set, making the joiner's retry +// loop safe. Any other repeat returns ErrJoinTokenAlreadyConsumed. +func (db *Database) RedeemJoinToken(ctx context.Context, tokenID string, nodeID int, fingerprint, certPEM string) ([]ClusterNodeCert, error) { + res, err := opRedeemJoinToken.Invoke(ctx, db, &redeemJoinTokenPayload{ + TokenID: tokenID, + NodeID: nodeID, + Fingerprint: fingerprint, + CertPEM: certPEM, + Now: time.Now().Unix(), + }) + if err != nil { + return nil, err + } + + if res == nil { + return nil, nil + } + + return res.Pins, nil +} + // DeleteStaleJoinTokens removes expired tokens and tokens consumed // more than an hour ago. func (db *Database) DeleteStaleJoinTokens(ctx context.Context, now time.Time) error { diff --git a/internal/db/cluster_pki_test.go b/internal/db/cluster_pki_test.go index f6dbdc839..f1dab1de3 100644 --- a/internal/db/cluster_pki_test.go +++ b/internal/db/cluster_pki_test.go @@ -172,3 +172,110 @@ func TestClusterJoinTokens_MintGetConsume(t *testing.T) { t.Fatalf("second consume should fail with ErrJoinTokenAlreadyConsumed, got %v", err) } } + +func mintTestToken(t *testing.T, database *db.Database, id string, nodeID int, ttl time.Duration) { + t.Helper() + + err := database.MintJoinTokenRecord(context.Background(), &db.ClusterJoinToken{ + ID: id, + NodeID: nodeID, + ClaimsJSON: "{}", + ExpiresAt: time.Now().Add(ttl).Unix(), + }) + if err != nil { + t.Fatalf("mint token record: %v", err) + } +} + +func TestRedeemJoinToken_ConsumesAndPinsTogether(t *testing.T) { + database := setupPKIDB(t) + ctx := context.Background() + + mintTestToken(t, database, "tok-1", 7, time.Hour) + + pins, err := database.RedeemJoinToken(ctx, "tok-1", 7, "sha256:aa", "PEM-7") + if err != nil { + t.Fatalf("redeem: %v", err) + } + + if len(pins) != 1 || pins[0].NodeID != 7 || pins[0].Fingerprint != "sha256:aa" { + t.Fatalf("unexpected pin snapshot: %+v", pins) + } + + row, err := database.GetJoinToken(ctx, "tok-1") + if err != nil { + t.Fatalf("get token: %v", err) + } + + if row.ConsumedAt == 0 || row.ConsumedBy != 7 { + t.Fatalf("token not consumed by node 7: %+v", row) + } +} + +// D-H7: a retry after the first attempt's outcome was lost must not +// find the token burnt. The joining node re-presents the identity it +// persisted before its first attempt, so the redemption replays. +func TestRedeemJoinToken_SameNodeSameCertReplays(t *testing.T) { + database := setupPKIDB(t) + ctx := context.Background() + + mintTestToken(t, database, "tok-2", 7, time.Hour) + + if _, err := database.RedeemJoinToken(ctx, "tok-2", 7, "sha256:aa", "PEM-7"); err != nil { + t.Fatalf("first redeem: %v", err) + } + + pins, err := database.RedeemJoinToken(ctx, "tok-2", 7, "sha256:aa", "PEM-7") + if err != nil { + t.Fatalf("replay must succeed, got: %v", err) + } + + if len(pins) != 1 || pins[0].Fingerprint != "sha256:aa" { + t.Fatalf("replay returned a different pin set: %+v", pins) + } +} + +func TestRedeemJoinToken_RejectsReplayWithDifferentCert(t *testing.T) { + database := setupPKIDB(t) + ctx := context.Background() + + mintTestToken(t, database, "tok-3", 7, time.Hour) + + if _, err := database.RedeemJoinToken(ctx, "tok-3", 7, "sha256:aa", "PEM-7"); err != nil { + t.Fatalf("first redeem: %v", err) + } + + _, err := database.RedeemJoinToken(ctx, "tok-3", 7, "sha256:bb", "PEM-7b") + if !errors.Is(err, db.ErrJoinTokenAlreadyConsumed) { + t.Fatalf("got %v, want ErrJoinTokenAlreadyConsumed", err) + } +} + +func TestRedeemJoinToken_RejectsWrongNodeAndExpiry(t *testing.T) { + database := setupPKIDB(t) + ctx := context.Background() + + mintTestToken(t, database, "tok-4", 7, time.Hour) + + if _, err := database.RedeemJoinToken(ctx, "tok-4", 8, "sha256:aa", "PEM-8"); !errors.Is(err, db.ErrJoinTokenNodeMismatch) { + t.Fatalf("wrong node: got %v, want ErrJoinTokenNodeMismatch", err) + } + + mintTestToken(t, database, "tok-5", 9, -time.Hour) + + if _, err := database.RedeemJoinToken(ctx, "tok-5", 9, "sha256:aa", "PEM-9"); !errors.Is(err, db.ErrJoinTokenExpired) { + t.Fatalf("expired: got %v, want ErrJoinTokenExpired", err) + } + + if _, err := database.GetClusterNodeCertByFingerprint(ctx, "sha256:aa"); !errors.Is(err, db.ErrNotFound) { + t.Fatal("a rejected redemption must not pin a cert") + } +} + +func TestRedeemJoinToken_UnknownTokenIsNotFound(t *testing.T) { + database := setupPKIDB(t) + + if _, err := database.RedeemJoinToken(context.Background(), "nope", 7, "sha256:aa", "PEM"); !errors.Is(err, db.ErrNotFound) { + t.Fatalf("got %v, want ErrNotFound", err) + } +} diff --git a/internal/db/db.go b/internal/db/db.go index 3534836b7..1a4abce22 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -270,7 +270,6 @@ type Database struct { countFlowReportsStmt *sqlair.Statement deleteOldFlowReportsStmt *sqlair.Statement deleteAllFlowReportsStmt *sqlair.Statement - getFlowReportByIDStmt *sqlair.Statement listFlowReportsByDayStmt *sqlair.Statement listFlowReportsBySubscriberStmt *sqlair.Statement flowReportProtocolCountsStmt *sqlair.Statement @@ -311,6 +310,7 @@ type Database struct { // Cluster PKI statements listNodeCertsStmt *sqlair.Statement getNodeCertByFPStmt *sqlair.Statement + getNodeCertByNodeStmt *sqlair.Statement upsertNodeCertStmt *sqlair.Statement deleteNodeCertByNodeStmt *sqlair.Statement insertJoinTokenStmt *sqlair.Statement @@ -1512,7 +1512,6 @@ func (db *Database) PrepareStatements() error { {&db.countFlowReportsStmt, fmt.Sprintf(countFlowReportsFilteredStmt, FlowReportsTableName), []any{FlowReportFilters{}, NumItems{}}}, {&db.deleteOldFlowReportsStmt, fmt.Sprintf(deleteOldFlowReportsStmt, FlowReportsTableName), []any{cutoffArgs{}}}, {&db.deleteAllFlowReportsStmt, fmt.Sprintf(deleteAllFlowReportsStmt, FlowReportsTableName), nil}, - {&db.getFlowReportByIDStmt, fmt.Sprintf(getFlowReportByIDStmt, FlowReportsTableName), []any{dbwriter.FlowReport{}}}, {&db.listFlowReportsByDayStmt, fmt.Sprintf(listFlowReportsFilteredByDayStmt, FlowReportsTableName), []any{FlowReportFilters{}, dbwriter.FlowReport{}}}, {&db.listFlowReportsBySubscriberStmt, fmt.Sprintf(listFlowReportsFilteredBySubscriberStmt, FlowReportsTableName), []any{FlowReportFilters{}, dbwriter.FlowReport{}}}, {&db.flowReportProtocolCountsStmt, fmt.Sprintf(flowReportProtocolCountsStmt, FlowReportsTableName), []any{FlowReportFilters{}, FlowReportProtocolCount{}}}, @@ -1553,6 +1552,7 @@ func (db *Database) PrepareStatements() error { // Cluster PKI (v12 fingerprint pinning) {&db.listNodeCertsStmt, fmt.Sprintf(listNodeCertsStmtStr, ClusterNodeCertsTableName), []any{ClusterNodeCert{}}}, {&db.getNodeCertByFPStmt, fmt.Sprintf(getNodeCertByFPStmtStr, ClusterNodeCertsTableName), []any{ClusterNodeCert{}}}, + {&db.getNodeCertByNodeStmt, fmt.Sprintf(getNodeCertByNodeStmtStr, ClusterNodeCertsTableName), []any{ClusterNodeCert{}}}, {&db.upsertNodeCertStmt, fmt.Sprintf(upsertNodeCertStmtStr, ClusterNodeCertsTableName), []any{ClusterNodeCert{}}}, {&db.deleteNodeCertByNodeStmt, fmt.Sprintf(deleteNodeCertByNodeStmtStr, ClusterNodeCertsTableName), []any{ClusterNodeCert{}}}, {&db.insertJoinTokenStmt, fmt.Sprintf(insertJoinTokenStmtStr, ClusterJoinTokensTableName), []any{ClusterJoinToken{}}}, diff --git a/internal/db/error.go b/internal/db/error.go index d41b6d172..475ef3c9b 100644 --- a/internal/db/error.go +++ b/internal/db/error.go @@ -22,6 +22,8 @@ var ( ErrOutcomeUnknown = ellaraft.ErrOutcomeUnknown ErrMigrationPending = errors.New("schema migration pending") ErrJoinTokenAlreadyConsumed = errors.New("join token already consumed") + ErrJoinTokenNodeMismatch = errors.New("join token is not registered to this node") + ErrJoinTokenExpired = errors.New("join token expired") ErrUnknownOperation = errors.New("unknown forwarded operation") ) diff --git a/internal/db/flow_reports.go b/internal/db/flow_reports.go index 863335c9e..951263cae 100644 --- a/internal/db/flow_reports.go +++ b/internal/db/flow_reports.go @@ -22,7 +22,6 @@ const FlowReportsTableName = "flow_reports" const ( insertFlowReportStmt = "INSERT INTO %s (subscriber_id, source_ip, destination_ip, source_port, destination_port, protocol, packets, bytes, start_time, end_time, direction, action) VALUES ($FlowReport.subscriber_id, $FlowReport.source_ip, $FlowReport.destination_ip, $FlowReport.source_port, $FlowReport.destination_port, $FlowReport.protocol, $FlowReport.packets, $FlowReport.bytes, $FlowReport.start_time, $FlowReport.end_time, $FlowReport.direction, $FlowReport.action)" - getFlowReportByIDStmt = "SELECT &FlowReport.* FROM %s WHERE id = $FlowReport.id" deleteOldFlowReportsStmt = "DELETE FROM %s WHERE end_time < $cutoffArgs.cutoff" deleteAllFlowReportsStmt = "DELETE FROM %s" ) diff --git a/internal/db/ip_leases.go b/internal/db/ip_leases.go index 396971266..325da87aa 100644 --- a/internal/db/ip_leases.go +++ b/internal/db/ip_leases.go @@ -375,8 +375,10 @@ func (db *Database) DeleteDynamicLease(ctx context.Context, leaseID string) erro return nil } -// DeleteAllDynamicLeases removes all dynamic leases. Called on startup to clean -// up stale leases from a previous process lifetime. Static leases are preserved. +// DeleteAllDynamicLeases removes all dynamic leases cluster-wide, +// preserving static ones. No production caller: under HA this would +// delete leases owned by every other node. Per-node cleanup on startup +// goes through DeleteDynamicLeasesByNode instead. func (db *Database) DeleteAllDynamicLeases(ctx context.Context) error { _, span := tracer.Start( ctx, diff --git a/internal/db/operations_register.go b/internal/db/operations_register.go index c260d89e2..c810c849d 100644 --- a/internal/db/operations_register.go +++ b/internal/db/operations_register.go @@ -158,6 +158,7 @@ var ( opDeleteNodeCert = registerChangesetOp("DeleteClusterNodeCert", (*Database).applyDeleteNodeCert, RequireSchema(12), AffectsTopic(TopicClusterNodeCerts)) opMintJoinToken = registerChangesetOp("MintJoinToken", (*Database).applyInsertJoinToken, RequireSchema(9)) opConsumeJoinToken = registerChangesetOp("ConsumeJoinToken", (*Database).applyConsumeJoinToken, RequireSchema(9)) + opRedeemJoinToken = registerChangesetOpReturning[redeemJoinTokenPayload, *RedeemJoinTokenResult]("RedeemJoinToken", (*Database).applyRedeemJoinToken, RequireSchema(12), AffectsTopic(TopicClusterNodeCerts)) opDeleteStaleJoinTokens = registerChangesetOp("DeleteStaleJoinTokens", (*Database).applyDeleteJoinTokensStale, RequireSchema(9)) opInitJoinHMAC = registerChangesetOp("InitClusterJoinHMACKey", (*Database).applyInitJoinHMAC, RequireSchema(12)) ) diff --git a/internal/db/operations_registry_internal_test.go b/internal/db/operations_registry_internal_test.go new file mode 100644 index 000000000..ac031f382 --- /dev/null +++ b/internal/db/operations_registry_internal_test.go @@ -0,0 +1,152 @@ +// SPDX-FileCopyrightText: Ella Networks Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package db + +import "testing" + +// Editing these maps is the deliberate act that adding, retiring or +// renaming a replicated operation requires. +var pinnedChangesetOps = map[string]int{ + "AdvanceSubscriberSQN": 1, + "AllocateIPLease": 12, + "AllocateIPv6Lease": 12, + "ClearDailyUsage": 1, + "ConsumeJoinToken": 9, + "CreateAPIToken": 1, + "CreateDataNetwork": 1, + "CreateHomeNetworkKey": 1, + "CreateLease": 9, + "CreateNetworkRule": 1, + "CreateNetworkSlice": 1, + "CreatePolicy": 1, + "CreatePolicyWithRules": 1, + "CreateProfile": 1, + "CreateSession": 1, + "CreateStaticLease": 13, + "CreateSubscriber": 1, + "CreateUser": 1, + "DeleteAPIToken": 1, + "DeleteAllSessions": 1, + "DeleteAllSessionsForUser": 1, + "DeleteClusterMember": 9, + "DeleteClusterNodeCert": 12, + "DeleteDataNetwork": 1, + "DeleteDynamicLease": 9, + "DeleteDynamicLeasesByNode": 9, + "DeleteHomeNetworkKey": 1, + "DeleteNetworkRule": 1, + "DeleteNetworkRulesByPolicy": 1, + "DeleteNetworkSlice": 1, + "DeleteOldestSessions": 1, + "DeletePolicy": 1, + "DeleteProfile": 1, + "DeleteSessionByTokenHash": 1, + "DeleteStaleJoinTokens": 9, + "DeleteStaticLease": 13, + "DeleteSubscriber": 1, + "DeleteUser": 1, + "EditSubscriberSeqNum": 1, + "IncrementDailyUsage": 1, + "InitClusterJoinHMACKey": 12, + "InitializeOperator": 1, + "InsertAuditLog": 1, + "MintJoinToken": 9, + "RedeemJoinToken": 12, + "ReleaseIPLease": 13, + "ReplaceFramedRoutes": 16, + "SetDefaultPolicy": 14, + "SetDrainState": 9, + "SetJWTSecret": 1, + "SetRetentionPolicy": 1, + "UpdateDataNetwork": 1, + "UpdateLeaseNode": 9, + "UpdateLeaseSession": 9, + "UpdateNetworkRule": 1, + "UpdateNetworkSlice": 1, + "UpdateOperatorAMFIdentity": 9, + "UpdateOperatorClusterID": 1, + "UpdateOperatorCode": 1, + "UpdateOperatorID": 1, + "UpdateOperatorSPN": 1, + "UpdateOperatorSecurityAlgorithms": 1, + "UpdateOperatorTracking": 1, + "UpdatePolicy": 1, + "UpdatePolicyWithRules": 1, + "UpdateProfile": 1, + "UpdateStaticLeaseAddress": 13, + "UpdateSubscriberProfile": 1, + "UpdateUser": 1, + "UpdateUserPassword": 1, + "UpsertClusterMember": 9, + "UpsertClusterNodeCert": 12, +} + +var pinnedIntentOps = map[string]int{ + "DeleteAllDynamicLeases": 1, + "DeleteExpiredSessions": 1, + "DeleteOldAuditLogs": 1, + "DeleteOldDailyUsage": 1, + "MigrateShared": 1, +} + +const operationRule = `A replicated operation was added, retired, renamed, or had its +RequireSchema changed. + +The operation name is the forwarded wire contract: a follower running an +older binary sends its own name to whichever node is leader, and the +leader dispatches by that name. Retiring a name breaks every write +forwarded by a not-yet-upgraded node during a rolling upgrade, which is +why an operation with no remaining Go caller must still stay registered. +Lowering RequireSchema lets an operation apply against a schema that +predates the columns it writes. + +Add or amend the pinned entry deliberately, in the same change.` + +func TestChangesetOpRegistryIsPinned(t *testing.T) { + assertRegistryPinned(t, pinnedChangesetOps, changesetOpSchemas()) +} + +func TestIntentOpRegistryIsPinned(t *testing.T) { + assertRegistryPinned(t, pinnedIntentOps, intentOpSchemas()) +} + +func changesetOpSchemas() map[string]int { + out := make(map[string]int, len(changesetOps)) + for name, h := range changesetOps { + out[name] = h.minSchema + } + + return out +} + +func intentOpSchemas() map[string]int { + out := make(map[string]int, len(intentOps)) + for name, h := range intentOps { + out[name] = h.minSchema + } + + return out +} + +func assertRegistryPinned(t *testing.T, pinned, registered map[string]int) { + t.Helper() + + for name, want := range pinned { + got, ok := registered[name] + if !ok { + t.Errorf("operation %q is pinned but no longer registered\n\n%s", name, operationRule) + continue + } + + if got != want { + t.Errorf("operation %q: RequireSchema is %d, pinned as %d\n\n%s", name, got, want, operationRule) + } + } + + for name := range registered { + if _, ok := pinned[name]; !ok { + t.Errorf("operation %q is registered but not pinned\n\n%s", name, operationRule) + } + } +} diff --git a/internal/pki/pki.go b/internal/pki/pki.go index fe4cd53ab..46cbea6a0 100644 --- a/internal/pki/pki.go +++ b/internal/pki/pki.go @@ -189,30 +189,6 @@ func EncodePrivateKeyPEM(key crypto.Signer) ([]byte, error) { return pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: der}), nil } -// ParsePrivateKeyPEM decodes a PKCS#8 PEM private key as a crypto.Signer. -func ParsePrivateKeyPEM(keyPEM []byte) (crypto.Signer, error) { - if len(keyPEM) == 0 { - return nil, fmt.Errorf("empty private key PEM") - } - - block, _ := pem.Decode(keyPEM) - if block == nil || block.Type != "PRIVATE KEY" { - return nil, fmt.Errorf("not a PRIVATE KEY PEM") - } - - k, err := x509.ParsePKCS8PrivateKey(block.Bytes) - if err != nil { - return nil, fmt.Errorf("parse PKCS#8: %w", err) - } - - signer, ok := k.(crypto.Signer) - if !ok { - return nil, fmt.Errorf("key is not a crypto.Signer") - } - - return signer, nil -} - // IdentityFromCert validates the SPIFFE URI SAN of a cluster cert // and returns its (clusterID, nodeID). func IdentityFromCert(cert *x509.Certificate) (clusterID string, nodeID int, err error) { diff --git a/internal/pki/tokens.go b/internal/pki/tokens.go index a1f5787db..3f1530cff 100644 --- a/internal/pki/tokens.go +++ b/internal/pki/tokens.go @@ -21,12 +21,29 @@ import ( // bootstrap TLS handshake), and the cluster's identity (so the // joiner mints a cert with the matching SPIFFE URI). type JoinClaims struct { - TokenID string `json:"id"` - NodeID int `json:"node_id"` - IssuedAt int64 `json:"iat"` - ExpiresAt int64 `json:"exp"` - LeaderCertPin string `json:"lcp"` - ClusterID string `json:"cid"` + TokenID string `json:"id"` + NodeID int `json:"node_id"` + IssuedAt int64 `json:"iat"` + ExpiresAt int64 `json:"exp"` + LeaderCertPin string `json:"lcp"` + ClusterID string `json:"cid"` + ClusterPins []string `json:"pins,omitempty"` +} + +// PinSet returns every cert fingerprint the joining node may pin its +// bootstrap handshake against. Tokens minted before ClusterPins +// existed carry only the minting leader's pin, so the joiner falls +// back to LeaderCertPin. +func (c *JoinClaims) PinSet() []string { + if len(c.ClusterPins) > 0 { + return c.ClusterPins + } + + if c.LeaderCertPin == "" { + return nil + } + + return []string{c.LeaderCertPin} } // joinTokenVersion is the first byte of every serialized token; bumped if diff --git a/internal/pki/tokens_test.go b/internal/pki/tokens_test.go index 54509d31b..9a82b3cfc 100644 --- a/internal/pki/tokens_test.go +++ b/internal/pki/tokens_test.go @@ -212,3 +212,23 @@ func TestMint_ShortKey(t *testing.T) { t.Fatal("short key must be rejected") } } + +func TestJoinClaims_PinSet(t *testing.T) { + multi := pki.JoinClaims{ + LeaderCertPin: "sha256:aa", + ClusterPins: []string{"sha256:aa", "sha256:bb"}, + } + if got := multi.PinSet(); len(got) != 2 { + t.Fatalf("PinSet() = %v, want both cluster pins", got) + } + + legacy := pki.JoinClaims{LeaderCertPin: "sha256:aa"} + if got := legacy.PinSet(); len(got) != 1 || got[0] != "sha256:aa" { + t.Fatalf("a token minted before ClusterPins must fall back to the leader pin, got %v", got) + } + + var empty pki.JoinClaims + if got := empty.PinSet(); got != nil { + t.Fatalf("PinSet() = %v, want nil when the token carries no pin", got) + } +} diff --git a/internal/raft/manager.go b/internal/raft/manager.go index 49eb6d022..a77e6650e 100644 --- a/internal/raft/manager.go +++ b/internal/raft/manager.go @@ -707,8 +707,8 @@ func (m *Manager) barrierFor(term uint64) *barrierAttempt { } // Snapshot triggers a user-requested Raft snapshot and blocks until it -// completes. Callers use this to force log truncation after large log -// entries so followers don't carry large blobs in their log indefinitely. +// completes. No production caller; reached only from db.ForceSnapshot, +// which tests use to exercise the snapshot-restore path. func (m *Manager) Snapshot() error { future := m.raft.Snapshot() if err := future.Error(); err != nil { diff --git a/pkg/runtime/pki.go b/pkg/runtime/pki.go index b7e84b76d..eecd31938 100644 --- a/pkg/runtime/pki.go +++ b/pkg/runtime/pki.go @@ -61,6 +61,13 @@ func (p *pkiState) ensureIssuer(dbInstance *db.Database) { } } +func (p *pkiState) Issuer() *pkiissuer.Service { + p.issuerMu.Lock() + defer p.issuerMu.Unlock() + + return p.issuer +} + func newPKIState(nodeID int, clusterID, dataDir string) *pkiState { return &pkiState{ agent: pkiagent.NewAgent(nodeID, clusterID, dataDir), diff --git a/pkg/runtime/pki_leader.go b/pkg/runtime/pki_leader.go index 636c0cd87..a1a942f61 100644 --- a/pkg/runtime/pki_leader.go +++ b/pkg/runtime/pki_leader.go @@ -9,8 +9,6 @@ import ( "sync" "time" - "github.com/ellanetworks/core/internal/api/server" - "github.com/ellanetworks/core/internal/cluster/listener" "github.com/ellanetworks/core/internal/db" "github.com/ellanetworks/core/internal/logger" "github.com/ellanetworks/core/internal/pki" @@ -26,24 +24,20 @@ type pkiLeaderCallback struct { ctx context.Context state *pkiState dbInstance *db.Database - clusterLn *listener.Listener nodeID int binaryVersion string needsDRSnapshot bool - bootstrapRegistered sync.Once - mu sync.Mutex leaderCancel context.CancelFunc } -func newPKILeaderCallback(ctx context.Context, state *pkiState, dbInstance *db.Database, ln *listener.Listener, nodeID int, binaryVersion string, needsDRSnapshot bool) *pkiLeaderCallback { +func newPKILeaderCallback(ctx context.Context, state *pkiState, dbInstance *db.Database, nodeID int, binaryVersion string, needsDRSnapshot bool) *pkiLeaderCallback { return &pkiLeaderCallback{ ctx: ctx, state: state, dbInstance: dbInstance, - clusterLn: ln, nodeID: nodeID, binaryVersion: binaryVersion, needsDRSnapshot: needsDRSnapshot, @@ -75,8 +69,6 @@ func (c *pkiLeaderCallback) OnBecameLeader() { return } - - c.onLeaderInitSuccess() } func (c *pkiLeaderCallback) OnLostLeadership() { @@ -122,7 +114,6 @@ func (c *pkiLeaderCallback) retryLeaderInit(ctx context.Context) { err := runLeaderInit(ctx, c.state, c.dbInstance, c.nodeID, c.binaryVersion) if err == nil { logger.EllaLog.Info("leader init recovered after retry") - c.onLeaderInitSuccess() return } @@ -138,14 +129,6 @@ func (c *pkiLeaderCallback) retryLeaderInit(ctx context.Context) { } } -func (c *pkiLeaderCallback) onLeaderInitSuccess() { - if c.clusterLn != nil && c.state != nil && c.state.issuer != nil { - c.bootstrapRegistered.Do(func() { - server.RegisterBootstrapALPN(c.clusterLn, c.state.issuer) - }) - } -} - // runLeaderInit is idempotent. func runLeaderInit(ctx context.Context, pki *pkiState, dbInstance *db.Database, nodeID int, binaryVersion string) error { if err := dbInstance.Initialize(ctx); err != nil { diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index 5ec782c8a..c3d7e1952 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -243,6 +243,16 @@ func Start(ctx context.Context, rc RuntimeConfig) error { stopClusterHTTP := server.StartClusterHTTP(dbInstance, clusterLn) defer stopClusterHTTP() + // Every voter serves the join endpoint, not just the node + // that happens to be leader: redeeming a token is a + // replicated op that forwards to the leader on its own, so + // a joiner whose token names several peers reaches the + // cluster through whichever one answers. + if pki != nil { + pki.ensureIssuer(dbInstance) + server.RegisterBootstrapALPN(clusterLn, pki.Issuer()) + } + if err := clusterLn.Start(ctx); err != nil { return fmt.Errorf("cluster listener: %w", err) } @@ -260,7 +270,7 @@ func Start(ctx context.Context, rc RuntimeConfig) error { observer.Register(server.NewLeadershipAuditCallback(dbInstance.NodeID())) if pki != nil { - observer.Register(newPKILeaderCallback(ctx, pki, dbInstance, clusterLn, cfg.Cluster.NodeID, ver.Version, restoredFromBundle)) + observer.Register(newPKILeaderCallback(ctx, pki, dbInstance, cfg.Cluster.NodeID, ver.Version, restoredFromBundle)) } } From 3853835620ece344df8b4fbe7af1f01f75c74d41 Mon Sep 17 00:00:00 2001 From: Guillaume Belanger Date: Wed, 26 Aug 2026 08:32:06 -0400 Subject: [PATCH 2/5] chore: tighten code comments Signed-off-by: Guillaume Belanger --- internal/cluster/pkiagent/agent.go | 10 +--------- internal/cluster/pkiagent/agent_test.go | 4 ---- internal/cluster/pkiissuer/service.go | 7 ------- internal/cluster/pkiissuer/service_test.go | 7 ------- internal/db/cluster_pki.go | 8 -------- internal/db/cluster_pki_test.go | 3 --- internal/db/ip_leases.go | 4 +--- internal/db/operations_registry_internal_test.go | 2 -- internal/pki/tokens.go | 4 ---- pkg/runtime/runtime.go | 5 ----- 10 files changed, 2 insertions(+), 52 deletions(-) diff --git a/internal/cluster/pkiagent/agent.go b/internal/cluster/pkiagent/agent.go index a872d26c5..26d282e0b 100644 --- a/internal/cluster/pkiagent/agent.go +++ b/internal/cluster/pkiagent/agent.go @@ -261,12 +261,6 @@ func (a *Agent) JoinFlow(ctx context.Context, serverAddr, token string) error { return nil } -// ensureJoinCert returns the node's join identity, generating and -// persisting it on first call and reloading the same keypair on every -// later call. The identity must survive both a retry and a restart: -// the leader pins the fingerprint it was shown when it consumed the -// token, and only a node presenting that same fingerprint can replay -// the redemption idempotently. func (a *Agent) ensureJoinCert() (certPEM, keyPEM []byte, cert *x509.Certificate, err error) { certPEM, err = os.ReadFile(a.path(joinCertFile)) // #nosec G304 -- under dataDir if err == nil { @@ -452,9 +446,7 @@ func atomicWrite(path string, data []byte, mode os.FileMode) error { // bootstrapHTTPClient returns an HTTP client that dials the bootstrap // ALPN without a client cert and pins the server cert to any -// fingerprint in expectedFingerprints. The join token carries every -// voter's pin, so a joiner is not tied to the node that minted it and -// can reach whichever node is leader when it retries. +// fingerprint in expectedFingerprints. func bootstrapHTTPClient(expectedFingerprints []string) (*http.Client, error) { raws := make([][]byte, 0, len(expectedFingerprints)) diff --git a/internal/cluster/pkiagent/agent_test.go b/internal/cluster/pkiagent/agent_test.go index 9449adea7..bd9f46e1f 100644 --- a/internal/cluster/pkiagent/agent_test.go +++ b/internal/cluster/pkiagent/agent_test.go @@ -175,10 +175,6 @@ func alwaysFailRegisterHandler() listener.ConnHandler { } } -// A joining node must present the same identity on every attempt: -// the leader pins the fingerprint it saw when it consumed the token, -// and only a node re-presenting that fingerprint can replay the -// redemption instead of finding the token burnt. func TestAgent_JoinFlow_ReusesIdentityAcrossRetries(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() diff --git a/internal/cluster/pkiissuer/service.go b/internal/cluster/pkiissuer/service.go index 7459db93c..3ab28a966 100644 --- a/internal/cluster/pkiissuer/service.go +++ b/internal/cluster/pkiissuer/service.go @@ -187,13 +187,6 @@ func (s *Service) pinsForToken(ctx context.Context, leaderNodeID int) (string, [ return "", nil, fmt.Errorf("leader node %d has no registered pin", leaderNodeID) } -// RedeemJoinToken authenticates tokenStr, validates certPEM, and then -// consumes the token and pins the cert in a single replicated apply. -// Consumption can no longer commit without the pin that it authorises, -// so a leadership change mid-join leaves the token usable rather than -// burnt. A repeat redemption by the same node presenting the same cert -// succeeds and returns the current pin set, making the joiner's retry -// loop safe; any other repeat returns db.ErrJoinTokenAlreadyConsumed. func (s *Service) RedeemJoinToken(ctx context.Context, tokenStr string, nodeID int, certPEM []byte) (string, []db.ClusterNodeCert, error) { hmacKey, err := s.store.GetClusterJoinHMACKey(ctx) if err != nil { diff --git a/internal/cluster/pkiissuer/service_test.go b/internal/cluster/pkiissuer/service_test.go index a3ea6077b..7f0ab7fb6 100644 --- a/internal/cluster/pkiissuer/service_test.go +++ b/internal/cluster/pkiissuer/service_test.go @@ -321,7 +321,6 @@ func TestService_MintAndVerifyJoinToken_RoundTrip(t *testing.T) { t.Fatalf("redeem returned fp=%q pins=%d, want non-empty fp and 2 pins", fp, len(pins)) } - // Replay by a different node must be rejected. otherPEM := nodeCertPEM(t, 6) if _, _, err := svc.RedeemJoinToken(context.Background(), token, 6, otherPEM); err == nil { t.Fatal("replay for a different node should be rejected") @@ -359,9 +358,6 @@ func TestService_NotLeader_RejectsMutations(t *testing.T) { } } -// RegisterCert no longer gates on leadership: the pin upsert is a -// replicated op that forwards to the leader on its own, so a rotation -// that lands on a follower is completed rather than refused. func TestService_RegisterCert_WorksOnNonLeader(t *testing.T) { store := newFakeStore("c") store.leader = false @@ -392,9 +388,6 @@ func TestService_Redeem_ReplayWithDifferentCertRejected(t *testing.T) { } } -// A joining node that retries after a leadership change presents the -// same cert it persisted before its first attempt, so the redemption -// replays as a no-op instead of reporting the token burnt. func TestService_Redeem_SameNodeSameCertIsIdempotent(t *testing.T) { store := newFakeStore("c") preregisterLeader(t, store, 1) diff --git a/internal/db/cluster_pki.go b/internal/db/cluster_pki.go index 219b4d8ae..84ffd8e0b 100644 --- a/internal/db/cluster_pki.go +++ b/internal/db/cluster_pki.go @@ -142,8 +142,6 @@ type redeemJoinTokenPayload struct { Now int64 `json:"now"` } -// RedeemJoinTokenResult is the post-commit pin snapshot returned to a -// joining node so it can seed its local pin map. type RedeemJoinTokenResult struct { Pins []ClusterNodeCert `json:"pins"` } @@ -331,12 +329,6 @@ func (db *Database) ConsumeJoinToken(ctx context.Context, id string, nodeID int) return err } -// RedeemJoinToken consumes a join token and pins the joining node's -// cert in one replicated apply, so a leadership change between the two -// can never burn the token without registering the cert. A repeat -// redemption by the same node presenting the same fingerprint is a -// no-op that returns the current pin set, making the joiner's retry -// loop safe. Any other repeat returns ErrJoinTokenAlreadyConsumed. func (db *Database) RedeemJoinToken(ctx context.Context, tokenID string, nodeID int, fingerprint, certPEM string) ([]ClusterNodeCert, error) { res, err := opRedeemJoinToken.Invoke(ctx, db, &redeemJoinTokenPayload{ TokenID: tokenID, diff --git a/internal/db/cluster_pki_test.go b/internal/db/cluster_pki_test.go index f1dab1de3..756c7e68d 100644 --- a/internal/db/cluster_pki_test.go +++ b/internal/db/cluster_pki_test.go @@ -212,9 +212,6 @@ func TestRedeemJoinToken_ConsumesAndPinsTogether(t *testing.T) { } } -// D-H7: a retry after the first attempt's outcome was lost must not -// find the token burnt. The joining node re-presents the identity it -// persisted before its first attempt, so the redemption replays. func TestRedeemJoinToken_SameNodeSameCertReplays(t *testing.T) { database := setupPKIDB(t) ctx := context.Background() diff --git a/internal/db/ip_leases.go b/internal/db/ip_leases.go index 325da87aa..a3dff13e8 100644 --- a/internal/db/ip_leases.go +++ b/internal/db/ip_leases.go @@ -376,9 +376,7 @@ func (db *Database) DeleteDynamicLease(ctx context.Context, leaseID string) erro } // DeleteAllDynamicLeases removes all dynamic leases cluster-wide, -// preserving static ones. No production caller: under HA this would -// delete leases owned by every other node. Per-node cleanup on startup -// goes through DeleteDynamicLeasesByNode instead. +// preserving static ones. No production caller. func (db *Database) DeleteAllDynamicLeases(ctx context.Context) error { _, span := tracer.Start( ctx, diff --git a/internal/db/operations_registry_internal_test.go b/internal/db/operations_registry_internal_test.go index ac031f382..92b6c519d 100644 --- a/internal/db/operations_registry_internal_test.go +++ b/internal/db/operations_registry_internal_test.go @@ -5,8 +5,6 @@ package db import "testing" -// Editing these maps is the deliberate act that adding, retiring or -// renaming a replicated operation requires. var pinnedChangesetOps = map[string]int{ "AdvanceSubscriberSQN": 1, "AllocateIPLease": 12, diff --git a/internal/pki/tokens.go b/internal/pki/tokens.go index 3f1530cff..ff1dacccd 100644 --- a/internal/pki/tokens.go +++ b/internal/pki/tokens.go @@ -30,10 +30,6 @@ type JoinClaims struct { ClusterPins []string `json:"pins,omitempty"` } -// PinSet returns every cert fingerprint the joining node may pin its -// bootstrap handshake against. Tokens minted before ClusterPins -// existed carry only the minting leader's pin, so the joiner falls -// back to LeaderCertPin. func (c *JoinClaims) PinSet() []string { if len(c.ClusterPins) > 0 { return c.ClusterPins diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index c3d7e1952..0d18da774 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -243,11 +243,6 @@ func Start(ctx context.Context, rc RuntimeConfig) error { stopClusterHTTP := server.StartClusterHTTP(dbInstance, clusterLn) defer stopClusterHTTP() - // Every voter serves the join endpoint, not just the node - // that happens to be leader: redeeming a token is a - // replicated op that forwards to the leader on its own, so - // a joiner whose token names several peers reaches the - // cluster through whichever one answers. if pki != nil { pki.ensureIssuer(dbInstance) server.RegisterBootstrapALPN(clusterLn, pki.Issuer()) From 8fefe40682f964011b9d39a06b9e12214f82c936 Mon Sep 17 00:00:00 2001 From: Guillaume Belanger Date: Wed, 26 Aug 2026 08:46:38 -0400 Subject: [PATCH 3/5] fix: pr review Signed-off-by: Guillaume Belanger --- internal/api/server/cluster_http_forward.go | 28 ++++++++++ internal/api/server/cluster_pki.go | 27 +++++++--- internal/cluster/pkiagent/agent.go | 48 +++++++++++++---- internal/cluster/pkiagent/agent_test.go | 54 +++++++++++++++++++ internal/db/cluster_pki.go | 13 +++-- internal/db/forward_sentinel_internal_test.go | 33 ++++++++++++ internal/db/operations.go | 23 ++++++++ .../db/operations_registry_internal_test.go | 4 ++ internal/raft/forward.go | 30 ++++++++++- internal/raft/forward_coded_error_test.go | 50 +++++++++++++++++ 10 files changed, 283 insertions(+), 27 deletions(-) create mode 100644 internal/db/forward_sentinel_internal_test.go create mode 100644 internal/raft/forward_coded_error_test.go diff --git a/internal/api/server/cluster_http_forward.go b/internal/api/server/cluster_http_forward.go index 64c60028c..cc26ddcbf 100644 --- a/internal/api/server/cluster_http_forward.go +++ b/internal/api/server/cluster_http_forward.go @@ -117,12 +117,40 @@ func mapApplyErrorToHTTP(ctx context.Context, w http.ResponseWriter, err error) writeProposeForwardError(ctx, w, http.StatusServiceUnavailable, "raft busy or shutting down", err) + case errors.Is(err, db.ErrMigrationPending): + writeProposeForwardCodedError(ctx, w, http.StatusServiceUnavailable, + "schema migration pending", ellaraft.ForwardCodeMigrationPend, err) + default: + if code := forwardCodeForDomainErr(err); code != "" { + writeProposeForwardCodedError(ctx, w, http.StatusConflict, + err.Error(), code, err) + + return + } + writeProposeForwardError(ctx, w, http.StatusInternalServerError, "apply failed", err) } } +func forwardCodeForDomainErr(err error) string { + switch { + case errors.Is(err, db.ErrJoinTokenAlreadyConsumed): + return ellaraft.ForwardCodeTokenConsumed + case errors.Is(err, db.ErrJoinTokenExpired): + return ellaraft.ForwardCodeTokenExpired + case errors.Is(err, db.ErrJoinTokenNodeMismatch): + return ellaraft.ForwardCodeTokenNodeMism + case errors.Is(err, db.ErrAlreadyExists): + return ellaraft.ForwardCodeAlreadyExists + case errors.Is(err, db.ErrNotFound): + return ellaraft.ForwardCodeNotFound + default: + return "" + } +} + func writeProposeForwardError(ctx context.Context, w http.ResponseWriter, status int, message string, cause error) { writeProposeForwardCodedError(ctx, w, status, message, "", cause) } diff --git a/internal/api/server/cluster_pki.go b/internal/api/server/cluster_pki.go index febf4a262..7f04ef4f6 100644 --- a/internal/api/server/cluster_pki.go +++ b/internal/api/server/cluster_pki.go @@ -26,6 +26,7 @@ import ( "github.com/ellanetworks/core/internal/db" "github.com/ellanetworks/core/internal/logger" "github.com/ellanetworks/core/internal/pki" + hraft "github.com/hashicorp/raft" "go.uber.org/zap" ) @@ -74,7 +75,7 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler { fp, pins, err = svc.RegisterCert(r.Context(), req.NodeID, []byte(req.CertPEM)) if err != nil { - writeError(r.Context(), w, http.StatusBadRequest, "register cert", err, logger.APILog) + writeError(r.Context(), w, clusterPKIStatus(err, http.StatusBadRequest), "register cert", err, logger.APILog) return } @@ -88,13 +89,7 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler { fp, pins, err = svc.RedeemJoinToken(r.Context(), req.Token, req.NodeID, []byte(req.CertPEM)) if err != nil { - status := http.StatusUnauthorized - if errors.Is(err, db.ErrMigrationPending) { - status = http.StatusServiceUnavailable - } - - writeError(r.Context(), w, status, "redeem join token", err, logger.APILog) - + writeError(r.Context(), w, clusterPKIStatus(err, http.StatusUnauthorized), "redeem join token", err, logger.APILog) return } } @@ -113,6 +108,22 @@ func ClusterPKIRegister(svc *pkiissuer.Service) http.Handler { }) } +func clusterPKIStatus(err error, deny int) int { + switch { + case errors.Is(err, db.ErrMigrationPending), + errors.Is(err, db.ErrProposeTimeout), + errors.Is(err, db.ErrNotFound), + errors.Is(err, hraft.ErrNotLeader): + return http.StatusServiceUnavailable + + case errors.Is(err, db.ErrOutcomeUnknown): + return http.StatusConflict + + default: + return deny + } +} + // RegisterBootstrapALPN dispatches POST /cluster/pki/register on // the bootstrap ALPN (no client cert) and closes the connection // after one request. diff --git a/internal/cluster/pkiagent/agent.go b/internal/cluster/pkiagent/agent.go index 26d282e0b..ca5162dfe 100644 --- a/internal/cluster/pkiagent/agent.go +++ b/internal/cluster/pkiagent/agent.go @@ -236,7 +236,7 @@ func (a *Agent) JoinFlow(ctx context.Context, serverAddr, token string) error { a.ClusterID = claims.ClusterID } - certPEM, keyPEM, cert, err := a.ensureJoinCert() + certPEM, keyPEM, cert, err := a.ensureJoinCert(claims.ClusterID) if err != nil { return fmt.Errorf("prepare cert: %w", err) } @@ -261,16 +261,10 @@ func (a *Agent) JoinFlow(ctx context.Context, serverAddr, token string) error { return nil } -func (a *Agent) ensureJoinCert() (certPEM, keyPEM []byte, cert *x509.Certificate, err error) { - certPEM, err = os.ReadFile(a.path(joinCertFile)) // #nosec G304 -- under dataDir - if err == nil { - keyPEM, err = os.ReadFile(a.path(joinKeyFile)) // #nosec G304 -- under dataDir - if err == nil { - cert, err = pki.ParseCertPEM(certPEM) - if err == nil { - return certPEM, keyPEM, cert, nil - } - } +func (a *Agent) ensureJoinCert(clusterID string) (certPEM, keyPEM []byte, cert *x509.Certificate, err error) { + certPEM, keyPEM, cert = a.loadJoinCert(clusterID) + if cert != nil { + return certPEM, keyPEM, cert, nil } certPEM, keyPEM, cert, err = a.prepareNewCert() @@ -293,6 +287,38 @@ func (a *Agent) ensureJoinCert() (certPEM, keyPEM []byte, cert *x509.Certificate return certPEM, keyPEM, cert, nil } +func (a *Agent) loadJoinCert(clusterID string) (certPEM, keyPEM []byte, cert *x509.Certificate) { + certPEM, err := os.ReadFile(a.path(joinCertFile)) // #nosec G304 -- under dataDir + if err != nil { + return nil, nil, nil + } + + keyPEM, err = os.ReadFile(a.path(joinKeyFile)) // #nosec G304 -- under dataDir + if err != nil { + return nil, nil, nil + } + + cert, err = pki.ParseCertPEM(certPEM) + if err != nil { + return nil, nil, nil + } + + certClusterID, certNodeID, err := pki.IdentityFromCert(cert) + if err != nil || certClusterID != clusterID || certNodeID != a.NodeID { + logger.EllaLog.Info("discarding pending join cert with a stale identity", + zap.String("want_cluster", clusterID), zap.Int("want_node", a.NodeID), + zap.String("got_cluster", certClusterID), zap.Int("got_node", certNodeID)) + + return nil, nil, nil + } + + if _, err := tls.X509KeyPair(certPEM, keyPEM); err != nil { + return nil, nil, nil + } + + return certPEM, keyPEM, cert +} + func (a *Agent) discardJoinCert() { for _, f := range []string{joinCertFile, joinKeyFile} { if err := os.Remove(a.path(f)); err != nil && !os.IsNotExist(err) { diff --git a/internal/cluster/pkiagent/agent_test.go b/internal/cluster/pkiagent/agent_test.go index bd9f46e1f..1edbd4d8f 100644 --- a/internal/cluster/pkiagent/agent_test.go +++ b/internal/cluster/pkiagent/agent_test.go @@ -242,3 +242,57 @@ func mintTestJoinToken(t *testing.T, nodeID int, clusterID, leaderPin string) st return token } + +func TestAgent_JoinFlow_DiscardsIdentityFromAnotherCluster(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + leader := newAgent(t, 1, "cluster-b") + + pinFn := func(fp string) listener.PinResult { + return listener.PinResult{Found: fp == pki.Fingerprint(leader.Leaf().Leaf), NodeID: leader.NodeID} + } + + _, leaderAddr := startListener(ctx, t, leader, pinFn, func(ln *listener.Listener) { + ln.Register(listener.ALPNPKIBootstrap, alwaysFailRegisterHandler()) + }) + + joiner := pkiagent.NewAgent(2, "", t.TempDir()) + joinCert := filepath.Join(joiner.DataDir, "cluster-tls", "join.crt") + + tokenA := mintTestJoinToken(t, 2, "cluster-a", pki.Fingerprint(leader.Leaf().Leaf)) + if err := joiner.JoinFlow(ctx, leaderAddr, tokenA); err == nil { + t.Fatal("JoinFlow should have failed; leader returns 500") + } + + stale, err := os.ReadFile(joinCert) + if err != nil { + t.Fatalf("read join.crt: %v", err) + } + + joiner.ClusterID = "" + + tokenB := mintTestJoinToken(t, 2, "cluster-b", pki.Fingerprint(leader.Leaf().Leaf)) + if err := joiner.JoinFlow(ctx, leaderAddr, tokenB); err == nil { + t.Fatal("JoinFlow should have failed; leader returns 500") + } + + fresh, err := os.ReadFile(joinCert) + if err != nil { + t.Fatalf("read join.crt after cluster change: %v", err) + } + + if bytes.Equal(stale, fresh) { + t.Fatal("kept an identity minted for another cluster; every retry would be rejected") + } + + cert, err := pki.ParseCertPEM(fresh) + if err != nil { + t.Fatalf("parse regenerated join.crt: %v", err) + } + + clusterID, _, err := pki.IdentityFromCert(cert) + if err != nil || clusterID != "cluster-b" { + t.Fatalf("regenerated cert is for %q, want cluster-b (err=%v)", clusterID, err) + } +} diff --git a/internal/db/cluster_pki.go b/internal/db/cluster_pki.go index 84ffd8e0b..773c20a43 100644 --- a/internal/db/cluster_pki.go +++ b/internal/db/cluster_pki.go @@ -139,7 +139,6 @@ type redeemJoinTokenPayload struct { NodeID int `json:"node_id"` Fingerprint string `json:"fingerprint"` CertPEM string `json:"cert_pem"` - Now int64 `json:"now"` } type RedeemJoinTokenResult struct { @@ -148,6 +147,7 @@ type RedeemJoinTokenResult struct { func (db *Database) applyRedeemJoinToken(ctx context.Context, p *redeemJoinTokenPayload) (any, error) { runner := db.runner(ctx) + now := time.Now().Unix() token := ClusterJoinToken{ID: p.TokenID} if err := runner.Query(ctx, db.getJoinTokenStmt, token).Get(&token); err != nil { @@ -162,7 +162,7 @@ func (db *Database) applyRedeemJoinToken(ctx context.Context, p *redeemJoinToken return nil, ErrJoinTokenNodeMismatch } - if token.ExpiresAt <= p.Now { + if token.ExpiresAt <= now { return nil, ErrJoinTokenExpired } @@ -183,7 +183,7 @@ func (db *Database) applyRedeemJoinToken(ctx context.Context, p *redeemJoinToken return db.pinSnapshot(ctx, runner) } - token.ConsumedAt = p.Now + token.ConsumedAt = now token.ConsumedBy = p.NodeID var outcome sqlair.Outcome @@ -204,7 +204,7 @@ func (db *Database) applyRedeemJoinToken(ctx context.Context, p *redeemJoinToken NodeID: p.NodeID, Fingerprint: p.Fingerprint, CertPEM: p.CertPEM, - AddedAt: p.Now, + AddedAt: now, } if err := runner.Query(ctx, db.upsertNodeCertStmt, cert).Run(); err != nil { return nil, fmt.Errorf("upsert node cert: %w", err) @@ -335,14 +335,13 @@ func (db *Database) RedeemJoinToken(ctx context.Context, tokenID string, nodeID NodeID: nodeID, Fingerprint: fingerprint, CertPEM: certPEM, - Now: time.Now().Unix(), }) if err != nil { return nil, err } - if res == nil { - return nil, nil + if res == nil || len(res.Pins) == 0 { + return nil, fmt.Errorf("redeem returned an empty pin snapshot") } return res.Pins, nil diff --git a/internal/db/forward_sentinel_internal_test.go b/internal/db/forward_sentinel_internal_test.go new file mode 100644 index 000000000..165cf662b --- /dev/null +++ b/internal/db/forward_sentinel_internal_test.go @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Ella Networks Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package db + +import ( + "errors" + "testing" + + ellaraft "github.com/ellanetworks/core/internal/raft" +) + +func TestSentinelForForwardCode_RoundTripsDomainErrors(t *testing.T) { + for _, tc := range []struct { + code string + want error + }{ + {ellaraft.ForwardCodeTokenConsumed, ErrJoinTokenAlreadyConsumed}, + {ellaraft.ForwardCodeTokenExpired, ErrJoinTokenExpired}, + {ellaraft.ForwardCodeTokenNodeMism, ErrJoinTokenNodeMismatch}, + {ellaraft.ForwardCodeMigrationPend, ErrMigrationPending}, + {ellaraft.ForwardCodeNotFound, ErrNotFound}, + {ellaraft.ForwardCodeAlreadyExists, ErrAlreadyExists}, + } { + if got := sentinelForForwardCode(tc.code); !errors.Is(got, tc.want) { + t.Errorf("code %q rehydrated as %v, want %v", tc.code, got, tc.want) + } + } + + if got := sentinelForForwardCode(""); got != nil { + t.Errorf("uncoded error rehydrated as %v, want nil", got) + } +} diff --git a/internal/db/operations.go b/internal/db/operations.go index c6e171090..2d03332aa 100644 --- a/internal/db/operations.go +++ b/internal/db/operations.go @@ -464,6 +464,25 @@ func (db *Database) leaderCaptureAndPropose(operation string, minSchema int, app // forwardOperation POSTs to the leader's /cluster/internal/propose // endpoint. Transient errors (no leader, leadership changed) become // ErrProposeTimeout so the API maps them to 503. +func sentinelForForwardCode(code string) error { + switch code { + case ellaraft.ForwardCodeTokenConsumed: + return ErrJoinTokenAlreadyConsumed + case ellaraft.ForwardCodeTokenExpired: + return ErrJoinTokenExpired + case ellaraft.ForwardCodeTokenNodeMism: + return ErrJoinTokenNodeMismatch + case ellaraft.ForwardCodeMigrationPend: + return ErrMigrationPending + case ellaraft.ForwardCodeAlreadyExists: + return ErrAlreadyExists + case ellaraft.ForwardCodeNotFound: + return ErrNotFound + default: + return nil + } +} + func (db *Database) forwardOperation(opName string, payload json.RawMessage) (*ellaraft.ProposeResult, error) { if db.raftManager == nil { return nil, hraft.ErrNotLeader @@ -482,6 +501,10 @@ func (db *Database) forwardOperation(opName string, payload json.RawMessage) (*e return nil, fmt.Errorf("%w: %v", ErrProposeTimeout, err) } + if sentinel := sentinelForForwardCode(ellaraft.ForwardErrorCode(err)); sentinel != nil { + return nil, fmt.Errorf("%w: %v", sentinel, err) + } + return nil, err } diff --git a/internal/db/operations_registry_internal_test.go b/internal/db/operations_registry_internal_test.go index 92b6c519d..08bbaed7c 100644 --- a/internal/db/operations_registry_internal_test.go +++ b/internal/db/operations_registry_internal_test.go @@ -99,6 +99,10 @@ why an operation with no remaining Go caller must still stay registered. Lowering RequireSchema lets an operation apply against a schema that predates the columns it writes. +Adding one is safe but ordered: a node running the new binary cannot +forward a new operation to a leader that predates it, so the leader must +be upgraded first. + Add or amend the pinned entry deliberately, in the same change.` func TestChangesetOpRegistryIsPinned(t *testing.T) { diff --git a/internal/raft/forward.go b/internal/raft/forward.go index da7c41e53..3e2203ba2 100644 --- a/internal/raft/forward.go +++ b/internal/raft/forward.go @@ -101,7 +101,31 @@ type ProposeForwardErrorBody struct { Code string `json:"code,omitempty"` } -const ForwardCodeOutcomeUnknown = "outcome_unknown" +const ( + ForwardCodeOutcomeUnknown = "outcome_unknown" + ForwardCodeNotFound = "not_found" + ForwardCodeAlreadyExists = "already_exists" + ForwardCodeMigrationPend = "migration_pending" + ForwardCodeTokenConsumed = "join_token_consumed" + ForwardCodeTokenExpired = "join_token_expired" + ForwardCodeTokenNodeMism = "join_token_node_mismatch" +) + +type ForwardCodedError struct { + Code string + Message string +} + +func (e *ForwardCodedError) Error() string { return e.Message } + +func ForwardErrorCode(err error) string { + var coded *ForwardCodedError + if errors.As(err, &coded) { + return coded.Code + } + + return "" +} var ErrOutcomeUnknown = errors.New("forwarded write outcome unknown") @@ -229,6 +253,10 @@ func decodeForwardError(body []byte, status int) error { return fmt.Errorf("%w: %s", ErrOutcomeUnknown, env.Message) } + if env.Code != "" { + return &ForwardCodedError{Code: env.Code, Message: env.Message} + } + return errors.New(env.Message) } diff --git a/internal/raft/forward_coded_error_test.go b/internal/raft/forward_coded_error_test.go new file mode 100644 index 000000000..8f2ac864e --- /dev/null +++ b/internal/raft/forward_coded_error_test.go @@ -0,0 +1,50 @@ +// SPDX-FileCopyrightText: Ella Networks Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package raft + +import ( + "encoding/json" + "errors" + "net/http" + "testing" +) + +func TestDecodeForwardError_PreservesCode(t *testing.T) { + body, err := json.Marshal(ProposeForwardErrorBody{ + Message: "join token already consumed", + Code: ForwardCodeTokenConsumed, + }) + if err != nil { + t.Fatal(err) + } + + decoded := decodeForwardError(body, http.StatusConflict) + + if got := ForwardErrorCode(decoded); got != ForwardCodeTokenConsumed { + t.Fatalf("ForwardErrorCode = %q, want %q", got, ForwardCodeTokenConsumed) + } + + if decoded.Error() != "join token already consumed" { + t.Fatalf("message not preserved: %q", decoded.Error()) + } +} + +func TestDecodeForwardError_OutcomeUnknownStillWins(t *testing.T) { + body, _ := json.Marshal(ProposeForwardErrorBody{ + Message: "may have committed", + Code: ForwardCodeOutcomeUnknown, + }) + + if !errors.Is(decodeForwardError(body, http.StatusConflict), ErrOutcomeUnknown) { + t.Fatal("outcome_unknown must still map to ErrOutcomeUnknown") + } +} + +func TestForwardErrorCode_UncodedIsEmpty(t *testing.T) { + body, _ := json.Marshal(ProposeForwardErrorBody{Message: "apply failed"}) + + if got := ForwardErrorCode(decodeForwardError(body, http.StatusInternalServerError)); got != "" { + t.Fatalf("ForwardErrorCode = %q, want empty", got) + } +} From e0fc079dd64cb718d6b33b5fa9c99e7c2ab859d7 Mon Sep 17 00:00:00 2001 From: Guillaume Belanger Date: Wed, 26 Aug 2026 09:26:45 -0400 Subject: [PATCH 4/5] chore: fix gosec issue Signed-off-by: Guillaume Belanger --- internal/raft/forward.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/internal/raft/forward.go b/internal/raft/forward.go index 3e2203ba2..e5fbf2cbd 100644 --- a/internal/raft/forward.go +++ b/internal/raft/forward.go @@ -106,9 +106,9 @@ const ( ForwardCodeNotFound = "not_found" ForwardCodeAlreadyExists = "already_exists" ForwardCodeMigrationPend = "migration_pending" - ForwardCodeTokenConsumed = "join_token_consumed" - ForwardCodeTokenExpired = "join_token_expired" - ForwardCodeTokenNodeMism = "join_token_node_mismatch" + ForwardCodeTokenConsumed = "join_token_consumed" // #nosec G101 -- response code, not a credential + ForwardCodeTokenExpired = "join_token_expired" // #nosec G101 -- response code, not a credential + ForwardCodeTokenNodeMism = "join_token_node_mismatch" // #nosec G101 -- response code, not a credential ) type ForwardCodedError struct { From c916942d43e9fda307e4b36424f3c45acb17cf8f Mon Sep 17 00:00:00 2001 From: Guillaume Belanger Date: Wed, 26 Aug 2026 09:46:42 -0400 Subject: [PATCH 5/5] fix(ha): mint join tokens against the cert the leader presents Signed-off-by: Guillaume Belanger --- internal/api/server/api_pki.go | 48 +++++++- internal/api/server/api_pki_test.go | 10 +- internal/api/server/cluster_pki_e2e_test.go | 4 +- internal/api/server/server.go | 2 +- internal/cluster/pkiissuer/service.go | 64 +++++++---- internal/cluster/pkiissuer/service_test.go | 120 +++++++++++++++++--- pkg/runtime/pki.go | 14 ++- 7 files changed, 209 insertions(+), 53 deletions(-) diff --git a/internal/api/server/api_pki.go b/internal/api/server/api_pki.go index 1359cb9fd..f9e8e4710 100644 --- a/internal/api/server/api_pki.go +++ b/internal/api/server/api_pki.go @@ -7,13 +7,14 @@ package server import ( + "context" "encoding/json" + "errors" "fmt" "net/http" "time" "github.com/ellanetworks/core/internal/cluster/pkiissuer" - "github.com/ellanetworks/core/internal/db" "github.com/ellanetworks/core/internal/logger" "github.com/ellanetworks/core/internal/pki" ) @@ -22,6 +23,15 @@ const ( PKIMintJoinTokenAction = "pki_mint_join_token" // #nosec G101 -- audit action name ) +// A leader that has just been promoted registers its own cluster +// certificate as part of leader init; minting before that commit +// lands would embed a stale pin, so the handler waits it out rather +// than failing a request that is about to become serviceable. +const ( + mintReadyWait = 15 * time.Second + mintReadyPoll = 100 * time.Millisecond +) + // pkiAdminEndpoint resolves the pkiissuer.Service at request time and // dispatches to build. Returns 503 until the issuer service has been // installed by runtime and its join-HMAC key is committed. @@ -57,10 +67,9 @@ type MintJoinTokenResponse struct { } // PKIMintJoinToken handles POST /api/v1/cluster/pki/join-tokens. -// The handler runs on the leader (the public API forwards to it), -// so dbInstance.NodeID() identifies the leader whose pin gets -// embedded in the token's claims. -func PKIMintJoinToken(dbInstance *db.Database, svc *pkiissuer.Service) http.Handler { +// Minting is leader-only and is not forwarded: a request that lands +// on a follower fails. +func PKIMintJoinToken(svc *pkiissuer.Service) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var req MintJoinTokenRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -80,9 +89,17 @@ func PKIMintJoinToken(dbInstance *db.Database, svc *pkiissuer.Service) http.Hand ttl = 30 * time.Minute } - token, err := svc.MintJoinToken(r.Context(), req.NodeID, ttl, dbInstance.NodeID()) + token, err := mintWhenReady(r.Context(), svc, req.NodeID, ttl) if err != nil { + if errors.Is(err, pkiissuer.ErrNotReady) { + w.Header().Set("Retry-After", "1") + writeError(r.Context(), w, http.StatusServiceUnavailable, "mint token", err, logger.APILog) + + return + } + writeError(r.Context(), w, http.StatusInternalServerError, "mint token", err, logger.APILog) + return } @@ -102,3 +119,22 @@ func PKIMintJoinToken(dbInstance *db.Database, svc *pkiissuer.Service) http.Hand }, http.StatusCreated, logger.APILog) }) } + +// mintWhenReady retries MintJoinToken while the issuer reports +// ErrNotReady, up to mintReadyWait. +func mintWhenReady(ctx context.Context, svc *pkiissuer.Service, nodeID int, ttl time.Duration) (string, error) { + deadline := time.Now().Add(mintReadyWait) + + for { + token, err := svc.MintJoinToken(ctx, nodeID, ttl) + if !errors.Is(err, pkiissuer.ErrNotReady) || time.Now().After(deadline) { + return token, err + } + + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-time.After(mintReadyPoll): + } + } +} diff --git a/internal/api/server/api_pki_test.go b/internal/api/server/api_pki_test.go index 2ea7dba78..a6ea22ac6 100644 --- a/internal/api/server/api_pki_test.go +++ b/internal/api/server/api_pki_test.go @@ -71,7 +71,9 @@ func TestPKIAdminEndpoints_MintToken(t *testing.T) { t.Fatal(err) } - issuer := pkiissuer.New(env.DB) + var leaderFP string + + issuer := pkiissuer.New(env.DB, func() string { return leaderFP }) if err := issuer.Bootstrap(context.Background()); err != nil { t.Fatalf("Bootstrap: %v", err) } @@ -83,9 +85,11 @@ func TestPKIAdminEndpoints_MintToken(t *testing.T) { t.Fatal(err) } + leaderFP = pki.Fingerprint(leaderCert) + if err := env.DB.UpsertClusterNodeCert(context.Background(), &db.ClusterNodeCert{ NodeID: 1, - Fingerprint: pki.Fingerprint(leaderCert), + Fingerprint: leaderFP, CertPEM: string(pki.EncodeCertPEM(leaderCert)), AddedAt: time.Now().Unix(), }); err != nil { @@ -149,7 +153,7 @@ func TestPKIAdminEndpoints_InstalledButNotBootstrapped503(t *testing.T) { t.Fatal(err) } - server.SetPKIIssuer(pkiissuer.New(env.DB)) + server.SetPKIIssuer(pkiissuer.New(env.DB, nil)) t.Cleanup(func() { server.SetPKIIssuer(nil) }) admin, err := initializeAndRefresh(env.Server.URL, env.Server.Client()) diff --git a/internal/api/server/cluster_pki_e2e_test.go b/internal/api/server/cluster_pki_e2e_test.go index 2384361fd..4c7284b22 100644 --- a/internal/api/server/cluster_pki_e2e_test.go +++ b/internal/api/server/cluster_pki_e2e_test.go @@ -62,7 +62,7 @@ func TestClusterPKI_JoinFlowEndToEnd(t *testing.T) { t.Fatalf("preregister leader pin: %v", err) } - issuer := pkiissuer.New(leaderDB) + issuer := pkiissuer.New(leaderDB, func() string { return pki.Fingerprint(leaderAgent.Leaf().Leaf) }) if err := issuer.Bootstrap(ctx); err != nil { t.Fatalf("issuer bootstrap: %v", err) } @@ -95,7 +95,7 @@ func TestClusterPKI_JoinFlowEndToEnd(t *testing.T) { t.Cleanup(leaderLn.Stop) // Mint a token for nodeID 2 (the joiner). Leader is nodeID 1. - token, err := issuer.MintJoinToken(ctx, 2, 5*time.Minute, 1) + token, err := issuer.MintJoinToken(ctx, 2, 5*time.Minute) if err != nil { t.Fatalf("mint join token: %v", err) } diff --git a/internal/api/server/server.go b/internal/api/server/server.go index c12bbf84e..b84177965 100644 --- a/internal/api/server/server.go +++ b/internal/api/server/server.go @@ -261,7 +261,7 @@ func NewHandler(cfg HandlerConfig) http.Handler { // request time (set by runtime after first-leader bootstrap), so // these routes can be registered before the issuer is ready. mux.HandleFunc("POST /api/v1/cluster/pki/join-tokens", Authenticate(jwtSecret, dbInstance, Authorize(PermManageCluster, pkiAdminEndpoint(func(svc *pkiissuer.Service) http.Handler { - return PKIMintJoinToken(dbInstance, svc) + return PKIMintJoinToken(svc) }))).ServeHTTP) // Fallback to UI diff --git a/internal/cluster/pkiissuer/service.go b/internal/cluster/pkiissuer/service.go index 3ab28a966..d52b90c9c 100644 --- a/internal/cluster/pkiissuer/service.go +++ b/internal/cluster/pkiissuer/service.go @@ -23,6 +23,15 @@ import ( "github.com/ellanetworks/core/internal/pki" ) +// ErrNotReady reports that this node cannot mint join tokens yet +// because the cluster certificate it presents is not committed in +// cluster_node_certs. Callers should retry. +var ErrNotReady = errors.New("cluster pki not ready") + +// LocalLeafFunc reports the SHA-256 pin of the cluster certificate +// this node is currently presenting, or "" when it has none yet. +type LocalLeafFunc func() string + // Store is the narrow DB surface the issuer needs. type Store interface { GetOperator(ctx context.Context) (*db.Operator, error) @@ -42,11 +51,14 @@ type Store interface { // Service runs on every voter. Bootstrap and MintJoinToken require // IsLeader; RedeemJoinToken and RegisterCert forward to the leader. type Service struct { - store Store + store Store + localLeaf LocalLeafFunc } -func New(store Store) *Service { - return &Service{store: store} +// New builds the issuer. localLeaf may be nil, in which case this +// node can register certs and redeem tokens but never mint one. +func New(store Store, localLeaf LocalLeafFunc) *Service { + return &Service{store: store, localLeaf: localLeaf} } // Bootstrap seeds the HMAC-key singleton on the leader-init path. @@ -87,10 +99,10 @@ func (s *Service) Ready(ctx context.Context) bool { } // MintJoinToken emits a single-use HMAC token bound to nodeID with -// the given TTL, embedding the cluster_node_certs pin owned by -// leaderNodeID so the joining node pins the bootstrap TLS -// handshake against the leader's certificate. -func (s *Service) MintJoinToken(ctx context.Context, nodeID int, ttl time.Duration, leaderNodeID int) (string, error) { +// the given TTL, embedding the pin set the joining node uses to +// pin its bootstrap TLS handshake. Returns ErrNotReady until this +// node's own certificate is committed cluster-wide. +func (s *Service) MintJoinToken(ctx context.Context, nodeID int, ttl time.Duration) (string, error) { if ttl < pki.DefaultJoinTokenMinTTL || ttl > pki.DefaultJoinTokenMaxTTL { return "", fmt.Errorf("join-token ttl %s outside [%s, %s]", ttl, pki.DefaultJoinTokenMinTTL, pki.DefaultJoinTokenMaxTTL) } @@ -113,7 +125,7 @@ func (s *Service) MintJoinToken(ctx context.Context, nodeID int, ttl time.Durati return "", fmt.Errorf("cluster id not yet populated") } - leaderPin, allPins, err := s.pinsForToken(ctx, leaderNodeID) + leaderPin, allPins, err := s.pinsForToken(ctx) if err != nil { return "", err } @@ -157,34 +169,42 @@ func (s *Service) MintJoinToken(ctx context.Context, nodeID int, ttl time.Durati return tokenStr, nil } -func (s *Service) pinsForToken(ctx context.Context, leaderNodeID int) (string, []string, error) { +// pinsForToken resolves the pin set a join token carries. The leader +// entry is the certificate this node is presenting right now, never a +// cluster_node_certs lookup by nodeID: after a restore the table still +// holds the pre-restore cluster's row for this nodeID, so a lookup +// mints a token pinning a certificate no node will ever present. +func (s *Service) pinsForToken(ctx context.Context) (string, []string, error) { + var leaderPin string + if s.localLeaf != nil { + leaderPin = s.localLeaf() + } + + if leaderPin == "" { + return "", nil, fmt.Errorf("%w: node has no cluster certificate", ErrNotReady) + } + rows, err := s.store.ListClusterNodeCerts(ctx) if err != nil { return "", nil, fmt.Errorf("list pins: %w", err) } all := make([]string, 0, len(rows)) + committed := false + for _, r := range rows { all = append(all, r.Fingerprint) - } - if leaderNodeID == 0 { - // Standalone or single-node test path: with exactly one - // registered pin, that pin belongs to the leader. - if len(rows) == 1 { - return rows[0].Fingerprint, all, nil + if r.Fingerprint == leaderPin { + committed = true } - - return "", nil, fmt.Errorf("leaderNodeID is zero and registry has %d pins", len(rows)) } - for _, r := range rows { - if r.NodeID == leaderNodeID { - return r.Fingerprint, all, nil - } + if !committed { + return "", nil, fmt.Errorf("%w: this node's certificate is not registered yet", ErrNotReady) } - return "", nil, fmt.Errorf("leader node %d has no registered pin", leaderNodeID) + return leaderPin, all, nil } func (s *Service) RedeemJoinToken(ctx context.Context, tokenStr string, nodeID int, certPEM []byte) (string, []db.ClusterNodeCert, error) { diff --git a/internal/cluster/pkiissuer/service_test.go b/internal/cluster/pkiissuer/service_test.go index 7f0ab7fb6..182cad3d0 100644 --- a/internal/cluster/pkiissuer/service_test.go +++ b/internal/cluster/pkiissuer/service_test.go @@ -197,10 +197,25 @@ func preregisterLeader(t *testing.T, store *fakeStore, nodeID int) string { return fp } +// leaderLeaf mirrors the runtime accessor: the pin of the cert this +// node presents. The fake leader is always node 1. +func leaderLeaf(store *fakeStore) pkiissuer.LocalLeafFunc { + return func() string { + store.mu.Lock() + defer store.mu.Unlock() + + if p := store.pins[1]; p != nil { + return p.Fingerprint + } + + return "" + } +} + func TestService_Bootstrap_SeedsHMACKey(t *testing.T) { store := newFakeStore("c") - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) if err := svc.Bootstrap(context.Background()); err != nil { t.Fatalf("bootstrap: %v", err) @@ -226,7 +241,7 @@ func TestService_Bootstrap_SeedsHMACKey(t *testing.T) { func TestService_RegisterCert_HappyPath(t *testing.T) { store := newFakeStore("c") - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) cert, _, err := pki.GenerateNodeCert(7, "c", time.Hour) if err != nil { @@ -254,7 +269,7 @@ func TestService_RegisterCert_HappyPath(t *testing.T) { func TestService_RegisterCert_RejectsCrossCluster(t *testing.T) { store := newFakeStore("c-a") - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) cert, _, err := pki.GenerateNodeCert(7, "c-b", time.Hour) if err != nil { @@ -269,7 +284,7 @@ func TestService_RegisterCert_RejectsCrossCluster(t *testing.T) { func TestService_RegisterCert_RejectsNodeIDMismatch(t *testing.T) { store := newFakeStore("c") - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) cert, _, err := pki.GenerateNodeCert(7, "c", time.Hour) if err != nil { @@ -286,13 +301,13 @@ func TestService_MintAndVerifyJoinToken_RoundTrip(t *testing.T) { leaderFP := preregisterLeader(t, store, 1) - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) if err := svc.Bootstrap(context.Background()); err != nil { t.Fatal(err) } - token, err := svc.MintJoinToken(context.Background(), 5, time.Minute*30, 1) + token, err := svc.MintJoinToken(context.Background(), 5, time.Minute*30) if err != nil { t.Fatalf("mint: %v", err) } @@ -331,14 +346,14 @@ func TestService_MintJoinToken_RejectsInvalidTTL(t *testing.T) { store := newFakeStore("c") preregisterLeader(t, store, 1) - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) _ = svc.Bootstrap(context.Background()) - if _, err := svc.MintJoinToken(context.Background(), 5, time.Second, 1); err == nil { + if _, err := svc.MintJoinToken(context.Background(), 5, time.Second); err == nil { t.Fatal("expected ttl < min to be rejected") } - if _, err := svc.MintJoinToken(context.Background(), 5, 48*time.Hour, 1); err == nil { + if _, err := svc.MintJoinToken(context.Background(), 5, 48*time.Hour); err == nil { t.Fatal("expected ttl > max to be rejected") } } @@ -347,13 +362,13 @@ func TestService_NotLeader_RejectsMutations(t *testing.T) { store := newFakeStore("c") store.leader = false - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) if err := svc.Bootstrap(context.Background()); err == nil { t.Fatal("Bootstrap should fail on non-leader") } - if _, err := svc.MintJoinToken(context.Background(), 5, time.Hour, 1); err == nil { + if _, err := svc.MintJoinToken(context.Background(), 5, time.Hour); err == nil { t.Fatal("MintJoinToken should fail on non-leader") } } @@ -362,7 +377,7 @@ func TestService_RegisterCert_WorksOnNonLeader(t *testing.T) { store := newFakeStore("c") store.leader = false - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) if _, _, err := svc.RegisterCert(context.Background(), 5, nodeCertPEM(t, 5)); err != nil { t.Fatalf("RegisterCert on a follower: %v", err) @@ -373,10 +388,10 @@ func TestService_Redeem_ReplayWithDifferentCertRejected(t *testing.T) { store := newFakeStore("c") preregisterLeader(t, store, 1) - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) _ = svc.Bootstrap(context.Background()) - tok, _ := svc.MintJoinToken(context.Background(), 5, time.Minute*10, 1) + tok, _ := svc.MintJoinToken(context.Background(), 5, time.Minute*10) if _, _, err := svc.RedeemJoinToken(context.Background(), tok, 5, nodeCertPEM(t, 5)); err != nil { t.Fatal(err) @@ -392,10 +407,10 @@ func TestService_Redeem_SameNodeSameCertIsIdempotent(t *testing.T) { store := newFakeStore("c") preregisterLeader(t, store, 1) - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) _ = svc.Bootstrap(context.Background()) - tok, _ := svc.MintJoinToken(context.Background(), 5, time.Minute*10, 1) + tok, _ := svc.MintJoinToken(context.Background(), 5, time.Minute*10) joinerPEM := nodeCertPEM(t, 5) fp1, pins1, err := svc.RedeemJoinToken(context.Background(), tok, 5, joinerPEM) @@ -418,10 +433,10 @@ func TestService_MintJoinToken_EmbedsEveryVoterPin(t *testing.T) { leaderFP := preregisterLeader(t, store, 1) peerFP := preregisterLeader(t, store, 2) - svc := pkiissuer.New(store) + svc := pkiissuer.New(store, leaderLeaf(store)) _ = svc.Bootstrap(context.Background()) - tok, err := svc.MintJoinToken(context.Background(), 5, time.Minute*10, 1) + tok, err := svc.MintJoinToken(context.Background(), 5, time.Minute*10) if err != nil { t.Fatal(err) } @@ -461,3 +476,72 @@ func nodeCertPEM(t *testing.T, nodeID int) []byte { return pki.EncodeCertPEM(cert) } + +// After a restore the table still holds the pre-restore cluster's row +// for this nodeID. Minting from that row hands the joiner a pin no +// node will ever present, and the joiner retries it forever. +func TestService_MintJoinToken_RefusesStalePinFromRestoredTable(t *testing.T) { + store := newFakeStore(testClusterID) + + stalePins := map[int]*db.ClusterNodeCert{} + + for _, nodeID := range []int{1, 2, 3} { + preregisterLeader(t, store, nodeID) + stalePins[nodeID] = store.pins[nodeID] + } + + // This node self-signs a fresh cert on restore; the row for its + // own nodeID has not been replaced yet. + freshCert, _, err := pki.GenerateNodeCert(1, testClusterID, time.Hour) + if err != nil { + t.Fatal(err) + } + + freshFP := pki.Fingerprint(freshCert) + + svc := pkiissuer.New(store, func() string { return freshFP }) + if err := svc.Bootstrap(context.Background()); err != nil { + t.Fatal(err) + } + + _, err = svc.MintJoinToken(context.Background(), 5, 10*time.Minute) + if !errors.Is(err, pkiissuer.ErrNotReady) { + t.Fatalf("mint before the fresh cert commits: got %v, want ErrNotReady", err) + } + + if _, _, err := svc.RegisterCert(context.Background(), 1, pki.EncodeCertPEM(freshCert)); err != nil { + t.Fatalf("register fresh leader cert: %v", err) + } + + tok, err := svc.MintJoinToken(context.Background(), 5, 10*time.Minute) + if err != nil { + t.Fatalf("mint after the fresh cert commits: %v", err) + } + + claims, err := pki.ExtractClaimsUnverified(tok) + if err != nil { + t.Fatal(err) + } + + if claims.LeaderCertPin != freshFP { + t.Fatalf("token pins %q, want the cert this node presents (%q)", claims.LeaderCertPin, freshFP) + } + + if claims.LeaderCertPin == stalePins[1].Fingerprint { + t.Fatal("token pins the pre-restore cert") + } +} + +func TestService_MintJoinToken_RefusesWithoutLocalLeaf(t *testing.T) { + store := newFakeStore(testClusterID) + preregisterLeader(t, store, 1) + + svc := pkiissuer.New(store, nil) + if err := svc.Bootstrap(context.Background()); err != nil { + t.Fatal(err) + } + + if _, err := svc.MintJoinToken(context.Background(), 5, 10*time.Minute); !errors.Is(err, pkiissuer.ErrNotReady) { + t.Fatalf("mint with no local leaf: got %v, want ErrNotReady", err) + } +} diff --git a/pkg/runtime/pki.go b/pkg/runtime/pki.go index eecd31938..4629aa7d0 100644 --- a/pkg/runtime/pki.go +++ b/pkg/runtime/pki.go @@ -56,11 +56,23 @@ func (p *pkiState) ensureIssuer(dbInstance *db.Database) { defer p.issuerMu.Unlock() if p.issuer == nil { - p.issuer = pkiissuer.New(dbInstance) + p.issuer = pkiissuer.New(dbInstance, p.leafFingerprint) server.SetPKIIssuer(p.issuer) } } +// leafFingerprint is the pkiissuer.LocalLeafFunc accessor: the pin of +// the cert this node presents on the cluster listener, "" until one +// exists. +func (p *pkiState) leafFingerprint() string { + leaf := p.agent.Leaf() + if leaf == nil { + return "" + } + + return pki.Fingerprint(leaf.Leaf) +} + func (p *pkiState) Issuer() *pkiissuer.Service { p.issuerMu.Lock() defer p.issuerMu.Unlock()