Merge remote-tracking branch 'origin/main' into fix_debug_upload_url_from_mgmt

# Conflicts:
#	shared/management/proto/management.pb.go
This commit is contained in:
riccardom
2026-10-09 17:12:38 +02:00
140 changed files with 12817 additions and 1661 deletions
+4
View File
@@ -143,6 +143,7 @@ var (
MgmtPort: mgmtPort,
MgmtMetricsPort: mgmtMetricsPort,
DisableLegacyManagementPort: disableLegacyManagementPort,
LetsEncryptListenAddress: mgmtLetsencryptListen,
DisableMetrics: disableMetrics,
DisableGeoliteUpdate: disableGeoliteUpdate,
UserDeleteFromIDPEnabled: userDeleteFromIDPEnabled,
@@ -367,6 +368,9 @@ func LogConfigInfo(cfg *nbconfig.Config) {
if cfg.Relay != nil {
log.Infof("Relay addresses: %v", cfg.Relay.Addresses)
}
if cfg.Signal != nil {
log.Infof("Signal addresses: %v", cfg.Signal.URI)
}
}
// EnsureEncryptionKey generates and saves a DataStoreEncryptionKey if not set
+2
View File
@@ -29,6 +29,7 @@ var (
mgmtMetricsPort int
disableLegacyManagementPort bool
mgmtLetsencryptDomain string
mgmtLetsencryptListen string
mgmtSingleAccModeDomain string
certFile string
certKey string
@@ -70,6 +71,7 @@ func init() {
mgmtCmd.Flags().StringVar(&mgmtDataDir, "datadir", defaultMgmtDataDir, "server data directory location")
mgmtCmd.Flags().StringVar(&nbconfig.MgmtConfigPath, "config", defaultMgmtConfig, "Netbird config file location. Config params specified via command line (e.g. datadir) have a precedence over configuration from this file")
mgmtCmd.Flags().StringVar(&mgmtLetsencryptDomain, "letsencrypt-domain", "", "a domain to issue Let's Encrypt certificate for. Enables TLS using Let's Encrypt. Will fetch and renew certificate, and run the server with TLS")
mgmtCmd.Flags().StringVar(&mgmtLetsencryptListen, "letsencrypt-listen-address", ":443", "address of the separate Let's Encrypt challenge listener, used when --port is not 443. Set it empty when public port 443 is forwarded to --port, which answers the challenges itself")
mgmtCmd.Flags().StringVar(&mgmtSingleAccModeDomain, "single-account-mode-domain", defaultSingleAccModeDomain, "Enables single account mode. This means that all the users will be under the same account grouped by the specified domain. If the installation has more than one account, the property is ineffective. Enabled by default with the default domain "+defaultSingleAccModeDomain)
mgmtCmd.Flags().BoolVar(&disableSingleAccMode, "disable-single-account-mode", false, "If set to true, disables single account mode. The --single-account-mode-domain property will be ignored and every new user will have a separate NetBird account.")
mgmtCmd.Flags().StringVar(&certFile, "cert-file", "", "Location of your SSL certificate. Can be used when you have an existing certificate and don't want a new certificate be generated automatically. If letsencrypt-domain is specified this property has no effect")
@@ -427,6 +427,9 @@ func accountPostureChecks(id string, pc *nmdata.PostureChecks, publicID string)
}
out.Checks.ProcessCheck = &posture.ProcessCheck{Processes: procs}
}
if def.CertificateCheck != nil {
out.Checks.CertificateCheck = &posture.CertificateCheck{CACertificates: def.CertificateCheck.CACertificates}
}
return out
}
@@ -2,6 +2,7 @@ package manager
import (
"context"
"errors"
"fmt"
"math/rand/v2"
"net"
@@ -1079,15 +1080,16 @@ func (m *Manager) ReloadAllServicesForAccount(ctx context.Context, accountID str
oidcCfg := m.proxyController.GetOIDCValidationConfig()
var errs []error
for _, s := range services {
err = m.replaceHostByLookup(ctx, accountID, s)
if err != nil {
return fmt.Errorf("failed to replace host by lookup for service %s: %w", s.ID, err)
if err := m.replaceHostByLookup(ctx, accountID, s); err != nil {
errs = append(errs, fmt.Errorf("replace host by lookup for service %s: %w", s.ID, err))
continue
}
m.proxyController.SendServiceUpdateToCluster(ctx, accountID, s.ToProtoMapping(service.Update, "", oidcCfg), s.ProxyCluster)
}
return nil
return errors.Join(errs...)
}
func (m *Manager) GetGlobalServices(ctx context.Context) ([]*service.Service, error) {
@@ -27,6 +27,7 @@ import (
"github.com/netbirdio/netbird/management/server/permissions/operations"
"github.com/netbirdio/netbird/management/server/store"
"github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/shared/management/proto"
"github.com/netbirdio/netbird/shared/management/status"
)
@@ -1481,3 +1482,42 @@ func TestReplaceHostByLookup_SkipsClusterTarget(t *testing.T) {
require.NoError(t, mgr.replaceHostByLookup(ctx, accountID, svc), "cluster target must not trigger peer/resource lookup")
assert.Equal(t, "127.0.0.1", svc.Targets[0].Host, "operator-supplied host must be preserved for cluster target")
}
func TestReloadAllServicesForAccount_ContinuesAfterFailedService(t *testing.T) {
ctx := context.Background()
ctrl := gomock.NewController(t)
mockStore := store.NewMockStore(ctrl)
proxyController := proxy.NewMockController(ctrl)
accountID := "test-account"
broken := &rpservice.Service{
ID: "svc-broken",
AccountID: accountID,
ProxyCluster: "eu.proxy.netbird.io",
Targets: []*rpservice.Target{{TargetId: "x", TargetType: "unknown"}},
}
healthy := &rpservice.Service{
ID: "svc-healthy",
AccountID: accountID,
ProxyCluster: "eu.proxy.netbird.io",
Targets: []*rpservice.Target{{TargetId: "peer-1", TargetType: rpservice.TargetTypePeer, Host: "100.64.0.1", Protocol: "http", Port: 8080, Enabled: true}},
}
mockStore.EXPECT().GetAccountServices(ctx, store.LockingStrengthNone, accountID).Return([]*rpservice.Service{broken, healthy}, nil)
mockStore.EXPECT().GetPeerByID(ctx, store.LockingStrengthNone, accountID, "peer-1").
Return(&nbpeer.Peer{ID: "peer-1", IP: netip.MustParseAddr("100.70.0.9")}, nil)
proxyController.EXPECT().GetOIDCValidationConfig().Return(proxy.OIDCValidationConfig{})
proxyController.EXPECT().SendServiceUpdateToCluster(ctx, accountID, gomock.Any(), "eu.proxy.netbird.io").
Do(func(_ context.Context, _ string, mapping *proto.ProxyMapping, _ string) {
assert.Equal(t, "svc-healthy", mapping.GetId())
require.Len(t, mapping.GetPath(), 1)
assert.Contains(t, mapping.GetPath()[0].GetTarget(), "100.70.0.9", "mapping must carry the peer's current IP")
})
mgr := &Manager{store: mockStore, proxyController: proxyController}
err := mgr.ReloadAllServicesForAccount(ctx, accountID)
require.Error(t, err)
assert.Contains(t, err.Error(), "svc-broken")
assert.NotContains(t, err.Error(), "svc-healthy")
}
@@ -12,7 +12,7 @@ const (
GetPeersQuery = `
select id, key, ssh_key, dns_label, extra_dns_labels, user_id, ssh_enabled, login_expiration_enabled, last_login, ip, ipv6,
peer_status_requires_approval, peer_status_connected, proxy_meta_embedded, proxy_meta_cluster,
meta_wt_version, meta_go_os, meta_os_version, meta_kernel_version, meta_network_addresses, meta_files, meta_capabilities, meta_flags, meta_sync_message_version,
meta_wt_version, meta_go_os, meta_os_version, meta_kernel_version, meta_network_addresses, meta_files, meta_certificates, meta_capabilities, meta_flags, meta_sync_message_version,
location_country_code, location_city_name, location_connection_ip
from peers
where account_id = $1
@@ -180,6 +180,7 @@ type Peer struct {
MetaKernelVersion sql.NullString `nmap:"skip"`
MetaNetworkAddresses []byte `nmap:"skip,json"`
MetaFiles []byte `nmap:"skip,json"`
MetaCertificates []byte `nmap:"skip,json"`
MetaCapabilities []byte `nmap:"skip,json"`
MetaFlags []byte `nmap:"skip,json"`
MetaSyncMessageVersion sql.NullInt64 `nmap:"skip"`
@@ -340,6 +341,12 @@ func ConvertToNmdataPeers(peers []Peer) ([]nmdata.Peer, map[string][]*nmdata.Pee
return toret, nil, err
}
}
if p.MetaCertificates != nil {
err := json.Unmarshal(p.MetaCertificates, &dp.Meta.Certificates)
if err != nil {
return toret, nil, err
}
}
if p.MetaCapabilities != nil {
err := json.Unmarshal(p.MetaCapabilities, &dp.Meta.Capabilities)
if err != nil {
@@ -11,7 +11,7 @@ const (
GetPeersQuery = `
select id, key, ssh_key, dns_label, extra_dns_labels, user_id, ssh_enabled, login_expiration_enabled, last_login, ip, ipv6,
peer_status_requires_approval, peer_status_connected, proxy_meta_embedded, proxy_meta_cluster,
meta_wt_version, meta_go_os, meta_os_version, meta_kernel_version, meta_network_addresses, meta_files, meta_capabilities, meta_flags, meta_sync_message_version,
meta_wt_version, meta_go_os, meta_os_version, meta_kernel_version, meta_network_addresses, meta_files, meta_certificates, meta_capabilities, meta_flags, meta_sync_message_version,
location_country_code, location_city_name, location_connection_ip
from peers
where account_id = ?
+33 -7
View File
@@ -68,6 +68,7 @@ type BaseServer struct {
mgmtMetricsPort int
mgmtPort int
disableLegacyManagementPort bool
letsEncryptListenAddress string
autoResolveDomains bool
proxyAuthClose func()
@@ -82,6 +83,8 @@ type BaseServer struct {
tlsConfig *tls.Config
certManager *autocert.Manager
update *version.Update
// certListener serves Let's Encrypt challenges when mgmtPort is not 443.
certListener net.Listener
errCh chan error
wg sync.WaitGroup
@@ -103,6 +106,9 @@ type Config struct {
UserDeleteFromIDPEnabled bool
AutoResolveDomains bool
TLSConfig *tls.Config
// LetsEncryptListenAddress is the separate Let's Encrypt challenge listener
// used when MgmtPort is not 443. Empty disables it.
LetsEncryptListenAddress string
}
// NewServer initializes and configures a new Server instance
@@ -117,6 +123,7 @@ func NewServer(cfg *Config) *BaseServer {
userDeleteFromIDPEnabled: cfg.UserDeleteFromIDPEnabled,
mgmtPort: cfg.MgmtPort,
disableLegacyManagementPort: cfg.DisableLegacyManagementPort,
letsEncryptListenAddress: cfg.LetsEncryptListenAddress,
mgmtMetricsPort: cfg.MgmtMetricsPort,
autoResolveDomains: cfg.AutoResolveDomains,
tlsConfig: cfg.TLSConfig,
@@ -210,19 +217,18 @@ func (s *BaseServer) start(ctx context.Context) error {
rootHandler := s.handlerFunc(srvCtx, s.GRPCServer(), s.APIHandler(), s.IDPHandler(), s.Metrics().GetMeter())
switch {
case s.certManager != nil:
// a call to certManager.Listener() always creates a new listener so we do it once
cml := s.certManager.Listener()
if s.mgmtPort == 443 {
// CertManager, HTTP and gRPC API all on the same port
rootHandler = s.certManager.HTTPHandler(rootHandler)
s.listener = cml
s.listener = s.certManager.Listener()
} else {
s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), s.certManager.TLSConfig())
if err != nil {
return fmt.Errorf("failed creating TLS listener on port %d: %v", s.mgmtPort, err)
}
log.WithContext(ctx).Infof("running HTTP server (LetsEncrypt challenge handler): %s", cml.Addr().String())
s.serveHTTP(ctx, cml, s.certManager.HTTPHandler(nil))
if err := s.serveLetsEncryptChallenges(ctx); err != nil {
return err
}
}
case s.tlsConfig != nil:
s.listener, err = tls.Listen("tcp", fmt.Sprintf(":%d", s.mgmtPort), s.tlsConfig)
@@ -309,8 +315,8 @@ func (s *BaseServer) Stop() error {
if s.listener != nil {
_ = s.listener.Close()
}
if s.certManager != nil {
_ = s.certManager.Listener().Close()
if s.certListener != nil {
_ = s.certListener.Close()
}
s.GRPCServer().Stop()
if s.proxyAuthClose != nil {
@@ -416,6 +422,26 @@ func (s *BaseServer) serveGRPC(ctx context.Context, grpcServer *grpc.Server, por
return listener, nil
}
// serveLetsEncryptChallenges starts the separate Let's Encrypt challenge listener
// unless it is disabled. The main TLS listener uses the cert manager's TLS
// config, so it still answers TLS-ALPN-01 challenges when public port 443 is
// forwarded to it.
func (s *BaseServer) serveLetsEncryptChallenges(ctx context.Context) error {
if s.letsEncryptListenAddress == "" {
log.WithContext(ctx).Infof("LetsEncrypt challenge server disabled, challenges are answered on port %d", s.mgmtPort)
return nil
}
cml, err := tls.Listen("tcp", s.letsEncryptListenAddress, s.certManager.TLSConfig())
if err != nil {
return fmt.Errorf("create LetsEncrypt challenge listener on %s: %w", s.letsEncryptListenAddress, err)
}
s.certListener = cml
log.WithContext(ctx).Infof("running HTTP server (LetsEncrypt challenge handler): %s", cml.Addr().String())
s.serveHTTP(ctx, cml, s.certManager.HTTPHandler(nil))
return nil
}
func (s *BaseServer) serveHTTP(ctx context.Context, httpListener net.Listener, handler http.Handler) {
s.wg.Add(1)
go func() {
@@ -0,0 +1,55 @@
package server
import (
"context"
"net"
"testing"
"time"
"github.com/stretchr/testify/require"
"golang.org/x/crypto/acme/autocert"
nbconfig "github.com/netbirdio/netbird/management/internals/server/config"
)
func newLetsEncryptTestServer(address string) *BaseServer {
srv := NewServer(&Config{NbConfig: &nbconfig.Config{}, MgmtPort: 8443, LetsEncryptListenAddress: address})
srv.certManager = &autocert.Manager{}
return srv
}
func TestServeLetsEncryptChallenges_Disabled(t *testing.T) {
srv := newLetsEncryptTestServer("")
require.NoError(t, srv.serveLetsEncryptChallenges(context.Background()))
require.Nil(t, srv.certListener, "no challenge listener should be created when the address is empty")
}
func TestServeLetsEncryptChallenges_CustomAddress(t *testing.T) {
srv := newLetsEncryptTestServer("127.0.0.1:0")
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(func() {
cancel()
if srv.certListener != nil {
_ = srv.certListener.Close()
}
srv.wg.Wait()
})
require.NoError(t, srv.serveLetsEncryptChallenges(ctx))
require.NotNil(t, srv.certListener, "challenge listener should be created on the configured address")
conn, err := net.DialTimeout("tcp", srv.certListener.Addr().String(), time.Second)
require.NoError(t, err)
require.NoError(t, conn.Close())
}
func TestServeLetsEncryptChallenges_BindFailure(t *testing.T) {
occupied, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
t.Cleanup(func() { _ = occupied.Close() })
srv := newLetsEncryptTestServer(occupied.Addr().String())
require.Error(t, srv.serveLetsEncryptChallenges(context.Background()))
require.Nil(t, srv.certListener, "no challenge listener should be stored when the bind fails")
}
@@ -0,0 +1,88 @@
package grpc
import (
"context"
"crypto/sha256"
"time"
log "github.com/sirupsen/logrus"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/shared/management/certposture"
"github.com/netbirdio/netbird/shared/management/proto"
)
const certChallengeKeyDomain = "netbird-cert-challenge-key"
// newCertChallenger derives the nonce secret from the data store encryption key:
// generated once, written back to the configuration, and read by every instance, so a
// nonce stays verifiable across a restart and between instances.
//
// Where none is configured it falls back to the server's WireGuard key, which is
// regenerated per process. Unpredictable is the property that has to hold either way: a
// peer that could guess the secret would mint future windows' nonces, sign them while
// its key is present, and keep passing after it is gone.
func newCertChallenger(encryptionKey string, serverKey wgtypes.Key) *certposture.Challenger {
secret := []byte(encryptionKey)
if len(secret) == 0 {
log.Warnf("no data store encryption key, deriving certificate challenges from the ephemeral server key: peers will be rejected once per restart and across instances")
secret = serverKey[:]
}
h := sha256.New()
h.Write([]byte(certChallengeKeyDomain))
h.Write(secret)
return certposture.NewChallenger(h.Sum(nil))
}
// stampCertificateChallenges fills the per-peer nonce into every certificate challenge
// right before the response is encrypted for that peer, reporting whether it issued one.
//
// On a sync stream the answer registers the peer for renewal. A nonce is stateless, but
// the renewal that keeps it fresh is local: only the instance holding a peer's stream
// can push to it, so an instance renews exactly the peers it streams nonces to.
func stampCertificateChallenges(checks []*proto.Checks, challenger *certposture.Challenger, peerKey wgtypes.Key) bool {
var nonce []byte
for _, check := range checks {
challenge := check.GetCertificateChallenge()
if challenge == nil {
continue
}
if nonce == nil {
nonce = challenger.Nonce(peerKey[:], time.Now())
}
challenge.Nonce = nonce
}
return nonce != nil
}
// verifiedCertificates turns the peer's proofs into PEM chains for its meta. Possession
// (nonce + signature) is verified here; trust against a check's CAs is evaluated by the
// posture check itself. Each proof stands for its own key, so an invalid one is dropped
// without the others: a user store answering with an unusable key must not cost the
// device the machine certificate it proved.
func (s *Server) verifiedCertificates(ctx context.Context, peerKey wgtypes.Key, proofs []*proto.CertificateProof) []string {
if len(proofs) == 0 {
return nil
}
now := time.Now()
chains := make([]string, 0, len(proofs))
for _, p := range proofs {
chain, err := s.challenger.Verify(certposture.Proof{
Nonce: p.GetNonce(),
Chain: p.GetChain(),
SigAlg: p.GetSigAlg(),
Signature: p.GetSignature(),
}, peerKey[:], now)
if err != nil {
log.WithContext(ctx).Warnf("dropping a certificate proof of peer %s: %v", peerKey, err)
continue
}
chains = append(chains, certposture.EncodeChainPEM(chain))
}
if len(chains) == 0 {
return nil
}
return chains
}
@@ -0,0 +1,91 @@
package grpc
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/shared/management/certposture"
"github.com/netbirdio/netbird/shared/management/certposture/certtest"
"github.com/netbirdio/netbird/shared/management/networkmap/nmdata"
"github.com/netbirdio/netbird/shared/management/proto"
)
func TestCertificateChallenge_StampAndVerifyRoundTrip(t *testing.T) {
const encryptionKey = "data-store-encryption-key"
challenger := newCertChallenger(encryptionKey, generateKey(t))
peerKey := generateKey(t).PublicKey()
ca := certtest.NewCA(t, "corp-root")
ctx := context.Background()
checks := toProtocolChecks(ctx, []*nmdata.PostureChecks{{
ID: "cert-check",
Checks: nmdata.ChecksDefinition{CertificateCheck: &nmdata.CertificateCheck{CACertificates: []string{ca.PEM}}},
}})
require.Len(t, checks, 1)
require.Equal(t, []string{ca.PEM}, checks[0].GetCertificateChallenge().GetCaCertificates())
require.Empty(t, checks[0].GetCertificateChallenge().GetNonce())
stampCertificateChallenges(checks, challenger, peerKey)
nonce := checks[0].GetCertificateChallenge().GetNonce()
require.NotEmpty(t, nonce)
deviceKey := certtest.ECDSAKey(t)
leaf := ca.Issue(t, deviceKey, "device")
sigAlg, sig, err := certposture.Sign(deviceKey, nonce, peerKey[:])
require.NoError(t, err)
proofs := []*proto.CertificateProof{{Nonce: nonce, Chain: [][]byte{leaf.Raw}, SigAlg: sigAlg, Signature: sig}}
s := &Server{challenger: challenger}
chains := s.verifiedCertificates(ctx, peerKey, proofs)
require.Len(t, chains, 1)
assert.True(t, certposture.ChainMatchesCAs(chains[0], []string{ca.PEM}, time.Now()))
t.Run("proof replayed by another peer is rejected", func(t *testing.T) {
assert.Nil(t, s.verifiedCertificates(ctx, generateKey(t).PublicKey(), proofs))
})
t.Run("nonce from another deployment's secret is rejected", func(t *testing.T) {
other := &Server{challenger: newCertChallenger("a-different-encryption-key", generateKey(t))}
assert.Nil(t, other.verifiedCertificates(ctx, peerKey, proofs))
})
t.Run("a second instance with the same encryption key accepts it", func(t *testing.T) {
// The point of deriving from the stored key: another instance, with its own
// freshly generated WireGuard key, verifies what this one issued. This is also
// what a restart looks like, and it used to reject.
peer := &Server{challenger: newCertChallenger(encryptionKey, generateKey(t))}
assert.Len(t, peer.verifiedCertificates(ctx, peerKey, proofs), 1)
})
t.Run("an invalid proof is dropped without the valid ones", func(t *testing.T) {
// A device proving its machine certificate while a user store answers with a key
// it cannot use must keep the machine certificate.
bad := &proto.CertificateProof{Nonce: nonce, Chain: [][]byte{leaf.Raw}, SigAlg: sigAlg, Signature: []byte("junk")}
chains := s.verifiedCertificates(ctx, peerKey, append([]*proto.CertificateProof{bad}, proofs...))
require.Len(t, chains, 1, "the valid proof survives the invalid one")
assert.True(t, certposture.ChainMatchesCAs(chains[0], []string{ca.PEM}, time.Now()))
})
t.Run("only invalid proofs yield no certificates", func(t *testing.T) {
bad := &proto.CertificateProof{Nonce: nonce, Chain: [][]byte{leaf.Raw}, SigAlg: sigAlg, Signature: []byte("junk")}
assert.Empty(t, s.verifiedCertificates(ctx, peerKey, []*proto.CertificateProof{bad}))
})
t.Run("no proofs yields no certificates", func(t *testing.T) {
assert.Nil(t, s.verifiedCertificates(ctx, peerKey, nil))
})
}
func TestStampCertificateChallenges_SkipsFileOnlyChecks(t *testing.T) {
checks := []*proto.Checks{{Files: []string{"/bin/agent"}}}
stampCertificateChallenges(checks, newCertChallenger("key", generateKey(t)), generateKey(t).PublicKey())
assert.Nil(t, checks[0].GetCertificateChallenge())
}
func generateKey(t *testing.T) wgtypes.Key {
t.Helper()
key, err := wgtypes.GeneratePrivateKey()
require.NoError(t, err)
return key
}
@@ -7,6 +7,7 @@ import (
"github.com/netbirdio/netbird/encryption"
"github.com/netbirdio/netbird/management/internals/controllers/network_map"
"github.com/netbirdio/netbird/management/server/telemetry"
"github.com/netbirdio/netbird/shared/management/certposture"
"github.com/netbirdio/netbird/shared/management/proto"
log "github.com/sirupsen/logrus"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
@@ -18,16 +19,20 @@ func PeerUpdateHandlerFactory(
peerKey wgtypes.Key,
updates chan *network_map.UpdateMessage,
secretsManager SecretsManager,
challenger *certposture.Challenger,
onChallengeStamped func(),
srv proto.ManagementService_SyncServer,
cleanupfunc func()) *PeerUpdateHandler {
return &PeerUpdateHandler{
peerKey: peerKey,
updates: updates,
secretsManager: secretsManager,
srv: srv,
encrypter: encryption.DefaultEncrypter{},
debouncer: NewUpdateDebouncer(1000 * time.Millisecond),
cleanupFunc: cleanupfunc,
peerKey: peerKey,
updates: updates,
secretsManager: secretsManager,
challenger: challenger,
onChallengeStamped: onChallengeStamped,
srv: srv,
encrypter: encryption.DefaultEncrypter{},
debouncer: NewUpdateDebouncer(1000 * time.Millisecond),
cleanupFunc: cleanupfunc,
}
}
@@ -40,10 +45,15 @@ type PeerUpdateHandler struct {
updates chan *network_map.UpdateMessage
appMetrics telemetry.AppMetrics
secretsManager SecretsManager
srv syncSender
encrypter encryption.Encrypter
debouncer Debouncer
cleanupFunc func()
challenger *certposture.Challenger
// onChallengeStamped registers the account for renewal. Every push reaches this
// path, including the ones a REST update on another instance broadcasts, which is
// what lets an instance pick up an account whose peers never reconnect.
onChallengeStamped func()
srv syncSender
encrypter encryption.Encrypter
debouncer Debouncer
cleanupFunc func()
}
func (pu *PeerUpdateHandler) WithMetrics(appMetrics telemetry.AppMetrics) *PeerUpdateHandler {
@@ -117,6 +127,9 @@ func (pu *PeerUpdateHandler) SendUpdate(ctx context.Context, update *network_map
return status.Errorf(codes.Internal, "failed processing update message")
}
if stampCertificateChallenges(update.Update.GetChecks(), pu.challenger, pu.peerKey) && pu.onChallengeStamped != nil {
pu.onChallengeStamped()
}
encryptedResp, err := pu.encrypter.EncryptMessage(pu.peerKey, key, update.Update)
if err != nil {
pu.cleanupFunc()
@@ -5,9 +5,9 @@ import (
"errors"
"testing"
"go.uber.org/mock/gomock"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"
"google.golang.org/grpc/codes"
grpcstatus "google.golang.org/grpc/status"
@@ -6,9 +6,9 @@ import (
"testing"
"time"
"go.uber.org/mock/gomock"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
+30 -5
View File
@@ -45,6 +45,7 @@ import (
"github.com/netbirdio/netbird/management/server/settings"
"github.com/netbirdio/netbird/management/server/telemetry"
"github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/shared/management/certposture"
"github.com/netbirdio/netbird/shared/management/networkmap/nmdata"
"github.com/netbirdio/netbird/shared/management/proto"
internalStatus "github.com/netbirdio/netbird/shared/management/status"
@@ -70,6 +71,7 @@ type Server struct {
peerLocks sync.Map
authManager auth.Manager
sessionStore *auth.SessionStore
challenger *certposture.Challenger
logBlockedPeers bool
blockPeersWithSameConfig bool
@@ -131,7 +133,13 @@ func NewServer(
}
}
serverKey, err := secretsManager.GetWGKey()
if err != nil {
return nil, fmt.Errorf("get server WireGuard key: %w", err)
}
return &Server{
challenger: newCertChallenger(config.DataStoreEncryptionKey, serverKey),
jobManager: jobManager,
accountManager: accountManager,
settingsManager: settingsManager,
@@ -246,6 +254,7 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
realIP := getRealIP(ctx)
sRealIP := realIP.String()
peerMeta := extractPeerMeta(ctx, syncReq.GetMeta())
peerMeta.Certificates = s.verifiedCertificates(ctx, peerKey, syncReq.GetMeta().GetCertificateProofs())
metahashed := metaHash(peerMeta)
if !s.loginFilter.allowLogin(peerKey.String(), metahashed) {
@@ -309,7 +318,8 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
return mapError(ctx, err)
}
err = s.sendInitialSync(ctx, peerKey, peer, netMap, postureChecks, srv, dnsFwdPort)
trackChallenge := func() { s.accountManager.TrackCertificateChallenges(ctx, accountID, peer.ID, syncStart) }
err = s.sendInitialSync(ctx, peerKey, peer, netMap, postureChecks, srv, dnsFwdPort, trackChallenge)
if err != nil {
log.WithContext(ctx).Debugf("error while sending initial sync for %s: %v", peerKey.String(), err)
s.syncSem.Add(-1)
@@ -337,7 +347,9 @@ func (s *Server) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_S
s.syncSem.Add(-1)
return PeerUpdateHandlerFactory(peerKey, updates, s.secretsManager, srv, func() { s.cancelPeerRoutines(ctx, accountID, peer, syncStart, updates) }).
return PeerUpdateHandlerFactory(peerKey, updates, s.secretsManager, s.challenger,
trackChallenge,
srv, func() { s.cancelPeerRoutines(ctx, accountID, peer, syncStart, updates) }).
WithMetrics(s.appMetrics).HandleUpdates(ctx)
}
@@ -451,6 +463,7 @@ func (s *Server) cancelPeerRoutinesWithoutLock(ctx context.Context, accountID st
return
}
s.secretsManager.CancelRefresh(peer.ID)
s.accountManager.UntrackCertificateChallenges(accountID, peer.ID, streamStartTime)
log.WithContext(ctx).Debugf("peer %s has been disconnected", peer.Key)
}
@@ -649,6 +662,7 @@ func (s *Server) Login(ctx context.Context, req *proto.EncryptedMessage) (*proto
}
peerMeta := extractPeerMeta(ctx, loginReq.GetMeta())
peerMeta.Certificates = s.verifiedCertificates(ctx, peerKey, loginReq.GetMeta().GetCertificateProofs())
metahashed := metaHash(peerMeta)
if !s.loginFilter.allowLogin(peerKey.String(), metahashed) {
if s.logBlockedPeers {
@@ -725,6 +739,8 @@ func (s *Server) Login(ctx context.Context, req *proto.EncryptedMessage) (*proto
return nil, status.Errorf(codes.Internal, "failed logging in peer")
}
// Renewal is tracked by the sync stream that follows, on whichever instance it lands.
stampCertificateChallenges(loginResp.Checks, s.challenger, peerKey)
encryptedResp, err := encryption.EncryptMessage(peerKey, key, loginResp)
if err != nil {
log.WithContext(ctx).Warnf("failed encrypting peer %s message", peer.ID)
@@ -904,7 +920,7 @@ func (s *Server) IsHealthy(ctx context.Context, req *proto.Empty) (*proto.Empty,
}
// sendInitialSync sends initial proto.SyncResponse to the peer requesting synchronization
func (s *Server) sendInitialSync(ctx context.Context, peerKey wgtypes.Key, peer *nbpeer.Peer, networkMap *types.NetworkMap, postureChecks []*nmdata.PostureChecks, srv proto.ManagementService_SyncServer, dnsFwdPort int64) error {
func (s *Server) sendInitialSync(ctx context.Context, peerKey wgtypes.Key, peer *nbpeer.Peer, networkMap *types.NetworkMap, postureChecks []*nmdata.PostureChecks, srv proto.ManagementService_SyncServer, dnsFwdPort int64, onChallengeStamped func()) error {
var err error
var turnToken *Token
@@ -976,6 +992,9 @@ func (s *Server) sendInitialSync(ctx context.Context, peerKey wgtypes.Key, peer
return status.Errorf(codes.Internal, "failed getting server key")
}
if stampCertificateChallenges(plainResp.Checks, s.challenger, peerKey) {
onChallengeStamped()
}
encryptedResp, err := encryption.EncryptMessage(peerKey, key, plainResp)
if err != nil {
return status.Errorf(codes.Internal, "error handling request")
@@ -1205,7 +1224,9 @@ func (s *Server) SyncMeta(ctx context.Context, req *proto.EncryptedMessage) (*pr
return nil, msg
}
err = s.accountManager.SyncPeerMeta(ctx, peerKey.String(), extractPeerMeta(ctx, syncMetaReq.GetMeta()), realIP)
peerMeta := extractPeerMeta(ctx, syncMetaReq.GetMeta())
peerMeta.Certificates = s.verifiedCertificates(ctx, peerKey, syncMetaReq.GetMeta().GetCertificateProofs())
err = s.accountManager.SyncPeerMeta(ctx, peerKey.String(), peerMeta, realIP)
if err != nil {
return nil, mapError(ctx, err)
}
@@ -1281,7 +1302,11 @@ func toProtocolCheck(postureCheck *nmdata.PostureChecks) *proto.Checks {
}
}
if len(protoCheck.Files) == 0 {
if check := postureCheck.Checks.CertificateCheck; check != nil {
protoCheck.CertificateChallenge = &proto.CertificateChallenge{CaCertificates: check.CACertificates}
}
if len(protoCheck.Files) == 0 && protoCheck.CertificateChallenge == nil {
return nil
}
@@ -40,6 +40,9 @@ func TestCancelPeerRoutines_SessionOwnership(t *testing.T) {
controller.EXPECT().OnPeerDisconnected(gomock.Any(), "account-1", peer.ID, tt.session).Return(tt.ownsPeer)
if tt.cancelRefresh {
secretsManager.EXPECT().CancelRefresh(peer.ID)
// The stream's challenge renewal ends with it; a stale session leaves the
// newer one's alone.
accountManager.EXPECT().UntrackCertificateChallenges("account-1", peer.ID, streamStart)
}
s.cancelPeerRoutines(context.Background(), "account-1", peer, streamStart, tt.session)
@@ -7,9 +7,9 @@ import (
"testing"
"time"
"go.uber.org/mock/gomock"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
@@ -10,8 +10,8 @@ import (
"testing"
"time"
"go.uber.org/mock/gomock"
"github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"
"github.com/netbirdio/netbird/management/internals/controllers/network_map"
"github.com/netbirdio/netbird/management/internals/controllers/network_map/update_channel"
+24
View File
@@ -100,6 +100,8 @@ type DefaultAccountManager struct {
peerInactivityExpiry Scheduler
certChallenges *certChallengeRefresher
// userDeleteFromIDPEnabled allows to delete user from IDP when user is deleted from account
userDeleteFromIDPEnabled bool
@@ -255,6 +257,8 @@ func BuildManager(
disableDefaultPolicy: disableDefaultPolicy,
}
am.certChallenges = newCertChallengeRefresher(am.refreshCertificateChallenges)
am.networkMapController.StartWarmup(ctx)
accountsCounter, err := store.GetAccountsCounter(ctx)
@@ -296,6 +300,10 @@ func BuildManager(
am.onPeersInvalidated(ctx, accountID, peerIDs)
})
// Started last: a manager that fails to build is never returned, so nothing would
// stop its refresher.
am.certChallenges.Start(ctx)
return am, nil
}
@@ -974,6 +982,7 @@ func (am *DefaultAccountManager) DeleteAccount(ctx context.Context, accountID, u
}
// cancel peer login expiry job
am.peerLoginExpiry.Cancel(ctx, []string{account.Id})
am.certChallenges.Forget(account.Id)
meta := map[string]any{"account_id": account.Id, "domain": account.Domain, "created_at": account.CreatedAt}
am.StoreEvent(ctx, userID, accountID, accountID, activity.AccountDeleted, meta)
@@ -2710,6 +2719,8 @@ func (am *DefaultAccountManager) UpdatePeerIP(ctx context.Context, accountID, us
}
if updateNetworkMap {
am.reloadServicesTargetingPeer(ctx, accountID, peerID)
peer, err := am.Store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
if err != nil {
return err
@@ -2724,6 +2735,19 @@ func (am *DefaultAccountManager) UpdatePeerIP(ctx context.Context, accountID, us
return nil
}
// reloadServicesTargetingPeer resends the account's reverse proxy services unless none of them targets the peer.
func (am *DefaultAccountManager) reloadServicesTargetingPeer(ctx context.Context, accountID, peerID string) {
serviceID, err := am.serviceManager.GetServiceIDByTargetID(ctx, accountID, peerID)
if err != nil {
log.WithContext(ctx).Warnf("failed to look up services targeting peer %s, reloading all: %v", peerID, err)
} else if serviceID == "" {
return
}
if err := am.serviceManager.ReloadAllServicesForAccount(ctx, accountID); err != nil {
log.WithContext(ctx).Warnf("failed to reload services for account %s: %v", accountID, err)
}
}
func (am *DefaultAccountManager) updatePeerIPInTransaction(ctx context.Context, accountID, userID, peerID string, newIP netip.Addr) (bool, error) {
var updateNetworkMap bool
err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
+2
View File
@@ -64,6 +64,8 @@ type Manager interface {
ListUsers(ctx context.Context, accountID string) ([]*types.User, error)
GetPeers(ctx context.Context, accountID, userID, nameFilter, ipFilter, macFilter string) ([]*nbpeer.Peer, error)
MarkPeerConnected(ctx context.Context, peerKey string, accountID string, sessionStartedAt int64, nmap *types.NetworkMap) error
TrackCertificateChallenges(ctx context.Context, accountID, peerID string, streamStart time.Time)
UntrackCertificateChallenges(accountID, peerID string, streamStart time.Time)
MarkPeerDisconnected(ctx context.Context, peerKey string, accountID string, sessionStartedAt int64) error
DeletePeer(ctx context.Context, accountID, peerID, userID string) error
UpdatePeer(ctx context.Context, accountID, userID string, p *nbpeer.Peer) (*nbpeer.Peer, error)
+24
View File
@@ -1632,6 +1632,30 @@ func (mr *MockManagerMockRecorder) SyncUserJWTGroups(ctx, userAuth any) *gomock.
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncUserJWTGroups", reflect.TypeOf((*MockManager)(nil).SyncUserJWTGroups), ctx, userAuth)
}
// TrackCertificateChallenges mocks base method.
func (m *MockManager) TrackCertificateChallenges(ctx context.Context, accountID, peerID string, streamStart time.Time) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "TrackCertificateChallenges", ctx, accountID, peerID, streamStart)
}
// TrackCertificateChallenges indicates an expected call of TrackCertificateChallenges.
func (mr *MockManagerMockRecorder) TrackCertificateChallenges(ctx, accountID, peerID, streamStart any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TrackCertificateChallenges", reflect.TypeOf((*MockManager)(nil).TrackCertificateChallenges), ctx, accountID, peerID, streamStart)
}
// UntrackCertificateChallenges mocks base method.
func (m *MockManager) UntrackCertificateChallenges(accountID, peerID string, streamStart time.Time) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "UntrackCertificateChallenges", accountID, peerID, streamStart)
}
// UntrackCertificateChallenges indicates an expected call of UntrackCertificateChallenges.
func (mr *MockManagerMockRecorder) UntrackCertificateChallenges(accountID, peerID, streamStart any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UntrackCertificateChallenges", reflect.TypeOf((*MockManager)(nil).UntrackCertificateChallenges), accountID, peerID, streamStart)
}
// UpdateAccountOnboarding mocks base method.
func (m *MockManager) UpdateAccountOnboarding(ctx context.Context, accountID, userID string, newOnboarding *types.AccountOnboarding) (*types.AccountOnboarding, error) {
m.ctrl.T.Helper()
+81
View File
@@ -3,6 +3,7 @@ package server
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net"
@@ -4460,6 +4461,86 @@ func TestDefaultAccountManager_UpdatePeerIP(t *testing.T) {
})
}
func TestDefaultAccountManager_UpdatePeerIP_ReloadsTargetingServices(t *testing.T) {
manager, _, err := createManager(t)
require.NoError(t, err, "unable to create account manager")
ctx := context.Background()
accountID, err := manager.GetAccountIDByUserID(ctx, auth.UserAuth{UserId: userID})
require.NoError(t, err, "unable to create an account")
key, err := wgtypes.GenerateKey()
require.NoError(t, err, "unable to generate WireGuard key")
peer, _, _, _, err := manager.AddPeer(ctx, "", "", userID, &nbpeer.Peer{
Key: key.PublicKey().String(),
Meta: nbpeer.PeerSystemMeta{Hostname: "proxied-peer"},
}, false)
require.NoError(t, err, "unable to add peer")
account, err := manager.Store.GetAccount(ctx, accountID)
require.NoError(t, err, "unable to get account")
network := netip.MustParsePrefix(account.Network.Net.String())
t.Run("targeted peer reloads services", func(t *testing.T) {
serviceManager := service.NewMockManager(gomock.NewController(t))
manager.SetServiceManager(serviceManager)
serviceManager.EXPECT().GetServiceIDByTargetID(gomock.Any(), accountID, peer.ID).Return("svc-1", nil)
serviceManager.EXPECT().ReloadAllServicesForAccount(gomock.Any(), accountID).Return(nil)
current, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peer.ID)
require.NoError(t, err)
newIP, err := types.AllocatePeerIP(network, []netip.Addr{current.IP})
require.NoError(t, err)
require.NoError(t, manager.UpdatePeerIP(ctx, accountID, userID, peer.ID, newIP))
})
t.Run("untargeted peer skips reload", func(t *testing.T) {
serviceManager := service.NewMockManager(gomock.NewController(t))
manager.SetServiceManager(serviceManager)
serviceManager.EXPECT().GetServiceIDByTargetID(gomock.Any(), accountID, peer.ID).Return("", nil)
current, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peer.ID)
require.NoError(t, err)
newIP, err := types.AllocatePeerIP(network, []netip.Addr{current.IP})
require.NoError(t, err)
require.NoError(t, manager.UpdatePeerIP(ctx, accountID, userID, peer.ID, newIP))
})
t.Run("lookup failure reloads services", func(t *testing.T) {
serviceManager := service.NewMockManager(gomock.NewController(t))
manager.SetServiceManager(serviceManager)
serviceManager.EXPECT().GetServiceIDByTargetID(gomock.Any(), accountID, peer.ID).Return("", errors.New("store unavailable"))
serviceManager.EXPECT().ReloadAllServicesForAccount(gomock.Any(), accountID).Return(nil)
current, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peer.ID)
require.NoError(t, err)
newIP, err := types.AllocatePeerIP(network, []netip.Addr{current.IP})
require.NoError(t, err)
require.NoError(t, manager.UpdatePeerIP(ctx, accountID, userID, peer.ID, newIP))
})
t.Run("network map notification failure still reloads services", func(t *testing.T) {
ctrl := gomock.NewController(t)
serviceManager := service.NewMockManager(ctrl)
manager.SetServiceManager(serviceManager)
serviceManager.EXPECT().GetServiceIDByTargetID(gomock.Any(), accountID, peer.ID).Return("svc-1", nil)
serviceManager.EXPECT().ReloadAllServicesForAccount(gomock.Any(), accountID).Return(nil)
originalController := manager.networkMapController
t.Cleanup(func() { manager.networkMapController = originalController })
networkMapController := network_map.NewMockController(ctrl)
networkMapController.EXPECT().GetDNSDomain(gomock.Any()).Return("netbird.cloud").AnyTimes()
networkMapController.EXPECT().OnPeersUpdated(gomock.Any(), accountID, []string{peer.ID}, gomock.Any()).Return(errors.New("notify failed"))
manager.networkMapController = networkMapController
current, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peer.ID)
require.NoError(t, err)
newIP, err := types.AllocatePeerIP(network, []netip.Addr{current.IP})
require.NoError(t, err)
require.Error(t, manager.UpdatePeerIP(ctx, accountID, userID, peer.ID, newIP))
})
}
func TestAddNewUserToDomainAccountWithApproval(t *testing.T) {
manager, _, err := createManager(t)
if err != nil {
+344
View File
@@ -0,0 +1,344 @@
package server
import (
"context"
"encoding/binary"
"hash/fnv"
"maps"
"slices"
"sync"
"time"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/management/server/store"
"github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/shared/management/certposture"
)
const (
minCertChallengeTick = time.Second
maxCertChallengeTick = 15 * time.Minute
maxCertChallengeRefresh = 30 * time.Second
)
// certChallengePeriod is how often an account whose policies carry a certificate
// posture check is pushed a fresh challenge. A nonce stays valid through the window
// after its own, so one issued at the very end of a window lives only one window. A
// third of that leaves a missed run well clear of the edge, where a half would put it
// exactly on it.
func certChallengePeriod() time.Duration {
return certposture.EffectiveWindow() / 3
}
// certChallengeTick is how often the refresher looks for accounts that are due. It is
// derived from the period rather than fixed, so shortening the challenge window for a
// test shortens this with it; the bounds keep a tiny window from spinning and a normal
// one from checking less often than is useful.
func certChallengeTick(period time.Duration) time.Duration {
return min(max(period/10, minCertChallengeTick), maxCertChallengeTick)
}
// certChallengeRefresh is how long one account's refresh is given before it is
// abandoned. Resolving the target peers reads the store and accounts are swept one
// after another, so an unbounded refresh lets one wedged read starve all the others.
func certChallengeRefresh(tick time.Duration) time.Duration {
return min(tick, maxCertChallengeRefresh)
}
// certChallengeRefresher pushes a fresh certificate challenge to the peers connected to
// this instance that answer one, from a single goroutine.
//
// A nonce only reaches a peer attached to a network map, and a quiet account sends no
// map, so without this the peer eventually re-sends an expired nonce, has its whole
// proof set rejected, and silently leaves the policies the check gates.
//
// Each instance renews only the peers whose sync stream it holds, so a peer is renewed
// once however many instances run, and an account is dropped as soon as none of its
// peers here answers a challenge.
type certChallengeRefresher struct {
mu sync.Mutex
accounts map[string]*certChallengeAccount
// refreshing holds the accounts whose refresh is running, and whether Track was
// called for one meanwhile, which makes its refresh's verdict stale.
refreshing map[string]bool
period time.Duration
tick time.Duration
timeout time.Duration
now func() time.Time
// refresh pushes an update to those of peerIDs that answer a certificate challenge,
// and reports whether any of them does.
refresh func(ctx context.Context, accountID string, peerIDs []string) bool
}
// certChallengeAccount is one account's renewal schedule and the peers it covers, each
// with the start of the sync stream it was stamped on.
type certChallengeAccount struct {
due time.Time
peers map[string]time.Time
}
// certChallengeDue is an account handed out for refresh, with its peers at that moment.
type certChallengeDue struct {
accountID string
peerIDs []string
}
func newCertChallengeRefresher(refresh func(ctx context.Context, accountID string, peerIDs []string) bool) *certChallengeRefresher {
period := certChallengePeriod()
tick := certChallengeTick(period)
return &certChallengeRefresher{
accounts: map[string]*certChallengeAccount{},
refreshing: map[string]bool{},
period: period,
tick: tick,
timeout: certChallengeRefresh(tick),
now: time.Now,
refresh: refresh,
}
}
// Start runs the refresh loop until ctx is done.
func (r *certChallengeRefresher) Start(ctx context.Context) {
go r.run(ctx)
}
// Track renews the challenge of peerID, which was stamped one on the sync stream that
// started at streamStart. An account seen for the first time has its first run spread
// over one period, so that a global window rollover does not fan out to every account in
// the same moment; an account already tracked keeps its schedule. A stamp from an older
// stream than the one recorded for the peer is ignored.
func (r *certChallengeRefresher) Track(ctx context.Context, accountID, peerID string, streamStart time.Time) {
r.mu.Lock()
defer r.mu.Unlock()
if _, ok := r.refreshing[accountID]; ok {
r.refreshing[accountID] = true
}
account, ok := r.accounts[accountID]
if !ok {
account = &certChallengeAccount{
due: r.now().Add(offsetWithin(accountID, r.period)),
peers: map[string]time.Time{},
}
r.accounts[accountID] = account
log.WithContext(ctx).Debugf("tracking certificate challenge refresh for account %s", accountID)
}
if started, ok := account.peers[peerID]; ok && streamStart.Before(started) {
return
}
account.peers[peerID] = streamStart
}
// Untrack stops renewing peerID once the sync stream that started at streamStart ends.
// A newer stream of the same peer keeps it tracked. The account is dropped with its
// last peer.
func (r *certChallengeRefresher) Untrack(accountID, peerID string, streamStart time.Time) {
r.mu.Lock()
defer r.mu.Unlock()
account, ok := r.accounts[accountID]
if !ok {
return
}
if started, ok := account.peers[peerID]; !ok || !started.Equal(streamStart) {
return
}
delete(account.peers, peerID)
if len(account.peers) == 0 {
delete(r.accounts, accountID)
}
}
// Forget stops refreshing accountID.
func (r *certChallengeRefresher) Forget(accountID string) {
r.mu.Lock()
defer r.mu.Unlock()
delete(r.accounts, accountID)
}
func (r *certChallengeRefresher) tracked(accountID string) bool {
r.mu.Lock()
defer r.mu.Unlock()
_, ok := r.accounts[accountID]
return ok
}
func (r *certChallengeRefresher) run(ctx context.Context) {
ticker := time.NewTicker(r.tick)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
for _, due := range r.takeDue() {
r.finish(due.accountID, r.refreshOne(ctx, due))
}
}
}
}
// refreshOne refreshes a single account under the refresh deadline. A refresh that runs
// out of time reports the account as still wanting challenges, since a deadline says
// nothing about the account's posture checks; the next sweep tries again.
func (r *certChallengeRefresher) refreshOne(ctx context.Context, due certChallengeDue) bool {
ctx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
return r.refresh(ctx, due.accountID, due.peerIDs)
}
// finish records the outcome of an account's refresh. An account none of whose peers
// answers a challenge any more is dropped, unless one was tracked again while the
// refresh ran: the refresh may have read the store before a certificate check was added.
// A dropped peer is tracked again the next time it is stamped a challenge.
func (r *certChallengeRefresher) finish(accountID string, wanted bool) {
r.mu.Lock()
defer r.mu.Unlock()
retracked := r.refreshing[accountID]
delete(r.refreshing, accountID)
if !wanted && !retracked {
delete(r.accounts, accountID)
}
}
// takeDue returns the accounts due now and books their next run straight away, so a
// slow refresh cannot make an account fall due twice, and so the refresh itself runs
// without the lock.
func (r *certChallengeRefresher) takeDue() []certChallengeDue {
r.mu.Lock()
defer r.mu.Unlock()
now := r.now()
var due []certChallengeDue
for accountID, account := range r.accounts {
if account.due.After(now) {
continue
}
due = append(due, certChallengeDue{accountID: accountID, peerIDs: slices.Collect(maps.Keys(account.peers))})
account.due = now.Add(r.period)
r.refreshing[accountID] = false
}
return due
}
// offsetWithin maps a key to a stable duration in [0, period).
func offsetWithin(key string, period time.Duration) time.Duration {
h := fnv.New64a()
//nolint:errcheck // hash.Write never returns an error
h.Write([]byte(key))
return time.Duration(binary.BigEndian.Uint64(h.Sum(nil)) % uint64(period))
}
// refreshCertificateChallenges pushes an update to those of peerIDs that answer a
// certificate challenge, so each is stamped with a nonce for the current window. It
// reports whether any of them answers one.
func (am *DefaultAccountManager) refreshCertificateChallenges(ctx context.Context, accountID string, peerIDs []string) bool {
targets, err := am.certificateChallengeTargets(ctx, accountID)
if err != nil {
log.WithContext(ctx).Debugf("cannot resolve the certificate challenge targets of account %s: %v", accountID, err)
// Keep the account tracked: a store error now says nothing about its checks.
return true
}
peerIDs = slices.DeleteFunc(peerIDs, func(id string) bool { return !slices.Contains(targets, id) })
if len(peerIDs) == 0 {
log.WithContext(ctx).Debugf("no peer of account %s on this instance answers a certificate challenge, stopping challenge refresh", accountID)
return false
}
log.WithContext(ctx).Debugf("refreshing certificate challenges for %d peers of account %s", len(peerIDs), accountID)
reason := types.UpdateReason{Resource: types.UpdateResourcePostureCheck, Operation: types.UpdateOperationRefresh}
if err := am.networkMapController.BufferUpdateAffectedPeers(ctx, accountID, peerIDs, reason); err != nil {
log.WithContext(ctx).Warnf("failed refreshing certificate challenges for account %s: %v", accountID, err)
}
return true
}
// certificateChallengeTargets returns the peers that are sent a certificate challenge.
// Only those peers hold a nonce, so pushing to the whole account would wake every peer
// that never uses the feature.
//
// It is the inverse of processPeerPostureChecks, which decides the same thing one peer
// at a time; TestCertificateChallengeTargets_MatchesThePerPeerRule holds them together.
func (am *DefaultAccountManager) certificateChallengeTargets(ctx context.Context, accountID string) ([]string, error) {
certCheckIDs, err := am.certificatePostureCheckIDs(ctx, accountID)
if err != nil {
return nil, err
}
if len(certCheckIDs) == 0 {
return nil, nil
}
policies, err := am.Store.GetAccountPolicies(ctx, store.LockingStrengthNone, accountID)
if err != nil {
return nil, err
}
groups, err := am.Store.GetAccountGroups(ctx, store.LockingStrengthNone, accountID)
if err != nil {
return nil, err
}
groupPeers := make(map[string][]string, len(groups))
for _, g := range groups {
groupPeers[g.ID] = g.Peers
}
return certificateChallengeTargets(policies, groupPeers, certCheckIDs), nil
}
// certificateChallengeTargets collects the source peers of every enabled policy whose
// posture checks include a certificate check.
func certificateChallengeTargets(policies []*types.Policy, groupPeers map[string][]string, certCheckIDs map[string]struct{}) []string {
targets := map[string]struct{}{}
for _, policy := range policies {
if !policy.Enabled || !slices.ContainsFunc(policy.SourcePostureChecks, func(id string) bool {
_, ok := certCheckIDs[id]
return ok
}) {
continue
}
for _, rule := range policy.Rules {
if !rule.Enabled {
continue
}
if rule.SourceResource.Type == types.ResourceTypePeer && rule.SourceResource.ID != "" {
targets[rule.SourceResource.ID] = struct{}{}
}
for _, groupID := range rule.Sources {
for _, peerID := range groupPeers[groupID] {
targets[peerID] = struct{}{}
}
}
}
}
return slices.Collect(maps.Keys(targets))
}
// certificatePostureCheckIDs returns the IDs of the account's posture checks that ask a
// peer to prove a certificate.
func (am *DefaultAccountManager) certificatePostureCheckIDs(ctx context.Context, accountID string) (map[string]struct{}, error) {
checks, err := am.Store.GetAccountPostureChecks(ctx, store.LockingStrengthNone, accountID)
if err != nil {
return nil, err
}
ids := map[string]struct{}{}
for _, check := range checks {
if check.Checks.CertificateCheck != nil {
ids[check.ID] = struct{}{}
}
}
return ids, nil
}
// TrackCertificateChallenges starts renewing the certificate challenge of a peer that was
// stamped one on the sync stream started at streamStart. It is called on every stamped
// update, so it has to stay cheap: no store access, just a map the refresher sweeps.
func (am *DefaultAccountManager) TrackCertificateChallenges(ctx context.Context, accountID, peerID string, streamStart time.Time) {
am.certChallenges.Track(ctx, accountID, peerID, streamStart)
}
// UntrackCertificateChallenges stops renewing the certificate challenge of a peer whose
// sync stream started at streamStart has ended.
func (am *DefaultAccountManager) UntrackCertificateChallenges(accountID, peerID string, streamStart time.Time) {
am.certChallenges.Untrack(accountID, peerID, streamStart)
}
@@ -0,0 +1,493 @@
package server
import (
"context"
"slices"
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/shared/management/certposture"
)
// fakeClock lets a test drive the refresher's schedule without waiting for it.
type fakeClock struct {
mu sync.Mutex
at time.Time
}
func (c *fakeClock) Now() time.Time {
c.mu.Lock()
defer c.mu.Unlock()
return c.at
}
func (c *fakeClock) Advance(d time.Duration) {
c.mu.Lock()
defer c.mu.Unlock()
c.at = c.at.Add(d)
}
// scheduleRefresher builds a refresher on a clock the test drives and never starts its
// loop, so the schedule can be checked by calling takeDue directly.
func scheduleRefresher(refresh func(accountID string) bool) (*certChallengeRefresher, *fakeClock) {
clock := &fakeClock{at: time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)}
r := newCertChallengeRefresher(func(_ context.Context, accountID string, _ []string) bool {
return refresh(accountID)
})
r.now = clock.Now
return r, clock
}
// runningRefresher builds a refresher whose period and tick are short enough to observe
// in a test, and starts its loop.
func runningRefresher(t *testing.T, refresh func(accountID string) bool) *certChallengeRefresher {
t.Helper()
r := newCertChallengeRefresher(func(_ context.Context, accountID string, _ []string) bool {
return refresh(accountID)
})
r.period = 10 * time.Millisecond
r.tick = time.Millisecond
r.Start(t.Context())
return r
}
// streamStart is the start of the sync stream the tests' peers are stamped on.
var streamStart = time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
// track tracks one peer of accountID, as a stamped sync stream does.
func track(r *certChallengeRefresher, accountID string) {
r.Track(context.Background(), accountID, accountID+"-peer", streamStart)
}
// dueAccounts returns the IDs of the accounts takeDue hands out.
func dueAccounts(r *certChallengeRefresher) []string {
var ids []string
for _, due := range r.takeDue() {
ids = append(ids, due.accountID)
}
return ids
}
func (r *certChallengeRefresher) dueAt(accountID string) time.Time {
r.mu.Lock()
defer r.mu.Unlock()
return r.accounts[accountID].due
}
func TestCertChallengePeriod_LeavesRoomForAMissedRun(t *testing.T) {
// A nonce issued at the very end of a window is accepted for that window and the
// next one only, so the shortest life a peer can be handed is one window. Renewing
// has to stay clear of that edge even when a run is missed.
assert.Less(t, 2*certChallengePeriod(), certposture.EffectiveWindow(),
"a missed refresh must still leave the peer's nonce valid, with margin")
}
func TestCertChallengeTick_FollowsTheWindow(t *testing.T) {
// An end-to-end test shortens the challenge window to watch a renewal happen. The
// tick has to come down with it, or the refresher would still be looking for due
// accounts every quarter of an hour and never renew anything in time.
short := certChallengeTick(30 * time.Second)
assert.Less(t, short, 30*time.Second, "the tick must be shorter than the period it serves")
assert.GreaterOrEqual(t, short, minCertChallengeTick, "the tick must not spin")
assert.Equal(t, maxCertChallengeTick, certChallengeTick(24*time.Hour),
"a long period must not stretch the tick without bound")
assert.Equal(t, minCertChallengeTick, certChallengeTick(time.Millisecond),
"a tiny period must not drive the tick below its floor")
}
func TestCertChallengeRefresher_SpreadsAccountsOverThePeriod(t *testing.T) {
// Every account on an instance shares the same global challenge window, so a
// restart arms them all at once. The offset is what keeps them from fanning out
// to their peers in the same moment.
r, clock := scheduleRefresher(func(string) bool { return true })
ids := []string{"account-a", "account-b", "account-c", "account-d", "account-e", "account-f"}
for _, id := range ids {
track(r, id)
}
start := clock.Now()
offsets := map[time.Duration]bool{}
for _, id := range ids {
offset := r.dueAt(id).Sub(start)
assert.GreaterOrEqual(t, offset, time.Duration(0), "account %s is due in the past", id)
assert.Less(t, offset, r.period, "account %s is due beyond one period", id)
offsets[offset] = true
}
assert.Greater(t, len(offsets), 1, "all accounts were given the same offset, which defeats the spreading")
}
func TestCertChallengeOffset_IsStablePerAccount(t *testing.T) {
assert.Equal(t, offsetWithin("account-a", certChallengePeriod()), offsetWithin("account-a", certChallengePeriod()),
"the same account must keep its slot across restarts")
assert.NotEqual(t, offsetWithin("account-a", certChallengePeriod()), offsetWithin("account-b", certChallengePeriod()),
"two accounts must not share a slot")
}
func TestCertChallengeRefresher_RenewsOncePerPeriod(t *testing.T) {
r, clock := scheduleRefresher(func(string) bool { return true })
track(r, "account-a")
assert.Empty(t, r.takeDue(), "an account just tracked is not due before its offset elapses")
clock.Advance(r.period)
assert.Equal(t, []string{"account-a"}, dueAccounts(r), "the account is due once its offset has elapsed")
clock.Advance(r.period / 2)
assert.Empty(t, r.takeDue(), "the next run is booked a full period out")
clock.Advance(r.period / 2)
assert.Equal(t, []string{"account-a"}, dueAccounts(r), "the account is due again one period later")
}
func TestCertChallengeRefresher_BooksTheNextRunBeforeRefreshing(t *testing.T) {
// takeDue reserves the next run while it holds the lock, so a refresh that outlives
// a tick cannot have the same account handed out twice.
r, clock := scheduleRefresher(func(string) bool { return true })
track(r, "account-a")
clock.Advance(r.period)
require.Len(t, r.takeDue(), 1, "the account is due")
assert.Empty(t, r.takeDue(), "a second pass at the same instant must not hand out the account again")
}
func TestCertChallengeRefresher_TrackIsIdempotent(t *testing.T) {
r, clock := scheduleRefresher(func(string) bool { return true })
track(r, "account-a")
first := r.dueAt("account-a")
clock.Advance(certChallengePeriod())
track(r, "account-a")
assert.Equal(t, first, r.dueAt("account-a"), "re-tracking must not push the next run further out")
}
func TestCertChallengeRefresher_RefreshesATrackedAccount(t *testing.T) {
refreshed := make(chan string, 4)
r := runningRefresher(t, func(accountID string) bool {
select {
case refreshed <- accountID:
default:
}
return true
})
track(r, "account-a")
select {
case got := <-refreshed:
assert.Equal(t, "account-a", got)
case <-time.After(3 * time.Second):
t.Fatal("a tracked account was never refreshed")
}
assert.True(t, r.tracked("account-a"), "an account that still wants challenges stays tracked")
}
func TestCertChallengeRefresher_DropsAnAccountThatNoLongerWantsChallenges(t *testing.T) {
var mu sync.Mutex
var calls int
r := runningRefresher(t, func(string) bool {
mu.Lock()
defer mu.Unlock()
calls++
return false
})
track(r, "account-a")
require.Eventually(t, func() bool { return !r.tracked("account-a") }, 3*time.Second, time.Millisecond,
"an account whose refresh reports it no longer wants challenges must be dropped")
time.Sleep(20 * r.tick)
mu.Lock()
defer mu.Unlock()
assert.Equal(t, 1, calls, "the account must not be refreshed again after being dropped")
}
func TestCertChallengeRefresher_RefreshesWithoutHoldingTheLock(t *testing.T) {
// The refresh fans out to every peer of the account, so holding the lock across it
// would stall every peer that connects meanwhile. Calling back into the refresher
// from inside the refresh deadlocks if the loop still holds it.
reentered := make(chan bool, 1)
var once sync.Once
r := newCertChallengeRefresher(nil)
r.period = 10 * time.Millisecond
r.tick = time.Millisecond
r.refresh = func(_ context.Context, accountID string, _ []string) bool {
once.Do(func() {
answered := make(chan bool, 1)
go func() { answered <- r.tracked(accountID) }()
select {
case got := <-answered:
reentered <- got
case <-time.After(2 * time.Second):
reentered <- false
}
})
return true
}
r.Start(t.Context())
track(r, "account-a")
select {
case ok := <-reentered:
assert.True(t, ok, "the refresher was not reachable while a refresh was in flight")
case <-time.After(5 * time.Second):
t.Fatal("the refresh never ran")
}
}
// TestCertificateChallengeTargets_MatchesThePerPeerRule holds the renewal to the same
// rule management uses when it hands out the challenge. The refresher asks "which peers
// get one", processPeerPostureChecks asks "does this peer get one", and the two have to
// answer identically: a peer the refresher forgets stops being renewed and silently
// falls out of its policies, which is the failure the refresher exists to prevent.
func TestCertificateChallengeTargets_MatchesThePerPeerRule(t *testing.T) {
const certCheck, osCheck = "check-cert", "check-os"
certCheckIDs := map[string]struct{}{certCheck: {}}
groupPeers := map[string][]string{
"g-devs": {"p1", "p2"},
"g-servers": {"p3"},
"g-empty": {},
"g-mixed": {"p2", "p4"},
}
peerGroups := map[string][]string{
"p1": {"g-devs"},
"p2": {"g-devs", "g-mixed"},
"p3": {"g-servers"},
"p4": {"g-mixed"},
"p5": {},
}
policies := []*types.Policy{
{
ID: "gated", Enabled: true, SourcePostureChecks: []string{certCheck},
Rules: []*types.PolicyRule{{Enabled: true, Sources: []string{"g-devs"}}},
},
{
// A disabled policy hands out nothing.
ID: "disabled-policy", Enabled: false, SourcePostureChecks: []string{certCheck},
Rules: []*types.PolicyRule{{Enabled: true, Sources: []string{"g-servers"}}},
},
{
// A disabled rule inside an enabled policy likewise.
ID: "disabled-rule", Enabled: true, SourcePostureChecks: []string{certCheck},
Rules: []*types.PolicyRule{{Enabled: false, Sources: []string{"g-servers"}}},
},
{
// Gated on something other than a certificate: no nonce to renew.
ID: "other-check", Enabled: true, SourcePostureChecks: []string{osCheck},
Rules: []*types.PolicyRule{{Enabled: true, Sources: []string{"g-mixed"}}},
},
{
// A peer named directly rather than through a group.
ID: "by-resource", Enabled: true, SourcePostureChecks: []string{osCheck, certCheck},
Rules: []*types.PolicyRule{{
Enabled: true,
SourceResource: types.Resource{Type: types.ResourceTypePeer, ID: "p4"},
}},
},
{
// Destinations are never filtered, so being one earns no challenge.
ID: "as-destination", Enabled: true, SourcePostureChecks: []string{certCheck},
Rules: []*types.PolicyRule{{Enabled: true, Sources: []string{"g-empty"}, Destinations: []string{"g-servers"}}},
},
}
got := certificateChallengeTargets(policies, groupPeers, certCheckIDs)
// The same question, asked one peer at a time the way the gRPC layer asks it.
var want []string
for peerID, groupIDs := range peerGroups {
for _, policy := range policies {
if !policy.Enabled {
continue
}
ids := processPeerPostureChecks(policy, peerID, groupIDs)
if slices.ContainsFunc(ids, func(id string) bool { _, ok := certCheckIDs[id]; return ok }) {
want = append(want, peerID)
break
}
}
}
slices.Sort(got)
slices.Sort(want)
assert.Equal(t, want, got, "the peers the refresher renews must be exactly the peers that are sent a challenge")
assert.Equal(t, []string{"p1", "p2", "p4"}, got, "p3 is only reachable through disabled rules, p5 is in no source group")
}
func TestCertChallengeRefresher_StopsWithItsContext(t *testing.T) {
// BuildManager hands the loop the manager's context. A refresher that detached from
// it could never be stopped: it would keep sweeping after the server it belongs to
// is gone, and leak a goroutine per manager a test builds.
var mu sync.Mutex
var calls int
r := newCertChallengeRefresher(func(_ context.Context, _ string, _ []string) bool {
mu.Lock()
defer mu.Unlock()
calls++
return true
})
r.period = time.Millisecond
r.tick = time.Millisecond
ctx, cancel := context.WithCancel(context.Background())
r.Start(ctx)
track(r, "account-a")
require.Eventually(t, func() bool {
mu.Lock()
defer mu.Unlock()
return calls > 0
}, 3*time.Second, time.Millisecond, "the refresher never ran")
cancel()
time.Sleep(20 * r.tick)
mu.Lock()
afterCancel := calls
mu.Unlock()
time.Sleep(20 * r.tick)
mu.Lock()
defer mu.Unlock()
assert.Equal(t, afterCancel, calls, "the refresher kept sweeping after its context was cancelled")
}
func TestCertChallengeRefresh_StaysWithinTheSweep(t *testing.T) {
assert.Equal(t, maxCertChallengeRefresh, certChallengeRefresh(time.Hour),
"a long tick must not grant a store read minutes")
assert.Equal(t, time.Second, certChallengeRefresh(time.Second),
"a short tick must bound the refresh to itself, so the sweep keeps its cadence")
}
func TestCertChallengeRefresher_AbandonsAWedgedAccount(t *testing.T) {
// Accounts are swept one after another and resolving the targets reads the store, so
// an account whose read never returns would hold up every other account on the
// instance for as long as the process lives.
entered := make(chan struct{}, 1)
refreshed := make(chan string, 4)
r := newCertChallengeRefresher(func(ctx context.Context, accountID string, _ []string) bool {
if accountID == "account-wedged" {
select {
case entered <- struct{}{}:
default:
}
<-ctx.Done()
return true
}
select {
case refreshed <- accountID:
default:
}
return true
})
r.period = 10 * time.Millisecond
r.tick = time.Millisecond
r.timeout = 20 * time.Millisecond
r.Start(t.Context())
track(r, "account-wedged")
select {
case <-entered:
case <-time.After(3 * time.Second):
t.Fatal("the wedged account was never refreshed")
}
// Tracked only once the sweep is already stuck, so reaching it proves the deadline
// released the loop rather than the map happening to hand this one out first.
track(r, "account-healthy")
select {
case got := <-refreshed:
assert.Equal(t, "account-healthy", got)
case <-time.After(3 * time.Second):
t.Fatal("a healthy account was never refreshed from behind a wedged one")
}
}
// TestCertChallengeRefresher_KeepsAnAccountTrackedDuringItsRefresh: a refresh reads the
// store before an admin adds a certificate check; the push that follows the change
// tracks the account again. The refresh then reports the account as not wanting
// challenges, which is stale, and must not untrack it.
func TestCertChallengeRefresher_KeepsAnAccountTrackedDuringItsRefresh(t *testing.T) {
r, clock := scheduleRefresher(func(string) bool { return false })
track(r, "account-a")
clock.Advance(r.period)
require.Equal(t, []string{"account-a"}, dueAccounts(r), "the account is due")
track(r, "account-a")
r.finish("account-a", false)
assert.True(t, r.tracked("account-a"), "an account tracked again during its refresh stays tracked")
clock.Advance(r.period)
require.Equal(t, []string{"account-a"}, dueAccounts(r), "the account is due again")
r.finish("account-a", false)
assert.False(t, r.tracked("account-a"), "an account nobody tracked again is dropped once it no longer wants challenges")
}
// TestCertChallengeRefresher_RenewsOnlyThePeersStreamedHere: every instance runs its
// own refresher, and a push fans out to all of them, so an instance must hand out only
// the peers whose sync stream it holds. Otherwise each peer is pushed once per instance.
func TestCertChallengeRefresher_RenewsOnlyThePeersStreamedHere(t *testing.T) {
r, clock := scheduleRefresher(func(string) bool { return true })
r.Track(context.Background(), "account-a", "peer-1", streamStart)
r.Track(context.Background(), "account-a", "peer-2", streamStart)
clock.Advance(r.period)
due := r.takeDue()
require.Len(t, due, 1, "the account is due")
assert.ElementsMatch(t, []string{"peer-1", "peer-2"}, due[0].peerIDs, "only the peers tracked here are renewed")
}
// TestCertChallengeRefresher_DropsAnAccountWithItsLastPeer: an instance whose peers of an
// account have all disconnected has nobody left to renew, and must not keep sweeping it.
func TestCertChallengeRefresher_DropsAnAccountWithItsLastPeer(t *testing.T) {
r, _ := scheduleRefresher(func(string) bool { return true })
r.Track(context.Background(), "account-a", "peer-1", streamStart)
r.Track(context.Background(), "account-a", "peer-2", streamStart)
r.Untrack("account-a", "peer-1", streamStart)
assert.True(t, r.tracked("account-a"), "the account stays tracked while a peer is still streamed here")
r.Untrack("account-a", "peer-2", streamStart)
assert.False(t, r.tracked("account-a"), "the account is dropped with its last peer")
}
// TestCertChallengeRefresher_StaleStreamKeepsTheNewerOne: a peer that reconnects to the
// same instance opens its new stream before the old one is torn down. The old stream's
// teardown must not untrack the peer the new stream is renewing.
func TestCertChallengeRefresher_StaleStreamKeepsTheNewerOne(t *testing.T) {
r, _ := scheduleRefresher(func(string) bool { return true })
newer := streamStart.Add(time.Minute)
r.Track(context.Background(), "account-a", "peer-1", streamStart)
r.Track(context.Background(), "account-a", "peer-1", newer)
r.Untrack("account-a", "peer-1", streamStart)
assert.True(t, r.tracked("account-a"), "the stale stream's teardown must leave the newer stream tracked")
r.Untrack("account-a", "peer-1", newer)
assert.False(t, r.tracked("account-a"), "the newer stream's teardown untracks the peer")
}
// TestCertChallengeRefresher_LateStampFromAStaleStream: the old stream of a reconnected
// peer can still deliver a stamped update after the new stream started. Its Track must
// not take the peer over, or the old stream's teardown would untrack the live one.
func TestCertChallengeRefresher_LateStampFromAStaleStream(t *testing.T) {
r, _ := scheduleRefresher(func(string) bool { return true })
newer := streamStart.Add(time.Minute)
r.Track(context.Background(), "account-a", "peer-1", newer)
r.Track(context.Background(), "account-a", "peer-1", streamStart)
r.Untrack("account-a", "peer-1", streamStart)
assert.True(t, r.tracked("account-a"), "a late stamp from the stale stream must leave the newer stream tracked")
}
@@ -133,6 +133,7 @@ type MockAccountManager struct {
GetOrCreateAccountByPrivateDomainFunc func(ctx context.Context, initiatorId, domain string) (*types.Account, bool, error)
AllowSyncFunc func(string, uint64) bool
TrackCertificateChallengesFunc func(ctx context.Context, accountID, peerID string, streamStart time.Time)
UpdateAccountPeersFunc func(ctx context.Context, accountID string, reason types.UpdateReason)
ExpandAndUpdateAffectedFunc func(ctx context.Context, accountID string, snap *affectedpeers.Snapshot, change affectedpeers.Change)
BufferUpdateAccountPeersFunc func(ctx context.Context, accountID string, reason types.UpdateReason)
@@ -206,6 +207,16 @@ func (am *MockAccountManager) UpdateGroups(ctx context.Context, accountID, userI
return status.Errorf(codes.Unimplemented, "method UpdateGroups is not implemented")
}
// TrackCertificateChallenges mocks TrackCertificateChallenges of the account manager.
func (am *MockAccountManager) TrackCertificateChallenges(ctx context.Context, accountID, peerID string, streamStart time.Time) {
if am.TrackCertificateChallengesFunc != nil {
am.TrackCertificateChallengesFunc(ctx, accountID, peerID, streamStart)
}
}
// UntrackCertificateChallenges mocks UntrackCertificateChallenges of the account manager.
func (am *MockAccountManager) UntrackCertificateChallenges(string, string, time.Time) {}
func (am *MockAccountManager) UpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) {
if am.UpdateAccountPeersFunc != nil {
am.UpdateAccountPeersFunc(ctx, accountID, reason)
+6 -1
View File
@@ -174,6 +174,7 @@ type PeerSystemMeta struct { //nolint:revive
Environment Environment `gorm:"serializer:json"`
Flags Flags `gorm:"serializer:json"`
Files []File `gorm:"serializer:json"`
Certificates []string `gorm:"serializer:json"`
Capabilities []int32 `gorm:"serializer:json"`
SyncMessageVersion int
}
@@ -199,7 +200,8 @@ func (p PeerSystemMeta) isEmpty() bool {
p.SystemManufacturer == "" &&
p.Environment.Cloud == "" &&
p.Environment.Platform == "" &&
len(p.Files) == 0
len(p.Files) == 0 &&
len(p.Certificates) == 0
}
// AddedWithSSOLogin indicates whether this peer has been added with an SSO login by a user.
@@ -418,6 +420,9 @@ func diffMeta(oldMeta, newMeta PeerSystemMeta, oldLocation, newLocation Location
if !sameMultiset(oldMeta.Files, newMeta.Files) {
add("files", fmt.Sprintf("%v", oldMeta.Files), fmt.Sprintf("%v", newMeta.Files))
}
if !sameMultiset(oldMeta.Certificates, newMeta.Certificates) {
add("certificates", len(oldMeta.Certificates), len(newMeta.Certificates))
}
if oldMeta.SyncMessageVersion != newMeta.SyncMessageVersion {
add("sync_meta_version", fmt.Sprintf("%d", oldMeta.SyncMessageVersion), fmt.Sprintf("%d", newMeta.SyncMessageVersion))
}
+36
View File
@@ -0,0 +1,36 @@
package posture
import (
"context"
"fmt"
"time"
nbpeer "github.com/netbirdio/netbird/management/server/peer"
"github.com/netbirdio/netbird/shared/management/certposture"
)
// CertificateCheck passes when the peer holds a certificate, proven at meta ingestion,
// that chains to one of the configured PEM encoded CA certificates.
type CertificateCheck struct {
CACertificates []string
}
var _ Check = (*CertificateCheck)(nil)
func (c *CertificateCheck) Check(_ context.Context, peer nbpeer.Peer) (bool, error) {
return certposture.AnyChainMatchesCAs(peer.Meta.Certificates, c.CACertificates, time.Now()), nil
}
func (c *CertificateCheck) Name() string {
return CertificateCheckName
}
func (c *CertificateCheck) Validate() error {
if len(c.CACertificates) == 0 {
return fmt.Errorf("%s ca certificates shouldn't be empty", c.Name())
}
if _, err := certposture.ParseCAs(c.CACertificates); err != nil {
return fmt.Errorf("%s: %w", c.Name(), err)
}
return nil
}
@@ -0,0 +1,68 @@
package posture
import (
"context"
"crypto/x509"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/management/server/peer"
"github.com/netbirdio/netbird/shared/management/certposture"
"github.com/netbirdio/netbird/shared/management/certposture/certtest"
)
func TestCertificateCheck_Check(t *testing.T) {
ca := certtest.NewCA(t, "corp-root")
otherCA := certtest.NewCA(t, "other-root")
chain := certposture.EncodeChainPEM([]*x509.Certificate{ca.Issue(t, certtest.ECDSAKey(t), "device")})
tests := []struct {
name string
certificates []string
cas []string
want bool
}{
{"chains to configured CA", []string{chain}, []string{ca.PEM}, true},
{"one of several CAs matches", []string{chain}, []string{otherCA.PEM, ca.PEM}, true},
{"unrelated CA", []string{chain}, []string{otherCA.PEM}, false},
{"no certificates proven", nil, []string{ca.PEM}, false},
{"garbage entry does not hide a valid one", []string{"garbage", chain}, []string{ca.PEM}, true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
check := CertificateCheck{CACertificates: tt.cas}
got, err := check.Check(context.Background(), peer.Peer{Meta: peer.PeerSystemMeta{Certificates: tt.certificates}})
require.NoError(t, err)
assert.Equal(t, tt.want, got)
})
}
}
func TestCertificateCheck_Validate(t *testing.T) {
ca := certtest.NewCA(t, "corp-root")
assert.Error(t, (&CertificateCheck{}).Validate())
assert.Error(t, (&CertificateCheck{CACertificates: []string{"not a pem"}}).Validate())
assert.NoError(t, (&CertificateCheck{CACertificates: []string{ca.PEM}}).Validate())
}
func TestChecks_CertificateCheckRegistered(t *testing.T) {
ca := certtest.NewCA(t, "corp-root")
checks := &Checks{Name: "cert", Checks: ChecksDefinition{CertificateCheck: &CertificateCheck{CACertificates: []string{ca.PEM}}}}
require.NoError(t, checks.Validate())
require.Len(t, checks.GetChecks(), 1)
assert.Equal(t, CertificateCheckName, checks.GetChecks()[0].Name())
copied := checks.Copy()
checks.Checks.CertificateCheck.CACertificates[0] = "mutated"
assert.Equal(t, ca.PEM, copied.Checks.CertificateCheck.CACertificates[0])
api := checks.ToAPIResponse()
require.NotNil(t, api.Checks.CertificateCheck)
roundTrip, err := NewChecksFromAPIPostureCheck(*api)
require.NoError(t, err)
assert.Equal(t, checks.Checks.CertificateCheck.CACertificates, roundTrip.Checks.CertificateCheck.CACertificates)
}
+19
View File
@@ -19,6 +19,7 @@ const (
GeoLocationCheckName = "GeoLocationCheck"
PeerNetworkRangeCheckName = "PeerNetworkRangeCheck"
ProcessCheckName = "ProcessCheck"
CertificateCheckName = "CertificateCheck"
CheckActionAllow string = "allow"
CheckActionDeny string = "deny"
@@ -61,6 +62,7 @@ type ChecksDefinition struct {
GeoLocationCheck *GeoLocationCheck `json:",omitempty"`
PeerNetworkRangeCheck *PeerNetworkRangeCheck `json:",omitempty"`
ProcessCheck *ProcessCheck `json:",omitempty"`
CertificateCheck *CertificateCheck `json:",omitempty"`
}
// Copy returns a copy of a checks definition.
@@ -113,6 +115,12 @@ func (cd ChecksDefinition) Copy() ChecksDefinition {
}
copy(cdCopy.ProcessCheck.Processes, processCheck.Processes)
}
if cd.CertificateCheck != nil {
cdCopy.CertificateCheck = &CertificateCheck{
CACertificates: make([]string, len(cd.CertificateCheck.CACertificates)),
}
copy(cdCopy.CertificateCheck.CACertificates, cd.CertificateCheck.CACertificates)
}
return cdCopy
}
@@ -157,6 +165,9 @@ func (pc *Checks) GetChecks() []Check {
if pc.Checks.ProcessCheck != nil {
checks = append(checks, pc.Checks.ProcessCheck)
}
if pc.Checks.CertificateCheck != nil {
checks = append(checks, pc.Checks.CertificateCheck)
}
return checks
}
@@ -212,6 +223,10 @@ func buildPostureCheck(postureChecksID string, name string, description string,
postureChecks.Checks.ProcessCheck = toProcessCheck(processCheck)
}
if certificateCheck := checks.CertificateCheck; certificateCheck != nil {
postureChecks.Checks.CertificateCheck = &CertificateCheck{CACertificates: certificateCheck.CaCertificates}
}
return &postureChecks, nil
}
@@ -246,6 +261,10 @@ func (pc *Checks) ToAPIResponse() *api.PostureCheck {
checks.ProcessCheck = toProcessCheckResponse(pc.Checks.ProcessCheck)
}
if pc.Checks.CertificateCheck != nil {
checks.CertificateCheck = &api.CertificateCheck{CaCertificates: pc.Checks.CertificateCheck.CACertificates}
}
return &api.PostureCheck{
Id: pc.ID,
Name: pc.Name,
+6 -3
View File
@@ -186,7 +186,7 @@ func (s *SqlStore) getPeers(ctx context.Context, accountID string) ([]nbpeer.Pee
inactivity_expiration_enabled, last_login, created_at, ephemeral, extra_dns_labels, allow_extra_dns_labels, meta_hostname,
meta_go_os, meta_kernel, meta_core, meta_platform, meta_os, meta_os_version, meta_wt_version, meta_ui_version,
meta_kernel_version, meta_network_addresses, meta_system_serial_number, meta_system_product_name, meta_system_manufacturer,
meta_environment, meta_flags, meta_files, meta_capabilities, peer_status_last_seen, peer_status_session_started_at,
meta_environment, meta_flags, meta_files, meta_certificates, meta_capabilities, peer_status_last_seen, peer_status_session_started_at,
peer_status_connected, peer_status_login_expired, peer_status_requires_approval, location_connection_ip,
location_country_code, location_city_name, location_geo_name_id, proxy_meta_embedded, proxy_meta_cluster, ipv6, meta_sync_message_version
FROM peers WHERE account_id = $1`
@@ -204,7 +204,7 @@ func (s *SqlStore) getPeers(ctx context.Context, accountID string) ([]nbpeer.Pee
peerStatusLastSeen sql.NullTime
peerStatusSessionStartedAt sql.NullInt64
peerStatusConnected, peerStatusLoginExpired, peerStatusRequiresApproval, proxyEmbedded sql.NullBool
ip, extraDNS, netAddr, env, flags, files, capabilities, connIP, ipv6 []byte
ip, extraDNS, netAddr, env, flags, files, certificates, capabilities, connIP, ipv6 []byte
metaHostname, metaGoOS, metaKernel, metaCore, metaPlatform sql.NullString
metaOS, metaOSVersion, metaWtVersion, metaUIVersion, metaKernelVersion sql.NullString
metaSystemSerialNumber, metaSystemProductName, metaSystemManufacturer sql.NullString
@@ -217,7 +217,7 @@ func (s *SqlStore) getPeers(ctx context.Context, accountID string) ([]nbpeer.Pee
&loginExpirationEnabled, &inactivityExpirationEnabled, &lastLogin, &createdAt, &ephemeral, &extraDNS,
&allowExtraDNSLabels, &metaHostname, &metaGoOS, &metaKernel, &metaCore, &metaPlatform,
&metaOS, &metaOSVersion, &metaWtVersion, &metaUIVersion, &metaKernelVersion, &netAddr,
&metaSystemSerialNumber, &metaSystemProductName, &metaSystemManufacturer, &env, &flags, &files, &capabilities,
&metaSystemSerialNumber, &metaSystemProductName, &metaSystemManufacturer, &env, &flags, &files, &certificates, &capabilities,
&peerStatusLastSeen, &peerStatusSessionStartedAt, &peerStatusConnected, &peerStatusLoginExpired,
&peerStatusRequiresApproval, &connIP, &locationCountryCode, &locationCityName, &locationGeoNameID,
&proxyEmbedded, &proxyCluster, &ipv6, &metaSyncMessageVersion)
@@ -334,6 +334,9 @@ func (s *SqlStore) getPeers(ctx context.Context, accountID string) ([]nbpeer.Pee
if files != nil {
_ = json.Unmarshal(files, &p.Meta.Files)
}
if certificates != nil {
_ = json.Unmarshal(certificates, &p.Meta.Certificates)
}
if capabilities != nil {
_ = json.Unmarshal(capabilities, &p.Meta.Capabilities)
}
@@ -54,7 +54,7 @@ func TestSqlStore_SavePeer(t *testing.T) {
numOfFields, err := populateFields.PopulateAll(reflectedMetadata)
assert.NoError(t, err)
assert.Equal(t, 33, numOfFields)
assert.Equal(t, 34, numOfFields)
// save status of non-existing peer
peer := &nbpeer.Peer{
@@ -198,6 +198,7 @@ func twinPeer(p *nbpeer.Peer) *nmdata.Peer {
KernelVersion: p.Meta.KernelVersion,
NetworkAddresses: networkAddresses,
Files: files,
Certificates: p.Meta.Certificates,
Capabilities: p.Meta.Capabilities,
SyncMessageVersion: p.Meta.SyncMessageVersion,
Flags: nmdata.Flags{
@@ -452,6 +453,9 @@ func TwinPostureChecks(pc *posture.Checks) *nmdata.PostureChecks {
}
out.Checks.ProcessCheck = &nmdata.ProcessCheck{Processes: procs}
}
if def.CertificateCheck != nil {
out.Checks.CertificateCheck = &nmdata.CertificateCheck{CACertificates: def.CertificateCheck.CACertificates}
}
return out
}
+3
View File
@@ -34,4 +34,7 @@ const (
UpdateOperationCreate UpdateOperation = "create"
UpdateOperationUpdate UpdateOperation = "update"
UpdateOperationDelete UpdateOperation = "delete"
// UpdateOperationRefresh is a periodic push that carries no change of its own, so
// it stays out of the counters that track what an administrator actually edited.
UpdateOperationRefresh UpdateOperation = "refresh"
)