Merge remote-tracking branch 'origin/main' into jnfrati/ubi-proxy

This commit is contained in:
jnfrati
2026-09-23 15:43:36 +02:00
44 changed files with 2456 additions and 231 deletions
+84 -14
View File
@@ -124,19 +124,9 @@ func newHostManager(wgInterface WGIface) (*registryConfigurator, error) {
return nil, err
}
var useGPO bool
k, err := registry.OpenKey(registry.LOCAL_MACHINE, GPODNSPolicyConfigRoot, registry.QUERY_VALUE)
if err != nil {
log.Debugf("failed to open GPO DNS policy root: %v", err)
} else {
closer(k)
useGPO = true
log.Infof("detected GPO DNS policy configuration, using policy store")
}
configurator := &registryConfigurator{
guid: guid,
gpo: useGPO,
gpo: useGPOPolicyStore(),
}
origNameservers, err := configurator.captureOriginalNameservers()
@@ -576,14 +566,22 @@ func (r *registryConfigurator) setInterfaceRegistryKeyStringValue(key, value str
return nil
}
// deleteInterfaceRegistryKeyProperty removes a value from the interface key.
// A value that is already gone, or an interface key that is, is not an error:
// the caller asked for the value not to be there, and a cleanup that runs twice
// has to reach its later steps on the second run as well.
func (r *registryConfigurator) deleteInterfaceRegistryKeyProperty(propertyKey string) error {
regKey, err := r.getInterfaceRegistryKey()
if err != nil {
switch {
case errors.Is(err, registry.ErrNotExist), errors.Is(err, syscall.ERROR_PATH_NOT_FOUND):
log.Debugf("interface key of %s does not exist, nothing to delete %s from", r.guid, propertyKey)
return nil
case err != nil:
return fmt.Errorf("get interface registry key: %w", err)
}
defer closer(regKey)
if err := regKey.DeleteValue(propertyKey); err != nil {
if err := regKey.DeleteValue(propertyKey); err != nil && !errors.Is(err, registry.ErrNotExist) {
return fmt.Errorf("delete registry key %s: %w", propertyKey, err)
}
return nil
@@ -612,7 +610,12 @@ func (r *registryConfigurator) restoreHostDNS() error {
go r.flushDNSCache()
return nil
// Last, and only on the way out, once no rule of ours is left: during a
// session the store is where the rules of this run live, and emptying it
// mid-session would have the next rule recreate it anyway. Propagated so a
// failure keeps the shutdown state for the next run to retry, rather than
// leaving the store to hold up every rule change from here on.
return removeEmptyGPOPolicyStore()
}
// removeDNSMatchPolicies deletes every NRPT rule this client may have created,
@@ -651,6 +654,73 @@ func (r *registryConfigurator) restoreUncleanShutdownDNS() error {
return r.restoreHostDNS()
}
// useGPOPolicyStore reports whether NRPT rules have to go into the group policy
// store, and clears an empty one out of the way first.
//
// The order is the point. A store left empty by an earlier run would otherwise
// decide this run too, sending its rules somewhere the resolver only reads when
// the policy engine next applies DNS client policy. Removing it before the
// choice is made leaves the local store authoritative for the whole session,
// including the first one after an upgrade.
func useGPOPolicyStore() bool {
if err := removeEmptyGPOPolicyStore(); err != nil {
// Nothing to retry against here: the worst case is the run going
// through the group policy store, which is where it would have gone
// before this check existed.
log.Warnf("%v", err)
}
k, err := registry.OpenKey(registry.LOCAL_MACHINE, GPODNSPolicyConfigRoot, registry.QUERY_VALUE)
if err != nil {
log.Debugf("failed to open GPO DNS policy root: %v", err)
return false
}
closer(k)
log.Infof("detected GPO DNS policy configuration, using policy store")
return true
}
// removeEmptyGPOPolicyStore deletes the group policy DnsPolicyConfig key once
// nothing is left in it. The key survives the deletion of the last rule it
// held, and the client treats its presence as "group policy configures the
// NRPT", so an empty one left behind keeps every later run writing rules there.
// Rules in that store reach the resolver only when the policy engine next
// applies DNS client policy, and a rule this client writes belongs to no GPO,
// so nothing schedules that application: both adding and removing a rule are
// held up by a minute or more, and for a removal that is a catch-all rule
// resolving every name over an interface that no longer exists. With the store
// absent the local one is authoritative and a change applies at once.
//
// A store that still holds rules, values or subkeys of somebody else's is left
// alone.
func removeEmptyGPOPolicyStore() error {
k, err := registry.OpenKey(registry.LOCAL_MACHINE, GPODNSPolicyConfigRoot, registry.QUERY_VALUE)
switch {
case errors.Is(err, registry.ErrNotExist), errors.Is(err, syscall.ERROR_PATH_NOT_FOUND):
return nil
case err != nil:
return fmt.Errorf("open HKEY_LOCAL_MACHINE\\%s: %w", GPODNSPolicyConfigRoot, err)
}
info, err := k.Stat()
closer(k)
if err != nil {
return fmt.Errorf("stat HKEY_LOCAL_MACHINE\\%s: %w", GPODNSPolicyConfigRoot, err)
}
if info.SubKeyCount != 0 || info.ValueCount != 0 {
return nil
}
if err := registry.DeleteKey(registry.LOCAL_MACHINE, GPODNSPolicyConfigRoot); err != nil {
return fmt.Errorf("delete empty HKEY_LOCAL_MACHINE\\%s: %w", GPODNSPolicyConfigRoot, err)
}
log.Infof("removed the empty GPO DNS policy store, leaving the local one authoritative")
return nil
}
// listNRPTRuleKeys returns the names of our NRPT rule keys under a policy store
// root. An absent root holds nothing to clean up, which is the normal state of
// the GPO store on a machine without DNS Client policy.
+129
View File
@@ -8,6 +8,8 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/sys/windows/registry"
"github.com/netbirdio/netbird/client/internal/winregistry"
)
// TestNRPTEntriesCleanupOnConfigChange tests that old NRPT entries are properly cleaned up
@@ -405,3 +407,130 @@ func TestNRPTDomainBatching(t *testing.T) {
})
}
}
// TestRemoveEmptyGPOPolicyStore verifies that cleanup takes the GPO policy
// store itself with it once our rules are gone, since the store existing keeps
// the local one from being applied, and that a store with somebody else's rule
// in it is left alone.
func TestRemoveEmptyGPOPolicyStore(t *testing.T) {
if testing.Short() {
t.Skip("skipping registry integration test in short mode")
}
t.Cleanup(func() { cleanupRegistryKeys(t) })
cleanupRegistryKeys(t)
testIP := netip.MustParseAddr("100.64.0.1")
cfg := &registryConfigurator{gpo: true}
// a store holding a rule of ours is kept, because the rule is still applied
require.NoError(t, cfg.addDNSMatchPolicy([]string{".example.com"}, testIP))
exists, err := registryKeyExists(gpoDnsPolicyConfigMatchPath + "-0")
require.NoError(t, err)
require.True(t, exists, "Should write the rule to the GPO policy store")
require.NoError(t, removeEmptyGPOPolicyStore())
exists, err = registryKeyExists(GPODNSPolicyConfigRoot)
require.NoError(t, err)
assert.True(t, exists, "Should keep a policy store that still holds a rule")
// once the rules are gone the store goes with them
require.NoError(t, cfg.removeDNSMatchPolicies())
require.NoError(t, removeEmptyGPOPolicyStore())
exists, err = registryKeyExists(GPODNSPolicyConfigRoot)
require.NoError(t, err)
assert.False(t, exists, "Should remove the GPO policy store once it is empty")
// A store is not ours to remove while somebody else has a rule in it. The
// rule is written volatile like our own: the rules above created the parent
// chain volatile, and Windows refuses a stable subkey under a volatile
// parent.
foreignRule := GPODNSPolicyConfigRoot + `\{2A3B4C5D-6E7F-4041-8283-84858687888A}`
foreignKey, _, err := winregistry.CreateVolatileKey(registry.LOCAL_MACHINE, foreignRule, registry.SET_VALUE)
require.NoError(t, err, "Should create a foreign GPO rule")
foreignKey.Close()
t.Cleanup(func() {
_ = registry.DeleteKey(registry.LOCAL_MACHINE, foreignRule)
_ = registry.DeleteKey(registry.LOCAL_MACHINE, GPODNSPolicyConfigRoot)
})
require.NoError(t, cfg.removeDNSMatchPolicies())
require.NoError(t, removeEmptyGPOPolicyStore())
exists, err = registryKeyExists(foreignRule)
require.NoError(t, err)
assert.True(t, exists, "Should not remove a foreign rule")
exists, err = registryKeyExists(GPODNSPolicyConfigRoot)
require.NoError(t, err)
assert.True(t, exists, "Should keep a policy store that still holds a foreign rule")
}
// TestDeleteInterfaceRegistryKeyPropertyTwice verifies that removing a value
// that is already gone, or one on an interface key that is, reports success.
// Teardown runs again after a failed cleanup, and the steps that follow this
// one have to be reached on that second run.
func TestDeleteInterfaceRegistryKeyPropertyTwice(t *testing.T) {
if testing.Short() {
t.Skip("skipping registry integration test in short mode")
}
testGUID := "{12345678-1234-1234-1234-123456789ABC}"
interfacePath := InterfaceConfigPath + `\` + testGUID
testKey, _, err := registry.CreateKey(registry.LOCAL_MACHINE, interfacePath, registry.SET_VALUE)
require.NoError(t, err, "Should create test interface registry key")
testKey.Close()
t.Cleanup(func() {
_ = registry.DeleteKey(registry.LOCAL_MACHINE, interfacePath)
})
cfg := &registryConfigurator{guid: testGUID}
require.NoError(t, cfg.setInterfaceRegistryKeyStringValue(interfaceConfigSearchListKey, "example.com"))
require.NoError(t, cfg.deleteInterfaceRegistryKeyProperty(interfaceConfigSearchListKey))
assert.NoError(t, cfg.deleteInterfaceRegistryKeyProperty(interfaceConfigSearchListKey),
"Should report success for a value that is already gone")
// and with the interface key itself gone, as it is once the adapter is
require.NoError(t, registry.DeleteKey(registry.LOCAL_MACHINE, interfacePath))
assert.NoError(t, cfg.deleteInterfaceRegistryKeyProperty(interfaceConfigSearchListKey),
"Should report success when the interface key does not exist")
}
// TestUseGPOPolicyStoreClearsEmptyStore verifies that the store is cleared
// before it is consulted, so an empty one left by an earlier run does not send
// this run's rules to the group policy store. A store somebody else has a rule
// in still decides where the rules go.
func TestUseGPOPolicyStoreClearsEmptyStore(t *testing.T) {
if testing.Short() {
t.Skip("skipping registry integration test in short mode")
}
t.Cleanup(func() { cleanupRegistryKeys(t) })
cleanupRegistryKeys(t)
// the leftover an earlier run used to keep, which the client read as
// "group policy configures the NRPT" for every run after it
emptyStore, _, err := winregistry.CreateVolatileKey(registry.LOCAL_MACHINE, GPODNSPolicyConfigRoot, registry.SET_VALUE)
require.NoError(t, err, "Should create the GPO policy store")
emptyStore.Close()
assert.False(t, useGPOPolicyStore(), "An empty store should not decide where the rules go")
exists, err := registryKeyExists(GPODNSPolicyConfigRoot)
require.NoError(t, err)
assert.False(t, exists, "Should clear the empty store before consulting it")
foreignRule := GPODNSPolicyConfigRoot + `\{2A3B4C5D-6E7F-4041-8283-84858687888A}`
foreignKey, _, err := winregistry.CreateVolatileKey(registry.LOCAL_MACHINE, foreignRule, registry.SET_VALUE)
require.NoError(t, err, "Should create a foreign GPO rule")
foreignKey.Close()
t.Cleanup(func() {
_ = registry.DeleteKey(registry.LOCAL_MACHINE, foreignRule)
_ = registry.DeleteKey(registry.LOCAL_MACHINE, GPODNSPolicyConfigRoot)
})
assert.True(t, useGPOPolicyStore(), "A store holding a rule should decide where the rules go")
exists, err = registryKeyExists(GPODNSPolicyConfigRoot)
require.NoError(t, err)
assert.True(t, exists, "Should keep a store that holds a rule")
}
+5 -1
View File
@@ -1061,7 +1061,11 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
// back to empty if the FQDN doesn't have the expected shape.
dnsName = extractDNSDomainFromFQDN(pc.GetFqdn())
}
result, err := nbnetworkmap.EnvelopeToNetworkMap(e.ctx, envelope, localKey, dnsName)
// With the firewall disabled there is no ACL manager to program, so
// RoutesFirewallRules would be built and then dropped. On a peer that
// routes many network resources that is the single most expensive
// step of the sync.
result, err := nbnetworkmap.EnvelopeToNetworkMap(e.ctx, envelope, localKey, dnsName, e.config.DisableFirewall)
if err != nil {
return fmt.Errorf("decode network map envelope: %w", err)
}
+25 -9
View File
@@ -135,9 +135,10 @@ type Conn struct {
// used to store the remote Rosenpass key for Relayed connection in case of connection update from ice
rosenpassRemoteKey []byte
wgProxyICE wgproxy.Proxy
wgProxyRelay wgproxy.Proxy
handshaker *Handshaker
wgProxyICE wgproxy.Proxy
wgProxyRelay wgproxy.Proxy
relayedConnRef *relayClient.Conn
handshaker *Handshaker
guard *guard.Guard
wg sync.WaitGroup
@@ -560,7 +561,7 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
conn.mu.Lock()
defer conn.mu.Unlock()
if conn.ctx.Err() != nil {
if conn.ctx.Err() != nil || rci.relayedConn.Context().Err() != nil {
if err := rci.relayedConn.Close(); err != nil {
conn.Log.Warnf("failed to close unnecessary relayed connection: %v", err)
}
@@ -575,7 +576,9 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
conn.Log.Errorf("failed to add relayed net.Conn to local proxy: %v", err)
return
}
wgProxy.SetDisconnectListener(conn.onRelayDisconnected)
wgProxy.SetDisconnectListener(func() {
conn.onRelayDisconnected(rci.relayedConn)
})
conn.dumpState.NewLocalProxy()
@@ -583,7 +586,7 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
if conn.isICEActive() {
conn.Log.Debugf("do not switch to relay because current priority is: %s", conn.currentConnPriority.String())
conn.setRelayedProxy(wgProxy)
conn.setRelayedProxy(wgProxy, rci.relayedConn)
conn.statusRelay.SetConnected()
conn.updateRelayStatus(rci.relayedConn.RemoteAddr().String(), rci.rosenpassPubKey, time.Now())
return
@@ -614,15 +617,26 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
conn.rosenpassRemoteKey = rci.rosenpassPubKey
conn.currentConnPriority = conntype.Relay
conn.statusRelay.SetConnected()
conn.setRelayedProxy(wgProxy)
conn.setRelayedProxy(wgProxy, rci.relayedConn)
conn.updateRelayStatus(rci.relayedConn.RemoteAddr().String(), rci.rosenpassPubKey, updateTime)
conn.Log.Infof("start to communicate with peer via relay")
conn.doOnConnected(rci.rosenpassPubKey, rci.rosenpassAddr, updateTime)
}
func (conn *Conn) onRelayDisconnected() {
// onRelayDisconnected reports the teardown of a relayed connection. relayedConn
// names the connection the signal belongs to, so a signal that arrives after
// its connection was replaced is ignored instead of tearing down its successor.
// A nil relayedConn means the caller does not track generations and the current
// connection is always torn down.
func (conn *Conn) onRelayDisconnected(relayedConn *relayClient.Conn) {
conn.mu.Lock()
defer conn.mu.Unlock()
if relayedConn != nil && conn.relayedConnRef != relayedConn {
conn.Log.Debugf("ignoring relay disconnect of a superseded connection")
return
}
conn.handleRelayDisconnectedLocked()
}
@@ -646,6 +660,7 @@ func (conn *Conn) handleRelayDisconnectedLocked() {
_ = conn.wgProxyRelay.CloseConn()
conn.wgProxyRelay = nil
}
conn.relayedConnRef = nil
changed := conn.statusRelay.Get() != worker.StatusDisconnected
if changed {
@@ -930,13 +945,14 @@ func (conn *Conn) logTraceConnState() {
}
}
func (conn *Conn) setRelayedProxy(proxy wgproxy.Proxy) {
func (conn *Conn) setRelayedProxy(proxy wgproxy.Proxy, relayedConn *relayClient.Conn) {
if conn.wgProxyRelay != nil {
if err := conn.wgProxyRelay.CloseConn(); err != nil {
conn.Log.Warnf("failed to close deprecated wg proxy conn: %v", err)
}
}
conn.wgProxyRelay = proxy
conn.relayedConnRef = relayedConn
}
// onWGHandshakeSuccess is called when the first WireGuard handshake is detected
+13 -14
View File
@@ -3,7 +3,6 @@ package peer
import (
"context"
"errors"
"net"
"net/netip"
"sync"
"sync/atomic"
@@ -14,7 +13,7 @@ import (
)
type RelayConnInfo struct {
relayedConn net.Conn
relayedConn *relayClient.Conn
rosenpassPubKey []byte
rosenpassAddr string
}
@@ -27,7 +26,7 @@ type WorkerRelay struct {
conn *Conn
relayManager *relayClient.Manager
relayedConn net.Conn
relayedConn *relayClient.Conn
relayLock sync.Mutex
relaySupportedOnRemotePeer atomic.Bool
@@ -80,12 +79,7 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
w.relayedConn = relayedConn
w.relayLock.Unlock()
err = w.relayManager.AddCloseListener(srv, w.onRelayClientDisconnected)
if err != nil {
log.Errorf("failed to add close listener: %s", err)
_ = relayedConn.Close()
return
}
go w.watchRelayedConn(relayedConn)
w.log.Debugf("peer conn opened via Relay: %s", srv)
go w.conn.onRelayConnectionIsReady(RelayConnInfo{
@@ -109,12 +103,15 @@ func (w *WorkerRelay) RelayIsSupportedLocally() bool {
func (w *WorkerRelay) CloseConn() {
w.relayLock.Lock()
defer w.relayLock.Unlock()
if w.relayedConn == nil {
conn := w.relayedConn
w.relayedConn = nil
w.relayLock.Unlock()
if conn == nil {
return
}
if err := w.relayedConn.Close(); err != nil {
if err := conn.Close(); err != nil {
w.log.Warnf("failed to close relay connection: %v", err)
}
}
@@ -133,6 +130,8 @@ func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress st
return remoteRelayAddress
}
func (w *WorkerRelay) onRelayClientDisconnected() {
go w.conn.onRelayDisconnected()
func (w *WorkerRelay) watchRelayedConn(relayedConn *relayClient.Conn) {
<-relayedConn.Context().Done()
w.conn.onRelayDisconnected(relayedConn)
}
@@ -245,7 +245,7 @@ func computeMode(t *testing.T, ctx context.Context, mode Mode, nmData *networkma
peerGroups := maps.Keys(nmData.GetPeerGroups(peerID))
resp := mgmtgrpc.ToComponentSyncResponse(ctx, nil, nil, nil, peer, nil, nil, components, nil,
dnsDomain, nil, nmData.AccountSettings, nil, peerGroups, dnsFwdPort)
res, err := networkmap.EnvelopeToNetworkMap(ctx, resp.NetworkMapEnvelope, peer.Key, dnsDomain)
res, err := networkmap.EnvelopeToNetworkMap(ctx, resp.NetworkMapEnvelope, peer.Key, dnsDomain, false)
require.NoError(t, err, "expand envelope")
return res.NetworkMap
default:
+13 -2
View File
@@ -102,7 +102,8 @@ type ProxyServiceServer struct {
mu sync.RWMutex
// Manager for reverse proxy operations
serviceManager rpservice.Manager
serviceManager rpservice.Manager
credentialLimits credentialVerificationLimiter
// agentNetworkSynth produces synthesised reverse-proxy services from
// Agent Network state. Optional — when nil the snapshot path only ships
// persisted services.
@@ -242,9 +243,10 @@ func (s *ProxyServiceServer) cleanupStaleProxies(ctx context.Context) {
}
}
// Close stops background goroutines.
// Close stops background goroutines and releases credential verification state.
func (s *ProxyServiceServer) Close() {
s.cancel()
s.credentialLimits.close()
}
// SetServiceManager sets the service manager. Must be called before serving.
@@ -1223,6 +1225,7 @@ func shallowCloneMapping(m *proto.ProxyMapping) *proto.ProxyMapping {
}
}
// Authenticate verifies service credentials and issues a session token.
func (s *ProxyServiceServer) Authenticate(ctx context.Context, req *proto.AuthenticateRequest) (*proto.AuthenticateResponse, error) {
if err := enforceAccountScope(ctx, req.GetAccountId()); err != nil {
return nil, err
@@ -1234,6 +1237,14 @@ func (s *ProxyServiceServer) Authenticate(ctx context.Context, req *proto.Authen
return nil, status.Errorf(codes.FailedPrecondition, "get service from store: %v", err)
}
switch req.GetRequest().(type) {
case *proto.AuthenticateRequest_Pin, *proto.AuthenticateRequest_Password:
key := credentialVerificationKey{accountID: credentialAccountID(service.AccountID), serviceID: credentialServiceID(service.ID)}
if err := s.credentialLimits.allow(key); err != nil {
return nil, err
}
}
authenticated, userId, method := s.authenticateRequest(ctx, req, service)
// Non-OIDC schemes (PIN/Password/Header) authenticate against per-service
@@ -0,0 +1,101 @@
package grpc
import (
"sync"
"time"
"golang.org/x/time/rate"
"google.golang.org/genproto/googleapis/rpc/errdetails"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/durationpb"
)
const (
credentialVerificationInterval = 6 * time.Second
credentialVerificationBurst = 5
credentialVerificationMaxServices = 4096
credentialVerificationIdleTimeout = 15 * time.Minute
credentialVerificationCleanupInterval = time.Minute
)
type credentialAccountID string
type credentialServiceID string
type credentialVerificationKey struct {
accountID credentialAccountID
serviceID credentialServiceID
}
type credentialVerificationBudget struct {
limiter *rate.Limiter
lastUsed time.Time
}
// The zero value is ready to use. Budgets are local to this Management process;
// proxy replicas reaching this process share a service's verification budget.
type credentialVerificationLimiter struct {
mu sync.Mutex
now func() time.Time
services map[credentialVerificationKey]*credentialVerificationBudget
nextCleanup time.Time
closed bool
}
func (l *credentialVerificationLimiter) allow(key credentialVerificationKey) error {
l.mu.Lock()
defer l.mu.Unlock()
if l.closed {
return status.Error(codes.Unavailable, "credential verification is closed")
}
now := time.Now()
if l.now != nil {
now = l.now()
}
l.cleanup(now)
budget := l.services[key]
if budget == nil {
if len(l.services) >= credentialVerificationMaxServices {
return credentialVerificationThrottled(credentialVerificationCleanupInterval)
}
if l.services == nil {
l.services = make(map[credentialVerificationKey]*credentialVerificationBudget)
}
budget = &credentialVerificationBudget{limiter: rate.NewLimiter(rate.Every(credentialVerificationInterval), credentialVerificationBurst)}
l.services[key] = budget
}
budget.lastUsed = now
if budget.limiter.AllowN(now, 1) {
return nil
}
delay := max(time.Nanosecond, time.Duration((1-budget.limiter.TokensAt(now))*float64(credentialVerificationInterval)))
return credentialVerificationThrottled(delay)
}
func (l *credentialVerificationLimiter) cleanup(now time.Time) {
if now.Before(l.nextCleanup) {
return
}
l.nextCleanup = now.Add(credentialVerificationCleanupInterval)
for key, budget := range l.services {
if now.Sub(budget.lastUsed) >= credentialVerificationIdleTimeout {
delete(l.services, key)
}
}
}
func (l *credentialVerificationLimiter) close() {
l.mu.Lock()
defer l.mu.Unlock()
l.closed = true
l.services = nil
}
func credentialVerificationThrottled(delay time.Duration) error {
s := status.New(codes.ResourceExhausted, "too many credential verification attempts")
withRetry, err := s.WithDetails(&errdetails.RetryInfo{RetryDelay: durationpb.New(delay)})
if err != nil {
return s.Err()
}
return withRetry.Err()
}
@@ -0,0 +1,79 @@
package grpc
import (
"strconv"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/genproto/googleapis/rpc/errdetails"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
func TestCredentialVerificationRefillAndIsolation(t *testing.T) {
now := time.Now()
l := credentialVerificationLimiter{now: func() time.Time { return now }}
key := credentialVerificationKey{accountID: "account", serviceID: "service"}
for range credentialVerificationBurst {
require.NoError(t, l.allow(key))
}
err := l.allow(key)
require.Equal(t, codes.ResourceExhausted, status.Code(err), "the burst must be bounded")
now = now.Add(3 * time.Second)
err = l.allow(key)
require.Equal(t, codes.ResourceExhausted, status.Code(err), "a partially refilled token must not permit a check")
details := status.Convert(err).Details()
require.Len(t, details, 1, "throttling must provide RetryInfo")
retry, ok := details[0].(*errdetails.RetryInfo)
require.True(t, ok, "retry details must use the standard message")
assert.Equal(t, 3*time.Second, retry.RetryDelay.AsDuration(), "retry hint must reflect time until the next check")
now = now.Add(3 * time.Second)
require.NoError(t, l.allow(key))
assert.Equal(t, codes.ResourceExhausted, status.Code(l.allow(key)), "only one check must refill every six seconds")
require.NoError(t, l.allow(credentialVerificationKey{accountID: "other-account", serviceID: key.serviceID}))
require.NoError(t, l.allow(credentialVerificationKey{accountID: key.accountID, serviceID: "other-service"}))
}
func TestCredentialVerificationCapacityAndExpiry(t *testing.T) {
now := time.Now()
l := credentialVerificationLimiter{now: func() time.Time { return now }}
for i := range credentialVerificationMaxServices {
require.NoError(t, l.allow(credentialVerificationKey{accountID: "account", serviceID: credentialServiceID(strconv.Itoa(i))}))
}
key := credentialVerificationKey{accountID: "account", serviceID: "new-service"}
assert.Equal(t, codes.ResourceExhausted, status.Code(l.allow(key)), "capacity exhaustion must deny new checks")
now = now.Add(credentialVerificationIdleTimeout)
for range credentialVerificationBurst {
require.NoError(t, l.allow(key))
}
assert.Equal(t, codes.ResourceExhausted, status.Code(l.allow(key)), "expiry must retain the normal burst bound")
}
func TestCredentialVerificationConcurrentChecksAndClose(t *testing.T) {
var l credentialVerificationLimiter
key := credentialVerificationKey{accountID: "account", serviceID: "service"}
var admitted atomic.Int32
var wg sync.WaitGroup
for range 100 {
wg.Go(func() {
if err := l.allow(key); err == nil {
admitted.Add(1)
} else {
assert.Equal(t, codes.ResourceExhausted, status.Code(err), "excess checks must be throttled")
}
})
}
wg.Wait()
assert.EqualValues(t, credentialVerificationBurst, admitted.Load(), "concurrent checks must share the burst")
for range 10 {
wg.Go(l.close)
wg.Go(func() { assert.Error(t, l.allow(key)) })
}
wg.Wait()
assert.Empty(t, l.services, "closing must release retained budgets")
assert.Equal(t, codes.Unavailable, status.Code(l.allow(key)), "checks after close must fail closed")
}
@@ -0,0 +1,18 @@
# Reverse proxy credential verification
The `ProxyService.Authenticate` RPC limits PIN and password checks before
verifying their Argon2 hashes. Both methods share one budget per account and
service: a burst of five checks, replenishing one check every six seconds
(ten per minute). Successful and failed checks consume the budget. Account
scope and service lookup run before the limiter.
Excess checks receive gRPC `ResourceExhausted` with a standard `RetryInfo` delay.
Updated proxies translate it to HTTP 429 and `Retry-After`. Older proxies show
an authentication-service error but cannot bypass the Management limit.
Budgets are held in memory per Management process and reset on restart. Proxy
replicas reaching the same Management process share its budgets. Multiple
Management processes have independent budgets; this is not a cluster-wide
limit. At most 4,096 service budgets are retained, with idle entries expiring
after fifteen minutes. Capacity exhaustion denies new checks until entries
expire. Closing the server releases the retained state.
@@ -0,0 +1,131 @@
package grpc_test
import (
"context"
"net"
"net/netip"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/genproto/googleapis/rpc/errdetails"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/peer"
"google.golang.org/grpc/status"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
servicemanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service/manager"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/sessionkey"
nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
"github.com/netbirdio/netbird/management/server/store"
"github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/shared/management/proto"
)
func credentialServer(t *testing.T) (*nbgrpc.ProxyServiceServer, context.Context, grpc.UnaryServerInterceptor) {
t.Helper()
ctx := context.Background()
s, err := store.NewStore(ctx, types.SqliteStoreEngine, t.TempDir(), nil, false)
require.NoError(t, err)
t.Cleanup(func() { assert.NoError(t, s.Close(ctx)) })
require.NoError(t, s.SaveAccount(ctx, &types.Account{Id: "account"}))
keys, err := sessionkey.GenerateKeyPair()
require.NoError(t, err)
for _, id := range []string{"service", "other-service"} {
svc := &service.Service{
ID: id, AccountID: "account", Name: id, Domain: id + ".example.com",
Enabled: true, SessionPrivateKey: keys.PrivateKey, SessionPublicKey: keys.PublicKey,
Auth: service.AuthConfig{
PinAuth: &service.PINAuthConfig{Enabled: true, Pin: "842716"},
PasswordAuth: &service.PasswordAuthConfig{Enabled: true, Password: "test-password"},
},
}
require.NoError(t, svc.Auth.HashSecrets())
require.NoError(t, s.CreateService(ctx, svc))
}
account := "account"
token, err := types.CreateNewProxyAccessToken("test proxy", time.Hour, &account, "admin")
require.NoError(t, err)
require.NoError(t, s.SaveProxyAccessToken(ctx, &token.ProxyAccessToken))
ctx = metadata.NewIncomingContext(ctx, metadata.Pairs("authorization", "Bearer "+string(token.PlainToken)))
ctx = peer.NewContext(ctx, &peer.Peer{Addr: net.TCPAddrFromAddrPort(netip.MustParseAddrPort("192.0.2.1:443"))})
server := nbgrpc.NewProxyServiceServer(nil, nil, nil, nbgrpc.ProxyOIDCConfig{}, nil, nil, nil, nil, nil)
t.Cleanup(server.Close)
server.SetServiceManager(servicemanager.NewManager(s, nil, nil, nil, nil, nil))
interceptor, _, closeInterceptor := nbgrpc.NewProxyAuthInterceptors(s)
t.Cleanup(closeInterceptor)
return server, ctx, interceptor
}
func TestAuthenticateCredentialRateLimit(t *testing.T) {
server, ctx, interceptor := credentialServer(t)
authenticate := func(req *proto.AuthenticateRequest) (*proto.AuthenticateResponse, error) {
response, err := interceptor(ctx, req, &grpc.UnaryServerInfo{FullMethod: "/management.ProxyService/Authenticate"}, func(ctx context.Context, req any) (any, error) {
return server.Authenticate(ctx, req.(*proto.AuthenticateRequest))
})
if err != nil {
return nil, err
}
return response.(*proto.AuthenticateResponse), nil
}
for i := range 5 {
req := &proto.AuthenticateRequest{AccountId: "account", Id: "service"}
if i%2 == 0 {
req.Request = &proto.AuthenticateRequest_Pin{Pin: &proto.PinRequest{Pin: "000000"}}
} else {
req.Request = &proto.AuthenticateRequest_Password{Password: &proto.PasswordRequest{Password: "wrong-password"}}
}
resp, err := authenticate(req)
require.NoError(t, err)
assert.False(t, resp.GetSuccess(), "incorrect PINs and passwords must be denied")
assert.Empty(t, resp.GetSessionToken(), "incorrect credentials must not issue a token")
}
req := &proto.AuthenticateRequest{AccountId: "account", Id: "service", Request: &proto.AuthenticateRequest_Pin{Pin: &proto.PinRequest{Pin: "842716"}}}
resp, err := authenticate(req)
assert.Nil(t, resp, "a throttled verification must not return a session")
require.Equal(t, codes.ResourceExhausted, status.Code(err), "PIN and password checks must share a service budget even with a valid proxy token")
details := status.Convert(err).Details()
require.Len(t, details, 1, "throttled responses must include a retry hint")
retry, ok := details[0].(*errdetails.RetryInfo)
require.True(t, ok, "the hint must use the standard RetryInfo message")
assert.Positive(t, retry.RetryDelay.AsDuration(), "the retry delay must be positive")
assert.LessOrEqual(t, retry.RetryDelay.AsDuration(), 6*time.Second, "the service must replenish one verification every six seconds")
req.AccountId = "another-account"
_, err = authenticate(req)
assert.Equal(t, codes.PermissionDenied, status.Code(err), "account scope must still be enforced before throttling")
req.AccountId = "account"
req.Id = "other-service"
resp, err = authenticate(req)
require.NoError(t, err)
assert.True(t, resp.GetSuccess(), "one service's throttle must not block another service")
assert.NotEmpty(t, resp.GetSessionToken(), "valid credentials on another service must issue a session")
}
func TestAuthenticateCredentialConcurrentLimit(t *testing.T) {
server, _, _ := credentialServer(t)
req := &proto.AuthenticateRequest{AccountId: "account", Id: "service", Request: &proto.AuthenticateRequest_Pin{Pin: &proto.PinRequest{Pin: "000000"}}}
var checked, throttled atomic.Int32
var wg sync.WaitGroup
for range 20 {
wg.Go(func() {
resp, err := server.Authenticate(context.Background(), req)
switch status.Code(err) {
case codes.OK:
checked.Add(1)
assert.False(t, resp.GetSuccess(), "incorrect credentials must be denied")
case codes.ResourceExhausted:
throttled.Add(1)
default:
assert.NoError(t, err)
}
})
}
wg.Wait()
assert.EqualValues(t, 5, checked.Load(), "only the burst budget may reach concurrent credential verification")
assert.EqualValues(t, 15, throttled.Load(), "excess concurrent checks must be throttled")
}
@@ -148,13 +148,10 @@ func (h *handler) updateGroup(w http.ResponseWriter, r *http.Request) {
peers = *req.Peers
}
resources := make([]types.Resource, 0)
if req.Resources != nil {
for _, res := range *req.Resources {
resource := types.Resource{}
resource.FromAPIRequest(&res)
resources = append(resources, resource)
}
resources, err := resourcesFromAPIRequest(req.Resources)
if err != nil {
util.WriteError(r.Context(), err, w)
return
}
group := types.Group{
@@ -210,13 +207,10 @@ func (h *handler) createGroup(w http.ResponseWriter, r *http.Request) {
peers = *req.Peers
}
resources := make([]types.Resource, 0)
if req.Resources != nil {
for _, res := range *req.Resources {
resource := types.Resource{}
resource.FromAPIRequest(&res)
resources = append(resources, resource)
}
resources, err := resourcesFromAPIRequest(req.Resources)
if err != nil {
util.WriteError(r.Context(), err, w)
return
}
group := types.Group{
@@ -335,11 +329,30 @@ func toGroupResponse(peers []*nbpeer.Peer, group *types.Group) *api.Group {
gr.PeersCount = len(gr.Peers)
for _, res := range group.Resources {
resResp := res.ToAPIResponse()
gr.Resources = append(gr.Resources, *resResp)
if resResp := res.ToAPIResponse(); resResp != nil {
gr.Resources = append(gr.Resources, *resResp)
}
}
gr.ResourcesCount = len(gr.Resources)
return &gr
}
func resourcesFromAPIRequest(req *[]api.Resource) ([]types.Resource, error) {
resources := make([]types.Resource, 0)
if req == nil {
return resources, nil
}
for _, res := range *req {
if res.Id == "" || !types.ResourceType(res.Type).Valid() {
return nil, status.Errorf(status.InvalidArgument, "resource id shouldn't be empty and type must be one of: peer, domain, host, subnet")
}
resource := types.Resource{}
resource.FromAPIRequest(&res)
resources = append(resources, resource)
}
return resources, nil
}
@@ -8,8 +8,8 @@ import (
"fmt"
"io"
"net/http"
"net/netip"
"net/http/httptest"
"net/netip"
"strings"
"testing"
@@ -208,6 +208,33 @@ func TestWriteGroup(t *testing.T) {
expectedStatus: http.StatusUnprocessableEntity,
expectedBody: false,
},
{
name: "Write Group POST Empty Resource",
requestType: http.MethodPost,
requestPath: "/api/groups",
requestBody: bytes.NewBuffer(
[]byte(`{"name":"With Resource","resources":[{}]}`)),
expectedStatus: http.StatusUnprocessableEntity,
expectedBody: false,
},
{
name: "Write Group PUT Empty Resource",
requestType: http.MethodPut,
requestPath: "/api/groups/id-existed",
requestBody: bytes.NewBuffer(
[]byte(`{"name":"With Resource","resources":[{"id":"","type":"host"}]}`)),
expectedStatus: http.StatusUnprocessableEntity,
expectedBody: false,
},
{
name: "Write Group POST Unknown Resource Type",
requestType: http.MethodPost,
requestPath: "/api/groups",
requestBody: bytes.NewBuffer(
[]byte(`{"name":"With Resource","resources":[{"id":"res-1","type":"banana"}]}`)),
expectedStatus: http.StatusUnprocessableEntity,
expectedBody: false,
},
{
name: "Write Group PUT OK",
requestType: http.MethodPut,
@@ -376,6 +403,20 @@ func TestGetAllGroups(t *testing.T) {
}
}
func TestToGroupResponseSkipsEmptyResource(t *testing.T) {
group := &types.Group{
ID: "id-resources",
Name: "Resources",
Issued: types.GroupIssuedAPI,
Resources: []types.Resource{{}, {ID: "res-1", Type: types.ResourceTypeHost}},
}
got := toGroupResponse(nil, group)
assert.Equal(t, 1, got.ResourcesCount)
assert.Equal(t, []api.Resource{{Id: "res-1", Type: api.ResourceType(types.ResourceTypeHost)}}, got.Resources)
}
func TestDeleteGroup(t *testing.T) {
tt := []struct {
name string
@@ -32,7 +32,7 @@ type NetworkResource struct {
ID string `gorm:"primaryKey"`
NetworkID string `gorm:"index"`
AccountID string `gorm:"index"`
PublicID string `json:"-"`
PublicID string `json:"-" gorm:"index"`
Name string
Description string
Type NetworkResourceType
+68
View File
@@ -58,6 +58,7 @@ const (
keyQueryCondition = "key = ?"
mysqlKeyQueryCondition = "`key` = ?"
accountAndIDQueryCondition = "account_id = ? and id = ?"
accountAndAnyIDQueryCondition = "account_id = ? and (id = ? or public_id = ?)"
accountAndPeerIDQueryCondition = "account_id = ? and peer_id = ?"
accountAndIDsQueryCondition = "account_id = ? AND id IN ?"
accountIDCondition = "account_id = ?"
@@ -4063,6 +4064,30 @@ func (s *SqlStore) GetPolicyByID(ctx context.Context, lockStrength LockingStreng
return policy, nil
}
// GetPolicyByIDOrPublicID retrieves a policy by either its ID or its PublicID. Peers report
// whichever of the two the network map they were served carries, so callers resolving a
// peer-reported reference cannot know upfront which namespace it belongs to.
func (s *SqlStore) GetPolicyByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID, policyID string) (*types.Policy, error) {
tx := s.db
if lockStrength != LockingStrengthNone {
tx = tx.Clauses(clause.Locking{Strength: string(lockStrength)})
}
var policy *types.Policy
result := tx.Preload(clause.Associations).
Take(&policy, accountAndAnyIDQueryCondition, accountID, policyID, policyID)
if err := result.Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, status.NewPolicyNotFoundError(policyID)
}
log.WithContext(ctx).Errorf("failed to get policy from store: %s", err)
return nil, status.Errorf(status.Internal, "failed to get policy from store")
}
return policy, nil
}
func (s *SqlStore) CreatePolicy(ctx context.Context, policy *types.Policy) error {
result := s.db.Create(policy)
if result.Error != nil {
@@ -4248,6 +4273,27 @@ func (s *SqlStore) GetRouteByID(ctx context.Context, lockStrength LockingStrengt
return route, nil
}
// GetRouteByIDOrPublicID retrieves a route by either its ID or its PublicID. See
// GetPolicyByIDOrPublicID for why peer-reported references need both.
func (s *SqlStore) GetRouteByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID string, routeID string) (*route.Route, error) {
tx := s.db
if lockStrength != LockingStrengthNone {
tx = tx.Clauses(clause.Locking{Strength: string(lockStrength)})
}
var route *route.Route
result := tx.Take(&route, accountAndAnyIDQueryCondition, accountID, routeID, routeID)
if err := result.Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, status.NewRouteNotFoundError(routeID)
}
log.WithContext(ctx).Errorf("failed to get route from the store: %s", err)
return nil, status.Errorf(status.Internal, "failed to get route from store")
}
return route, nil
}
// SaveRoute saves a route to the database.
func (s *SqlStore) SaveRoute(ctx context.Context, route *route.Route) error {
result := s.db.Save(route)
@@ -4642,6 +4688,28 @@ func (s *SqlStore) GetNetworkResourceByID(ctx context.Context, lockStrength Lock
return netResources, nil
}
// GetNetworkResourceByIDOrPublicID retrieves a network resource by either its ID or its
// PublicID. See GetPolicyByIDOrPublicID for why peer-reported references need both.
func (s *SqlStore) GetNetworkResourceByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID, resourceID string) (*resourceTypes.NetworkResource, error) {
tx := s.db
if lockStrength != LockingStrengthNone {
tx = tx.Clauses(clause.Locking{Strength: string(lockStrength)})
}
var netResources *resourceTypes.NetworkResource
result := tx.
Take(&netResources, accountAndAnyIDQueryCondition, accountID, resourceID, resourceID)
if result.Error != nil {
if errors.Is(result.Error, gorm.ErrRecordNotFound) {
return nil, status.NewNetworkResourceNotFoundError(resourceID)
}
log.WithContext(ctx).Errorf("failed to get network resource from store: %v", result.Error)
return nil, status.Errorf(status.Internal, "failed to get network resource from store")
}
return netResources, nil
}
func (s *SqlStore) GetNetworkResourceByName(ctx context.Context, lockStrength LockingStrength, accountID, resourceName string) (*resourceTypes.NetworkResource, error) {
tx := s.db
if lockStrength != LockingStrengthNone {
+78
View File
@@ -1972,6 +1972,32 @@ func TestSqlStore_GetPolicyByID(t *testing.T) {
}
}
func TestSqlStore_GetPolicyByIDOrPublicID(t *testing.T) {
store, cleanup, err := NewTestStoreFromSQL(context.Background(), "../testdata/store.sql", t.TempDir())
t.Cleanup(cleanup)
require.NoError(t, err)
accountID := "bf1c8084-ba50-4ce7-9439-34653001fc3b"
policyID := "cs1tnh0hhcjnqoiuebf0"
policy, err := store.GetPolicyByID(context.Background(), LockingStrengthNone, accountID, policyID)
require.NoError(t, err)
require.NotEmpty(t, policy.PublicID)
for _, id := range []string{policyID, policy.PublicID} {
policy, err := store.GetPolicyByIDOrPublicID(context.Background(), LockingStrengthNone, accountID, id)
require.NoError(t, err)
require.Equal(t, policyID, policy.ID)
}
policy, err = store.GetPolicyByIDOrPublicID(context.Background(), LockingStrengthNone, accountID, "non-existing")
require.Error(t, err)
sErr, ok := status.FromError(err)
require.True(t, ok)
require.Equal(t, sErr.Type(), status.NotFound)
require.Nil(t, policy)
}
func TestSqlStore_CreatePolicy(t *testing.T) {
store, cleanup, err := NewTestStoreFromSQL(context.Background(), "../testdata/store.sql", t.TempDir())
t.Cleanup(cleanup)
@@ -2631,6 +2657,32 @@ func TestSqlStore_GetNetworkResourceByID(t *testing.T) {
}
}
func TestSqlStore_GetNetworkResourceByIDOrPublicID(t *testing.T) {
store, cleanup, err := NewTestStoreFromSQL(context.Background(), "../testdata/store.sql", t.TempDir())
t.Cleanup(cleanup)
require.NoError(t, err)
accountID := "bf1c8084-ba50-4ce7-9439-34653001fc3b"
netResourceID := "ctc4nci7qv9061u6ilfg"
netResource, err := store.GetNetworkResourceByID(context.Background(), LockingStrengthNone, accountID, netResourceID)
require.NoError(t, err)
require.NotEmpty(t, netResource.PublicID)
for _, id := range []string{netResourceID, netResource.PublicID} {
netResource, err := store.GetNetworkResourceByIDOrPublicID(context.Background(), LockingStrengthNone, accountID, id)
require.NoError(t, err)
require.Equal(t, netResourceID, netResource.ID)
}
netResource, err = store.GetNetworkResourceByIDOrPublicID(context.Background(), LockingStrengthNone, accountID, "non-existing")
require.Error(t, err)
sErr, ok := status.FromError(err)
require.True(t, ok)
require.Equal(t, sErr.Type(), status.NotFound)
require.Nil(t, netResource)
}
func TestSqlStore_SaveNetworkResource(t *testing.T) {
store, cleanup, err := NewTestStoreFromSQL(context.Background(), "../testdata/store.sql", t.TempDir())
t.Cleanup(cleanup)
@@ -3756,6 +3808,32 @@ func TestSqlStore_GetRouteByID(t *testing.T) {
}
}
func TestSqlStore_GetRouteByIDOrPublicID(t *testing.T) {
store, cleanup, err := NewTestStoreFromSQL(context.Background(), "../testdata/extended-store.sql", t.TempDir())
t.Cleanup(cleanup)
require.NoError(t, err)
accountID := "bf1c8084-ba50-4ce7-9439-34653001fc3b"
routeID := "ct03t427qv97vmtmglog"
route, err := store.GetRouteByID(context.Background(), LockingStrengthNone, accountID, routeID)
require.NoError(t, err)
require.NotEmpty(t, route.PublicID)
for _, id := range []string{routeID, route.PublicID} {
route, err := store.GetRouteByIDOrPublicID(context.Background(), LockingStrengthNone, accountID, id)
require.NoError(t, err)
require.Equal(t, routeID, string(route.ID))
}
route, err = store.GetRouteByIDOrPublicID(context.Background(), LockingStrengthNone, accountID, "non-existing")
require.Error(t, err)
sErr, ok := status.FromError(err)
require.True(t, ok)
require.Equal(t, sErr.Type(), status.NotFound)
require.Nil(t, route)
}
func TestSqlStore_SaveRoute(t *testing.T) {
store, cleanup, err := NewTestStoreFromSQL(context.Background(), "../testdata/extended-store.sql", t.TempDir())
t.Cleanup(cleanup)
+3
View File
@@ -138,6 +138,7 @@ type Store interface {
GetAccountPolicies(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*types.Policy, error)
GetPolicyByID(ctx context.Context, lockStrength LockingStrength, accountID, policyID string) (*types.Policy, error)
GetPolicyByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID, policyID string) (*types.Policy, error)
CreatePolicy(ctx context.Context, policy *types.Policy) error
SavePolicy(ctx context.Context, policy *types.Policy) error
DeletePolicy(ctx context.Context, accountID, policyID string) error
@@ -208,6 +209,7 @@ type Store interface {
GetAccountRoutes(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*route.Route, error)
GetRouteByID(ctx context.Context, lockStrength LockingStrength, accountID, routeID string) (*route.Route, error)
GetRouteByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID, routeID string) (*route.Route, error)
SaveRoute(ctx context.Context, route *route.Route) error
DeleteRoute(ctx context.Context, accountID, routeID string) error
@@ -248,6 +250,7 @@ type Store interface {
GetNetworkResourcesByNetID(ctx context.Context, lockStrength LockingStrength, accountID, netID string) ([]*resourceTypes.NetworkResource, error)
GetNetworkResourcesByAccountID(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*resourceTypes.NetworkResource, error)
GetNetworkResourceByID(ctx context.Context, lockStrength LockingStrength, accountID, resourceID string) (*resourceTypes.NetworkResource, error)
GetNetworkResourceByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID, resourceID string) (*resourceTypes.NetworkResource, error)
GetNetworkResourceByName(ctx context.Context, lockStrength LockingStrength, accountID, resourceName string) (*resourceTypes.NetworkResource, error)
SaveNetworkResource(ctx context.Context, resource *resourceTypes.NetworkResource) error
DeleteNetworkResource(ctx context.Context, accountID, resourceID string) error
+45
View File
@@ -2166,6 +2166,21 @@ func (mr *MockStoreMockRecorder) GetNetworkResourceByID(ctx, lockStrength, accou
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNetworkResourceByID", reflect.TypeOf((*MockStore)(nil).GetNetworkResourceByID), ctx, lockStrength, accountID, resourceID)
}
// GetNetworkResourceByIDOrPublicID mocks base method.
func (m *MockStore) GetNetworkResourceByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID, resourceID string) (*types0.NetworkResource, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetNetworkResourceByIDOrPublicID", ctx, lockStrength, accountID, resourceID)
ret0, _ := ret[0].(*types0.NetworkResource)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetNetworkResourceByIDOrPublicID indicates an expected call of GetNetworkResourceByIDOrPublicID.
func (mr *MockStoreMockRecorder) GetNetworkResourceByIDOrPublicID(ctx, lockStrength, accountID, resourceID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNetworkResourceByIDOrPublicID", reflect.TypeOf((*MockStore)(nil).GetNetworkResourceByIDOrPublicID), ctx, lockStrength, accountID, resourceID)
}
// GetNetworkResourceByName mocks base method.
func (m *MockStore) GetNetworkResourceByName(ctx context.Context, lockStrength LockingStrength, accountID, resourceName string) (*types0.NetworkResource, error) {
m.ctrl.T.Helper()
@@ -2496,6 +2511,21 @@ func (mr *MockStoreMockRecorder) GetPolicyByID(ctx, lockStrength, accountID, pol
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPolicyByID", reflect.TypeOf((*MockStore)(nil).GetPolicyByID), ctx, lockStrength, accountID, policyID)
}
// GetPolicyByIDOrPublicID mocks base method.
func (m *MockStore) GetPolicyByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID, policyID string) (*types3.Policy, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetPolicyByIDOrPublicID", ctx, lockStrength, accountID, policyID)
ret0, _ := ret[0].(*types3.Policy)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetPolicyByIDOrPublicID indicates an expected call of GetPolicyByIDOrPublicID.
func (mr *MockStoreMockRecorder) GetPolicyByIDOrPublicID(ctx, lockStrength, accountID, policyID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPolicyByIDOrPublicID", reflect.TypeOf((*MockStore)(nil).GetPolicyByIDOrPublicID), ctx, lockStrength, accountID, policyID)
}
// GetPolicyRulesByResourceID mocks base method.
func (m *MockStore) GetPolicyRulesByResourceID(ctx context.Context, lockStrength LockingStrength, accountID, peerID string) ([]*types3.PolicyRule, error) {
m.ctrl.T.Helper()
@@ -2676,6 +2706,21 @@ func (mr *MockStoreMockRecorder) GetRouteByID(ctx, lockStrength, accountID, rout
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRouteByID", reflect.TypeOf((*MockStore)(nil).GetRouteByID), ctx, lockStrength, accountID, routeID)
}
// GetRouteByIDOrPublicID mocks base method.
func (m *MockStore) GetRouteByIDOrPublicID(ctx context.Context, lockStrength LockingStrength, accountID, routeID string) (*route.Route, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetRouteByIDOrPublicID", ctx, lockStrength, accountID, routeID)
ret0, _ := ret[0].(*route.Route)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetRouteByIDOrPublicID indicates an expected call of GetRouteByIDOrPublicID.
func (mr *MockStoreMockRecorder) GetRouteByIDOrPublicID(ctx, lockStrength, accountID, routeID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRouteByIDOrPublicID", reflect.TypeOf((*MockStore)(nil).GetRouteByIDOrPublicID), ctx, lockStrength, accountID, routeID)
}
// GetRoutingPeerNetworks mocks base method.
func (m *MockStore) GetRoutingPeerNetworks(ctx context.Context, accountID, peerID string) ([]string, error) {
m.ctrl.T.Helper()
@@ -175,6 +175,63 @@ func TestNetworkMapComponents_NetworkResourceRoutes_RouterPeer(t *testing.T) {
assert.NotEmpty(t, nm.RoutesFirewallRules, "router peer should have route firewall rules for the resource")
}
// A receiver without a firewall asks Calculate to skip the route firewall
// rules. Everything the rest of the sync consumes — routes, peers, peer
// firewall rules — must come out unchanged.
func TestNetworkMapComponents_SkipRouteFirewallRules(t *testing.T) {
ctx := context.Background()
account := createComponentTestAccount()
// The shared fixture leaves peer-router-1 out of every peer ACL, so its
// FirewallRules would be empty and the comparison below vacuous. Give the
// router a policy of its own.
account.Policies = append(account.Policies, &types.Policy{
ID: "policy-router", Name: "Router connectivity", Enabled: true,
Rules: []*types.PolicyRule{{
ID: "rule-router", Name: "Allow all <-> router", Enabled: true,
Action: types.PolicyTrafficActionAccept, Protocol: types.PolicyRuleProtocolALL,
Bidirectional: true,
Sources: []string{"group-all"}, Destinations: []string{"group-all"},
}},
})
validated := allPeersValidated(account)
components := account.GetPeerNetworkMapComponents(
ctx,
"peer-router-1",
account.GetPeersCustomZone(ctx, "netbird.io"),
nil,
validated,
account.GetResourcePoliciesMap(),
account.GetResourceRoutersMap(),
account.GetActiveGroupUsers(),
)
full := components.Calculate(ctx)
require.NotEmpty(t, full.RoutesFirewallRules, "baseline: router peer must get route firewall rules")
require.NotEmpty(t, full.FirewallRules, "baseline: router peer must get peer firewall rules")
components.SkipRouteFirewallRules = true
skipped := components.Calculate(ctx)
assert.Empty(t, skipped.RoutesFirewallRules, "route firewall rules must not be computed when skipped")
assert.ElementsMatch(t, routeNetworks(full.Routes), routeNetworks(skipped.Routes),
"skipping route firewall rules must not change the routes")
assert.ElementsMatch(t, peerIDs(full.Peers), peerIDs(skipped.Peers),
"skipping route firewall rules must not change the peers to connect")
assert.Equal(t, full.FirewallRules, skipped.FirewallRules,
"peer firewall rules are unrelated and must come out unchanged")
}
func routeNetworks(routes []*nmdata.Route) []string {
networks := make([]string, 0, len(routes))
for _, r := range routes {
networks = append(networks, r.Network.String())
}
return networks
}
func TestNetworkMapComponents_NetworkResourceRoutes_UnrelatedPeer(t *testing.T) {
account := createComponentTestAccount()
validated := allPeersValidated(account)
+1 -1
View File
@@ -29,7 +29,7 @@ type Policy struct {
// ID of the policy'
ID string `gorm:"primaryKey"`
PublicID string `json:"-"`
PublicID string `json:"-" gorm:"index"`
// AccountID is a reference to Account that this object belongs
AccountID string `json:"-" gorm:"index"`
+19
View File
@@ -285,6 +285,25 @@ func (u *User) EncryptSensitiveData(enc *crypt.FieldEncrypt) error {
return nil
}
func MaskEmail(email string) string {
local, domain, found := strings.Cut(email, "@")
if !found || local == "" || domain == "" {
return ""
}
// Runes, not bytes, so a non-ASCII local part is not cut mid-character.
runes := []rune(local)
// Keeping the first two and the last needs a local part of at least four to
// hide anything at all: at three or fewer those are the whole of it, and the
// address would be recoverable in full from what is meant to conceal it.
if len(runes) < 4 {
return "****@" + domain
}
return string(runes[:2]) + "****" + string(runes[len(runes)-1]) + "@" + domain
}
// DecryptSensitiveData decrypts the user's sensitive fields (Email and Name) in place.
func (u *User) DecryptSensitiveData(enc *crypt.FieldEncrypt) error {
if enc == nil {
+141
View File
@@ -296,3 +296,144 @@ func TestUser_EncryptDecryptRoundTrip(t *testing.T) {
})
}
}
func TestMaskEmail(t *testing.T) {
testCases := []struct {
name string
email string
expected string
}{
{
name: "ordinary address keeps the first two, the last, and the domain",
email: "admin@example.com",
expected: "ad****n@example.com",
},
{
name: "four characters is the shortest local part that reveals anything",
email: "abcd@example.com",
expected: "ab****d@example.com",
},
{
name: "three character local part is masked whole, since a lead and tail would be all of it",
email: "abc@example.com",
expected: "****@example.com",
},
{
name: "two character local part is masked whole",
email: "ab@example.com",
expected: "****@example.com",
},
{
name: "single character local part is masked whole",
email: "a@b.co",
expected: "****@b.co",
},
{
name: "mask width does not report the length it stands in for",
email: "a.very.long.local.part@example.com",
expected: "a.****t@example.com",
},
{
name: "a local part far longer than the mask is still reduced to three characters",
email: "finance.department.notifications.owner.account@example.com",
expected: "fi****t@example.com",
},
{
name: "plus addressing is masked along with the rest of the local part",
email: "admin+netbird@example.com",
expected: "ad****d@example.com",
},
{
name: "separators inside the local part are not treated specially",
email: "first.last-name_x@example.com",
expected: "fi****x@example.com",
},
{
name: "case is preserved rather than normalised",
email: "Admin@Example.COM",
expected: "Ad****n@Example.COM",
},
{
name: "subdomains stay intact",
email: "owner@mail.corp.example.com",
expected: "ow****r@mail.corp.example.com",
},
{
name: "german umlauts count as single characters",
email: "müller@example.de",
expected: "mü****r@example.de",
},
{
name: "cyrillic local part is cut on runes",
email: "иванов@example.ru",
expected: "ив****в@example.ru",
},
{
name: "cjk local part of three runes is masked whole, counted in runes not bytes",
email: "用户名@example.cn",
expected: "****@example.cn",
},
{
name: "cjk local part of four runes reveals the first two and the last",
email: "用户名字@example.cn",
expected: "用户****字@example.cn",
},
{
name: "arabic local part is cut on runes",
email: "مستخدم@example.sa",
expected: "مس****م@example.sa",
},
{
name: "two rune non-ascii local part is masked whole",
email: "ää@example.de",
expected: "****@example.de",
},
{
name: "astral plane runes are not split into surrogates",
email: "a🎉bc@example.com",
expected: "a🎉****c@example.com",
},
{
name: "a non-ascii domain is left alone",
email: "admin@münchen.example",
expected: "ad****n@münchen.example",
},
{
name: "only the first separator splits, so a second stays in the domain",
email: "a@b@example.com",
expected: "****@b@example.com",
},
{
name: "empty email has nothing to mask",
email: "",
expected: "",
},
{
name: "value without a separator is not an address",
email: "not-an-email",
expected: "",
},
{
name: "missing local part is not an address",
email: "@example.com",
expected: "",
},
{
name: "missing domain is not an address",
email: "admin@",
expected: "",
},
{
name: "a bare separator is not an address",
email: "@",
expected: "",
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
assert.Equal(t, tc.expected, MaskEmail(tc.email))
})
}
}
+27
View File
@@ -1448,6 +1448,25 @@ func (am *DefaultAccountManager) deleteRegularUser(ctx context.Context, accountI
return updateAccountPeers, nil
}
// pendingApprovalError refuses a user awaiting approval, naming the owner who
// can approve them when their address resolves. Failing to resolve one is not a
// reason to withhold the refusal, so the lookup is best effort.
func (am *DefaultAccountManager) pendingApprovalError(ctx context.Context, accountID string) error {
owner, err := am.GetOwnerInfo(ctx, accountID)
if err != nil {
log.WithContext(ctx).Debugf("pending approval refusal: owner of account %s did not resolve: %v", accountID, err)
return status.NewUserPendingApprovalError()
}
masked := types.MaskEmail(owner.Email)
if masked == "" {
log.WithContext(ctx).Debugf("pending approval refusal: no address found for the owner of account %s", accountID)
return status.NewUserPendingApprovalError()
}
return status.NewUserPendingApprovalByOwnerError(masked)
}
// GetOwnerInfo retrieves the owner information for a given account ID.
func (am *DefaultAccountManager) GetOwnerInfo(ctx context.Context, accountID string) (*types.UserInfo, error) {
owner, err := am.Store.GetAccountOwner(ctx, store.LockingStrengthNone, accountID)
@@ -1505,6 +1524,14 @@ func (am *DefaultAccountManager) GetCurrentUserInfo(ctx context.Context, userAut
return nil, err
}
// A user pending approval is blocked too, and the dashboard needs to tell
// the two apart: one is a dead end, the other resolves by itself once the
// owner acts. Naming that owner needs the address the IdP holds, which is
// why this is answered here rather than in the permission gate.
if user.IsBlocked() && user.PendingApproval {
return nil, am.pendingApprovalError(ctx, user.AccountID)
}
if user.IsBlocked() {
return nil, status.NewUserBlockedError()
}
+64
View File
@@ -1779,6 +1779,42 @@ func TestDefaultAccountManager_GetCurrentUserInfo(t *testing.T) {
}
require.NoError(t, store.SaveAccount(context.Background(), account2))
account3 := newAccountWithId(context.Background(), "account3", "account3Owner", "", "owner@example.com", "", false)
account3.Users["pending-user"] = &types.User{
Id: "pending-user",
AccountID: account3.Id,
Role: types.UserRoleUser,
Blocked: true,
PendingApproval: true,
}
require.NoError(t, store.SaveAccount(context.Background(), account3))
// The owner has no address to name, so the refusal falls back to the generic one.
account4 := newAccountWithId(context.Background(), "account4", "account4Owner", "", "", "", false)
account4.Users["pending-user-without-owner-email"] = &types.User{
Id: "pending-user-without-owner-email",
AccountID: account4.Id,
Role: types.UserRoleUser,
Blocked: true,
PendingApproval: true,
}
require.NoError(t, store.SaveAccount(context.Background(), account4))
// No user holds the owner role, so the owner lookup itself fails.
account5 := newAccountWithId(context.Background(), "account5", "account5Admin", "", "", "", false)
account5.Users["account5Admin"].Role = types.UserRoleAdmin
account5.Users["pending-user-without-owner"] = &types.User{
Id: "pending-user-without-owner",
AccountID: account5.Id,
Role: types.UserRoleUser,
Blocked: true,
PendingApproval: true,
}
require.NoError(t, store.SaveAccount(context.Background(), account5))
account6 := newAccountWithId(context.Background(), "account6", "account6Owner", "", "stranger@example.com", "", false)
require.NoError(t, store.SaveAccount(context.Background(), account6))
permissionsManager := permissions.NewManager(store)
am := DefaultAccountManager{
Store: store,
@@ -1812,6 +1848,34 @@ func TestDefaultAccountManager_GetCurrentUserInfo(t *testing.T) {
userAuth: auth.UserAuth{AccountId: account1.Id, UserId: "service-user"},
expectedErr: status.NewPermissionDeniedError(),
},
{
name: "pending approval names the owner",
userAuth: auth.UserAuth{AccountId: account3.Id, UserId: "pending-user"},
expectedErr: status.NewUserPendingApprovalByOwnerError("ow****r@example.com"),
},
{
name: "pending approval without an owner address",
userAuth: auth.UserAuth{AccountId: account4.Id, UserId: "pending-user-without-owner-email"},
expectedErr: status.NewUserPendingApprovalError(),
},
{
name: "pending approval without an owner",
userAuth: auth.UserAuth{AccountId: account5.Id, UserId: "pending-user-without-owner"},
expectedErr: status.NewUserPendingApprovalError(),
},
{
// The account claim points at an account the caller is not in. The
// owner named has to be the one of the account holding the caller's
// own record, never the one the claim asks for.
name: "pending approval ignores a mismatched account claim",
userAuth: auth.UserAuth{AccountId: account6.Id, UserId: "pending-user"},
expectedErr: status.NewUserPendingApprovalByOwnerError("ow****r@example.com"),
},
{
name: "blocked user answers before the account claim is validated",
userAuth: auth.UserAuth{AccountId: account6.Id, UserId: "blocked-user"},
expectedErr: status.NewUserBlockedError(),
},
{
name: "owner user",
userAuth: auth.UserAuth{AccountId: account1.Id, UserId: "account1Owner"},
+26
View File
@@ -0,0 +1,26 @@
# PIN and password authentication limits
PIN and password credentials are accepted only in a POST form body. Query-string
credentials and credentials on other HTTP methods are ignored.
The proxy permits a burst of five credential checks per account and service,
then replenishes one check every six seconds (ten per minute). PIN and password
checks share the same budget. Five failed checks from one client IP in a
rolling five-minute window block that source for fifteen minutes. In-flight checks
reserve failure slots; blocked requests do not extend the cooldown. Successful
authentication clears that source's failure history. Infrastructure failures
consume the service budget without counting as incorrect credentials.
Throttled requests return HTTP 429 with a `Retry-After` delay in seconds. The
login page displays that delay. Existing authenticated sessions and other
authentication methods do not consume these credential budgets.
The client IP comes from the existing trusted-proxy resolution. Deployments
behind a load balancer must configure trusted proxies correctly; otherwise
visitors share the load balancer's source budget. Visitors behind the same NAT
also share a source budget for a service.
State is held in memory per proxy process and resets on restart. Multiple
replicas have independent budgets. State is bounded to 16,384 source entries and
4,096 service entries; when capacity is exhausted, new checks are denied until
idle entries expire. Active blocks are never evicted to admit a new source.
+100
View File
@@ -0,0 +1,100 @@
package auth
import (
"errors"
"math"
"net/http"
"strconv"
"time"
"google.golang.org/genproto/googleapis/rpc/errdetails"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"github.com/netbirdio/netbird/proxy/auth"
"github.com/netbirdio/netbird/proxy/internal/proxy"
)
var errCredentialClientIP = errors.New("invalid client address")
type credentialLimitError struct {
retryAfter time.Duration
}
func (e *credentialLimitError) Error() string {
return "too many authentication attempts"
}
func credentialFormValue(r *http.Request, field string) string {
if r.Method != http.MethodPost {
return ""
}
return r.PostFormValue(field)
}
func (mw *Middleware) authenticateScheme(r *http.Request, config DomainConfig, scheme Scheme) (string, string, error) {
method := scheme.Type()
if (method != auth.MethodPIN && method != auth.MethodPassword) || !wasCredentialSubmitted(r, method) {
return scheme.Authenticate(r)
}
ip := mw.resolveClientIP(r).Unmap()
if !ip.IsValid() {
return "", "", errCredentialClientIP
}
source, retry := mw.credentials.begin(credentialSourceKey{
service: credentialServiceKey{accountID: config.AccountID, serviceID: config.ServiceID},
ip: ip,
})
if retry > 0 {
return "", "", &credentialLimitError{retryAfter: retry}
}
token, prompt, err := scheme.Authenticate(r)
outcome := credentialUnavailable
if err == nil {
outcome = credentialRejected
if token != "" {
outcome = credentialAccepted
}
}
mw.credentials.finish(source, outcome)
return token, prompt, err
}
func credentialRetryAfter(err error) time.Duration {
var limitErr *credentialLimitError
if errors.As(err, &limitErr) {
return limitErr.retryAfter
}
s := status.Convert(err)
if s.Code() != codes.ResourceExhausted {
return 0
}
for _, detail := range s.Details() {
if info, ok := detail.(*errdetails.RetryInfo); ok && info.RetryDelay != nil && info.RetryDelay.CheckValid() == nil {
if delay := info.RetryDelay.AsDuration(); delay > 0 {
return delay
}
}
}
return credentialCheckInterval
}
func (mw *Middleware) writeAuthenticationError(w http.ResponseWriter, r *http.Request, method auth.Method, err error) {
if cd := proxy.CapturedDataFromContext(r.Context()); cd != nil {
cd.SetOrigin(proxy.OriginAuth)
cd.SetAuthMethod(method.String())
}
if retry := credentialRetryAfter(err); retry > 0 {
// RFC 6585 section 4 forbids caching 429 responses.
w.Header().Set("Cache-Control", "no-store")
w.Header().Set("Retry-After", strconv.FormatInt(int64(math.Ceil(retry.Seconds())), 10))
http.Error(w, "too many authentication attempts; try again later", http.StatusTooManyRequests)
return
}
if errors.Is(err, errCredentialClientIP) {
http.Error(w, "invalid client address", http.StatusBadRequest)
return
}
mw.logger.WithField("scheme", method.String()).Warnf("authentication infrastructure error: %v", err)
http.Error(w, "authentication service unavailable", http.StatusBadGateway)
}
+169
View File
@@ -0,0 +1,169 @@
package auth
import (
"net/netip"
"sync"
"time"
"golang.org/x/time/rate"
"github.com/netbirdio/netbird/proxy/internal/types"
)
const (
credentialFailureLimit = 5
credentialFailureWindow = 5 * time.Minute
credentialBlockDuration = 15 * time.Minute
credentialCheckInterval = 6 * time.Second
credentialCheckBurst = 5
credentialMaxSources = 16384
credentialMaxServices = 4096
credentialCleanupInterval = time.Minute
)
type credentialServiceKey struct {
accountID types.AccountID
serviceID types.ServiceID
}
type credentialSourceKey struct {
service credentialServiceKey
ip netip.Addr
}
type credentialSource struct {
failures []time.Time
pending int
expiresAt time.Time
blockedUntil time.Time
}
type credentialService struct {
limiter *rate.Limiter
lastUsed time.Time
}
type credentialOutcome string
const (
credentialUnavailable credentialOutcome = "unavailable"
credentialRejected credentialOutcome = "rejected"
credentialAccepted credentialOutcome = "accepted"
)
// State is local to this proxy process. Active blocks are never evicted to
// make room for a new source; exhausting capacity denies new checks.
type credentialLimiter struct {
mu sync.Mutex
now func() time.Time
sources map[credentialSourceKey]*credentialSource
services map[credentialServiceKey]*credentialService
nextCleanup time.Time
}
func newCredentialLimiter() *credentialLimiter {
return &credentialLimiter{
now: time.Now,
sources: make(map[credentialSourceKey]*credentialSource),
services: make(map[credentialServiceKey]*credentialService),
}
}
func (l *credentialLimiter) begin(key credentialSourceKey) (*credentialSource, time.Duration) {
l.mu.Lock()
defer l.mu.Unlock()
now := l.now()
l.cleanup(now)
source := l.sources[key]
if source != nil {
if now.Before(source.blockedUntil) {
return nil, source.blockedUntil.Sub(now)
}
if source.pending == 0 && !now.Before(source.expiresAt) {
*source = credentialSource{}
}
source.expireFailures(now)
// Reserve the failure budget before verification so concurrent guesses
// cannot all pass a check against the same completed failure count.
if len(source.failures)+source.pending >= credentialFailureLimit {
return nil, time.Second
}
} else if len(l.sources) >= credentialMaxSources {
return nil, credentialCleanupInterval
}
if retry := l.allowService(key.service, now); retry > 0 {
return nil, retry
}
if source == nil {
source = &credentialSource{}
l.sources[key] = source
}
if source.expiresAt.IsZero() {
source.expiresAt = now.Add(credentialFailureWindow)
}
source.pending++
return source, 0
}
func (l *credentialLimiter) allowService(key credentialServiceKey, now time.Time) time.Duration {
service := l.services[key]
if service == nil {
if len(l.services) >= credentialMaxServices {
return credentialCleanupInterval
}
service = &credentialService{limiter: rate.NewLimiter(rate.Every(credentialCheckInterval), credentialCheckBurst)}
l.services[key] = service
}
service.lastUsed = now
if service.limiter.AllowN(now, 1) {
return 0
}
return max(time.Nanosecond, time.Duration((1-service.limiter.TokensAt(now))*float64(credentialCheckInterval)))
}
func (l *credentialLimiter) finish(source *credentialSource, outcome credentialOutcome) {
l.mu.Lock()
defer l.mu.Unlock()
source.pending--
now := l.now()
source.expireFailures(now)
switch outcome {
case credentialRejected:
source.failures = append(source.failures, now)
source.expiresAt = now.Add(credentialFailureWindow)
if len(source.failures) >= credentialFailureLimit && source.blockedUntil.IsZero() {
source.blockedUntil = now.Add(credentialBlockDuration)
source.expiresAt = source.blockedUntil
}
case credentialAccepted:
if !now.Before(source.blockedUntil) {
source.failures = nil
source.expiresAt = now.Add(credentialFailureWindow)
}
case credentialUnavailable:
// Transport failures consume the service budget, but are not bad guesses.
}
}
func (s *credentialSource) expireFailures(now time.Time) {
for len(s.failures) > 0 && !now.Before(s.failures[0].Add(credentialFailureWindow)) {
s.failures = s.failures[1:]
}
}
func (l *credentialLimiter) cleanup(now time.Time) {
if now.Before(l.nextCleanup) {
return
}
l.nextCleanup = now.Add(credentialCleanupInterval)
for key, source := range l.sources {
if source.pending == 0 && !now.Before(source.expiresAt) {
delete(l.sources, key)
}
}
for key, service := range l.services {
if now.Sub(service.lastUsed) >= credentialBlockDuration {
delete(l.services, key)
}
}
}
@@ -0,0 +1,190 @@
package auth
import (
"net/netip"
"strconv"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/proxy/internal/types"
)
func TestCredentialLimiterCooldown(t *testing.T) {
l := newCredentialLimiter()
now := time.Now()
l.now = func() time.Time { return now }
key := credentialSourceKey{service: credentialServiceKey{"account", "service"}, ip: netip.MustParseAddr("192.0.2.1")}
for range credentialFailureLimit {
attempt, retry := l.begin(key)
require.Zero(t, retry, "initial guesses must reach verification")
l.finish(attempt, credentialRejected)
}
_, retry := l.begin(key)
assert.Equal(t, credentialBlockDuration, retry, "five failures must start a fifteen-minute block")
now = now.Add(credentialBlockDuration - time.Second)
_, retry = l.begin(key)
assert.Equal(t, time.Second, retry, "blocked requests must not extend the deadline")
now = now.Add(time.Second)
attempt, retry := l.begin(key)
require.Zero(t, retry, "the source must recover when its block expires")
l.finish(attempt, credentialAccepted)
}
func TestCredentialLimiterFailureWindowAndSuccess(t *testing.T) {
for _, outcome := range []credentialOutcome{credentialAccepted, credentialUnavailable} {
t.Run(map[credentialOutcome]string{credentialAccepted: "success", credentialUnavailable: "infrastructure error"}[outcome], func(t *testing.T) {
l := newCredentialLimiter()
now := time.Now()
l.now = func() time.Time { return now }
key := credentialSourceKey{service: credentialServiceKey{"account", "service"}, ip: netip.MustParseAddr("192.0.2.1")}
for range 4 {
attempt, retry := l.begin(key)
require.Zero(t, retry, "four failures must fit the budget")
l.finish(attempt, credentialRejected)
}
attempt, retry := l.begin(key)
require.Zero(t, retry, "fifth check must be allowed")
l.finish(attempt, outcome)
now = now.Add(credentialCheckInterval)
attempt, retry = l.begin(key)
require.Zero(t, retry, "success or infrastructure error must not start a block")
l.finish(attempt, credentialRejected)
now = now.Add(credentialCheckInterval)
attempt, retry = l.begin(key)
if outcome == credentialUnavailable {
assert.Greater(t, retry, time.Duration(0), "infrastructure errors must preserve earlier failures")
return
}
require.Zero(t, retry, "success must clear earlier failures")
l.finish(attempt, credentialRejected)
now = now.Add(credentialFailureWindow)
for range credentialFailureLimit {
attempt, retry = l.begin(key)
require.Zero(t, retry, "old failures must expire")
l.finish(attempt, credentialRejected)
}
})
}
}
func TestCredentialLimiterRollingWindow(t *testing.T) {
l := newCredentialLimiter()
now := time.Now()
l.now = func() time.Time { return now }
key := credentialSourceKey{service: credentialServiceKey{"account", "service"}, ip: netip.MustParseAddr("192.0.2.1")}
attempt, retry := l.begin(key)
require.Zero(t, retry, "the first failure starts the history")
l.finish(attempt, credentialRejected)
now = now.Add(4 * time.Minute)
for range 3 {
attempt, retry = l.begin(key)
require.Zero(t, retry, "three more failures must fit the budget")
l.finish(attempt, credentialRejected)
}
now = now.Add(time.Minute + time.Second)
for range 2 {
attempt, retry = l.begin(key)
require.Zero(t, retry, "only the oldest failure must have expired")
l.finish(attempt, credentialRejected)
}
_, retry = l.begin(key)
assert.Equal(t, credentialBlockDuration, retry, "five recent failures must block even across the first window boundary")
}
func TestCredentialLimiterServiceBudget(t *testing.T) {
l := newCredentialLimiter()
now := time.Now()
l.now = func() time.Time { return now }
key := credentialSourceKey{service: credentialServiceKey{"account", "service"}, ip: netip.MustParseAddr("192.0.2.1")}
for range credentialCheckBurst {
attempt, retry := l.begin(key)
require.Zero(t, retry, "initial checks must fit the service burst")
l.finish(attempt, credentialAccepted)
key.ip = key.ip.Next()
}
_, retry := l.begin(key)
assert.Equal(t, credentialCheckInterval, retry, "changing IP must not bypass the service budget")
other := key
other.service.accountID = "another-account"
attempt, retry := l.begin(other)
require.Zero(t, retry, "accounts must have separate budgets")
l.finish(attempt, credentialAccepted)
other = key
other.service.serviceID = "another-service"
attempt, retry = l.begin(other)
require.Zero(t, retry, "services must have separate budgets")
l.finish(attempt, credentialAccepted)
now = now.Add(credentialCheckInterval)
attempt, retry = l.begin(key)
require.Zero(t, retry, "one check must refill every six seconds")
l.finish(attempt, credentialAccepted)
_, retry = l.begin(key)
assert.Equal(t, credentialCheckInterval, retry, "refill must only grant one new check")
}
func TestCredentialLimiterConcurrentReservations(t *testing.T) {
l := newCredentialLimiter()
now := time.Now()
l.now = func() time.Time { return now }
key := credentialSourceKey{service: credentialServiceKey{"account", "service"}, ip: netip.MustParseAddr("192.0.2.1")}
var attempts []*credentialSource
for range credentialFailureLimit {
attempt, retry := l.begin(key)
require.Zero(t, retry, "initial requests must reserve the failure budget")
attempts = append(attempts, attempt)
}
// Refill the service budget while earlier verification calls are still running.
now = now.Add(time.Minute)
var admitted atomic.Int32
var wg sync.WaitGroup
for range 100 {
wg.Go(func() {
attempt, retry := l.begin(key)
if retry == 0 {
admitted.Add(1)
l.finish(attempt, credentialRejected)
}
})
}
wg.Wait()
assert.Zero(t, admitted.Load(), "in-flight guesses must reserve the failure budget despite a refilled service budget")
for _, attempt := range attempts {
wg.Go(func() { l.finish(attempt, credentialRejected) })
}
wg.Wait()
_, retry := l.begin(key)
assert.Equal(t, credentialBlockDuration, retry, "concurrent failures must activate the block")
}
func TestCredentialLimiterCapacityAndCleanup(t *testing.T) {
for _, fullSources := range []bool{true, false} {
t.Run(map[bool]string{true: "sources", false: "services"}[fullSources], func(t *testing.T) {
l := newCredentialLimiter()
now := time.Now()
l.now = func() time.Time { return now }
key := credentialSourceKey{service: credentialServiceKey{"account", "service"}, ip: netip.MustParseAddr("192.0.2.1")}
if fullSources {
ip := netip.MustParseAddr("198.18.0.1")
for range credentialMaxSources {
l.sources[credentialSourceKey{service: key.service, ip: ip}] = &credentialSource{expiresAt: now.Add(credentialBlockDuration), blockedUntil: now.Add(credentialBlockDuration)}
ip = ip.Next()
}
} else {
for i := range credentialMaxServices {
l.services[credentialServiceKey{serviceID: key.service.serviceID, accountID: types.AccountID(strconv.Itoa(i))}] = &credentialService{lastUsed: now}
}
}
_, retry := l.begin(key)
assert.Positive(t, retry, "full state must deny new checks without evicting active entries")
now = now.Add(credentialBlockDuration)
attempt, retry := l.begin(key)
require.Zero(t, retry, "expired state must release capacity")
l.finish(attempt, credentialAccepted)
})
}
}
+196
View File
@@ -0,0 +1,196 @@
package auth
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"net/netip"
"net/url"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/genproto/googleapis/rpc/errdetails"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/durationpb"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
servicemanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service/manager"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/sessionkey"
nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
"github.com/netbirdio/netbird/management/server/store"
mgmttypes "github.com/netbirdio/netbird/management/server/types"
proxyauth "github.com/netbirdio/netbird/proxy/auth"
"github.com/netbirdio/netbird/proxy/internal/proxy"
"github.com/netbirdio/netbird/shared/management/proto"
)
// localCredentialClient replaces the transport while keeping the real service
// store, credential verification, and session signing.
type localCredentialClient struct {
server *nbgrpc.ProxyServiceServer
}
func (c localCredentialClient) Authenticate(ctx context.Context, req *proto.AuthenticateRequest, _ ...grpc.CallOption) (*proto.AuthenticateResponse, error) {
return c.server.Authenticate(ctx, req)
}
func credentialHandler(t *testing.T, field string) (*Middleware, http.Handler) {
t.Helper()
ctx := context.Background()
s, err := store.NewStore(ctx, mgmttypes.SqliteStoreEngine, t.TempDir(), nil, false)
require.NoError(t, err)
t.Cleanup(func() { assert.NoError(t, s.Close(ctx)) })
require.NoError(t, s.SaveAccount(ctx, &mgmttypes.Account{Id: "account"}))
keys := generateTestKeyPair(t)
svc := &service.Service{
ID: "service", AccountID: "account", Name: "test", Domain: "example.com",
Enabled: true, SessionPrivateKey: keys.PrivateKey, SessionPublicKey: keys.PublicKey,
Auth: service.AuthConfig{
PinAuth: &service.PINAuthConfig{Enabled: true, Pin: "842716"},
PasswordAuth: &service.PasswordAuthConfig{Enabled: true, Password: "842716"},
},
}
require.NoError(t, svc.Auth.HashSecrets())
require.NoError(t, s.CreateService(ctx, svc))
server := nbgrpc.NewProxyServiceServer(nil, nil, nil, nbgrpc.ProxyOIDCConfig{}, nil, nil, nil, nil, nil)
t.Cleanup(server.Close)
server.SetServiceManager(servicemanager.NewManager(s, nil, nil, nil, nil, nil))
client := localCredentialClient{server: server}
var scheme Scheme = NewPin(client, "service", "account")
if field == "password" {
scheme = NewPassword(client, "service", "account")
}
mw := NewMiddleware(nil, nil, nil)
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, keys.PublicKey, time.Hour, "account", "service", nil, false, nil))
return mw, mw.Protect(newPassthroughHandler())
}
func credentialRequest(method, field, value string) *http.Request {
r := httptest.NewRequest(method, "https://example.com/", strings.NewReader(url.Values{field: {value}}.Encode()))
r.Header.Set("Content-Type", "application/x-www-form-urlencoded")
r.RemoteAddr = "198.51.100.25:12345"
return r
}
func TestCredentialAuthPOSTOnly(t *testing.T) {
for _, field := range []string{"pin", "password"} {
t.Run(field, func(t *testing.T) {
_, handler := credentialHandler(t, field)
for _, method := range []string{http.MethodGet, http.MethodPut, http.MethodPatch, http.MethodDelete, http.MethodPost} {
r := credentialRequest(method, field, "")
r.URL.RawQuery = url.Values{field: {"842716"}}.Encode()
resp := httptest.NewRecorder()
handler.ServeHTTP(resp, r)
assert.Equal(t, http.StatusUnauthorized, resp.Code, "%s query credentials must not authenticate", method)
assert.Empty(t, resp.Result().Cookies(), "query credentials must not issue a session")
}
for _, method := range []string{http.MethodGet, http.MethodPut, http.MethodPatch, http.MethodDelete} {
resp := httptest.NewRecorder()
handler.ServeHTTP(resp, credentialRequest(method, field, "842716"))
assert.Equal(t, http.StatusUnauthorized, resp.Code, "%s body credentials must not authenticate", method)
}
resp := httptest.NewRecorder()
handler.ServeHTTP(resp, credentialRequest(http.MethodPost, field, "842716"))
assert.Equal(t, http.StatusSeeOther, resp.Code, "POST body credentials must authenticate")
})
}
}
func TestCredentialAuthThrottling(t *testing.T) {
for _, field := range []string{"pin", "password"} {
t.Run(field, func(t *testing.T) {
_, handler := credentialHandler(t, field)
for range 5 {
resp := httptest.NewRecorder()
handler.ServeHTTP(resp, credentialRequest(http.MethodPost, field, "000000"))
require.Equal(t, http.StatusUnauthorized, resp.Code, "initial wrong credentials must be rejected")
}
resp := httptest.NewRecorder()
handler.ServeHTTP(resp, credentialRequest(http.MethodPost, field, "842716"))
assert.Equal(t, http.StatusTooManyRequests, resp.Code, "even correct credentials must wait for the block to expire")
assert.Equal(t, "900", resp.Header().Get("Retry-After"), "five failures must block the source for fifteen minutes")
assert.Empty(t, resp.Result().Cookies(), "blocked credentials must not issue a session")
})
}
}
func TestCredentialAuthSessionAndClientIP(t *testing.T) {
keys := generateTestKeyPair(t)
token, err := sessionkey.SignToken(keys.PrivateKey, "pin-user", "", "example.com", proxyauth.MethodPIN, nil, nil, time.Hour)
require.NoError(t, err)
mw := NewMiddleware(nil, nil, nil)
now := time.Now()
mw.credentials.now = func() time.Time { return now }
scheme := &stubScheme{method: proxyauth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, keys.PublicKey, 0, "account", "service", nil, false, nil))
handler := mw.Protect(newPassthroughHandler())
for range credentialFailureLimit {
resp := httptest.NewRecorder()
handler.ServeHTTP(resp, credentialRequest(http.MethodPost, "pin", "000000"))
require.Equal(t, http.StatusUnauthorized, resp.Code, "bad PIN must consume the failure budget")
}
now = now.Add(credentialCheckInterval)
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, keys.PublicKey, 0, "account", "service", nil, false, nil))
r := credentialRequest(http.MethodPost, "pin", "000000")
r.RemoteAddr = "[::ffff:198.51.100.25]:45678"
r.Header.Set("X-Forwarded-For", "192.0.2.5")
r.Header.Set("X-Real-IP", "192.0.2.6")
resp := httptest.NewRecorder()
handler.ServeHTTP(resp, r)
assert.Equal(t, http.StatusTooManyRequests, resp.Code, "mapped addresses and untrusted forwarding headers must not bypass the source block")
assert.Equal(t, "no-store", resp.Header().Get("Cache-Control"), "rate limits must not be cached")
r.AddCookie(&http.Cookie{Name: proxyauth.SessionCookieName, Value: token})
resp = httptest.NewRecorder()
handler.ServeHTTP(resp, r)
assert.Equal(t, http.StatusOK, resp.Code, "an existing session must pass even with credentials in the request")
assert.Equal(t, "backend", resp.Body.String(), "the authenticated request must reach the application")
r = credentialRequest(http.MethodPost, "pin", "000000")
cd := proxy.NewCapturedData("test")
cd.SetClientIP(netip.MustParseAddr("192.0.2.9"))
r = r.WithContext(proxy.WithCapturedData(r.Context(), cd))
resp = httptest.NewRecorder()
handler.ServeHTTP(resp, r)
assert.Equal(t, http.StatusUnauthorized, resp.Code, "a client resolved by the trusted-proxy middleware must get its own source budget")
r = credentialRequest(http.MethodPost, "pin", "000000")
r.RemoteAddr = "invalid"
resp = httptest.NewRecorder()
handler.ServeHTTP(resp, r)
assert.Equal(t, http.StatusBadRequest, resp.Code, "an unresolvable client address must fail closed")
now = now.Add(credentialBlockDuration)
scheme.token = token
resp = httptest.NewRecorder()
handler.ServeHTTP(resp, credentialRequest(http.MethodPost, "pin", "842716"))
assert.Equal(t, http.StatusSeeOther, resp.Code, "credentials must work again after cooldown")
}
func TestCredentialAuthManagementThrottling(t *testing.T) {
s, err := status.New(codes.ResourceExhausted, "rate limited").WithDetails(&errdetails.RetryInfo{RetryDelay: durationpb.New(2500 * time.Millisecond)})
require.NoError(t, err)
for _, tc := range []struct {
name string
err error
code int
retry string
}{
{"retry info", fmt.Errorf("authenticate PIN: %w", s.Err()), http.StatusTooManyRequests, "3"},
{"missing retry info", status.Error(codes.ResourceExhausted, "rate limited"), http.StatusTooManyRequests, "6"},
{"unavailable", status.Error(codes.Unavailable, "unavailable"), http.StatusBadGateway, ""},
} {
t.Run(tc.name, func(t *testing.T) {
keys := generateTestKeyPair(t)
mw := NewMiddleware(nil, nil, nil)
scheme := &stubScheme{method: proxyauth.MethodPIN, authFn: func(*http.Request) (string, string, error) { return "", "", tc.err }}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, keys.PublicKey, 0, "account", "service", nil, false, nil))
resp := httptest.NewRecorder()
mw.Protect(newPassthroughHandler()).ServeHTTP(resp, credentialRequest(http.MethodPost, "pin", "000000"))
assert.Equal(t, tc.code, resp.Code, "management errors must keep their HTTP meaning")
assert.Equal(t, tc.retry, resp.Header().Get("Retry-After"), "retry hints must round up to whole seconds")
})
}
}
+29 -11
View File
@@ -87,6 +87,7 @@ type Middleware struct {
sessionValidator SessionValidator
geo restrict.GeoResolver
tunnelCache *tunnelValidationCache
credentials *credentialLimiter
}
// NewMiddleware creates a new authentication middleware. The sessionValidator is
@@ -101,6 +102,7 @@ func NewMiddleware(logger *log.Logger, sessionValidator SessionValidator, geo re
sessionValidator: sessionValidator,
geo: geo,
tunnelCache: newTunnelValidationCache(),
credentials: newCredentialLimiter(),
}
}
@@ -133,7 +135,7 @@ func (mw *Middleware) Protect(next http.Handler) http.Handler {
if mw.forwardWithTunnelPeer(w, r, host, config, next) {
return
}
http.Error(w, "Forbidden", http.StatusForbidden)
denyPrivate(w)
return
}
@@ -228,7 +230,7 @@ func (mw *Middleware) checkIPRestrictions(w http.ResponseWriter, r *http.Request
clientIP := mw.resolveClientIP(r)
if !clientIP.IsValid() {
mw.logger.Debugf("IP restriction: cannot resolve client address for %q, denying", r.RemoteAddr)
http.Error(w, "Forbidden", http.StatusForbidden)
denyForbidden(w, config)
return false
}
@@ -263,10 +265,30 @@ func (mw *Middleware) checkIPRestrictions(w http.ResponseWriter, r *http.Request
reason := verdict.String()
mw.blockIPRestriction(r, reason)
http.Error(w, "Forbidden", http.StatusForbidden)
denyForbidden(w, config)
return false
}
// denyForbidden writes a 403, dropping the client connection when the
// domain is private so a later retry cannot reuse it.
func denyForbidden(w http.ResponseWriter, config DomainConfig) {
if config.Private {
denyPrivate(w)
return
}
http.Error(w, "Forbidden", http.StatusForbidden)
}
// denyPrivate writes a 403 and closes the connection, so a client refused
// before joining the overlay cannot keep retrying on the same warm socket.
// Go's HTTP/2 server turns the exact lowercase "close" token into a GOAWAY.
func denyPrivate(w http.ResponseWriter) {
h := w.Header()
h.Set("Connection", "close")
h.Set("Cache-Control", "no-store")
http.Error(w, "Forbidden", http.StatusForbidden)
}
// resolveClientIP extracts the real client IP from CapturedData, falling back to r.RemoteAddr.
func (mw *Middleware) resolveClientIP(r *http.Request) netip.Addr {
if cd := proxy.CapturedDataFromContext(r.Context()); cd != nil {
@@ -523,13 +545,9 @@ func (mw *Middleware) authenticateWithSchemes(w http.ResponseWriter, r *http.Req
var attemptedMethod string
for _, scheme := range config.Schemes {
token, promptData, err := scheme.Authenticate(r)
token, promptData, err := mw.authenticateScheme(r, config, scheme)
if err != nil {
mw.logger.WithField("scheme", scheme.Type().String()).Warnf("authentication infrastructure error: %v", err)
if cd := proxy.CapturedDataFromContext(r.Context()); cd != nil {
cd.SetOrigin(proxy.OriginAuth)
}
http.Error(w, "authentication service unavailable", http.StatusBadGateway)
mw.writeAuthenticationError(w, r, scheme.Type(), err)
return
}
@@ -630,9 +648,9 @@ func setSessionCookie(w http.ResponseWriter, token string, expiration time.Durat
func wasCredentialSubmitted(r *http.Request, method auth.Method) bool {
switch method {
case auth.MethodPIN:
return r.FormValue("pin") != ""
return credentialFormValue(r, pinFormId) != ""
case auth.MethodPassword:
return r.FormValue("password") != ""
return credentialFormValue(r, passwordFormId) != ""
case auth.MethodOIDC:
return r.URL.Query().Get("session_token") != ""
}
+1 -1
View File
@@ -35,7 +35,7 @@ func (Password) Type() auth.Method {
// so that it can be injected into a request from the UI so that
// authentication may be successful.
func (p Password) Authenticate(r *http.Request) (string, string, error) {
password := r.FormValue(passwordFormId)
password := credentialFormValue(r, passwordFormId)
if password == "" {
// No password submitted; return the form ID so the UI can prompt the user.
+1 -1
View File
@@ -35,7 +35,7 @@ func (Pin) Type() auth.Method {
// so that it can be injected into a request from the UI so that
// authentication may be successful.
func (p Pin) Authenticate(r *http.Request) (string, string, error) {
pin := r.FormValue(pinFormId)
pin := credentialFormValue(r, pinFormId)
if pin == "" {
// No PIN submitted; return the form ID so the UI can prompt the user.
+272
View File
@@ -0,0 +1,272 @@
package auth
import (
"context"
"errors"
"io"
"net/http"
"net/http/httptest"
"net/http/httptrace"
"net/netip"
"sync"
"testing"
"time"
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc"
"github.com/netbirdio/netbird/proxy/internal/proxy"
"github.com/netbirdio/netbird/proxy/internal/restrict"
"github.com/netbirdio/netbird/shared/management/proto"
)
// switchableTunnelValidator flips the ValidateTunnelPeer verdict between requests.
type switchableTunnelValidator struct {
mu sync.Mutex
valid bool
}
func (s *switchableTunnelValidator) setValid(v bool) {
s.mu.Lock()
defer s.mu.Unlock()
s.valid = v
}
func (s *switchableTunnelValidator) ValidateSession(context.Context, *proto.ValidateSessionRequest, ...grpc.CallOption) (*proto.ValidateSessionResponse, error) {
return nil, errors.New("not used in this test")
}
func (s *switchableTunnelValidator) ValidateTunnelPeer(context.Context, *proto.ValidateTunnelPeerRequest, ...grpc.CallOption) (*proto.ValidateTunnelPeerResponse, error) {
s.mu.Lock()
defer s.mu.Unlock()
if !s.valid {
return &proto.ValidateTunnelPeerResponse{Valid: false, DeniedReason: "not_in_group"}, nil
}
return &proto.ValidateTunnelPeerResponse{
Valid: true,
UserId: "user-1",
SessionToken: "tunnel-session-token",
}, nil
}
// testServerHost is the domain key Protect derives from the httptest listener.
const testServerHost = "127.0.0.1"
var testTunnelIP = netip.MustParseAddr("100.90.1.14")
// startProtectedServer serves mw.Protect and stamps requests as overlay traffic.
func startProtectedServer(t *testing.T, mw *Middleware, clientIP netip.Addr, lookup TunnelLookupFunc, h2 bool) *httptest.Server {
t.Helper()
protected := mw.Protect(newPassthroughHandler())
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
cd := proxy.NewCapturedData("")
cd.SetClientIP(clientIP)
ctx := proxy.WithCapturedData(r.Context(), cd)
ctx = WithTunnelLookup(ctx, lookup)
protected.ServeHTTP(w, r.WithContext(ctx))
})
srv := httptest.NewUnstartedServer(handler)
if h2 {
srv.EnableHTTP2 = true
srv.StartTLS()
} else {
srv.Start()
}
t.Cleanup(srv.Close)
return srv
}
// tracedResponse is what a test observes from one client round trip.
type tracedResponse struct {
status int
protoMajor int
close bool
connection string
cacheControl string
reused bool
}
// doTraced GETs url and reports whether the connection that served it was reused.
func doTraced(t *testing.T, client *http.Client, url string) tracedResponse {
t.Helper()
var reused bool
trace := &httptrace.ClientTrace{
GotConn: func(info httptrace.GotConnInfo) { reused = info.Reused },
}
req, err := http.NewRequestWithContext(httptrace.WithClientTrace(context.Background(), trace), http.MethodGet, url, nil)
require.NoError(t, err)
resp, err := client.Do(req)
require.NoError(t, err)
defer func() { require.NoError(t, resp.Body.Close()) }()
_, err = io.Copy(io.Discard, resp.Body)
require.NoError(t, err)
return tracedResponse{
status: resp.StatusCode,
protoMajor: resp.ProtoMajor,
close: resp.Close,
connection: resp.Header.Get("Connection"),
cacheControl: resp.Header.Get("Cache-Control"),
reused: reused,
}
}
func acceptAllLookup(_ netip.Addr) (PeerIdentity, bool) {
return PeerIdentity{TunnelIP: testTunnelIP}, true
}
func newPrivateMiddleware(t *testing.T, validator SessionValidator, ipRestrictions *restrict.Filter) *Middleware {
t.Helper()
mw := NewMiddleware(log.StandardLogger(), validator, nil)
kp := generateTestKeyPair(t)
require.NoError(t, mw.AddDomain(testServerHost, nil, kp.PublicKey, time.Hour, "acct-1", "svc-1", ipRestrictions, true, nil))
return mw
}
// A rejected tunnel peer must emit the exact lowercase "close" token h2 matches on.
func TestProtect_PrivateService_DeniedSetsCloseHeaders(t *testing.T) {
mw := newPrivateMiddleware(t, &switchableTunnelValidator{}, nil)
handler := mw.Protect(newPassthroughHandler())
cd := proxy.NewCapturedData("")
cd.SetClientIP(testTunnelIP)
req := httptest.NewRequest(http.MethodGet, "http://"+testServerHost+"/", nil)
req.RemoteAddr = testTunnelIP.String() + ":5000"
req = req.WithContext(WithTunnelLookup(proxy.WithCapturedData(req.Context(), cd), acceptAllLookup))
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, req)
assert.Equal(t, http.StatusForbidden, rec.Code)
assert.Equal(t, "close", rec.Header().Get("Connection"), "private denial must ask the client to drop the connection")
assert.Equal(t, "no-store", rec.Header().Get("Cache-Control"), "private denial must not be cacheable")
}
// A denied client must not keep reusing the warm socket after joining the overlay.
func TestPrivateDeny_HTTP1_ClosesConnection(t *testing.T) {
validator := &switchableTunnelValidator{}
mw := newPrivateMiddleware(t, validator, nil)
srv := startProtectedServer(t, mw, testTunnelIP, acceptAllLookup, false)
client := srv.Client()
resp := doTraced(t, client, srv.URL)
assert.Equal(t, http.StatusForbidden, resp.status)
assert.Equal(t, 1, resp.protoMajor, "plain httptest server must speak HTTP/1.1")
// The Go client folds "Connection: close" into resp.close and drops the header.
assert.True(t, resp.close, "private denial must make the client mark the connection as not reusable")
assert.Equal(t, "no-store", resp.cacheControl, "private denial must not be cacheable")
validator.setValid(true)
resp2 := doTraced(t, client, srv.URL)
assert.Equal(t, http.StatusOK, resp2.status, "the retry must reach the upstream once the peer is valid")
assert.False(t, resp2.reused, "the retry must open a new connection")
}
// On HTTP/2 the header becomes a GOAWAY and the retry must use a new connection.
func TestPrivateDeny_HTTP2_SendsGoAway(t *testing.T) {
validator := &switchableTunnelValidator{}
mw := newPrivateMiddleware(t, validator, nil)
srv := startProtectedServer(t, mw, testTunnelIP, acceptAllLookup, true)
client := srv.Client()
resp := doTraced(t, client, srv.URL)
require.Equal(t, 2, resp.protoMajor, "test client must negotiate HTTP/2")
assert.Equal(t, http.StatusForbidden, resp.status)
assert.Empty(t, resp.connection, "HTTP/2 must not carry a Connection header on the wire")
assert.Equal(t, "no-store", resp.cacheControl)
validator.setValid(true)
resp2 := doTraced(t, client, srv.URL)
assert.Equal(t, 2, resp2.protoMajor)
assert.Equal(t, http.StatusOK, resp2.status, "the retry must reach the upstream once the peer is valid")
assert.False(t, resp2.reused, "GOAWAY must retire the connection so the retry opens a new one")
}
// Legitimate private traffic keeps its keep-alive connection.
func TestPrivateAllow_KeepsConnection(t *testing.T) {
validator := &switchableTunnelValidator{valid: true}
mw := newPrivateMiddleware(t, validator, nil)
srv := startProtectedServer(t, mw, testTunnelIP, acceptAllLookup, false)
client := srv.Client()
resp := doTraced(t, client, srv.URL)
assert.Equal(t, http.StatusOK, resp.status)
assert.Empty(t, resp.connection, "an allowed private request must not close the connection")
resp2 := doTraced(t, client, srv.URL)
assert.Equal(t, http.StatusOK, resp2.status)
assert.True(t, resp2.reused, "allowed private traffic must keep reusing the connection")
}
// Public denials keep the connection open; only private services change.
func TestPublicDeny_KeepsConnection(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
filter := restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"10.0.0.0/8"}})
require.NoError(t, mw.AddDomain(testServerHost, nil, "", 0, "acct-1", "svc-1", filter, false, nil))
srv := startProtectedServer(t, mw, netip.MustParseAddr("192.168.1.1"), nil, false)
client := srv.Client()
resp := doTraced(t, client, srv.URL)
assert.Equal(t, http.StatusForbidden, resp.status)
assert.Empty(t, resp.connection, "public denial must not close the connection")
assert.Empty(t, resp.cacheControl, "public denial must not gain cache headers")
resp2 := doTraced(t, client, srv.URL)
assert.Equal(t, http.StatusForbidden, resp2.status)
assert.True(t, resp2.reused, "public denials must keep reusing the connection")
}
// IP restriction denials on a private service must close the connection too.
func TestCheckIPRestrictions_PrivateDenialClosesConnection(t *testing.T) {
filter := restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"100.64.0.0/16"}})
mw := newPrivateMiddleware(t, &switchableTunnelValidator{valid: true}, filter)
handler := mw.Protect(newPassthroughHandler())
tests := []struct {
name string
remoteAddr string
}{
{"denied by CIDR", "100.65.5.6:5000"},
{"unresolvable client address", "not-an-ip:1234"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "http://"+testServerHost+"/", nil)
req.RemoteAddr = tt.remoteAddr
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, req)
assert.Equal(t, http.StatusForbidden, rec.Code)
assert.Equal(t, "close", rec.Header().Get("Connection"), "private IP-restriction denial must close the connection")
assert.Equal(t, "no-store", rec.Header().Get("Cache-Control"), "private IP-restriction denial must not be cacheable")
})
}
}
func TestCheckIPRestrictions_PublicDenialKeepsConnection(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
filter := restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"10.0.0.0/8"}})
require.NoError(t, mw.AddDomain(testServerHost, nil, "", 0, "acct-1", "svc-1", filter, false, nil))
handler := mw.Protect(newPassthroughHandler())
tests := []struct {
name string
remoteAddr string
}{
{"denied by CIDR", "192.168.1.1:5000"},
{"unresolvable client address", "not-an-ip:1234"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "http://"+testServerHost+"/", nil)
req.RemoteAddr = tt.remoteAddr
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, req)
assert.Equal(t, http.StatusForbidden, rec.Code)
assert.Empty(t, rec.Header().Get("Connection"), "public IP-restriction denial must not close the connection")
assert.Empty(t, rec.Header().Get("Cache-Control"), "public IP-restriction denial must not gain cache headers")
})
}
}
+6 -6
View File
File diff suppressed because one or more lines are too long
+6
View File
@@ -68,6 +68,12 @@ function App() {
if (res.type === "opaqueredirect" || res.status === 0) {
setSubmitting("redirect");
globalThis.location.reload();
} else if (res.status === 429) {
const seconds = Number(res.headers.get("Retry-After"));
const wait = Number.isFinite(seconds) && seconds > 0
? ` Try again in ${Math.ceil(seconds)} seconds.`
: " Please try again later.";
handleAuthError(method, `Too many authentication attempts.${wait}`);
} else {
handleAuthError(method, "Authentication failed. Please try again.");
}
+1 -1
View File
@@ -95,7 +95,7 @@ type Route struct {
ID ID `gorm:"primaryKey"`
// AccountID is a reference to Account that this object belongs
AccountID string `gorm:"index"`
PublicID string `json:"-"`
PublicID string `json:"-" gorm:"index"`
// Network and Domains are mutually exclusive
Network netip.Prefix `gorm:"serializer:json"`
Domains domain.List `gorm:"serializer:json"`
+7 -1
View File
@@ -35,7 +35,12 @@ type EnvelopeResult struct {
//
// dnsName is the account's DNS domain ("netbird.cloud" etc.); used when
// rebuilding the per-peer FQDNs that proto.RemotePeerConfig carries.
func EnvelopeToNetworkMap(ctx context.Context, env *proto.NetworkMapEnvelope, localPeerKey, dnsName string) (*EnvelopeResult, error) {
//
// skipRouteFirewallRules leaves RoutesFirewallRules empty. Callers that have
// no firewall to program pass true: the rules are the most expensive part of
// Calculate on a peer that routes many network resources, and nothing reads
// them afterwards.
func EnvelopeToNetworkMap(ctx context.Context, env *proto.NetworkMapEnvelope, localPeerKey, dnsName string, skipRouteFirewallRules bool) (*EnvelopeResult, error) {
components, err := DecodeEnvelope(ctx, env)
if err != nil {
return nil, fmt.Errorf("decode envelope: %w", err)
@@ -53,6 +58,7 @@ func EnvelopeToNetworkMap(ctx context.Context, env *proto.NetworkMapEnvelope, lo
return nil, fmt.Errorf("receiving peer (wg_key prefix %q) not found among %d decoded peers — components have no PeerID, Calculate would return empty", trimKey(localPeerKey), len(components.Peers))
}
components.PeerID = canonicalKey
components.SkipRouteFirewallRules = skipRouteFirewallRules
includeIPv6 := localPeer.SupportsIPv6() && localPeer.IPv6.IsValid()
useSourcePrefixes := localPeer.SupportsSourcePrefixes()
+116 -8
View File
@@ -9,6 +9,7 @@ import (
"net/netip"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
goproto "google.golang.org/protobuf/proto"
@@ -37,7 +38,7 @@ func TestEnvelopeToNetworkMap_RoundTrip(t *testing.T) {
var decoded proto.NetworkMapEnvelope
require.NoError(t, goproto.Unmarshal(wire, &decoded), "unmarshal envelope")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud", false)
require.NoError(t, err, "EnvelopeToNetworkMap")
require.NotNil(t, result)
require.NotNil(t, result.NetworkMap, "decoded NetworkMap must be non-nil")
@@ -78,7 +79,7 @@ func TestCalculate_FirewallRuleProtocol_NeverNetbirdSSH(t *testing.T) {
var decoded proto.NetworkMapEnvelope
require.NoError(t, goproto.Unmarshal(wire, &decoded))
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud", false)
require.NoError(t, err)
require.NotEmpty(t, result.NetworkMap.FirewallRules, "ssh policy should produce firewall rules")
for i, fr := range result.NetworkMap.FirewallRules {
@@ -88,13 +89,13 @@ func TestCalculate_FirewallRuleProtocol_NeverNetbirdSSH(t *testing.T) {
}
func TestEnvelopeToNetworkMap_NilEnvelope(t *testing.T) {
_, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), nil, "key", "netbird.cloud")
_, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), nil, "key", "netbird.cloud", false)
require.Error(t, err, "nil envelope must produce an error rather than panic")
}
func TestEnvelopeToNetworkMap_FullPayloadMissing(t *testing.T) {
env := &proto.NetworkMapEnvelope{}
_, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), env, "key", "netbird.cloud")
_, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), env, "key", "netbird.cloud", false)
require.Error(t, err, "envelope with no Full payload must produce an error")
}
@@ -126,7 +127,7 @@ func TestDecodeEnvelope_MalformedWgKeyPeerSkipped(t *testing.T) {
var decoded proto.NetworkMapEnvelope
require.NoError(t, goproto.Unmarshal(wire, &decoded), "unmarshal envelope")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud", false)
require.NoError(t, err, "EnvelopeToNetworkMap must tolerate one bad peer key")
require.NotNil(t, result)
require.NotNil(t, result.Components)
@@ -195,7 +196,7 @@ func TestEnvelopeRoundTrip_AllGroupShortCircuitParity(t *testing.T) {
var decodedEnv proto.NetworkMapEnvelope
require.NoError(t, goproto.Unmarshal(wire, &decodedEnv), "unmarshal envelope")
result, err := nbnetworkmap.EnvelopeToNetworkMap(ctx, &decodedEnv, peers["peer-T"].Key, "netbird.cloud")
result, err := nbnetworkmap.EnvelopeToNetworkMap(ctx, &decodedEnv, peers["peer-T"].Key, "netbird.cloud", false)
require.NoError(t, err, "EnvelopeToNetworkMap")
clientNM := result.NetworkMap
@@ -253,7 +254,7 @@ func TestEnvelopeToNetworkMap_EmptyComponents(t *testing.T) {
var decoded proto.NetworkMapEnvelope
require.NoError(t, goproto.Unmarshal(wire, &decoded), "unmarshal envelope")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud", false)
require.NoError(t, err, "EnvelopeToNetworkMap must degrade gracefully on empty components")
require.Equal(t, uint64(7), result.NetworkMap.Serial)
require.Empty(t, result.NetworkMap.RemotePeers, "unvalidated peer connects to nobody")
@@ -276,7 +277,7 @@ func TestEnvelopeToNetworkMap_MissingNetwork(t *testing.T) {
var decoded proto.NetworkMapEnvelope
require.NoError(t, goproto.Unmarshal(wire, &decoded), "unmarshal envelope")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud")
result, err := nbnetworkmap.EnvelopeToNetworkMap(context.Background(), &decoded, localPeerKey, "netbird.cloud", false)
require.NoError(t, err, "a missing AccountNetwork must not panic the client")
require.NotNil(t, result.Components.Network)
require.NotEmpty(t, result.NetworkMap.RemotePeers, "the rest of the snapshot stays usable")
@@ -353,3 +354,110 @@ func randomWgKey(t *testing.T) string {
require.NoError(t, err)
return base64.StdEncoding.EncodeToString(raw[:])
}
// TestEnvelopeToNetworkMap_SkipRouteFirewallRules covers the flag end to end,
// through the envelope rather than by poking Calculate directly. The
// RoutesFirewallRulesIsEmpty derivation is the part that matters: the client's
// legacy-management probe reads an empty rule list together with that bit, so
// skipping the rules must set it rather than leave it false.
func TestEnvelopeToNetworkMap_SkipRouteFirewallRules(t *testing.T) {
ctx := context.Background()
c, routerKey := buildRoutedResourceComponents(t)
envelope := mgmtgrpc.EncodeNetworkMapEnvelope(mgmtgrpc.ComponentsEnvelopeInput{
Components: c,
DNSDomain: "netbird.cloud",
})
wire, err := goproto.Marshal(envelope)
require.NoError(t, err, "marshal envelope")
var decoded proto.NetworkMapEnvelope
require.NoError(t, goproto.Unmarshal(wire, &decoded), "unmarshal envelope")
full, err := nbnetworkmap.EnvelopeToNetworkMap(ctx, &decoded, routerKey, "netbird.cloud", false)
require.NoError(t, err, "EnvelopeToNetworkMap without skip")
require.NotEmpty(t, full.NetworkMap.RoutesFirewallRules,
"baseline: the router peer must receive route firewall rules")
require.False(t, full.NetworkMap.RoutesFirewallRulesIsEmpty,
"baseline: the empty bit must be false when rules are present")
var decodedSkip proto.NetworkMapEnvelope
require.NoError(t, goproto.Unmarshal(wire, &decodedSkip), "unmarshal envelope")
skipped, err := nbnetworkmap.EnvelopeToNetworkMap(ctx, &decodedSkip, routerKey, "netbird.cloud", true)
require.NoError(t, err, "EnvelopeToNetworkMap with skip")
assert.Empty(t, skipped.NetworkMap.RoutesFirewallRules,
"route firewall rules must not be computed when skipped")
assert.True(t, skipped.NetworkMap.RoutesFirewallRulesIsEmpty,
"the empty bit must be derived from the skipped list, or the client misreads it as legacy management")
assert.Len(t, skipped.NetworkMap.Routes, len(full.NetworkMap.Routes),
"skipping route firewall rules must not change the routes")
assert.Len(t, skipped.NetworkMap.RemotePeers, len(full.NetworkMap.RemotePeers),
"skipping route firewall rules must not change the remote peers")
}
// buildRoutedResourceComponents returns components in which the local peer is
// the routing peer for one enabled network resource, reachable by a second
// peer through a resource policy — the minimum shape that yields a non-empty
// RoutesFirewallRules. It also returns the local peer's WG key.
func buildRoutedResourceComponents(t *testing.T) (*types.NetworkMapComponents, string) {
t.Helper()
routerKey := randomWgKey(t)
peers := map[string]*nmdata.Peer{
"peer-R": {
ID: "peer-R", Key: routerKey, DNSLabel: "router",
IP: netip.AddrFrom4([4]byte{100, 64, 0, 1}),
Meta: nmdata.PeerSystemMeta{WtVersion: "0.40.0"},
},
"peer-S": {
ID: "peer-S", Key: randomWgKey(t), DNSLabel: "source",
IP: netip.AddrFrom4([4]byte{100, 64, 0, 2}),
Meta: nmdata.PeerSystemMeta{WtVersion: "0.40.0"},
},
}
resourcePolicy := &nmdata.Policy{
ID: "pol-res", PublicID: "10", Enabled: true,
Rules: []*nmdata.PolicyRule{{
ID: "rule-res",
Enabled: true,
Action: string(types.PolicyTrafficActionAccept),
Protocol: string(types.PolicyRuleProtocolALL),
Sources: []string{"g-src"},
}},
}
c := &types.NetworkMapComponents{
PeerID: "peer-R",
Network: &nmdata.Network{
Identifier: "net-routed-resource",
Net: net.IPNet{IP: net.IP{100, 64, 0, 0}, Mask: net.CIDRMask(10, 32)},
Serial: 1,
},
AccountSettings: &nmdata.AccountSettingsInfo{},
DNSSettings: &nmdata.DNSSettings{},
Peers: peers,
Groups: map[string]*nmdata.Group{
"g-src": {PublicID: "1", Name: "sources", Peers: []string{"peer-S"}},
"g-routers": {PublicID: "2", Name: "routers", Peers: []string{"peer-R"}},
},
NetworkResources: []*nmdata.NetworkResource{{
ID: "res-1", NetworkID: "netid-1", PublicID: "100", Name: "res1",
Type: "subnet",
Prefix: netip.MustParsePrefix("10.200.0.0/24"),
Enabled: true,
}},
RoutersMap: map[string]map[string]*nmdata.NetworkRouter{
"netid-1": {"peer-R": {
PublicID: "200", PeerGroups: []string{"g-routers"}, Metric: 9999, Enabled: true,
}},
},
ResourcePoliciesMap: map[string][]*nmdata.Policy{
"res-1": {resourcePolicy},
},
Policies: []*nmdata.Policy{resourcePolicy},
NetworkXIDToPublicID: map[string]string{"netid-1": "1"},
}
return c, routerKey
}
+5
View File
@@ -135,6 +135,11 @@ func NewUserPendingApprovalError() error {
return Errorf(PermissionDenied, "user is pending approval")
}
// NewUserPendingApprovalByOwnerError creates a new Error with PermissionDenied type for a blocked user pending approval, naming the masked address of the owner who can approve them
func NewUserPendingApprovalByOwnerError(ownerEmail string) error {
return Errorf(PermissionDenied, "user is pending approval by owner %s", ownerEmail)
}
// NewPeerNotRegisteredError creates a new Error with Unauthenticated type unregistered peer
func NewPeerNotRegisteredError() error {
return Errorf(Unauthenticated, "peer is not registered")
@@ -58,6 +58,13 @@ type NetworkMapComponents struct {
// domain targets.
ForceRoutingPeerDNSResolution bool
// SkipRouteFirewallRules drops the route firewall rule computation from
// Calculate. A receiver without a firewall manager never reads
// RoutesFirewallRules, and on a routing peer with many network resources
// building them dominates the cost of a sync. Defaults to false so the
// management server keeps producing them.
SkipRouteFirewallRules bool
routesByPeerOnce sync.Once
routesByPeerIdx map[string][]routeIndexEntry
@@ -149,11 +156,15 @@ func (c *NetworkMapComponents) Calculate(ctx context.Context) *NetworkMap {
includeIPv6 = p.SupportsIPv6() && p.IPv6.IsValid()
}
routesUpdate := filterAndExpandRoutes(c.getRoutesToSync(targetPeerID, peersToConnect, peerGroups), includeIPv6)
routesFirewallRules := c.getPeerRoutesFirewallRules(ctx, targetPeerID, includeIPv6)
var routesFirewallRules []*RouteFirewallRule
if !c.SkipRouteFirewallRules {
routesFirewallRules = c.getPeerRoutesFirewallRules(ctx, targetPeerID, includeIPv6)
}
isRouter, networkResourcesRoutes, sourcePeers := c.getNetworkResourcesRoutesToSync(targetPeerID)
var networkResourcesFirewallRules []*RouteFirewallRule
if isRouter {
if isRouter && !c.SkipRouteFirewallRules {
networkResourcesFirewallRules = c.getPeerNetworkResourceFirewallRules(ctx, targetPeerID, networkResourcesRoutes, includeIPv6)
}
+19 -16
View File
@@ -30,6 +30,12 @@ const (
var (
ErrConnAlreadyExists = fmt.Errorf("connection already exists")
// ErrServerDisconnected is the cancellation cause of a relayed Conn when the
// client lost the connection to the relay server.
ErrServerDisconnected = fmt.Errorf("relay server disconnected")
// ErrPeerDisconnected is the cancellation cause of a relayed Conn when the
// remote peer went offline.
ErrPeerDisconnected = fmt.Errorf("remote peer disconnected")
)
type internalStopFlag struct {
@@ -74,16 +80,17 @@ type connContainer struct {
msgChanLock sync.Mutex
closed bool // flag to check if channel is closed
ctx context.Context
cancel context.CancelFunc
cancel context.CancelCauseFunc
}
func newConnContainer(log *log.Entry, c *Client, peerID messages.PeerID, instanceURL *RelayAddr) *connContainer {
ctx, cancel := context.WithCancel(context.Background())
ctx, cancel := context.WithCancelCause(context.Background())
msgChan := make(chan Msg, connChannelSize)
cn := &Conn{
dstID: peerID,
messageChan: msgChan,
instanceURL: instanceURL,
ctx: ctx,
}
cc := &connContainer{
log: log,
@@ -106,10 +113,6 @@ func newConnContainer(log *log.Entry, c *Client, peerID messages.PeerID, instanc
return cc
}
func (cc *connContainer) netConn() net.Conn {
return cc.conn
}
func (cc *connContainer) writeMsg(msg Msg) {
cc.msgChanLock.Lock()
defer cc.msgChanLock.Unlock()
@@ -128,8 +131,8 @@ func (cc *connContainer) writeMsg(msg Msg) {
}
}
func (cc *connContainer) close() {
cc.cancel()
func (cc *connContainer) close(cause error) {
cc.cancel(cause)
cc.msgChanLock.Lock()
defer cc.msgChanLock.Unlock()
@@ -293,12 +296,12 @@ func (c *Client) Connect(ctx context.Context) error {
return nil
}
// OpenConn create a new net.Conn for the destination peer ID. In case if the connection is in progress
// OpenConn create a new Conn for the destination peer ID. In case if the connection is in progress
// to the relay server, the function will block until the connection is established or timed out. Otherwise,
// it will return immediately.
// It block until the server confirm the peer is online.
// todo: what should happen if call with the same peerID with multiple times?
func (c *Client) OpenConn(ctx context.Context, dstPeerID string) (net.Conn, error) {
func (c *Client) OpenConn(ctx context.Context, dstPeerID string) (*Conn, error) {
peerID := messages.HashID(dstPeerID)
c.mu.Lock()
@@ -335,7 +338,7 @@ func (c *Client) OpenConn(ctx context.Context, dstPeerID string) (net.Conn, erro
delete(c.conns, peerID)
}
c.mu.Unlock()
container.close()
container.close(err)
return nil, err
}
@@ -345,13 +348,13 @@ func (c *Client) OpenConn(ctx context.Context, dstPeerID string) (net.Conn, erro
delete(c.conns, peerID)
}
c.mu.Unlock()
container.close()
container.close(ErrServerDisconnected)
return nil, fmt.Errorf("relay connection is not established")
}
c.mu.Unlock()
c.log.Infof("remote peer is available: %s", peerID)
return container.netConn(), nil
return container.conn, nil
}
// ServerInstanceURL returns the address of the relay server. It could change after the close and reopen the connection.
@@ -773,7 +776,7 @@ func (c *Client) serverInstanceAddress() (string, netip.Addr, error) {
func (c *Client) closeAllConns() {
for _, container := range c.conns {
container.close()
container.close(ErrServerDisconnected)
}
c.conns = make(map[messages.PeerID]*connContainer)
@@ -793,7 +796,7 @@ func (c *Client) closeConnsByPeerID(peerIDs []messages.PeerID) {
}
container.log.Infof("remote peer has been disconnected, free up connection: %s", peerID)
container.close()
container.close(ErrPeerDisconnected)
delete(c.conns, peerID)
}
@@ -821,7 +824,7 @@ func (c *Client) closeConn(containerRef *connContainer, id messages.PeerID) erro
c.log.Infof("free up connection to peer: %s", id)
delete(c.conns, id)
current.close()
current.close(net.ErrClosed)
return nil
}
+10
View File
@@ -1,6 +1,7 @@
package client
import (
"context"
"net"
"time"
@@ -12,11 +13,20 @@ type Conn struct {
dstID messages.PeerID
messageChan chan Msg
instanceURL *RelayAddr
ctx context.Context
writeFn func(messages.PeerID, []byte) (int, error)
closeFn func(messages.PeerID) error
localAddrFn func() net.Addr
}
// Context returns a context that is cancelled when the connection is torn down,
// either by Close or by the relay client losing the server connection. The
// cancellation cause carries the reason, see ErrServerDisconnected and
// ErrPeerDisconnected.
func (c *Conn) Context() context.Context {
return c.ctx
}
func (c *Conn) Write(p []byte) (n int, err error) {
return c.writeFn(c.dstID, p)
}
+8 -74
View File
@@ -1,12 +1,9 @@
package client
import (
"container/list"
"context"
"fmt"
"net"
"net/netip"
"reflect"
"sync"
"time"
@@ -43,8 +40,6 @@ func NewRelayTrack() *RelayTrack {
}
}
type OnServerCloseListener func()
// ManagerOption configures a Manager at construction time.
type ManagerOption func(*Manager)
@@ -91,7 +86,6 @@ type Manager struct {
relayClients map[string]*RelayTrack
relayClientsMutex sync.RWMutex
onDisconnectedListeners map[string]*list.List
onReconnectedListenerFn func()
listenerLock sync.Mutex
@@ -126,10 +120,9 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
ConnectionTimeout: defaultConnectionTimeout,
TransportFallback: tf,
},
relayClients: make(map[string]*RelayTrack),
onDisconnectedListeners: make(map[string]*list.List),
cleanupInterval: relayCleanupInterval,
keepUnusedServerTime: keepUnusedServerTime,
relayClients: make(map[string]*RelayTrack),
cleanupInterval: relayCleanupInterval,
keepUnusedServerTime: keepUnusedServerTime,
}
for _, opt := range opts {
opt(m)
@@ -168,11 +161,11 @@ func (m *Manager) Serve() error {
// OpenConn opens a connection to the given peer key. If the peer is on the same relay server, the connection will be
// established via the relay server. If the peer is on a different relay server, the manager will establish a new
// connection to the relay server. It returns back with a net.Conn what represent the remote peer connection.
// connection to the relay server. It returns the relayed connection to the remote peer.
//
// serverIP, when valid and serverAddress is foreign, is used as a dial target if the FQDN-based dial fails.
// Ignored for the local home-server path. TLS verification still uses the FQDN via SNI.
func (m *Manager) OpenConn(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (net.Conn, error) {
func (m *Manager) OpenConn(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (*Conn, error) {
m.relayClientMu.RLock()
defer m.relayClientMu.RUnlock()
@@ -185,9 +178,7 @@ func (m *Manager) OpenConn(ctx context.Context, serverAddress, peerKey string, s
return nil, err
}
var (
netConn net.Conn
)
var netConn *Conn
if !foreign {
log.Debugf("open peer connection via permanent server: %s", peerKey)
netConn, err = m.relayClient.OpenConn(ctx, peerKey)
@@ -220,31 +211,6 @@ func (m *Manager) SetOnReconnectedListener(f func()) {
m.onReconnectedListenerFn = f
}
// AddCloseListener adds a listener to the given server instance address. The listener will be called if the connection
// closed.
func (m *Manager) AddCloseListener(serverAddress string, onClosedListener OnServerCloseListener) error {
m.relayClientMu.RLock()
defer m.relayClientMu.RUnlock()
if m.relayClient == nil {
return ErrRelayClientNotConnected
}
foreign, err := m.isForeignServer(serverAddress)
if err != nil {
return err
}
var listenerAddr string
if foreign {
listenerAddr = serverAddress
} else {
listenerAddr = m.relayClient.connectionURL
}
m.addListener(listenerAddr, onClosedListener)
return nil
}
// RelayInstanceAddress returns the address and resolved IP of the permanent relay server. It could change if the
// network connection is lost. The address is sent to the target peer to choose the common relay server for the
// communication; the IP is sent alongside so remote peers can dial directly without their own DNS lookup. Both
@@ -330,7 +296,7 @@ func (m *Manager) UpdateToken(token *relayAuth.Token) error {
return m.tokenStore.UpdateToken(token)
}
func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (net.Conn, error) {
func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (*Conn, error) {
// check if already has a connection to the desired relay server
m.relayClientsMutex.RLock()
rt, ok := m.relayClients[serverAddress]
@@ -383,7 +349,7 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
// waiting for the dial started by another openConnVia call to finish. It waits
// on rt.ready rather than the track lock, so it neither holds nor contends the
// track lock across the dial.
func (m *Manager) openConnOnTrack(ctx context.Context, rt *RelayTrack, peerKey string) (net.Conn, error) {
func (m *Manager) openConnOnTrack(ctx context.Context, rt *RelayTrack, peerKey string) (*Conn, error) {
select {
case <-rt.ready:
case <-ctx.Done():
@@ -428,8 +394,6 @@ func (m *Manager) onServerDisconnected(serverAddress string) {
if !isHome {
m.evictForeignRelay(serverAddress)
}
m.notifyOnDisconnectListeners(serverAddress)
}
func (m *Manager) evictForeignRelay(serverAddress string) {
@@ -523,36 +487,6 @@ func (m *Manager) cleanUpUnusedRelays() {
}
}
func (m *Manager) addListener(serverAddress string, onClosedListener OnServerCloseListener) {
m.listenerLock.Lock()
defer m.listenerLock.Unlock()
l, ok := m.onDisconnectedListeners[serverAddress]
if !ok {
l = list.New()
}
for e := l.Front(); e != nil; e = e.Next() {
if reflect.ValueOf(e.Value).Pointer() == reflect.ValueOf(onClosedListener).Pointer() {
return
}
}
l.PushBack(onClosedListener)
m.onDisconnectedListeners[serverAddress] = l
}
func (m *Manager) notifyOnDisconnectListeners(serverAddress string) {
m.listenerLock.Lock()
defer m.listenerLock.Unlock()
l, ok := m.onDisconnectedListeners[serverAddress]
if !ok {
return
}
for e := l.Front(); e != nil; e = e.Next() {
go e.Value.(OnServerCloseListener)()
}
delete(m.onDisconnectedListeners, serverAddress)
}
func relayConnState(c *Client) RelayConnState {
addr, err := c.ServerInstanceURL()
if err != nil {
+107 -50
View File
@@ -2,7 +2,9 @@ package client
import (
"context"
"errors"
"fmt"
"net"
"net/netip"
"testing"
"time"
@@ -291,35 +293,29 @@ func TestForeignAutoClose(t *testing.T) {
t.Fatalf("failed to serve manager: %s", err)
}
// Set up a disconnect listener to track when foreign server disconnects
foreignServerURL := toURL(srvCfg2)[0]
disconnected := make(chan struct{})
onDisconnect := func() {
select {
case disconnected <- struct{}{}:
default:
}
}
t.Log("open connection to another peer")
if _, err = mgr.OpenConn(ctx, foreignServerURL, "anotherpeer", netip.Addr{}); err == nil {
t.Fatalf("should have failed to open connection to another peer")
}
// Add the disconnect listener after the connection attempt
if err := mgr.AddCloseListener(foreignServerURL, onDisconnect); err != nil {
t.Logf("failed to add close listener (expected if connection failed): %s", err)
}
// Wait for cleanup to happen
timeout := relayCleanupInterval + keepUnusedServerTime + 2*time.Second
t.Logf("waiting for relay cleanup: %s", timeout)
select {
case <-disconnected:
t.Log("foreign relay connection cleaned up successfully")
case <-time.After(timeout):
t.Log("timeout waiting for cleanup - this might be expected if connection never established")
deadline := time.After(timeout)
for {
mgr.relayClientsMutex.RLock()
_, tracked := mgr.relayClients[foreignServerURL]
mgr.relayClientsMutex.RUnlock()
if !tracked {
t.Log("foreign relay connection cleaned up successfully")
break
}
select {
case <-deadline:
t.Fatal("foreign relay was not cleaned up")
case <-time.After(200 * time.Millisecond):
}
}
t.Logf("closing manager")
@@ -413,23 +409,24 @@ func waitForReady(ctx context.Context, m *Manager, timeout time.Duration) error
return fmt.Errorf("manager not ready within %s", timeout)
}
func TestNotifierDoubleAdd(t *testing.T) {
func toURL(address server.ListenerConfig) []string {
return []string{"rel://" + address.Address}
}
func TestConnContextCancelledOnServerDisconnect(t *testing.T) {
ctx := context.Background()
listenerCfg1 := server.ListenerConfig{
Address: "localhost:52501",
}
srv, err := server.NewServer(newManagerTestServerConfig(listenerCfg1.Address))
srvCfg := server.ListenerConfig{Address: "localhost:52601"}
srv, err := server.NewServer(newManagerTestServerConfig(srvCfg.Address))
if err != nil {
t.Fatalf("failed to create server: %s", err)
}
errChan := make(chan error, 1)
go func() {
if err := srv.Listen(listenerCfg1); err != nil {
if err := srv.Listen(srvCfg); err != nil {
errChan <- err
}
}()
defer func() {
if err := srv.Shutdown(ctx); err != nil {
t.Errorf("failed to close server: %s", err)
@@ -440,46 +437,106 @@ func TestNotifierDoubleAdd(t *testing.T) {
t.Fatalf("failed to start server: %s", err)
}
log.Debugf("connect by alice")
mCtx, cancel := context.WithCancel(ctx)
defer cancel()
clientBob := NewManager(mCtx, toURL(listenerCfg1), "bob", iface.DefaultMTU)
if err = clientBob.Serve(); err != nil {
mgrBob := NewManager(mCtx, toURL(srvCfg), "bob", iface.DefaultMTU)
if err := mgrBob.Serve(); err != nil {
t.Fatalf("failed to serve bob manager: %s", err)
}
mgr := NewManager(mCtx, toURL(srvCfg), "alice", iface.DefaultMTU)
if err := mgr.Serve(); err != nil {
t.Fatalf("failed to serve manager: %s", err)
}
clientAlice := NewManager(mCtx, toURL(listenerCfg1), "alice", iface.DefaultMTU)
if err = clientAlice.Serve(); err != nil {
ra, _, err := mgr.RelayInstanceAddress()
if err != nil {
t.Fatalf("failed to get relay address: %s", err)
}
relayedConn, err := mgr.OpenConn(ctx, ra, "bob", netip.Addr{})
if err != nil {
t.Fatalf("failed to open conn: %s", err)
}
select {
case <-relayedConn.Context().Done():
t.Fatal("conn context cancelled while the relay is still up")
default:
}
_ = mgr.relayClient.relayConn.Close()
select {
case <-relayedConn.Context().Done():
case <-time.After(15 * time.Second):
t.Fatal("conn context was not cancelled after the relay connection dropped")
}
if cause := context.Cause(relayedConn.Context()); !errors.Is(cause, ErrServerDisconnected) {
t.Errorf("unexpected cancellation cause: %v, want %v", cause, ErrServerDisconnected)
}
}
func TestConnContextCauseOnLocalClose(t *testing.T) {
ctx := context.Background()
srvCfg := server.ListenerConfig{Address: "localhost:52602"}
srv, err := server.NewServer(newManagerTestServerConfig(srvCfg.Address))
if err != nil {
t.Fatalf("failed to create server: %s", err)
}
errChan := make(chan error, 1)
go func() {
if err := srv.Listen(srvCfg); err != nil {
errChan <- err
}
}()
defer func() {
if err := srv.Shutdown(ctx); err != nil {
t.Errorf("failed to close server: %s", err)
}
}()
if err := waitForServerToStart(errChan); err != nil {
t.Fatalf("failed to start server: %s", err)
}
mCtx, cancel := context.WithCancel(ctx)
defer cancel()
mgrBob := NewManager(mCtx, toURL(srvCfg), "bob", iface.DefaultMTU)
if err := mgrBob.Serve(); err != nil {
t.Fatalf("failed to serve bob manager: %s", err)
}
mgr := NewManager(mCtx, toURL(srvCfg), "alice", iface.DefaultMTU)
if err := mgr.Serve(); err != nil {
t.Fatalf("failed to serve manager: %s", err)
}
conn1, err := clientAlice.OpenConn(ctx, clientAlice.ServerURLs()[0], "bob", netip.Addr{})
ra, _, err := mgr.RelayInstanceAddress()
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
t.Fatalf("failed to get relay address: %s", err)
}
fnCloseListener := OnServerCloseListener(func() {
log.Infof("close listener")
})
err = clientAlice.AddCloseListener(clientAlice.ServerURLs()[0], fnCloseListener)
relayedConn, err := mgr.OpenConn(ctx, ra, "bob", netip.Addr{})
if err != nil {
t.Fatalf("failed to add close listener: %s", err)
t.Fatalf("failed to open conn: %s", err)
}
err = clientAlice.AddCloseListener(clientAlice.ServerURLs()[0], fnCloseListener)
if err != nil {
t.Fatalf("failed to add close listener: %s", err)
if err := relayedConn.Close(); err != nil {
t.Fatalf("failed to close conn: %s", err)
}
err = conn1.Close()
if err != nil {
t.Errorf("failed to close connection: %s", err)
select {
case <-relayedConn.Context().Done():
case <-time.After(5 * time.Second):
t.Fatal("conn context was not cancelled after a local close")
}
}
func toURL(address server.ListenerConfig) []string {
return []string{"rel://" + address.Address}
if cause := context.Cause(relayedConn.Context()); !errors.Is(cause, net.ErrClosed) {
t.Errorf("unexpected cancellation cause after a local close: %v, want %v", cause, net.ErrClosed)
}
}