Merge branch 'reverse-proxy-allow-match-or' into reverse-proxy-crowdsec-appsec

# Conflicts:
#	proxy/internal/auth/middleware.go
#	proxy/internal/auth/middleware_test.go
#	proxy/internal/auth/tunnel_lookup_test.go
#	proxy/management_integration_test.go
#	proxy/server.go
#	shared/management/proto/proxy_service.pb.go
This commit is contained in:
Viktor Liu
2026-09-23 07:53:31 +02:00
968 changed files with 72582 additions and 16724 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25-alpine AS builder
FROM golang:1.26.7-alpine AS builder
WORKDIR /app
RUN echo "netbird:x:1000:1000:netbird:/var/lib/netbird:/sbin/nologin" > /tmp/passwd && \
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25-alpine AS builder
FROM golang:1.26.7-alpine AS builder
WORKDIR /app
COPY go.mod go.sum ./
+12 -4
View File
@@ -2,12 +2,14 @@ package acme
import (
"context"
"fmt"
"path/filepath"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/proxy/internal/flock"
"github.com/netbirdio/netbird/proxy/internal/k8s"
"github.com/netbirdio/netbird/shared/management/domain"
)
// certLocker provides distributed mutual exclusion for certificate operations.
@@ -74,9 +76,15 @@ func newFlockLocker(certDir string, logger *log.Logger) *flockLocker {
return &flockLocker{certDir: certDir, logger: logger}
}
// Lock acquires an advisory file lock for the given domain.
func (l *flockLocker) Lock(ctx context.Context, domain string) (func(), error) {
lockPath := filepath.Join(l.certDir, domain+".lock")
// Lock acquires an advisory file lock for the given domain. The domain must
// be a valid hostname so the lock file always resolves to a direct child of
// certDir; anything else is rejected before touching the filesystem.
func (l *flockLocker) Lock(ctx context.Context, name string) (func(), error) {
if !domain.IsValidDomainNoWildcard(name) {
return nil, fmt.Errorf("invalid domain %q for lock file", name)
}
lockPath := filepath.Join(l.certDir, name+".lock")
lockFile, err := flock.Lock(ctx, lockPath)
if err != nil {
return nil, err
@@ -89,7 +97,7 @@ func (l *flockLocker) Lock(ctx context.Context, domain string) (func(), error) {
return func() {
if err := flock.Unlock(lockFile); err != nil {
l.logger.Debugf("release cert lock for domain %q: %v", domain, err)
l.logger.Debugf("release cert lock for domain %q: %v", name, err)
}
}, nil
}
+30
View File
@@ -63,3 +63,33 @@ func TestNewCertLockerK8sFallsBackToFlock(t *testing.T) {
_, ok := locker.(*flockLocker)
assert.True(t, ok, "k8s-lease without SA should fall back to flockLocker")
}
func TestFlockLockerRejectsUnsafeDomain(t *testing.T) {
root := t.TempDir()
certDir := filepath.Join(root, "certs")
require.NoError(t, os.Mkdir(certDir, 0o700))
locker := newFlockLocker(certDir, nil)
for _, d := range []string{
"",
".",
"..",
"../escape",
"../../etc/cron.d/attacker",
"sub/dir.example.com",
`back\slash.example.com`,
"*.example.com",
} {
unlock, err := locker.Lock(context.Background(), d)
assert.Error(t, err, "domain %q", d)
assert.Nil(t, unlock, "domain %q", d)
}
assert.NoFileExists(t, filepath.Join(root, "escape.lock"))
certEntries, err := os.ReadDir(certDir)
require.NoError(t, err)
assert.Empty(t, certEntries)
rootEntries, err := os.ReadDir(root)
require.NoError(t, err)
assert.Len(t, rootEntries, 1)
}
+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", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: keys.PublicKey, SessionExpiration: time.Hour, AccountID: "account", ServiceID: "service"}))
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", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: keys.PublicKey, AccountID: "account", ServiceID: "service"}))
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", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: keys.PublicKey, AccountID: "account", ServiceID: "service"}))
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", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: keys.PublicKey, AccountID: "account", ServiceID: "service"}))
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")
})
}
}
+94 -12
View File
@@ -73,6 +73,11 @@ type DomainConfig struct {
redactBodyFields []string
redactHeaders []string
redactQueryParams []string
// AllowedGroups holds the group ids that may reach the service through an
// OIDC identity. When non-empty, a session cookie is honoured only if its
// groups claim intersects this set. Empty means group membership does not
// restrict access.
AllowedGroups map[string]struct{}
}
type validationResult struct {
@@ -96,6 +101,7 @@ type Middleware struct {
sessionValidator SessionValidator
geo restrict.GeoResolver
tunnelCache *tunnelValidationCache
credentials *credentialLimiter
// appsec is the shared CrowdSec AppSec client, nil when the proxy has no
// AppSec endpoint configured. Set once during startup, before serving.
appsec *appsec.Client
@@ -113,6 +119,7 @@ func NewMiddleware(logger *log.Logger, sessionValidator SessionValidator, geo re
sessionValidator: sessionValidator,
geo: geo,
tunnelCache: newTunnelValidationCache(),
credentials: newCredentialLimiter(),
}
}
@@ -159,7 +166,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
}
@@ -254,7 +261,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
}
@@ -289,7 +296,7 @@ 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
}
@@ -333,7 +340,7 @@ func (mw *Middleware) checkAppSec(w http.ResponseWriter, r *http.Request, config
mw.markDenied(r, verdict.String())
mw.logger.Debugf("AppSec: %s for %s %s", verdict, r.Host, r.RemoteAddr)
http.Error(w, "Forbidden", http.StatusForbidden)
denyForbidden(w, config)
return false, release
}
@@ -421,6 +428,26 @@ func credentialHeaders(schemes []Scheme) []string {
return names
}
// 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 {
@@ -481,6 +508,9 @@ func (mw *Middleware) handleOAuthCallbackError(w http.ResponseWriter, r *http.Re
// forwardWithSessionCookie checks for a valid session cookie and, if found,
// sets the user identity on the request context and forwards to the next handler.
// A signature-valid cookie is not on its own a grant: an OIDC session must also
// carry a group the service allows, so a token cannot be replayed past the
// group check that gated the login it came from.
func (mw *Middleware) forwardWithSessionCookie(w http.ResponseWriter, r *http.Request, host string, config DomainConfig, next http.Handler) bool {
cookie, err := r.Cookie(auth.SessionCookieName)
if err != nil {
@@ -500,6 +530,14 @@ func (mw *Middleware) forwardWithSessionCookie(w http.ResponseWriter, r *http.Re
return false
}
if !sessionGroupsAllowed(config.AllowedGroups, auth.Method(method), groups) {
mw.logger.WithFields(log.Fields{
"host": host,
"user_id": userID,
}).Debug("session cookie rejected: groups claim does not intersect the service's allowed groups")
return false
}
if cd := proxy.CapturedDataFromContext(r.Context()); cd != nil {
cd.SetUserID(userID)
cd.SetUserEmail(email)
@@ -672,13 +710,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
}
@@ -779,9 +813,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(sessionTokenParam) != ""
}
@@ -802,6 +836,9 @@ type DomainSettings struct {
// of the schemes list.
Private bool
AppSecMode restrict.AppSecMode
// AllowedGroups restricts OIDC sessions to the given group ids; empty means
// unrestricted.
AllowedGroups []string
}
// AddDomain registers authentication schemes for the given domain. With schemes
@@ -814,6 +851,7 @@ func (mw *Middleware) AddDomain(domain string, settings DomainSettings) error {
IPRestrictions: settings.IPRestrictions,
Private: settings.Private,
AppSecMode: settings.AppSecMode,
AllowedGroups: groupSet(settings.AllowedGroups),
redactBodyFields: credentialFields,
redactHeaders: credentialHeaders(settings.Schemes),
// A credential can arrive in the query too: r.FormValue merges the URL
@@ -888,6 +926,50 @@ func (mw *Middleware) validateSessionToken(ctx context.Context, host, token stri
return &validationResult{UserID: userID, UserEmail: email, Valid: true, Groups: groups, GroupNames: groupNames}, nil
}
// groupSet builds the lookup set the cookie path consults, returning nil for an
// empty list so callers can test membership restriction with len().
func groupSet(groups []string) map[string]struct{} {
if len(groups) == 0 {
return nil
}
set := make(map[string]struct{}, len(groups))
for _, g := range groups {
if g != "" {
set[g] = struct{}{}
}
}
if len(set) == 0 {
return nil
}
return set
}
// sessionGroupsAllowed reports whether a session token's groups claim satisfies
// the service's allowed groups. Only OIDC sessions are gated: password, PIN and
// header credentials carry no group identity and are authorised by the secret
// itself, which mirrors how management validates them. A token minted before the
// groups claim existed carries none and is therefore denied on a group-restricted
// service, which sends the user back through login for a fresh decision. A method
// this build doesn't know carries no such argument, so it is denied.
func sessionGroupsAllowed(allowed map[string]struct{}, method auth.Method, groups []string) bool {
if len(allowed) == 0 {
return true
}
switch method {
case auth.MethodPassword, auth.MethodPIN, auth.MethodHeader:
return true
case auth.MethodOIDC:
for _, g := range groups {
if _, ok := allowed[g]; ok {
return true
}
}
return false
default:
return false
}
}
// stripSessionTokenParam returns the request URI with the session_token query
// parameter removed so it doesn't linger in the browser's address bar or history.
func stripSessionTokenParam(u *url.URL) string {
+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, DomainSettings{SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acct-1", ServiceID: "svc-1", IPRestrictions: ipRestrictions, Private: true}))
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, DomainSettings{AccountID: "acct-1", ServiceID: "svc-1", IPRestrictions: filter}))
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, DomainSettings{AccountID: "acct-1", ServiceID: "svc-1", IPRestrictions: filter}))
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")
})
}
}
+167
View File
@@ -0,0 +1,167 @@
package auth
import (
"context"
"crypto/tls"
"net/http"
"net/http/httptest"
"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/management/internals/modules/reverseproxy/sessionkey"
"github.com/netbirdio/netbird/proxy/auth"
"github.com/netbirdio/netbird/proxy/internal/proxy"
"github.com/netbirdio/netbird/shared/management/proto"
)
// denyingSessionValidator mimics management for a user who completed OIDC login
// but is outside the service's distribution groups: ValidateSession denies.
type denyingSessionValidator struct {
calls int
}
func (d *denyingSessionValidator) ValidateSession(context.Context, *proto.ValidateSessionRequest, ...grpc.CallOption) (*proto.ValidateSessionResponse, error) {
d.calls++
return &proto.ValidateSessionResponse{Valid: false, UserId: "user-1", DeniedReason: "not_in_group"}, nil
}
func (d *denyingSessionValidator) ValidateTunnelPeer(context.Context, *proto.ValidateTunnelPeerRequest, ...grpc.CallOption) (*proto.ValidateTunnelPeerResponse, error) {
return &proto.ValidateTunnelPeerResponse{Valid: false}, nil
}
// TestProtect_SelfInstalledCookieCannotBypassGroupCheck is the regression guard
// for the group-authorisation bypass: a user denied at login still holds the raw
// session token from the ?session_token= redirect, so pasting it into the
// nb_session cookie must not buy access. The cookie path validated only the JWT
// signature, which turned the token management had already refused into a bearer
// credential for the service.
func TestProtect_SelfInstalledCookieCannotBypassGroupCheck(t *testing.T) {
validator := &denyingSessionValidator{}
mw := NewMiddleware(log.StandardLogger(), validator, nil)
kp := generateTestKeyPair(t)
oidc := &stubScheme{method: auth.MethodOIDC, authFn: func(r *http.Request) (string, string, error) {
return r.URL.Query().Get("session_token"), "https://idp.example/authorize", nil
}}
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{oidc}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acct-1", ServiceID: "svc-1", AllowedGroups: []string{"grp-allowed"}}))
// The token a denied user gets to see: validly signed for this service and
// domain, but carrying no group the service allows.
token, err := sessionkey.SignToken(kp.PrivateKey, "user-1", "john.doe@example.com", "example.com", auth.MethodOIDC, nil, nil, time.Hour)
require.NoError(t, err)
backendHits := 0
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
backendHits++
w.WriteHeader(http.StatusOK)
}))
t.Run("token in the callback URL is denied", func(t *testing.T) {
rec := serveWithCookie(t, handler, "https://example.com/?session_token="+token, nil)
assert.Equal(t, http.StatusForbidden, rec.Code, "group check must deny the login")
assert.Empty(t, rec.Result().Cookies(), "a denied login must not install a session cookie")
})
t.Run("same token pasted into the session cookie is denied", func(t *testing.T) {
rec := serveWithCookie(t, handler, "https://example.com/", &http.Cookie{Name: auth.SessionCookieName, Value: token})
assert.NotEqual(t, http.StatusOK, rec.Code, "a self-installed cookie must not reach the backend")
assert.Equal(t, 0, backendHits, "backend must never be reached without an allowed group")
})
}
// TestProtect_SessionCookieWithAllowedGroupPassesThrough is the positive half of
// the group gate: a member of an allowed group keeps the cookie fast-path, with
// no management round-trip.
func TestProtect_SessionCookieWithAllowedGroupPassesThrough(t *testing.T) {
validator := &denyingSessionValidator{}
mw := NewMiddleware(log.StandardLogger(), validator, nil)
kp := generateTestKeyPair(t)
oidc := &stubScheme{method: auth.MethodOIDC}
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{oidc}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acct-1", ServiceID: "svc-1", AllowedGroups: []string{"grp-other", "grp-allowed"}}))
token, err := sessionkey.SignToken(kp.PrivateKey, "user-2", "jane@example.com", "example.com", auth.MethodOIDC,
[]string{"grp-unrelated", "grp-allowed"}, []string{"Unrelated", "Allowed"}, time.Hour)
require.NoError(t, err)
handler := mw.Protect(newPassthroughHandler())
rec := serveWithCookie(t, handler, "https://example.com/", &http.Cookie{Name: auth.SessionCookieName, Value: token})
assert.Equal(t, http.StatusOK, rec.Code, "a cookie carrying an allowed group must pass through")
assert.Equal(t, 0, validator.calls, "the cookie fast-path must not call management")
}
// TestProtect_NonOIDCSessionCookieIgnoresGroupRestriction locks the scope of the
// gate: PIN, password and header credentials carry no group identity and are
// authorised by the secret itself, exactly as management validates them.
func TestProtect_NonOIDCSessionCookieIgnoresGroupRestriction(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acct-1", ServiceID: "svc-1", AllowedGroups: []string{"grp-allowed"}}))
token, err := sessionkey.SignToken(kp.PrivateKey, "pin-user", "", "example.com", auth.MethodPIN, nil, nil, time.Hour)
require.NoError(t, err)
handler := mw.Protect(newPassthroughHandler())
rec := serveWithCookie(t, handler, "https://example.com/", &http.Cookie{Name: auth.SessionCookieName, Value: token})
assert.Equal(t, http.StatusOK, rec.Code, "a PIN session must not be gated on OIDC group membership")
}
func TestSessionGroupsAllowed(t *testing.T) {
allowed := groupSet([]string{"a", "b"})
tests := []struct {
name string
allowed map[string]struct{}
method auth.Method
groups []string
want bool
}{
{"unrestricted service allows a groupless token", nil, auth.MethodOIDC, nil, true},
{"restricted service allows an intersecting token", allowed, auth.MethodOIDC, []string{"c", "b"}, true},
{"restricted service denies a disjoint token", allowed, auth.MethodOIDC, []string{"c"}, false},
{"restricted service denies a groupless token", allowed, auth.MethodOIDC, nil, false},
{"restricted service ignores a pin token", allowed, auth.MethodPIN, nil, true},
{"restricted service ignores a password token", allowed, auth.MethodPassword, nil, true},
{"restricted service ignores a header token", allowed, auth.MethodHeader, nil, true},
{"restricted service denies an unknown method", allowed, auth.Method("totp"), []string{"a"}, false},
{"restricted service denies a token with no method", allowed, auth.Method(""), []string{"a"}, false},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
assert.Equal(t, tc.want, sessionGroupsAllowed(tc.allowed, tc.method, tc.groups))
})
}
}
func TestGroupSetDropsEmptyEntries(t *testing.T) {
assert.Nil(t, groupSet(nil), "no groups means unrestricted")
assert.Nil(t, groupSet([]string{"", ""}), "blank ids must not restrict access to nothing reachable")
assert.Equal(t, map[string]struct{}{"a": {}}, groupSet([]string{"a", ""}))
}
// serveWithCookie drives the middleware over TLS with captured data attached,
// optionally carrying a session cookie.
func serveWithCookie(t *testing.T, handler http.Handler, url string, cookie *http.Cookie) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodGet, url, nil)
req.TLS = &tls.ConnectionState{}
if cookie != nil {
req.AddCookie(cookie)
}
req = req.WithContext(proxy.WithCapturedData(req.Context(), proxy.NewCapturedData("")))
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, req)
return rec
}
+176 -13
View File
@@ -105,6 +105,20 @@ type Handler struct {
startTime time.Time
templates *template.Template
templateMu sync.RWMutex
// setPerformance applies a buffer cap to one client. Held as a field so
// tests can drive applyBufferCap without a live embedded client.
setPerformance func(*nbembed.Client, uint32) error
perfMu sync.Mutex
perfInflight map[types.AccountID]*perfWorker
}
// perfWorker is the single in-flight retune for one account. err is valid once
// done is closed.
type perfWorker struct {
done chan struct{}
err error
}
// NewHandler creates a new debug handler.
@@ -113,10 +127,11 @@ func NewHandler(provider clientProvider, healthChecker healthChecker, logger *lo
logger = log.StandardLogger()
}
h := &Handler{
provider: provider,
health: healthChecker,
logger: logger,
startTime: time.Now(),
provider: provider,
health: healthChecker,
logger: logger,
startTime: time.Now(),
setPerformance: setClientPerformance,
}
if err := h.loadTemplates(); err != nil {
logger.Errorf("failed to load embedded templates: %v", err)
@@ -716,15 +731,7 @@ func (h *Handler) handlePerf(w http.ResponseWriter, r *http.Request) {
}
capN := uint32(n)
applied := 0
failed := map[string]string{}
for accountID, client := range h.provider.ListClientsForStartup() {
if err := client.SetPerformance(nbembed.Performance{PreallocatedBuffersPerPool: &capN}); err != nil {
failed[string(accountID)] = err.Error()
continue
}
applied++
}
applied, failed, inFlight := h.applyBufferCap(capN)
resp := map[string]any{
"success": true,
@@ -734,9 +741,165 @@ func (h *Handler) handlePerf(w http.ResponseWriter, r *http.Request) {
if len(failed) > 0 {
resp["failed"] = failed
}
if len(inFlight) > 0 {
resp["in_flight"] = inFlight
}
h.writeJSON(w, resp)
}
// perfApplyTimeout bounds the whole apply, however many clients are registered.
// A var, not a const, so tests can shorten the wait.
var perfApplyTimeout = 5 * time.Second
type perfResult struct {
accountID types.AccountID
err error
}
// setClientPerformance is the production implementation behind Handler.setPerformance.
func setClientPerformance(client *nbembed.Client, capN uint32) error {
return client.SetPerformance(nbembed.Performance{PreallocatedBuffersPerPool: &capN})
}
// collectBuffered takes every result already sitting in the channel, removing
// those accounts from pending, and returns how many of them succeeded. It is
// called when the deadline fires: select picks at random among ready cases, so
// a result that landed in time would otherwise be reported as a timeout.
func collectBuffered(results <-chan perfResult, pending map[types.AccountID]*perfWorker, failed map[string]string) int {
applied := 0
for {
select {
case res := <-results:
delete(pending, res.accountID)
if res.err != nil {
failed[string(res.accountID)] = res.err.Error()
continue
}
applied++
default:
return applied
}
}
}
// resolvePending closes out the accounts still pending when the deadline fires.
// A worker whose done channel is closed has finished, whatever the results
// channel has managed to deliver, so its own error is the truth; the rest are
// genuinely still running and are reported as timed out. Returns how many of
// them had in fact succeeded.
func resolvePending(pending map[types.AccountID]*perfWorker, failed map[string]string) int {
applied := 0
for accountID, w := range pending {
select {
case <-w.done:
if w.err != nil {
failed[string(accountID)] = w.err.Error()
continue
}
applied++
default:
failed[string(accountID)] = fmt.Sprintf("timed out after %s waiting for the client", perfApplyTimeout)
}
}
return applied
}
// startPerfWorker returns the in-flight retune for the account, starting one if
// there is none. The bool reports whether this call started it.
//
// At most one retune runs per account at a time. A client wedged inside its own
// lock never returns, so without this a caller could add one permanently blocked
// goroutine per request just by retrying the endpoint.
func (h *Handler) startPerfWorker(accountID types.AccountID, client *nbembed.Client, capN uint32, results chan<- perfResult) (*perfWorker, bool) {
h.perfMu.Lock()
defer h.perfMu.Unlock()
if w, ok := h.perfInflight[accountID]; ok {
return w, false
}
w := &perfWorker{done: make(chan struct{})}
if h.perfInflight == nil {
h.perfInflight = make(map[types.AccountID]*perfWorker)
}
h.perfInflight[accountID] = w
go func() {
err := h.setPerformance(client, capN)
w.err = err
close(w.done)
// Publish before touching the registry: perfMu is taken once per
// account by every caller walking the fleet, so a finishing worker
// can queue behind a long apply and miss its own deadline.
results <- perfResult{accountID: accountID, err: err}
h.perfMu.Lock()
delete(h.perfInflight, accountID)
h.perfMu.Unlock()
}()
return w, true
}
// applyBufferCap sets the WireGuard buffer pool cap on every registered client
// and reports how many took it, a per-account error for those that did not, and
// the accounts whose earlier retune has not come back yet.
//
// Clients are handled concurrently and the wait is bounded: SetPerformance goes
// through the embedded client's lock, which Start and Stop hold for as long as
// they take - and on a wedged client Stop never returns. This endpoint is the
// recovery path for exactly that fleet, so one stuck account must neither delay
// the others nor accumulate goroutines across retries.
func (h *Handler) applyBufferCap(capN uint32) (int, map[string]string, []string) {
clients := h.provider.ListClientsForStartup()
results := make(chan perfResult, len(clients))
applied := 0
failed := map[string]string{}
var inFlight []string
pending := make(map[types.AccountID]*perfWorker, len(clients))
for accountID, client := range clients {
w, started := h.startPerfWorker(accountID, client, capN, results)
if started {
pending[accountID] = w
continue
}
// Another request owns this account's retune. Take its result if it
// has already landed, otherwise report it as still running instead of
// waiting on it again.
select {
case <-w.done:
if w.err != nil {
failed[string(accountID)] = w.err.Error()
continue
}
applied++
default:
inFlight = append(inFlight, string(accountID))
}
}
deadline := time.After(perfApplyTimeout)
for range len(pending) {
select {
case res := <-results:
delete(pending, res.accountID)
if res.err != nil {
failed[string(res.accountID)] = res.err.Error()
continue
}
applied++
case <-deadline:
applied += collectBuffered(results, pending, failed)
applied += resolvePending(pending, failed)
return applied, failed, inFlight
}
}
return applied, failed, inFlight
}
// handleRuntime returns cheap runtime and process stats. Safe to hit on a
// running proxy; does not read pprof profiles.
func (h *Handler) handleRuntime(w http.ResponseWriter, _ *http.Request) {
+158
View File
@@ -0,0 +1,158 @@
package debug
import (
"context"
"errors"
"sync/atomic"
"testing"
"time"
nbembed "github.com/netbirdio/netbird/client/embed"
"github.com/netbirdio/netbird/proxy/internal/health"
"github.com/netbirdio/netbird/proxy/internal/roundtrip"
"github.com/netbirdio/netbird/proxy/internal/types"
)
// perfProvider serves a fixed set of accounts. The clients are nil: the tests
// drive Handler.setPerformance, which never dereferences them.
type perfProvider struct {
accounts []types.AccountID
}
func (p *perfProvider) GetClient(types.AccountID) (*nbembed.Client, bool) { return nil, false }
func (p *perfProvider) ListClientsForDebug() map[types.AccountID]roundtrip.ClientDebugInfo {
return nil
}
func (p *perfProvider) ListClientsForStartup() map[types.AccountID]*nbembed.Client {
out := make(map[types.AccountID]*nbembed.Client, len(p.accounts))
for _, id := range p.accounts {
out[id] = nil
}
return out
}
type stubHealth struct{}
func (stubHealth) ReadinessProbe() bool { return true }
func (stubHealth) StartupProbe(context.Context) bool { return true }
func (stubHealth) CheckClientsConnected(context.Context) (bool, map[types.AccountID]health.ClientHealth) {
return true, nil
}
func shortenPerfTimeout(t *testing.T, d time.Duration) {
t.Helper()
prev := perfApplyTimeout
perfApplyTimeout = d
t.Cleanup(func() { perfApplyTimeout = prev })
}
// TestCollectBufferedCountsResultsReadyAtTheDeadline covers the select-ordering
// trap: when the deadline fires, results already buffered must be counted, not
// reported as timeouts. Driving collectBuffered directly keeps it deterministic
// - through applyBufferCap the two select cases race by construction.
func TestCollectBufferedCountsResultsReadyAtTheDeadline(t *testing.T) {
results := make(chan perfResult, 3)
results <- perfResult{accountID: "ok"}
results <- perfResult{accountID: "broken", err: errors.New("boom")}
pending := map[types.AccountID]*perfWorker{
"ok": {done: make(chan struct{})},
"broken": {done: make(chan struct{})},
"wedged": {done: make(chan struct{})},
}
failed := map[string]string{}
applied := collectBuffered(results, pending, failed)
if applied != 1 {
t.Fatalf("applied = %d, want 1", applied)
}
if failed["broken"] != "boom" {
t.Fatalf("failed = %v, want the error recorded for \"broken\"", failed)
}
if _, ok := pending["wedged"]; !ok || len(pending) != 1 {
t.Fatalf("pending = %v, want only the account that never answered", pending)
}
}
// TestApplyBufferCapSingleFlightPerAccount covers the goroutine accumulation
// reported on PR #7452: repeated calls against a client stuck in its own lock
// must not start a second attempt for the same account.
func TestApplyBufferCapSingleFlightPerAccount(t *testing.T) {
shortenPerfTimeout(t, 50*time.Millisecond)
release := make(chan struct{})
t.Cleanup(func() { close(release) })
var calls atomic.Int32
h := &Handler{
provider: &perfProvider{accounts: []types.AccountID{"wedged"}},
health: stubHealth{},
setPerformance: func(_ *nbembed.Client, _ uint32) error {
calls.Add(1)
<-release
return nil
},
}
for i := range 5 {
applied, failed, inFlight := h.applyBufferCap(4096)
if applied != 0 {
t.Fatalf("call %d: applied = %d, want 0", i, applied)
}
if i == 0 {
if len(failed) != 1 {
t.Fatalf("first call: failed = %v, want the account reported as timed out", failed)
}
continue
}
if len(inFlight) != 1 {
t.Fatalf("call %d: inFlight = %v, want the account reported as still running", i, inFlight)
}
if len(failed) != 0 {
t.Fatalf("call %d: failed = %v, want empty while the retune is in flight", i, failed)
}
}
if got := calls.Load(); got != 1 {
t.Fatalf("setPerformance called %d times, want 1: each retry started another blocked worker", got)
}
}
// TestResolvePendingTrustsFinishedWorkers covers the reporting race cubic
// flagged on PR #7452: a retune that finished just before the deadline must be
// reported by its outcome, not as a timeout, whatever the results channel has
// delivered so far.
func TestResolvePendingTrustsFinishedWorkers(t *testing.T) {
ok := &perfWorker{done: make(chan struct{})}
close(ok.done)
broken := &perfWorker{done: make(chan struct{}), err: errors.New("boom")}
close(broken.done)
stillRunning := &perfWorker{done: make(chan struct{})}
pending := map[types.AccountID]*perfWorker{
"ok": ok,
"broken": broken,
"running": stillRunning,
}
failed := map[string]string{}
applied := resolvePending(pending, failed)
if applied != 1 {
t.Fatalf("applied = %d, want 1", applied)
}
if failed["broken"] != "boom" {
t.Fatalf("failed[broken] = %q, want the worker's own error", failed["broken"])
}
if _, ok := failed["ok"]; ok {
t.Fatalf("failed = %v, want no entry for the account that succeeded", failed)
}
if got := failed["running"]; got == "" || got == "boom" {
t.Fatalf("failed[running] = %q, want the timeout message", got)
}
}
@@ -115,3 +115,20 @@ func TestRouter_BedrockNamespacedInferenceProfilesStripsPrefix(t *testing.T) {
assert.Equal(t, "/bedrock", out.Mutations.RewriteUpstream.StripPathPrefix,
"the namespace prefix must not reach the real Bedrock endpoint")
}
// TestRouteClaimsModel_VertexNormalizesCandidate is the Vertex counterpart of
// the Bedrock case above: the parser strips the "@version" suffix from the
// path model, so a provider registered with the versioned form must still
// match the normalized request model.
func TestRouteClaimsModel_VertexNormalizesCandidate(t *testing.T) {
route := ProviderRoute{Vertex: true, Models: []string{"claude-sonnet-4-5@20250929"}}
assert.True(t, routeClaimsModel(route, "claude-sonnet-4-5"),
"raw @version Vertex model must match the normalized request model")
assert.False(t, routeClaimsModel(route, "claude-opus-4-8"),
"a model outside the provider's list must not match")
// Non-Vertex routes keep exact matching (no @version stripping).
openai := ProviderRoute{Models: []string{"gpt-4o@2024"}}
assert.False(t, routeClaimsModel(openai, "gpt-4o"),
"non-Vertex routes must not strip an @version suffix")
}
@@ -36,7 +36,10 @@ type ProviderRoute struct {
// request on a same-vendor route so catch-all gateways of a different
// vendor can't swallow it. Empty disables vendor filtering for this
// route.
Vendor string `json:"vendor,omitempty"`
Vendor string `json:"vendor,omitempty"`
// Vendors lists every parser surface a multi-surface gateway accepts.
// Vendor remains supported for existing single-surface configurations.
Vendors []string `json:"vendors,omitempty"`
Models []string `json:"models"`
UpstreamScheme string `json:"upstream_scheme"`
UpstreamHost string `json:"upstream_host"`
@@ -331,6 +331,11 @@ func discoverableModels(route ProviderRoute, userGroups []string) ([]string, boo
intersection[m] = struct{}{}
}
}
if route.Vertex {
if _, ok := permitted[llm.NormalizeVertexModel(m)]; ok {
intersection[m] = struct{}{}
}
}
}
return sortedModels(intersection), true
}
@@ -409,7 +414,7 @@ func stripBedrockNamespace(out *middleware.Output) {
// peer, return matchOutcomeUnauthorised so the caller can emit
// the dedicated no_authorised_provider deny code.
// 3. Vendor precedence: when the request carries a detected vendor
// (llm.provider) and at least one candidate is the same vendor,
// (llm.provider) and at least one candidate declares that vendor,
// drop the rest — a vendor-tagged request must never cross to
// another vendor's route (e.g. an Anthropic call landing on an
// OpenAI-compatible gateway that also claims the model).
@@ -432,9 +437,9 @@ func (m *Middleware) matchRoute(model, vendor, reqPath string, userGroups []stri
// Vendor pinning runs BEFORE the group filter so a request the parser
// tagged with a vendor can never cross to another vendor's route — not
// even an authorised one. Narrow to same-vendor routes when any
// model-matched route declares that vendor; setups with no vendor tag on
// any route fall through unchanged. After narrowing, if no same-vendor
// even an authorised one. Narrow to supporting routes when any
// model-matched route declares that vendor; setups with no matching vendor
// declaration fall through unchanged. After narrowing, if no supporting
// route authorises the caller, that's matchOutcomeUnauthorised (no
// cross-vendor fallback).
if vendor != "" {
@@ -805,21 +810,31 @@ func authorisingGroupsCSV(routeGroups, userGroups []string) string {
return strings.Join(out, ",")
}
// matchingVendor returns the subset of routes whose Vendor equals the
// request's detected vendor. Routes with an empty Vendor never match — an
// untagged route can't be asserted to speak the request's surface, so it
// stays out of the vendor-filtered set (but remains eligible via the
// fall-through when no route matches the vendor at all).
// matchingVendor returns the routes that declare the request's detected
// vendor through either the legacy singular field or the multi-vendor field.
// Untagged routes remain eligible only when no route declares the vendor.
func matchingVendor(routes []ProviderRoute, vendor string) []ProviderRoute {
var out []ProviderRoute
for _, r := range routes {
if r.Vendor == vendor {
if routeSupportsVendor(r, vendor) {
out = append(out, r)
}
}
return out
}
func routeSupportsVendor(route ProviderRoute, vendor string) bool {
if route.Vendor == vendor {
return true
}
for _, candidate := range route.Vendors {
if candidate == vendor {
return true
}
}
return false
}
// explicitlyClaiming returns the subset of routes whose Models list
// names the model exactly. Catch-all routes (empty Models) are excluded,
// so callers can prefer a provider that genuinely declares the model over
@@ -859,6 +874,11 @@ func routeClaimsModel(route ProviderRoute, model string) bool {
if route.Bedrock && llm.NormalizeBedrockModel(candidate) == model {
return true
}
// Vertex likewise: the parser strips the "@version" suffix from the
// path model, while the operator may register the versioned form.
if route.Vertex && llm.NormalizeVertexModel(candidate) == model {
return true
}
// A client may pin a dated Anthropic id ("claude-sonnet-4-5-20250929")
// where the operator registered the undated one. Only an undated
// registration absorbs a dated request: normalising both sides would
@@ -412,6 +412,50 @@ func TestRouter_VendorKeepsOpenAIOffAnthropic(t *testing.T) {
assert.Equal(t, "api.openai.com", out.Mutations.RewriteUpstream.Host, "openai vendor must pin to the openai route despite anthropic being declared first")
}
func TestRouter_MultiVendorGatewayAcceptsBothSurfaces(t *testing.T) {
gateway := ProviderRoute{
ID: "agentgateway",
Vendors: []string{"openai", "anthropic"},
Models: nil,
AllowedGroupIDs: []string{defaultTestGroup},
UpstreamScheme: "https",
UpstreamHost: "gateway.example.com",
}
other := ProviderRoute{
ID: "other-vendor",
Vendor: "mistral",
Models: nil,
AllowedGroupIDs: []string{defaultTestGroup},
UpstreamScheme: "https",
UpstreamHost: "mistral.example.com",
}
mw := New(Config{Providers: []ProviderRoute{other, gateway}})
for _, tc := range []struct {
name string
vendor string
model string
path string
}{
{name: "OpenAI", vendor: "openai", model: "gpt-4o-mini", path: "/v1/chat/completions"},
{name: "Anthropic", vendor: "anthropic", model: "claude-sonnet-4-5", path: "/v1/messages"},
} {
t.Run(tc.name, func(t *testing.T) {
out, err := mw.Invoke(context.Background(), newInputVendorModelURL(tc.vendor, tc.model, tc.path))
require.NoError(t, err)
require.NotNil(t, out)
assert.Equal(t, middleware.DecisionAllow, out.Decision,
"supported vendor must route through the multi-surface gateway")
require.NotNil(t, out.Mutations)
require.NotNil(t, out.Mutations.RewriteUpstream)
assert.Equal(t, "gateway.example.com", out.Mutations.RewriteUpstream.Host)
provider, _ := metaValue(t, out.Metadata, middleware.KeyLLMResolvedProviderID)
assert.Equal(t, "agentgateway", provider)
})
}
}
// TestRouter_VendorAbsentFallsBackToModelPath confirms vendor filtering is
// inert when the request carries no detected vendor: routing then relies on
// model/path as before.
@@ -692,6 +736,23 @@ func TestRouter_FactoryRejectsBadJSON(t *testing.T) {
require.Error(t, err, "malformed JSON config must be rejected at chain build time")
}
func TestRouter_FactoryDecodesLegacyAndMultiVendorFields(t *testing.T) {
raw := []byte(`{"providers":[` +
`{"id":"legacy","vendor":"openai","models":[],"upstream_scheme":"https","upstream_host":"openai.example.com","auth_header_name":"Authorization","auth_header_value":"Bearer legacy","allowed_group_ids":["group"]},` +
`{"id":"multi","vendors":["openai","anthropic"],"models":[],"upstream_scheme":"https","upstream_host":"gateway.example.com","auth_header_name":"Authorization","auth_header_value":"Bearer multi","allowed_group_ids":["group"]}` +
`]}`)
resolved, err := Factory{}.New(raw)
require.NoError(t, err)
router, ok := resolved.(*Middleware)
require.True(t, ok, "factory must return the concrete router middleware")
require.Len(t, router.cfg.Providers, 2)
assert.Equal(t, "openai", router.cfg.Providers[0].Vendor,
"the legacy singular field must keep decoding")
assert.Equal(t, []string{"openai", "anthropic"}, router.cfg.Providers[1].Vendors,
"the multi-vendor field must decode both supported surfaces")
}
func TestRouter_FactoryAcceptsEmptyShapes(t *testing.T) {
cases := [][]byte{nil, []byte(""), []byte(" "), []byte("null"), []byte("{}"), []byte("[]")}
for _, raw := range cases {
+1 -1
View File
@@ -264,7 +264,7 @@ func applyMutations(ctx context.Context, d *Dispatcher, spec Spec, r *http.Reque
if m == nil {
return
}
add, remove, blocked := FilterHeaderMutations(m)
add, remove, blocked := filterHeaderMutations(m, spec.ID)
for _, h := range blocked {
d.metrics.IncHeaderMutationBlocked(ctx, spec.ID, h)
}
+59
View File
@@ -2,6 +2,7 @@ package middleware
import (
"context"
"net/http"
"strconv"
"testing"
@@ -278,6 +279,64 @@ func TestChain_ApplyMutations_RewriteGatedOnCanMutate(t *testing.T) {
assert.Nil(t, rewrite, "rewrite must be filtered when CanMutate=false")
}
func TestChain_IdentityInjectReplacesReservedNetBirdHeaders(t *testing.T) {
mw := &fakeMiddleware{
id: "llm_identity_inject",
slot: SlotOnRequest,
mutationsSupported: true,
canMutate: true,
mutations: &Mutations{
HeadersRemove: []string{"x-netbird-user-id", "x-netbird-groups"},
HeadersAdd: []KV{
{Key: "x-netbird-user-id", Value: "trusted-user"},
{Key: "x-netbird-groups", Value: "trusted-group"},
},
},
}
c := chainFor(t, mw)
req, err := http.NewRequest(http.MethodGet, "https://gateway.example.com/v1/models", nil)
require.NoError(t, err)
req.Header.Set("x-netbird-user-id", "spoofed-user")
req.Header.Set("x-netbird-groups", "spoofed-group")
denied, _, _, err := c.RunRequest(context.Background(), req, &Input{}, NewAccumulator(0))
require.NoError(t, err)
assert.Nil(t, denied, "identity injection must not deny the request")
assert.Equal(t, "trusted-user", req.Header.Get("x-netbird-user-id"),
"the built-in identity middleware must replace a spoofed user header")
assert.Equal(t, "trusted-group", req.Header.Get("x-netbird-groups"),
"the built-in identity middleware must replace spoofed groups")
}
func TestChain_OtherMiddlewareCannotReplaceReservedNetBirdHeaders(t *testing.T) {
mw := &fakeMiddleware{
id: "untrusted-middleware",
slot: SlotOnRequest,
mutationsSupported: true,
canMutate: true,
mutations: &Mutations{
HeadersRemove: []string{"x-netbird-user-id", "x-netbird-groups"},
HeadersAdd: []KV{
{Key: "x-netbird-user-id", Value: "replacement-user"},
{Key: "x-netbird-groups", Value: "replacement-group"},
},
},
}
c := chainFor(t, mw)
req, err := http.NewRequest(http.MethodGet, "https://gateway.example.com/v1/models", nil)
require.NoError(t, err)
req.Header.Set("x-netbird-user-id", "original-user")
req.Header.Set("x-netbird-groups", "original-group")
denied, _, _, err := c.RunRequest(context.Background(), req, &Input{}, NewAccumulator(0))
require.NoError(t, err)
assert.Nil(t, denied, "blocked mutations must not deny the request")
assert.Equal(t, "original-user", req.Header.Get("x-netbird-user-id"),
"other middleware must remain unable to mutate reserved identity headers")
assert.Equal(t, "original-group", req.Header.Get("x-netbird-groups"),
"other middleware must remain unable to mutate reserved identity headers")
}
// TestChain_RunRequest_PropagatesUserGroups asserts the chain forwards
// Input.UserGroups verbatim through cloneInputFor so policy-aware
// middlewares (e.g. llm_policy_check) can authorise without an extra
+16 -2
View File
@@ -2,6 +2,8 @@ package middleware
import "strings"
const trustedIdentityMiddlewareID = "llm_identity_inject"
var denyHeaders = []string{
"Authorization",
"Connection",
@@ -78,18 +80,22 @@ func isHeaderFieldName(name string) bool {
// header names so the dispatcher can increment the blocked-header
// metric.
func FilterHeaderMutations(m *Mutations) (filteredAdd []KV, filteredRemove []string, blocked []string) {
return filterHeaderMutations(m, "")
}
func filterHeaderMutations(m *Mutations, middlewareID string) (filteredAdd []KV, filteredRemove []string, blocked []string) {
if m == nil {
return nil, nil, nil
}
for _, kv := range m.HeadersAdd {
if IsHeaderMutable(kv.Key) {
if IsHeaderMutable(kv.Key) || isTrustedIdentityHeader(middlewareID, kv.Key) {
filteredAdd = append(filteredAdd, kv)
continue
}
blocked = append(blocked, kv.Key)
}
for _, name := range m.HeadersRemove {
if IsHeaderMutable(name) {
if IsHeaderMutable(name) || isTrustedIdentityHeader(middlewareID, name) {
filteredRemove = append(filteredRemove, name)
continue
}
@@ -97,3 +103,11 @@ func FilterHeaderMutations(m *Mutations) (filteredAdd []KV, filteredRemove []str
}
return filteredAdd, filteredRemove, blocked
}
func isTrustedIdentityHeader(middlewareID, name string) bool {
if middlewareID != trustedIdentityMiddlewareID {
return false
}
return strings.EqualFold(name, "x-netbird-user-id") ||
strings.EqualFold(name, "x-netbird-groups")
}
@@ -0,0 +1,26 @@
package middleware
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestFilterHeaderMutationsDoesNotTrustReservedHeaders(t *testing.T) {
mutations := &Mutations{
HeadersAdd: []KV{
{Key: "x-request-label", Value: "allowed"},
{Key: "x-netbird-user-id", Value: "spoofed-user"},
},
HeadersRemove: []string{"x-request-label", "x-netbird-groups"},
}
filteredAdd, filteredRemove, blocked := FilterHeaderMutations(mutations)
assert.Equal(t, []KV{{Key: "x-request-label", Value: "allowed"}}, filteredAdd,
"the public filter should retain mutable additions")
assert.Equal(t, []string{"x-request-label"}, filteredRemove,
"the public filter should retain mutable removals")
assert.ElementsMatch(t, []string{"x-netbird-user-id", "x-netbird-groups"}, blocked,
"the public filter must not grant the identity middleware exception")
}
+4 -5
View File
@@ -26,8 +26,8 @@ import (
// branch at all), construct the MultiTransport via NewDirectOnly.
type MultiTransport struct {
embedded http.RoundTripper
direct *http.Transport
insecure *http.Transport
direct *upstreamTransport
insecure *upstreamTransport
}
// errNoEmbeddedTransport is returned when a request reaches the
@@ -53,7 +53,6 @@ func NewMultiTransport(embedded http.RoundTripper, logger *log.Logger) *MultiTra
}
direct := &http.Transport{
DialContext: dialWithTimeout(dialer.DialContext),
ForceAttemptHTTP2: true,
MaxIdleConns: cfg.maxIdleConns,
MaxIdleConnsPerHost: cfg.maxIdleConnsPerHost,
MaxConnsPerHost: cfg.maxConnsPerHost,
@@ -70,8 +69,8 @@ func NewMultiTransport(embedded http.RoundTripper, logger *log.Logger) *MultiTra
return &MultiTransport{
embedded: embedded,
direct: direct,
insecure: insecure,
direct: newUpstreamTransport(direct, cfg.upstreamHTTPVersion, logger),
insecure: newUpstreamTransport(insecure, cfg.upstreamHTTPVersion, logger),
}
}
+60 -4
View File
@@ -5,6 +5,7 @@ import (
"io"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"time"
@@ -75,16 +76,71 @@ func TestMultiTransport_AppliesEnvOverridesToDirect(t *testing.T) {
mt := NewMultiTransport(&stubRoundTripper{body: "embedded"}, nil)
assert.Equal(t, 42, mt.direct.MaxIdleConns,
assert.Equal(t, 42, mt.direct.primary.MaxIdleConns,
"NB_PROXY_MAX_IDLE_CONNS must propagate to the direct transport")
assert.Equal(t, 11*time.Second, mt.direct.IdleConnTimeout,
assert.Equal(t, 11*time.Second, mt.direct.primary.IdleConnTimeout,
"NB_PROXY_IDLE_CONN_TIMEOUT must propagate to the direct transport")
assert.Equal(t, 7*time.Second, mt.direct.TLSHandshakeTimeout,
assert.Equal(t, 7*time.Second, mt.direct.primary.TLSHandshakeTimeout,
"NB_PROXY_TLS_HANDSHAKE_TIMEOUT must propagate to the direct transport")
assert.Equal(t, 42, mt.insecure.MaxIdleConns,
assert.Equal(t, 42, mt.insecure.primary.MaxIdleConns,
"env tuning must also apply to the insecure-skip-verify direct transport")
}
// TestMultiTransport_UpstreamHTTPVersion pins the protocol actually
// negotiated with an HTTPS upstream that offers both h2 and http/1.1.
// The request rides the insecure clone, so this also covers the version
// surviving http.Transport.Clone.
func TestMultiTransport_UpstreamHTTPVersion(t *testing.T) {
tests := []struct {
name string
env string
wantProto string
}{
{name: "unset negotiates h2", env: "", wantProto: "HTTP/2.0"},
{name: "auto negotiates h2", env: "auto", wantProto: "HTTP/2.0"},
{name: "1.1 pins http/1.1", env: "1.1", wantProto: "HTTP/1.1"},
{name: "2 negotiates h2", env: "2", wantProto: "HTTP/2.0"},
{name: "unsupported value keeps the default", env: "http3", wantProto: "HTTP/2.0"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
// t.Setenv registers the restore for whatever the process
// inherited; unsetting afterwards lets the default row
// exercise a genuinely absent variable.
t.Setenv(EnvUpstreamHTTPVersion, tc.env)
if tc.env == "" {
require.NoError(t, os.Unsetenv(EnvUpstreamHTTPVersion))
}
// The test server's certificate isn't in any root pool, so the
// request rides the insecure branch via WithSkipTLSVerify.
srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, r.Proto)
}))
srv.EnableHTTP2 = true
srv.StartTLS()
defer srv.Close()
mt := NewDirectOnly(nil)
ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background()))
req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil)
require.NoError(t, err)
resp, err := mt.RoundTrip(req)
require.NoError(t, err)
body, err := io.ReadAll(resp.Body)
_ = resp.Body.Close()
require.NoError(t, err)
assert.Equal(t, tc.wantProto, resp.Proto,
"client-side protocol must follow %s=%q", EnvUpstreamHTTPVersion, tc.env)
assert.Equal(t, tc.wantProto, string(body),
"the upstream must see the same protocol the client negotiated")
})
}
}
// TestMultiTransport_NilEmbeddedErrorsWhenWGPathRequested guards
// against the previous silent fallback: a MultiTransport constructed
// without an embedded transport must reject requests that don't
+30 -13
View File
@@ -30,6 +30,12 @@ import (
const deviceNamePrefix = "ingress-proxy-"
// envProxyRosenpass toggles Rosenpass (permissive) on the embedded proxy client. Defaults to on.
const envProxyRosenpass = "NB_PROXY_ROSENPASS" //nolint:gosec // env var name, not a credential
// envProxyClientLogLevel sets the embedded NetBird client's log level.
const envProxyClientLogLevel = "NB_PROXY_CLIENT_LOG_LEVEL"
const clientStopTimeout = 30 * time.Second
const createProxyPeerTimeout = 30 * time.Second
@@ -76,10 +82,10 @@ type serviceNotification struct {
// clientEntry holds an embedded NetBird client and tracks which services use it.
type clientEntry struct {
client *embed.Client
transport *http.Transport
transport *upstreamTransport
// insecureTransport is a clone of transport with TLS verification disabled,
// used when per-target skip_tls_verify is set.
insecureTransport *http.Transport
insecureTransport *upstreamTransport
services map[ServiceKey]serviceInfo
createdAt time.Time
started bool
@@ -353,11 +359,11 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account
// NB_PROXY_CLIENT_LOG_LEVEL (e.g. "trace") to surface the embedded NetBird
// client's relay / signal / handshake detail for local debugging.
clientLogLevel := log.WarnLevel.String()
if v := strings.TrimSpace(os.Getenv("NB_PROXY_CLIENT_LOG_LEVEL")); v != "" {
if v := strings.TrimSpace(os.Getenv(envProxyClientLogLevel)); v != "" {
if lvl, err := log.ParseLevel(v); err == nil {
clientLogLevel = lvl.String()
} else {
n.logger.Warnf("invalid NB_PROXY_CLIENT_LOG_LEVEL %q, using %q: %v", v, clientLogLevel, err)
n.logger.Warnf("invalid %s %q, using %q: %v", envProxyClientLogLevel, v, clientLogLevel, err)
}
}
@@ -367,15 +373,26 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account
}
})
// Rosenpass runs in permissive mode by default so the embedded proxy can
// establish connections with Rosenpass-enabled peers (which otherwise fail
// on a PSK mismatch) while still falling back to plain WireGuard for peers
// that do not run Rosenpass. Set NB_PROXY_ROSENPASS=false to disable it.
rosenpassEnabled := true
if v, ok := envBool(envProxyRosenpass, n.logger); ok {
rosenpassEnabled = v
}
// Create embedded NetBird client with the generated private key.
// The peer has already been created via CreateProxyPeer RPC with the public key.
wgPort := int(n.clientCfg.WGPort)
embedOpts := embed.Options{
DeviceName: deviceNamePrefix + n.proxyID,
ManagementURL: n.clientCfg.MgmtAddr,
PrivateKey: privateKey.String(),
LogLevel: clientLogLevel,
BlockInbound: n.clientCfg.BlockInbound,
DeviceName: deviceNamePrefix + n.proxyID,
ManagementURL: n.clientCfg.MgmtAddr,
PrivateKey: privateKey.String(),
LogLevel: clientLogLevel,
BlockInbound: n.clientCfg.BlockInbound,
EnableRosenpass: rosenpassEnabled,
RosenpassPermissive: rosenpassEnabled,
// The embedded proxy peer must never be a stepping stone into
// the proxy host's LAN: it only exists to reach NetBird mesh
// targets or, when direct_upstream is set, the host network
@@ -397,7 +414,6 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account
// not work with reverse proxied requests.
transport := &http.Transport{
DialContext: dialWithTimeout(client.DialContext),
ForceAttemptHTTP2: true,
MaxIdleConns: n.transportCfg.maxIdleConns,
MaxIdleConnsPerHost: n.transportCfg.maxIdleConnsPerHost,
MaxConnsPerHost: n.transportCfg.maxConnsPerHost,
@@ -409,15 +425,14 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account
ReadBufferSize: n.transportCfg.readBufferSize,
DisableCompression: n.transportCfg.disableCompression,
}
insecureTransport := transport.Clone()
insecureTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec
return &clientEntry{
client: client,
services: map[ServiceKey]serviceInfo{key: si},
transport: transport,
insecureTransport: insecureTransport,
transport: newUpstreamTransport(transport, n.transportCfg.upstreamHTTPVersion, n.logger),
insecureTransport: newUpstreamTransport(insecureTransport, n.transportCfg.upstreamHTTPVersion, n.logger),
createdAt: time.Now(),
started: false,
inflightMap: make(map[backendKey]chan struct{}),
@@ -899,6 +914,8 @@ func logEmbedOptions(logger *log.Logger, accountID types.AccountID, serviceID ty
"mtu": mtu,
"block_inbound": opts.BlockInbound,
"block_lan_access": opts.BlockLANAccess,
"rosenpass_enabled": opts.EnableRosenpass,
"rosenpass_permissive": opts.RosenpassPermissive,
"disable_ipv6": opts.DisableIPv6,
"disable_client_routes": opts.DisableClientRoutes,
"no_userspace": opts.NoUserspace,
+108
View File
@@ -1,8 +1,11 @@
package roundtrip
import (
"crypto/tls"
"net/http"
"os"
"strconv"
"strings"
"time"
log "github.com/sirupsen/logrus"
@@ -21,6 +24,30 @@ const (
EnvReadBufferSize = "NB_PROXY_READ_BUFFER_SIZE"
EnvDisableCompression = "NB_PROXY_DISABLE_COMPRESSION"
EnvMaxInflight = "NB_PROXY_MAX_INFLIGHT"
EnvUpstreamHTTPVersion = "NB_PROXY_UPSTREAM_HTTP_VERSION"
)
// upstreamHTTPVersion selects the HTTP version the proxy uses towards an
// upstream. The explicit values are absolute: they mean the same thing
// however the transports are dialled and whatever the default becomes,
// so operator configuration survives a change of default.
type upstreamHTTPVersion string
const (
// upstreamHTTPAuto leaves the choice to the upstream: h2 is offered
// alongside http/1.1 in the TLS handshake and the upstream picks.
// An upstream that picks h2 and then fails to serve it is moved to
// HTTP/1.1 on its own (see upstreamTransport), which is the part
// ALPN cannot express. This is the only value whose meaning tracks
// the proxy's default.
upstreamHTTPAuto upstreamHTTPVersion = "auto"
// upstreamHTTP11 never offers h2, so the upstream sees HTTP/1.1.
upstreamHTTP11 upstreamHTTPVersion = "1.1"
// upstreamHTTP2 offers h2 in the TLS handshake and keeps it there:
// an upstream that negotiates h2 and then breaks is never moved to
// HTTP/1.1. Cleartext upstreams stay on HTTP/1.1 regardless: the
// proxy speaks no h2c.
upstreamHTTP2 upstreamHTTPVersion = "2"
)
// transportConfig holds tunable parameters for the per-account HTTP transport.
@@ -37,6 +64,11 @@ type transportConfig struct {
disableCompression bool
// maxInflight limits per-backend concurrent requests. 0 means unlimited.
maxInflight int
// upstreamHTTPVersion selects the HTTP version used towards HTTPS
// upstreams. The default negotiates it with each upstream; the
// explicit values are for backends whose advertised h2 support is
// unusable and whose failure mode the negotiation cannot see.
upstreamHTTPVersion upstreamHTTPVersion
}
func defaultTransportConfig() transportConfig {
@@ -47,6 +79,7 @@ func defaultTransportConfig() transportConfig {
idleConnTimeout: 90 * time.Second,
tlsHandshakeTimeout: 10 * time.Second,
expectContinueTimeout: 1 * time.Second,
upstreamHTTPVersion: upstreamHTTPAuto,
}
}
@@ -86,6 +119,9 @@ func loadTransportConfig(logger *log.Logger) transportConfig {
if v, ok := envInt(EnvMaxInflight, logger); ok {
cfg.maxInflight = v
}
if v, ok := envUpstreamHTTPVersion(EnvUpstreamHTTPVersion, logger); ok {
cfg.upstreamHTTPVersion = v
}
logger.WithFields(log.Fields{
"max_idle_conns": cfg.maxIdleConns,
@@ -99,11 +135,83 @@ func loadTransportConfig(logger *log.Logger) transportConfig {
"read_buffer_size": cfg.readBufferSize,
"disable_compression": cfg.disableCompression,
"max_inflight": cfg.maxInflight,
"upstream_http_version": cfg.upstreamHTTPVersion,
}).Debug("backend transport configuration")
return cfg
}
// applyUpstreamHTTPVersion configures t's ALPN offer for the requested
// HTTP version. It is the single place that decides which protocols a
// transport offers, so changing the proxy's default only touches this
// function and leaves every explicit operator setting intact. What
// happens when a negotiated h2 upstream then fails belongs to
// upstreamTransport, which owns the runtime half of "auto".
//
// HTTP/1.1 is pinned by clearing ForceAttemptHTTP2 and installing an
// empty TLSNextProto, which disables h2 regardless of how the transport
// is dialled. Relying on net/http's conservative default (h2 off
// whenever a custom dialer is set) would silently start negotiating h2
// again the day a transport switches to DialTLSContext.
func applyUpstreamHTTPVersion(t *http.Transport, version upstreamHTTPVersion) {
if version == upstreamHTTP11 {
t.ForceAttemptHTTP2 = false
t.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{}
t.TLSClientConfig = withoutHTTP2ALPN(t.TLSClientConfig)
return
}
t.ForceAttemptHTTP2 = true
}
// withoutHTTP2ALPN drops h2 from the ALPN offer. Configuring h2 makes
// net/http append h2 to the transport's TLSClientConfig, so a transport
// cloned from one that already served a request carries that offer with
// it. Left in place, the upstream would select a protocol this
// transport then refuses to speak, and the response would come back as
// h2 frames parsed as an HTTP/1.1 message.
func withoutHTTP2ALPN(cfg *tls.Config) *tls.Config {
// A nil config offers no ALPN at all, which is already HTTP/1.1.
if cfg == nil || len(cfg.NextProtos) == 0 {
return cfg
}
protos := make([]string, 0, len(cfg.NextProtos))
for _, proto := range cfg.NextProtos {
if proto == "h2" {
continue
}
protos = append(protos, proto)
}
if len(protos) == len(cfg.NextProtos) {
return cfg
}
// Clone rather than edit in place: the caller may share this config
// with the transport it was cloned from.
stripped := cfg.Clone()
stripped.NextProtos = protos
return stripped
}
// envUpstreamHTTPVersion reads an upstream HTTP version from the
// environment. An unrecognised value warns and leaves the default in
// place rather than guessing at the operator's intent.
func envUpstreamHTTPVersion(key string, logger *log.Logger) (upstreamHTTPVersion, bool) {
s := strings.TrimSpace(os.Getenv(key))
if s == "" {
return "", false
}
switch v := upstreamHTTPVersion(strings.ToLower(s)); v {
case upstreamHTTPAuto, upstreamHTTP11, upstreamHTTP2:
return v, true
default:
logger.Warnf("ignoring unsupported %s=%q, expected one of %q, %q, %q",
key, s, upstreamHTTPAuto, upstreamHTTP11, upstreamHTTP2)
return "", false
}
}
func envInt(key string, logger *log.Logger) (int, bool) {
s := os.Getenv(key)
if s == "" {
+455
View File
@@ -0,0 +1,455 @@
package roundtrip
import (
"errors"
"net"
"net/http"
"net/netip"
"net/url"
"strings"
"sync"
"time"
log "github.com/sirupsen/logrus"
)
// upstreamDowngradeTTL is how long an upstream stays pinned to HTTP/1.1
// after an h2 failure that only implied it cannot serve h2. Bounded so a
// fixed or replaced backend returns to h2 without restarting the proxy.
// A pin the upstream asked for itself does not expire — see downgrade.
const upstreamDowngradeTTL = 10 * time.Minute
// downgrade is an upstream's HTTP/1.1 pin.
type downgrade struct {
// expiry is when the pin lapses and the upstream is offered h2
// again. The zero time means it never does: the upstream answered
// HTTP_1_1_REQUIRED, which is a statement about how it is
// configured, not a fault that may clear on its own. Re-probing
// that every upstreamDowngradeTTL would buy nothing but a failed
// request per interval, so the pin holds until the transport goes
// away with the proxy or the account's client.
expiry time.Time
}
// permanent reports whether the upstream asked for this pin itself.
func (d downgrade) permanent() bool {
return d.expiry.IsZero()
}
// active reports whether the pin still stands at now.
func (d downgrade) active(now time.Time) bool {
return d.permanent() || now.Before(d.expiry)
}
// upstreamTransport carries requests to a single upstream family (one
// TLS configuration) and implements what upstreamHTTPAuto means.
//
// ALPN already lets the upstream pick the protocol: primary offers both
// h2 and http/1.1 and the server chooses. What ALPN cannot express is
// an upstream that selects h2 and then fails to speak it — the case
// this type handles. The first h2-level failure for a host pins that
// host to fallback, an HTTP/1.1-only clone of primary, and the request
// is retried there when it can be replayed.
//
// The downgrade is per upstream host, not per transport: one broken
// backend must not drop every other backend to HTTP/1.1.
type upstreamTransport struct {
// primary is the configured transport: h2 offered in ALPN for
// upstreamHTTPAuto and upstreamHTTP2, HTTP/1.1-only for
// upstreamHTTP11.
primary *http.Transport
// version decides whether a downgrade may happen at all. Only
// upstreamHTTPAuto downgrades; the explicit values are absolute.
version upstreamHTTPVersion
logger *log.Logger
// fallbackMu guards the lazy fallback clone: most deployments never
// hit a broken h2 upstream and should not pay for a second
// connection pool.
fallbackMu sync.Mutex
fallback *http.Transport
mu sync.RWMutex
// downgraded maps an upstream host to its HTTP/1.1 pin.
downgraded map[string]downgrade
}
// newUpstreamTransport wraps base for the requested HTTP version. base
// must not be used directly afterwards: the wrapper owns it, including
// its connection pool.
func newUpstreamTransport(base *http.Transport, version upstreamHTTPVersion, logger *log.Logger) *upstreamTransport {
if logger == nil {
logger = log.StandardLogger()
}
applyUpstreamHTTPVersion(base, version)
return &upstreamTransport{
primary: base,
version: version,
logger: logger,
downgraded: make(map[string]downgrade),
}
}
// RoundTrip implements http.RoundTripper.
func (t *upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error) {
if !t.mayDowngrade(req) {
return t.primary.RoundTrip(req)
}
host := upstreamKey(req.URL)
if t.isDowngraded(host) {
return t.http1().RoundTrip(req)
}
resp, err := t.primary.RoundTrip(req)
if err == nil || !isHTTP2ProtocolError(err) {
return resp, err
}
// HTTP_1_1_REQUIRED is the upstream saying it will not serve this
// request over h2 however often it is asked — IIS answers it for
// Windows Authentication and for client-certificate sites, where
// the cause is site configuration rather than a passing fault.
t.markDowngraded(host, isHTTP11Required(err))
if !safeToRetry(req, err) {
// The upstream may have carried out the request before failing
// to answer over h2, and repeating it could duplicate whatever
// it did. The host is pinned either way, so the next request
// goes out over HTTP/1.1.
t.logger.WithFields(log.Fields{
"upstream": host,
"method": req.Method,
}).Debug("not retrying over HTTP/1.1: the upstream may already have applied this request")
return nil, err
}
retry, ok := replayable(req)
if !ok {
// The body is already consumed and cannot be regenerated, so
// this request fails, and the pin carries the next one.
return nil, err
}
return t.http1().RoundTrip(retry)
}
// CloseIdleConnections closes idle connections on both pools.
func (t *upstreamTransport) CloseIdleConnections() {
t.primary.CloseIdleConnections()
if fallback := t.existingHTTP1(); fallback != nil {
fallback.CloseIdleConnections()
}
}
// mayDowngrade reports whether a failed request is a downgrade
// candidate. Only upstreamHTTPAuto downgrades, and only for TLS
// upstreams: the proxy speaks no h2c, so a cleartext upstream is
// already on HTTP/1.1 and an error there says nothing about h2.
func (t *upstreamTransport) mayDowngrade(req *http.Request) bool {
return t.version == upstreamHTTPAuto && req.URL != nil && req.URL.Scheme == "https"
}
func (t *upstreamTransport) isDowngraded(host string) bool {
t.mu.RLock()
pin, ok := t.downgraded[host]
t.mu.RUnlock()
if !ok {
return false
}
if pin.active(time.Now()) {
return true
}
t.mu.Lock()
defer t.mu.Unlock()
// Re-read under the write lock rather than trusting the expired pin
// from above: a concurrent request may have re-pinned the host since,
// and that pin decides this request too. Reporting the stale read
// would send one request back to h2 against a live pin.
pin, ok = t.downgraded[host]
if !ok {
return false
}
if pin.active(time.Now()) {
return true
}
delete(t.downgraded, host)
return false
}
// markDowngraded pins host to HTTP/1.1. permanent marks a pin the
// upstream asked for; anything else lapses after upstreamDowngradeTTL so
// a repaired backend is offered h2 again.
func (t *upstreamTransport) markDowngraded(host string, permanent bool) {
now := time.Now()
pin := downgrade{expiry: now.Add(upstreamDowngradeTTL)}
if permanent {
pin = downgrade{}
}
t.mu.Lock()
previous, pinned := t.downgraded[host]
// A permanent pin is never weakened back into an expiring one: the
// upstream has already said h2 is not on offer.
promoted := pinned && !previous.permanent() && permanent
if !pinned || !previous.permanent() {
t.downgraded[host] = pin
}
for h, existing := range t.downgraded {
if !existing.active(now) {
delete(t.downgraded, h)
}
}
t.mu.Unlock()
// Log a new pin, and a pin the upstream has since asked to make
// permanent — otherwise an operator would only ever see the "for the
// next 10m" line and never learn the upstream settled the question.
if pinned && !promoted {
return
}
entry := t.logger.WithField("upstream", host)
if permanent {
entry.Warnf("upstream answered HTTP_1_1_REQUIRED, using HTTP/1.1 for it from now on")
return
}
entry.Warnf("upstream negotiated HTTP/2 but failed to serve it, using HTTP/1.1 for the next %s (set %s=1.1 to pin it)",
upstreamDowngradeTTL, EnvUpstreamHTTPVersion)
}
// http1 returns the HTTP/1.1-only clone, creating it on first use.
func (t *upstreamTransport) http1() *http.Transport {
t.fallbackMu.Lock()
defer t.fallbackMu.Unlock()
if t.fallback == nil {
fallback := t.primary.Clone()
applyUpstreamHTTPVersion(fallback, upstreamHTTP11)
t.fallback = fallback
}
return t.fallback
}
// existingHTTP1 returns the fallback transport only if it was already
// created, so housekeeping never allocates a second connection pool for
// an upstream that never needed one.
func (t *upstreamTransport) existingHTTP1() *http.Transport {
t.fallbackMu.Lock()
defer t.fallbackMu.Unlock()
return t.fallback
}
// upstreamKey normalizes an authority for use as a pin key, so one
// upstream cannot end up with two independent pins. DNS labels compare
// case-insensitively, and the default HTTPS port is implied — every
// downgrade path is TLS-only, so a bare host and the same host on :443
// are the same upstream.
func upstreamKey(u *url.URL) string {
host := normalizeUpstreamHost(u.Hostname())
port := u.Port()
if port == "" || port == "443" {
return host
}
// JoinHostPort rather than concatenation: an IPv6 literal needs its
// brackets back after Hostname stripped them.
return net.JoinHostPort(host, port)
}
// normalizeUpstreamHost folds the spellings of one host onto a single
// key. An IP literal goes through netip so that the several textual
// forms of one address (case, leading zeroes, a compressed run) collapse
// and a v4-mapped address keys as the v4 address it is. A zone
// identifier is left exactly as written: it names an interface, and
// interface names are case-sensitive on the systems that have them, so
// %eth0 and %ETH0 may be different links and must not share a pin.
// Anything that is not an IP literal is a DNS name, which compares
// case-insensitively.
func normalizeUpstreamHost(host string) string {
if addr, err := netip.ParseAddr(host); err == nil {
return addr.Unmap().String()
}
return strings.ToLower(host)
}
// safeToRetry reports whether req may be sent a second time over
// HTTP/1.1 after err ended its h2 attempt.
//
// A failure at the h2 layer does not say whether the upstream already
// carried out the request, so replaying one that changes state could
// duplicate it. Two cases are safe: a request whose repetition is
// harmless by definition, and an upstream that told us it processed
// nothing on the connection. The second is what makes the IIS case work
// for every method — a site requiring HTTP/1.1 refuses at stream 0,
// before the request is looked at.
func safeToRetry(req *http.Request, err error) bool {
return idempotent(req) || upstreamProcessedNothing(err)
}
// idempotent reports whether repeating req is defined to be harmless.
// It mirrors net/http's own retry rule (Request.isReplayable): a method
// with no side effects, or a caller that promised the upstream
// deduplicates by key.
func idempotent(req *http.Request) bool {
if req.Header.Get("Idempotency-Key") != "" || req.Header.Get("X-Idempotency-Key") != "" {
return true
}
switch req.Method {
// An empty method means GET, as in net/http.
case "", http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodTrace:
return true
}
return false
}
// upstreamProcessedNothing reports whether err describes a GOAWAY that
// named this request's stream as one the upstream had not received, so
// it cannot have acted on it. A stream error says the opposite: the
// stream was open, so the request had been delivered.
func upstreamProcessedNothing(err error) bool {
if err == nil {
return false
}
msg := transportError(err).Error()
return strings.Contains(msg, goAwayStreamNotReceivedMarker) ||
strings.Contains(msg, goAwayNothingProcessedMarker)
}
// replayable returns a request that can be sent a second time, or
// ok=false when the body is gone. A RoundTripper consumes and closes
// the body it was given, so a retry needs either no body at all or
// GetBody to produce a fresh one.
func replayable(req *http.Request) (*http.Request, bool) {
if req.Body == nil || req.Body == http.NoBody {
return req, true
}
if req.GetBody == nil {
return nil, false
}
body, err := req.GetBody()
if err != nil {
return nil, false
}
retry := req.Clone(req.Context())
retry.Body = body
return retry, true
}
// http2ErrorMarkers are the substrings that identify an HTTP/2 protocol
// failure. net/http bundles its own private copy of the http2 package,
// so its errors cannot be matched by type from here: http2.StreamError
// and friends in x/net are different types from the ones a
// bundled-h2 transport returns. The strings below are the formats those
// bundled errors print, and they are specific to h2 framing — a
// downgrade must never be triggered by an ordinary network or TLS
// error, which retrying on HTTP/1.1 would not fix.
var http2ErrorMarkers = []string{
// Transport-level h2 failures, e.g.
// "http2: server sent GOAWAY and closed the connection".
"http2:",
// http2.StreamError, e.g. "stream error: stream ID 1; PROTOCOL_ERROR".
"stream error: stream ID",
// http2.ConnectionError, e.g. "connection error: PROTOCOL_ERROR".
"connection error: ",
// The code an upstream sends to say the request must be retried
// over HTTP/1.1, as a GOAWAY or on the stream.
http11RequiredMarker,
}
const (
// http11RequiredMarker is the error code an upstream sends to say the
// request belongs on HTTP/1.1. Unlike the other markers it is not a
// fault: the upstream is describing its own configuration.
http11RequiredMarker = "HTTP_1_1_REQUIRED"
// A GOAWAY carrying NO_ERROR closes a connection without complaint:
// a server draining before shutdown, recycling an application pool,
// capping requests per connection. The upstream speaks h2 perfectly
// well, so this must never pin it. The two spellings are the two
// formats the bundled transport prints the code in.
goAwayNoErrorEqualsMarker = "ErrCode=NO_ERROR"
goAwayNoErrorColonMarker = "ErrCode:NO_ERROR"
// gracefulGoAwayMarker is errClientConnGotGoAway, which the bundled
// transport raises for a stream the server never received on a
// connection it is shutting down gracefully. It normally retries
// those itself on a new connection and this never surfaces.
gracefulGoAwayMarker = "Transport received Server's graceful shutdown GOAWAY"
// goAwayStreamNotReceivedMarker is the bundled transport's abort for
// the first stream on a connection whose GOAWAY carried a real error
// code — the IIS case. It sits in the same "streamID > LastStreamID"
// branch as the graceful abort, so the server had not received the
// stream (see net/http's h2_bundle.go).
goAwayStreamNotReceivedMarker = "Transport received GOAWAY from server ErrCode:"
// goAwayNothingProcessedMarker is a GoAwayError naming stream 0 as
// the last one received, which says the same thing. The trailing
// comma keeps it from matching LastStreamID=10 and the rest.
goAwayNothingProcessedMarker = "LastStreamID=0,"
)
// isHTTP11Required reports whether the upstream itself asked for
// HTTP/1.1, rather than merely failing at h2.
func isHTTP11Required(err error) bool {
return err != nil && strings.Contains(transportError(err).Error(), http11RequiredMarker)
}
// isHTTP2ProtocolError reports whether err says the upstream cannot
// serve the h2 it negotiated.
func isHTTP2ProtocolError(err error) bool {
if err == nil {
return false
}
msg := transportError(err).Error()
// A graceful GOAWAY is routine connection management, not an
// upstream that cannot serve h2.
if isGracefulGoAway(msg) {
return false
}
for _, marker := range http2ErrorMarkers {
if strings.Contains(msg, marker) {
return true
}
}
return false
}
// isGracefulGoAway reports whether msg describes a GOAWAY sent to close
// a healthy connection rather than to report an inability to serve h2.
func isGracefulGoAway(msg string) bool {
return strings.Contains(msg, goAwayNoErrorEqualsMarker) ||
strings.Contains(msg, goAwayNoErrorColonMarker) ||
strings.Contains(msg, gracefulGoAwayMarker)
}
// transportError strips a *url.Error wrapper, which prefixes the request
// URL to the message. Markers are matched as substrings, so a URL left
// in place could classify an ordinary dial or TLS failure as an h2 one
// on the strength of the path alone.
func transportError(err error) error {
var urlErr *url.Error
if errors.As(err, &urlErr) && urlErr.Err != nil {
return urlErr.Err
}
return err
}
+587
View File
@@ -0,0 +1,587 @@
package roundtrip
import (
"bufio"
"context"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"errors"
"fmt"
"io"
"math/big"
"net"
"net/http"
"net/url"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/net/http2"
)
// TestUpstreamTransport_AutoFallsBackOnBrokenHTTP2 covers the case ALPN
// cannot express: the upstream advertises h2, picks it, and then cannot
// serve it. The request must still succeed, over HTTP/1.1, and the
// upstream must stay on HTTP/1.1 for the requests that follow.
func TestUpstreamTransport_AutoFallsBackOnBrokenHTTP2(t *testing.T) {
t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTPAuto))
srv := startBrokenHTTP2Server(t)
mt := NewDirectOnly(nil)
ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background()))
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil)
require.NoError(t, err)
resp, err := mt.RoundTrip(req)
require.NoError(t, err, "a replayable request must be retried on HTTP/1.1 instead of failing")
body, err := io.ReadAll(resp.Body)
_ = resp.Body.Close()
require.NoError(t, err)
assert.Equal(t, "HTTP/1.1", resp.Proto, "the retry must ride the HTTP/1.1 transport")
assert.Equal(t, "http/1.1", string(body), "the upstream must see an http/1.1 ALPN offer on the retry")
assert.True(t, mt.insecure.isDowngraded(srv.addr),
"the upstream must stay pinned to HTTP/1.1 after proving it cannot serve h2")
mt.insecure.mu.RLock()
pin := mt.insecure.downgraded[srv.addr]
mt.insecure.mu.RUnlock()
assert.True(t, pin.permanent(),
"an upstream answering HTTP_1_1_REQUIRED must not be re-probed for h2")
// The second request must not repeat the h2 attempt: the server
// counts h2 handshakes, so a repeat would show up here.
h2Attempts := srv.http2Handshakes()
req, err = http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil)
require.NoError(t, err)
resp, err = mt.RoundTrip(req)
require.NoError(t, err)
_ = resp.Body.Close()
assert.Equal(t, "HTTP/1.1", resp.Proto, "a pinned upstream must go straight to HTTP/1.1")
assert.Equal(t, h2Attempts, srv.http2Handshakes(),
"a pinned upstream must not be probed for h2 again until the pin expires")
}
// TestUpstreamTransport_ExplicitHTTP2NeverDowngrades pins the promise
// that the explicit values are absolute: an operator who asked for h2
// keeps h2, broken upstream or not.
func TestUpstreamTransport_ExplicitHTTP2NeverDowngrades(t *testing.T) {
t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTP2))
srv := startBrokenHTTP2Server(t)
mt := NewDirectOnly(nil)
ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background()))
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil)
require.NoError(t, err)
resp, err := mt.RoundTrip(req)
if err == nil {
_ = resp.Body.Close()
}
require.Error(t, err, "NB_PROXY_UPSTREAM_HTTP_VERSION=2 must not fall back to HTTP/1.1")
assert.False(t, mt.insecure.isDowngraded(srv.addr), "an explicit version must never pin an upstream")
}
// TestUpstreamTransport_AutoDoesNotReplayUnsafeRequests covers the
// other half of the fallback: an h2 failure says nothing about whether
// the upstream already applied the request, so a state-changing one is
// not replayed. The host is still pinned, so the next request rides
// HTTP/1.1 without a second h2 attempt.
func TestUpstreamTransport_AutoDoesNotReplayUnsafeRequests(t *testing.T) {
t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTPAuto))
srv := startBrokenHTTP2Server(t)
// A stream error means the stream was open, so the upstream had the
// request in hand — unlike the GOAWAY at stream 0 the fake server
// sends, which states it processed nothing.
streamErr := http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol}
mt := NewDirectOnly(nil)
transport := mt.insecure
ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background()))
assert.False(t, safeToRetry(newTestRequest(t, ctx, http.MethodPost, srv.addr), streamErr),
"a POST must not be replayed after a failure that may have been applied")
assert.True(t, safeToRetry(newTestRequest(t, ctx, http.MethodGet, srv.addr), streamErr),
"a GET is safe to replay whatever the failure was")
// The fake server's GOAWAY names stream 0, so even a POST is safe
// there and the request must succeed over HTTP/1.1.
resp, err := transport.RoundTrip(newTestRequest(t, ctx, http.MethodPost, srv.addr))
require.NoError(t, err, "a GOAWAY at stream 0 means the upstream applied nothing, so the POST may be replayed")
body, err := io.ReadAll(resp.Body)
_ = resp.Body.Close()
require.NoError(t, err)
assert.Equal(t, "http/1.1", string(body), "the retry must reach the upstream over http/1.1")
h2Attempts := srv.http2Handshakes()
resp, err = transport.RoundTrip(newTestRequest(t, ctx, http.MethodPost, srv.addr))
require.NoError(t, err)
_ = resp.Body.Close()
assert.Equal(t, h2Attempts, srv.http2Handshakes(),
"the pin must carry later requests without another h2 attempt")
}
func newTestRequest(t *testing.T, ctx context.Context, method, addr string) *http.Request {
t.Helper()
req, err := http.NewRequestWithContext(ctx, method, "https://"+addr, strings.NewReader("payload"))
require.NoError(t, err)
return req
}
func TestUpstreamKey(t *testing.T) {
tests := []struct {
name string
url string
want string
}{
{name: "host", url: "https://backend.invalid/path", want: "backend.invalid"},
// DNS is case-insensitive, so these are one upstream and must
// share one pin.
{name: "mixed case", url: "https://Backend.INVALID/path", want: "backend.invalid"},
// The default port is implied on every path that can downgrade.
{name: "explicit default port", url: "https://backend.invalid:443/", want: "backend.invalid"},
{name: "non-default port", url: "https://backend.invalid:8443/", want: "backend.invalid:8443"},
// An IPv6 literal needs its brackets back after Hostname strips
// them, or the key is not a dialable authority.
{name: "ipv6 default port", url: "https://[2001:db8::1]/", want: "2001:db8::1"},
{name: "ipv6 with port", url: "https://[2001:db8::1]:8443/", want: "[2001:db8::1]:8443"},
// One address in three spellings: hex case, a leading zero and an
// uncompressed zero run are all the same upstream.
{name: "ipv6 upper case", url: "https://[2001:DB8::1]/", want: "2001:db8::1"},
{name: "ipv6 leading zero", url: "https://[2001:0db8::1]/", want: "2001:db8::1"},
{name: "ipv6 uncompressed", url: "https://[2001:db8:0:0:0:0:0:1]/", want: "2001:db8::1"},
// A v4-mapped address is the v4 address, not a second upstream.
{name: "v4-mapped", url: "https://[::ffff:192.0.2.1]/", want: "192.0.2.1"},
// A zone names an interface, and interface names are
// case-sensitive, so these two are different links.
{name: "ipv6 zone", url: "https://[fe80::1%25eth0]/", want: "fe80::1%eth0"},
{name: "ipv6 zone upper case", url: "https://[fe80::1%25ETH0]/", want: "fe80::1%ETH0"},
// The address before the zone still normalizes.
{name: "ipv6 zone with upper-case address", url: "https://[FE80::1%25eth0]/", want: "fe80::1%eth0"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
parsed, err := url.Parse(tc.url)
require.NoError(t, err)
assert.Equal(t, tc.want, upstreamKey(parsed))
})
}
}
func TestSafeToRetry(t *testing.T) {
streamErr := http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol}
goAwayAtZero := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`)
goAwayLater := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=11, ErrCode=PROTOCOL_ERROR, debug=""`)
tests := []struct {
name string
method string
headers map[string]string
err error
want bool
}{
{name: "get", method: http.MethodGet, err: streamErr, want: true},
{name: "head", method: http.MethodHead, err: streamErr, want: true},
{name: "options", method: http.MethodOptions, err: streamErr, want: true},
{name: "trace", method: http.MethodTrace, err: streamErr, want: true},
{name: "post", method: http.MethodPost, err: streamErr, want: false},
{name: "put", method: http.MethodPut, err: streamErr, want: false},
{name: "patch", method: http.MethodPatch, err: streamErr, want: false},
{name: "delete", method: http.MethodDelete, err: streamErr, want: false},
// The upstream reported it handled nothing, so repeating the
// request cannot duplicate anything.
{name: "post with goaway at stream 0", method: http.MethodPost, err: goAwayAtZero, want: true},
// What the bundled transport actually raises for the first
// stream on a connection the upstream GOAWAYs with a real error
// code, which is the shape a real IIS site produces.
{
name: "post with first-stream goaway abort",
method: http.MethodPost,
err: errors.New("http2: Transport received GOAWAY from server ErrCode:HTTP_1_1_REQUIRED"),
want: true,
},
// It handled earlier streams, so this one may have been applied.
{name: "post with goaway after other streams", method: http.MethodPost, err: goAwayLater, want: false},
{
name: "post with idempotency key",
method: http.MethodPost,
headers: map[string]string{"Idempotency-Key": "abc"},
err: streamErr,
want: true,
},
{
name: "post with prefixed idempotency key",
method: http.MethodPost,
headers: map[string]string{"X-Idempotency-Key": "abc"},
err: streamErr,
want: true,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
req, err := http.NewRequest(tc.method, "https://backend.invalid", nil)
require.NoError(t, err)
for k, v := range tc.headers {
req.Header.Set(k, v)
}
assert.Equal(t, tc.want, safeToRetry(req, tc.err))
})
}
}
func TestUpstreamTransport_MayDowngrade(t *testing.T) {
tests := []struct {
name string
version upstreamHTTPVersion
url string
want bool
}{
{name: "auto over TLS", version: upstreamHTTPAuto, url: "https://backend.invalid", want: true},
// The proxy speaks no h2c, so a cleartext upstream is already on
// HTTP/1.1 and its failures say nothing about h2.
{name: "auto cleartext", version: upstreamHTTPAuto, url: "http://backend.invalid", want: false},
{name: "explicit 1.1", version: upstreamHTTP11, url: "https://backend.invalid", want: false},
{name: "explicit 2", version: upstreamHTTP2, url: "https://backend.invalid", want: false},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
transport := newUpstreamTransport(&http.Transport{}, tc.version, nil)
req, err := http.NewRequest(http.MethodGet, tc.url, nil)
require.NoError(t, err)
assert.Equal(t, tc.want, transport.mayDowngrade(req))
})
}
}
func TestUpstreamTransport_DowngradeExpires(t *testing.T) {
transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil)
transport.markDowngraded("backend.invalid:443", false)
require.True(t, transport.isDowngraded("backend.invalid:443"))
transport.mu.Lock()
transport.downgraded["backend.invalid:443"] = downgrade{expiry: time.Now().Add(-time.Second)}
transport.mu.Unlock()
assert.False(t, transport.isDowngraded("backend.invalid:443"),
"an expired pin must let the upstream be offered h2 again")
transport.mu.RLock()
_, stillTracked := transport.downgraded["backend.invalid:443"]
transport.mu.RUnlock()
assert.False(t, stillTracked, "an expired pin must not be kept around")
}
func TestUpstreamTransport_DowngradeIsPerUpstream(t *testing.T) {
transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil)
transport.markDowngraded("broken.invalid:443", false)
assert.True(t, transport.isDowngraded("broken.invalid:443"))
assert.False(t, transport.isDowngraded("healthy.invalid:443"),
"one broken upstream must not drop the others to HTTP/1.1")
}
// TestUpstreamTransport_HTTP11RequiredPinIsPermanent covers the IIS
// case: HTTP_1_1_REQUIRED describes how the upstream is configured
// (Windows Authentication, client certificates), so re-probing it every
// upstreamDowngradeTTL would only buy a failed request per interval.
func TestUpstreamTransport_HTTP11RequiredPinIsPermanent(t *testing.T) {
transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil)
transport.markDowngraded("iis.invalid:443", true)
transport.mu.RLock()
pin := transport.downgraded["iis.invalid:443"]
transport.mu.RUnlock()
assert.True(t, pin.permanent(), "an upstream that asked for HTTP/1.1 must not be re-probed")
assert.True(t, pin.active(time.Now().Add(100*upstreamDowngradeTTL)),
"a permanent pin must outlive any TTL")
}
func TestUpstreamTransport_PermanentPinSurvivesLaterFailures(t *testing.T) {
transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil)
transport.markDowngraded("iis.invalid:443", true)
// A later ambiguous failure for the same upstream must not turn the
// permanent pin into an expiring one.
transport.markDowngraded("iis.invalid:443", false)
transport.mu.RLock()
pin := transport.downgraded["iis.invalid:443"]
transport.mu.RUnlock()
assert.True(t, pin.permanent(), "a permanent pin must never be weakened")
}
func TestIsHTTP11Required(t *testing.T) {
tests := []struct {
name string
err error
want bool
}{
{
name: "goaway",
err: errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`),
want: true,
},
{
name: "stream error",
err: http2.StreamError{StreamID: 1, Code: http2.ErrCodeHTTP11Required},
want: true,
},
{
name: "other h2 failure",
err: http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol},
want: false,
},
{name: "nil", err: nil, want: false},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
assert.Equal(t, tc.want, isHTTP11Required(tc.err))
})
}
}
func TestIsHTTP2ProtocolError(t *testing.T) {
tests := []struct {
name string
err error
want bool
}{
{
name: "goaway demanding http/1.1",
err: errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`),
want: true,
},
{
name: "stream error",
err: http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol},
want: true,
},
{
name: "connection error",
err: http2.ConnectionError(http2.ErrCodeProtocol),
want: true,
},
{
name: "wrapped h2 error",
err: errors.New("Get \"https://backend.invalid\": http2: client connection lost"),
want: true,
},
// A GOAWAY with NO_ERROR is a server draining a connection —
// recycling an application pool, capping requests per
// connection, shutting down gracefully. It speaks h2 fine.
{
name: "graceful goaway",
err: errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=9, ErrCode=NO_ERROR, debug=""`),
want: false,
},
{
name: "graceful shutdown abort",
err: errors.New("http2: Transport received Server's graceful shutdown GOAWAY"),
want: false,
},
// Markers are substrings, so a URL carried by a *url.Error must
// not be able to classify a plain failure as an h2 one.
{
name: "url error whose path looks like a marker",
err: &url.Error{
Op: "Get",
URL: "https://backend.invalid/http2:/connection error: x",
Err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused"),
},
want: false,
},
// Retrying these on HTTP/1.1 fixes nothing, so they must never
// pin an upstream.
{name: "dial failure", err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused"), want: false},
{name: "tls failure", err: errors.New("tls: failed to verify certificate: x509: certificate signed by unknown authority"), want: false},
{name: "context cancelled", err: context.Canceled, want: false},
{name: "eof", err: io.EOF, want: false},
{name: "nil", err: nil, want: false},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
assert.Equal(t, tc.want, isHTTP2ProtocolError(tc.err))
})
}
}
func TestReplayable(t *testing.T) {
t.Run("bodyless request", func(t *testing.T) {
req, err := http.NewRequest(http.MethodGet, "https://backend.invalid", nil)
require.NoError(t, err)
retry, ok := replayable(req)
require.True(t, ok)
assert.Same(t, req, retry, "a bodyless request needs no clone")
})
t.Run("request with GetBody", func(t *testing.T) {
req, err := http.NewRequest(http.MethodPost, "https://backend.invalid", strings.NewReader("payload"))
require.NoError(t, err)
// Consume the body the way a failed RoundTrip would.
_, err = io.ReadAll(req.Body)
require.NoError(t, err)
retry, ok := replayable(req)
require.True(t, ok)
body, err := io.ReadAll(retry.Body)
require.NoError(t, err)
assert.Equal(t, "payload", string(body), "the retry must carry a fresh copy of the body")
})
t.Run("streamed request", func(t *testing.T) {
req, err := http.NewRequest(http.MethodPost, "https://backend.invalid", io.NopCloser(strings.NewReader("payload")))
require.NoError(t, err)
require.Nil(t, req.GetBody, "an opaque reader must not get a GetBody")
_, ok := replayable(req)
assert.False(t, ok, "a body that cannot be regenerated must not be replayed")
})
}
// brokenHTTP2Server advertises h2 in ALPN, accepts it, and then refuses
// to serve it — the upstream behaviour that motivated the fallback. Over
// http/1.1 it answers normally, so a downgraded request succeeds.
type brokenHTTP2Server struct {
addr string
handshakes chan struct{}
}
func (s *brokenHTTP2Server) http2Handshakes() int {
return len(s.handshakes)
}
func startBrokenHTTP2Server(t *testing.T) *brokenHTTP2Server {
t.Helper()
ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{
Certificates: []tls.Certificate{selfSignedCert(t)},
NextProtos: []string{"h2", "http/1.1"},
MinVersion: tls.VersionTLS12,
})
require.NoError(t, err)
t.Cleanup(func() { _ = ln.Close() })
srv := &brokenHTTP2Server{
addr: ln.Addr().String(),
// Buffered well past what the test drives so a stuck server
// never blocks the accept loop.
handshakes: make(chan struct{}, 64),
}
go func() {
for {
conn, err := ln.Accept()
if err != nil {
return
}
go srv.handle(conn)
}
}()
return srv
}
func (s *brokenHTTP2Server) handle(conn net.Conn) {
defer func() { _ = conn.Close() }()
tlsConn, ok := conn.(*tls.Conn)
if !ok {
return
}
if err := tlsConn.Handshake(); err != nil {
return
}
proto := tlsConn.ConnectionState().NegotiatedProtocol
if proto == "h2" {
select {
case s.handshakes <- struct{}{}:
default:
}
s.refuseHTTP2(tlsConn)
return
}
s.serveHTTP1(tlsConn, proto)
}
// refuseHTTP2 completes just enough of the h2 handshake for the client
// to accept the connection, then sends the GOAWAY an upstream uses to
// say the request belongs on HTTP/1.1.
//
// The client is still writing its preface and request while the GOAWAY
// goes out, so the connection is drained before the caller closes it.
// Closing a socket with unread bytes still in its receive buffer makes
// the kernel answer with RST, which reaches the client as a write error
// rather than the GOAWAY — no h2 error, so no downgrade, and the test
// fails on the error the client saw first.
func (s *brokenHTTP2Server) refuseHTTP2(conn net.Conn) {
framer := http2.NewFramer(conn, conn)
if err := framer.WriteSettings(); err != nil {
return
}
if err := framer.WriteGoAway(0, http2.ErrCodeHTTP11Required, nil); err != nil {
return
}
// The client closes its side once it has read the GOAWAY, which ends
// the drain; the deadline is only a backstop against a client that
// never does.
_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
_, _ = io.Copy(io.Discard, conn)
}
// serveHTTP1 answers a single request with the ALPN protocol the
// upstream actually settled on, so a test asserting on the body is
// checking what the upstream saw rather than a constant.
func (s *brokenHTTP2Server) serveHTTP1(conn net.Conn, alpn string) {
reader := bufio.NewReader(conn)
if _, err := http.ReadRequest(reader); err != nil {
return
}
_, _ = fmt.Fprintf(conn,
"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s",
len(alpn), alpn)
}
func selfSignedCert(t *testing.T) tls.Certificate {
t.Helper()
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
require.NoError(t, err)
template := x509.Certificate{
SerialNumber: big.NewInt(1),
Subject: pkix.Name{CommonName: "127.0.0.1"},
NotBefore: time.Now().Add(-time.Hour),
NotAfter: time.Now().Add(time.Hour),
KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
IPAddresses: []net.IP{net.ParseIP("127.0.0.1")},
IsCA: true,
}
der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
require.NoError(t, err)
return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
}
+4 -3
View File
@@ -567,9 +567,10 @@ func TestIntegration_ProxyConnection_ReconnectDoesNotDuplicateState(t *testing.T
// Apply to real auth middleware (idempotent)
err := authMw.AddDomain(mapping.GetDomain(), auth.DomainSettings{
AccountID: proxytypes.AccountID(mapping.GetAccountId()),
ServiceID: proxytypes.ServiceID(mapping.GetId()),
Private: mapping.GetPrivate(),
AccountID: proxytypes.AccountID(mapping.GetAccountId()),
ServiceID: proxytypes.ServiceID(mapping.GetId()),
Private: mapping.GetPrivate(),
AllowedGroups: mapping.GetAuth().GetAllowedGroupIds(),
})
require.NoError(t, err)
+1
View File
@@ -2169,6 +2169,7 @@ func (s *Server) updateMapping(ctx context.Context, mapping *proto.ProxyMapping)
IPRestrictions: ipRestrictions,
Private: mapping.GetPrivate(),
AppSecMode: s.appSecMode(mapping),
AllowedGroups: mapping.GetAuth().GetAllowedGroupIds(),
}
if err := s.auth.AddDomain(mapping.GetDomain(), settings); err != nil {
return fmt.Errorf("auth setup for domain %s: %w", mapping.GetDomain(), err)
+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.");
}