Compare commits

..

2 Commits

Author SHA1 Message Date
Viktor Liu
663468e199 Add CrowdSec AppSec request inspection to the reverse proxy 2026-07-28 10:51:54 +02:00
Viktor Liu
d681670a9d [misc] Restore the rootless-latest docker tag (#6914) 2026-07-27 12:07:58 +02:00
61 changed files with 3908 additions and 4497 deletions

View File

@@ -273,8 +273,8 @@ dockers_v2:
- netbirdio/netbird
- ghcr.io/netbirdio/netbird
tags:
- "v{{ .Version }}-rootless"
- "{{ if eq .Env.SKIP_PUBLISH \"false\" }}latest{{ end }}"
- "{{ .Version }}-rootless"
- "{{ if eq .Env.SKIP_PUBLISH \"false\" }}rootless-latest{{ end }}"
dockerfile: client/Dockerfile-rootless
extra_files:
- client/netbird-entrypoint.sh

View File

@@ -228,6 +228,8 @@ read_enable_crowdsec() {
echo "CrowdSec checks client IPs against a community threat intelligence database" > /dev/stderr
echo "and blocks known malicious sources before they reach your services." > /dev/stderr
echo "A local CrowdSec LAPI container will be added to your deployment." > /dev/stderr
echo "It also enables the AppSec (WAF) endpoint, so services can inspect HTTP" > /dev/stderr
echo "requests for exploits. Both stay off per service until you enable them." > /dev/stderr
echo -n "Enable CrowdSec? [y/N]: " > /dev/stderr
read -r CHOICE < /dev/tty
@@ -497,7 +499,8 @@ generate_configuration_files() {
# TCP ServersTransport for PROXY protocol v2 to the proxy backend
render_traefik_dynamic > traefik-dynamic.yaml
if [[ "$ENABLE_CROWDSEC" == "true" ]]; then
mkdir -p crowdsec
mkdir -p crowdsec/acquis.d
render_crowdsec_appsec_acquis > crowdsec/acquis.d/appsec.yaml
fi
fi
;;
@@ -531,6 +534,23 @@ generate_configuration_files() {
return 0
}
# The AppSec (WAF) listener only exists if an appsec acquisition datasource is
# configured. One datasource is one listener carrying one merged rule set: the
# protocol has no rule-set selector, so per-service rule variation would need
# either a second datasource on another port or pre_eval hooks filtering on
# req.Host.
render_crowdsec_appsec_acquis() {
cat <<EOF
source: appsec
listen_addr: 0.0.0.0:7422
appsec_configs:
- crowdsecurity/appsec-default
labels:
type: appsec
EOF
return 0
}
start_services_and_show_instructions() {
# For built-in Traefik, start containers immediately
# For NPM, start containers first (NPM needs services running to create proxy)
@@ -742,7 +762,11 @@ render_docker_compose_traefik_builtin() {
restart: unless-stopped
networks: [netbird]
environment:
COLLECTIONS: crowdsecurity/linux
# appsec-generic-rules is required alongside appsec-virtual-patching:
# the appsec-default config references crowdsecurity/generic-* and
# crowdsecurity/experimental-*, which only that collection provides, and
# the engine exits at startup if they are missing.
COLLECTIONS: crowdsecurity/linux crowdsecurity/appsec-virtual-patching crowdsecurity/appsec-generic-rules
volumes:
- ./crowdsec:/etc/crowdsec
- crowdsec_db:/var/lib/crowdsec/data
@@ -1007,6 +1031,11 @@ EOF
cat <<EOF
NB_PROXY_CROWDSEC_API_URL=http://crowdsec:8080
NB_PROXY_CROWDSEC_API_KEY=$CROWDSEC_BOUNCER_KEY
# AppSec (WAF) request inspection. Separate endpoint from the LAPI above and
# validated with the same bouncer key. Setting it makes the proxy advertise the
# AppSec capability, which is what lets a service select appsec_mode; nothing is
# inspected until a service opts in.
NB_PROXY_CROWDSEC_APPSEC_URL=http://crowdsec:7422/
EOF
fi

View File

@@ -25,8 +25,8 @@ type Model struct {
// These typically need NetBird identity stamped onto upstream
// requests so the gateway's analytics and budgets attribute to the
// real caller; that's what IdentityInjection is for.
// - KindCustom: named and generic OpenAI-compatible self-hosted
// endpoints (vLLM, Ollama, custom inference servers).
// - KindCustom: the catch-all "OpenAI-compatible self-hosted endpoint"
// entry (vLLM, Ollama, custom inference servers).
//
// Frontend uses Kind to group the provider Select in the modal so an
// operator can spot at a glance which catalog entries proxy other
@@ -40,18 +40,6 @@ const (
KindCustom ProviderKind = "custom"
)
// AuthMode describes whether a catalog provider accepts an upstream
// credential. The zero value intentionally resolves to AuthModeRequired so
// existing and newly-added catalog entries fail closed unless they explicitly
// opt into a less restrictive mode.
type AuthMode string
const (
AuthModeRequired AuthMode = "required"
AuthModeOptional AuthMode = "optional"
AuthModeNone AuthMode = "none"
)
// Provider is the in-memory representation of a catalog provider.
type Provider struct {
ID string
@@ -60,9 +48,6 @@ type Provider struct {
DefaultHost string
// Kind groups this entry for UI presentation; see ProviderKind.
Kind ProviderKind
// AuthMode declares whether the provider requires, optionally accepts, or
// does not support an upstream API key. Empty defaults to required.
AuthMode AuthMode
// AuthHeaderName is the HTTP header the provider's API expects
// the credential under (e.g. "Authorization" for OpenAI,
// "x-api-key" for Anthropic). Combined with AuthHeaderTemplate
@@ -101,17 +86,6 @@ type Provider struct {
// upstream provider + credentials on Portkey's hosted side).
ExtraHeaders []ExtraHeader
Models []Model
// ModelDiscovery configures provider-specific discovery behavior. A nil
// profile means discovery is unsupported. The profile never carries
// caller-supplied paths: the proxy owns the fixed endpoint allowlist.
ModelDiscovery *ModelDiscovery
}
// ModelDiscovery describes the safe, catalog-owned fallback behavior for a
// discoverable provider. Every discovery starts with OpenAI-compatible
// /v1/models; OllamaFallback permits /api/tags only when that route is absent.
type ModelDiscovery struct {
OllamaFallback bool
}
// ExtraHeader names a single optional per-provider routing/config
@@ -487,27 +461,6 @@ var providers = []Provider{
{ID: "kimi-k3", Label: "Kimi K3", InputPer1k: 0.003, OutputPer1k: 0.015, ContextWindow: 1000000},
},
},
{
// Direct Ollama Cloud uses the hosted OpenAI-compatible /v1 API.
// Models are discovered dynamically because the Cloud catalog changes
// independently of NetBird releases. ParserID intentionally remains
// empty to preserve the routing behavior of Ollama, vLLM, and custom
// OpenAI-compatible providers.
ID: "ollama_cloud",
Kind: KindProvider,
AuthMode: AuthModeRequired,
Name: "Ollama Cloud",
Description: "Hosted Ollama models via the OpenAI-compatible API",
DefaultHost: "ollama.com",
AuthHeaderName: "Authorization",
AuthHeaderTemplate: "Bearer ${API_KEY}",
DefaultContentType: "application/json",
BrandColor: "#000000",
Models: []Model{},
ModelDiscovery: &ModelDiscovery{
OllamaFallback: true,
},
},
{
ID: "litellm_proxy",
Kind: KindGateway,
@@ -748,31 +701,11 @@ var providers = []Provider{
BrandColor: "#30A2FF",
Models: []Model{},
},
{
// Ollama exposes an OpenAI-compatible /v1 API. Like vLLM, it gets a
// dedicated catalog id for provider-specific setup guidance while
// retaining generic custom-provider routing. Authentication is optional
// because local Ollama is authless but protected front ends may use it.
ID: "ollama",
Kind: KindCustom,
AuthMode: AuthModeOptional,
Name: "Ollama",
Description: "Self-hosted Ollama (OpenAI-compatible)",
DefaultHost: "",
AuthHeaderName: "Authorization",
AuthHeaderTemplate: "Bearer ${API_KEY}",
DefaultContentType: "application/json",
BrandColor: "#000000",
Models: []Model{},
ModelDiscovery: &ModelDiscovery{
OllamaFallback: true,
},
},
{
ID: "custom",
Kind: KindCustom,
Name: "Custom / Self-hosted",
Description: "Other OpenAI-compatible endpoint",
Description: "OpenAI-compatible endpoint (vLLM, Ollama, …)",
DefaultHost: "",
AuthHeaderName: "Authorization",
AuthHeaderTemplate: "Bearer ${API_KEY}",
@@ -805,15 +738,6 @@ func IsKnown(id string) bool {
return ok
}
// EffectiveAuthMode returns the provider's configured authentication mode.
// Treating the zero value as required keeps older catalog entries fail closed.
func (p Provider) EffectiveAuthMode() AuthMode {
if p.AuthMode == "" {
return AuthModeRequired
}
return p.AuthMode
}
// IsVertexPathStyle reports whether a provider uses the Google Vertex AI
// request shape — the model is carried in the URL path
// (/v1/projects/{p}/locations/{r}/publishers/{pub}/models/{model}:{action})
@@ -850,17 +774,15 @@ func (p Provider) ToAPIResponse() api.AgentNetworkCatalogProvider {
kind = api.AgentNetworkCatalogProviderKindCustom
}
resp := api.AgentNetworkCatalogProvider{
Id: p.ID,
Name: p.Name,
Description: p.Description,
DefaultHost: p.DefaultHost,
Kind: kind,
AuthMode: api.AgentNetworkCatalogProviderAuthMode(p.EffectiveAuthMode()),
AuthHeaderTemplate: p.AuthHeaderTemplate,
DefaultContentType: p.DefaultContentType,
BrandColor: p.BrandColor,
Models: models,
SupportsModelDiscovery: p.ModelDiscovery != nil,
Id: p.ID,
Name: p.Name,
Description: p.Description,
DefaultHost: p.DefaultHost,
Kind: kind,
AuthHeaderTemplate: p.AuthHeaderTemplate,
DefaultContentType: p.DefaultContentType,
BrandColor: p.BrandColor,
Models: models,
}
if len(p.ExtraHeaders) > 0 {
extras := make([]api.AgentNetworkCatalogExtraHeader, 0, len(p.ExtraHeaders))

View File

@@ -1,96 +0,0 @@
package catalog
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/shared/management/http/api"
)
func TestOllamaCatalogEntry(t *testing.T) {
entry, ok := Lookup("ollama")
require.True(t, ok, "Ollama must be available as a dedicated catalog provider")
assert.Equal(t, KindCustom, entry.Kind)
assert.Equal(t, AuthModeOptional, entry.EffectiveAuthMode())
assert.Equal(t, "Ollama", entry.Name)
assert.Equal(t, "Self-hosted Ollama (OpenAI-compatible)", entry.Description)
assert.Empty(t, entry.DefaultHost, "the dashboard owns Ollama's scheme-aware HTTP placeholder")
assert.Equal(t, "Authorization", entry.AuthHeaderName)
assert.Equal(t, "Bearer ${API_KEY}", entry.AuthHeaderTemplate)
assert.Equal(t, "application/json", entry.DefaultContentType)
assert.Empty(t, entry.ParserID, "Ollama preserves the untagged vLLM/custom routing behavior")
assert.Empty(t, entry.Models, "Ollama models are installed dynamically on the configured endpoint")
require.NotNil(t, entry.ModelDiscovery)
assert.True(t, entry.ModelDiscovery.OllamaFallback)
wire := entry.ToAPIResponse()
assert.Equal(t, "ollama", wire.Id)
assert.Equal(t, api.AgentNetworkCatalogProviderKindCustom, wire.Kind)
assert.Equal(t, api.AgentNetworkCatalogProviderAuthModeOptional, wire.AuthMode)
assert.True(t, wire.SupportsModelDiscovery)
assert.NotNil(t, wire.Models)
assert.Empty(t, wire.Models)
}
func TestOllamaCloudCatalogEntry(t *testing.T) {
entry, ok := Lookup("ollama_cloud")
require.True(t, ok, "Ollama Cloud must be available as a dedicated catalog provider")
assert.Equal(t, KindProvider, entry.Kind)
assert.Equal(t, AuthModeRequired, entry.EffectiveAuthMode())
assert.Equal(t, "Ollama Cloud", entry.Name)
assert.Equal(t, "Hosted Ollama models via the OpenAI-compatible API", entry.Description)
assert.Equal(t, "ollama.com", entry.DefaultHost)
assert.Equal(t, "Authorization", entry.AuthHeaderName)
assert.Equal(t, "Bearer ${API_KEY}", entry.AuthHeaderTemplate)
assert.Equal(t, "application/json", entry.DefaultContentType)
assert.Empty(t, entry.ParserID, "Ollama Cloud preserves the untagged Ollama/vLLM/custom routing behavior")
assert.Empty(t, entry.Models, "Ollama Cloud models are discovered dynamically")
require.NotNil(t, entry.ModelDiscovery)
assert.True(t, entry.ModelDiscovery.OllamaFallback)
wire := entry.ToAPIResponse()
assert.Equal(t, "ollama_cloud", wire.Id)
assert.Equal(t, api.AgentNetworkCatalogProviderKindProvider, wire.Kind)
assert.Equal(t, api.AgentNetworkCatalogProviderAuthModeRequired, wire.AuthMode)
assert.Equal(t, "ollama.com", wire.DefaultHost)
assert.True(t, wire.SupportsModelDiscovery)
assert.NotNil(t, wire.Models)
assert.Empty(t, wire.Models)
}
func TestOnlyOllamaProvidersSupportModelDiscovery(t *testing.T) {
discoverable := map[string]bool{
"ollama": true,
"ollama_cloud": true,
}
for _, entry := range All() {
supportsDiscovery := entry.ModelDiscovery != nil
assert.Equal(t, discoverable[entry.ID], supportsDiscovery, entry.ID)
assert.Equal(t, supportsDiscovery, entry.ToAPIResponse().SupportsModelDiscovery, entry.ID)
}
}
func TestCatalogAuthenticationModes(t *testing.T) {
openAI, ok := Lookup("openai_api")
require.True(t, ok)
assert.Empty(t, openAI.AuthMode, "existing entries use the fail-closed default")
assert.Equal(t, AuthModeRequired, openAI.EffectiveAuthMode())
assert.Equal(t, api.AgentNetworkCatalogProviderAuthModeRequired, openAI.ToAPIResponse().AuthMode)
for _, entry := range All() {
switch entry.EffectiveAuthMode() {
case AuthModeRequired, AuthModeOptional:
assert.NotEmpty(t, entry.AuthHeaderName, "%s must declare an auth header name", entry.ID)
assert.NotEmpty(t, entry.AuthHeaderTemplate, "%s must declare an auth header template", entry.ID)
case AuthModeNone:
assert.Empty(t, entry.AuthHeaderName, "%s must not declare an auth header name", entry.ID)
assert.Empty(t, entry.AuthHeaderTemplate, "%s must not declare an auth header template", entry.ID)
default:
t.Errorf("%s has invalid auth mode %q", entry.ID, entry.AuthMode)
}
}
}

View File

@@ -1,72 +0,0 @@
package handlers
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"github.com/gorilla/mux"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
nbcontext "github.com/netbirdio/netbird/management/server/context"
"github.com/netbirdio/netbird/shared/auth"
"github.com/netbirdio/netbird/shared/management/http/api"
)
type discoveryManagerStub struct {
agentnetwork.Manager
result *agentNetworkTypes.ModelDiscoveryResult
err error
accountID string
userID string
providerID string
}
func (m *discoveryManagerStub) DiscoverProviderModels(_ context.Context, accountID, userID, providerID string) (*agentNetworkTypes.ModelDiscoveryResult, error) {
m.accountID = accountID
m.userID = userID
m.providerID = providerID
return m.result, m.err
}
func TestDiscoverProviderModelsHandler(t *testing.T) {
manager := &discoveryManagerStub{
Manager: agentnetwork.NewManagerMock(),
result: &agentNetworkTypes.ModelDiscoveryResult{
RequestID: "probe-123",
Source: "ollama_api_tags",
ProxyCluster: "private.example.com",
Models: []agentNetworkTypes.DiscoveredModel{
{ID: "llama3.2:latest", Label: "llama3.2:latest"},
},
},
}
router := mux.NewRouter()
RegisterEndpoints(manager, router)
req := httptest.NewRequest(http.MethodPost, "/agent-network/providers/provider-1/discover-models", nil)
req = nbcontext.SetUserAuthInRequest(req, auth.UserAuth{
UserId: testUserID,
AccountId: testAccountID,
})
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
assert.Equal(t, "no-store", rec.Header().Get("Cache-Control"))
var response api.AgentNetworkModelDiscoveryResponse
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &response))
assert.Equal(t, testAccountID, manager.accountID)
assert.Equal(t, testUserID, manager.userID)
assert.Equal(t, "provider-1", manager.providerID)
assert.Equal(t, "probe-123", response.RequestId)
assert.Equal(t, "private.example.com", response.ProxyCluster)
assert.Equal(t, api.AgentNetworkModelDiscoveryResponseSourceOllamaApiTags, response.Source)
require.Len(t, response.Models, 1)
assert.Equal(t, "llama3.2:latest", response.Models[0].Id)
}

View File

@@ -35,7 +35,6 @@ func RegisterEndpoints(manager agentnetwork.Manager, router *mux.Router) {
router.HandleFunc("/agent-network/providers/{providerId}", h.getProvider).Methods("GET", "OPTIONS")
router.HandleFunc("/agent-network/providers/{providerId}", h.updateProvider).Methods("PUT", "OPTIONS")
router.HandleFunc("/agent-network/providers/{providerId}", h.deleteProvider).Methods("DELETE", "OPTIONS")
router.HandleFunc("/agent-network/providers/{providerId}/discover-models", h.discoverProviderModels).Methods("POST", "OPTIONS")
h.addPolicyEndpoints(router)
h.addGuardrailEndpoints(router)
h.addSettingsEndpoints(router)
@@ -99,41 +98,6 @@ func (h *handler) getProvider(w http.ResponseWriter, r *http.Request) {
util.WriteJSONObject(r.Context(), w, provider.ToAPIResponse())
}
func (h *handler) discoverProviderModels(w http.ResponseWriter, r *http.Request) {
userAuth, err := nbcontext.GetUserAuthFromContext(r.Context())
if err != nil {
util.WriteError(r.Context(), err, w)
return
}
providerID := strings.TrimSpace(mux.Vars(r)["providerId"])
if providerID == "" {
util.WriteError(r.Context(), status.Errorf(status.InvalidArgument, "provider ID is required"), w)
return
}
result, err := h.manager.DiscoverProviderModels(r.Context(), userAuth.AccountId, userAuth.UserId, providerID)
if err != nil {
util.WriteError(r.Context(), err, w)
return
}
models := make([]api.AgentNetworkDiscoveredModel, 0, len(result.Models))
for _, model := range result.Models {
models = append(models, api.AgentNetworkDiscoveredModel{
Id: model.ID,
Label: model.Label,
})
}
w.Header().Set("Cache-Control", "no-store")
util.WriteJSONObject(r.Context(), w, api.AgentNetworkModelDiscoveryResponse{
Models: models,
Source: api.AgentNetworkModelDiscoveryResponseSource(result.Source),
ProxyCluster: result.ProxyCluster,
RequestId: result.RequestID,
})
}
func (h *handler) createProvider(w http.ResponseWriter, r *http.Request) {
userAuth, err := nbcontext.GetUserAuthFromContext(r.Context())
if err != nil {
@@ -229,12 +193,11 @@ func (h *handler) deleteProvider(w http.ResponseWriter, r *http.Request) {
util.WriteJSONObject(r.Context(), w, util.EmptyObject{})
}
func validate(req *api.AgentNetworkProviderRequest, creating bool) error {
func validate(req *api.AgentNetworkProviderRequest, requireAPIKey bool) error {
if strings.TrimSpace(req.ProviderId) == "" {
return status.Errorf(status.InvalidArgument, "provider_id is required")
}
entry, ok := catalog.Lookup(req.ProviderId)
if !ok {
if !catalog.IsKnown(req.ProviderId) {
return status.Errorf(status.InvalidArgument, "provider_id %q is not a known catalog provider", req.ProviderId)
}
if strings.TrimSpace(req.Name) == "" {
@@ -247,27 +210,8 @@ func validate(req *api.AgentNetworkProviderRequest, creating bool) error {
if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
return status.Errorf(status.InvalidArgument, "upstream_url must be a full http(s) URL")
}
return validateProviderAPIKey(entry, req.ApiKey, creating)
}
func validateProviderAPIKey(entry catalog.Provider, apiKey *string, creating bool) error {
apiKeyBlank := apiKey == nil || strings.TrimSpace(*apiKey) == ""
switch entry.EffectiveAuthMode() {
case catalog.AuthModeRequired:
// Updates may omit the key to preserve it, but an explicitly empty
// value cannot clear a credential required by the selected provider.
if (creating || apiKey != nil) && apiKeyBlank {
return status.Errorf(status.InvalidArgument, "api_key is required for provider_id %q", entry.ID)
}
case catalog.AuthModeOptional:
// Both omitted and explicitly empty values are valid. The manager
// distinguishes preserve from clear during update.
case catalog.AuthModeNone:
if !apiKeyBlank {
return status.Errorf(status.InvalidArgument, "api_key is not supported for provider_id %q", entry.ID)
}
default:
return status.Errorf(status.InvalidArgument, "provider_id %q has an invalid authentication mode", entry.ID)
if requireAPIKey && (req.ApiKey == nil || strings.TrimSpace(*req.ApiKey) == "") {
return status.Errorf(status.InvalidArgument, "api_key is required")
}
return nil
}

View File

@@ -1,47 +0,0 @@
package handlers
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/catalog"
)
func TestValidateProviderAPIKey(t *testing.T) {
key := "secret"
empty := ""
tests := []struct {
name string
mode catalog.AuthMode
apiKey *string
creating bool
wantErr string
}{
{name: "required create with key", mode: catalog.AuthModeRequired, apiKey: &key, creating: true},
{name: "required create omitted", mode: catalog.AuthModeRequired, creating: true, wantErr: "required"},
{name: "required update omitted preserves", mode: catalog.AuthModeRequired},
{name: "required update explicit empty", mode: catalog.AuthModeRequired, apiKey: &empty, wantErr: "required"},
{name: "optional create omitted", mode: catalog.AuthModeOptional, creating: true},
{name: "optional update explicit empty", mode: catalog.AuthModeOptional, apiKey: &empty},
{name: "optional with key", mode: catalog.AuthModeOptional, apiKey: &key, creating: true},
{name: "none omitted", mode: catalog.AuthModeNone, creating: true},
{name: "none explicit empty", mode: catalog.AuthModeNone, apiKey: &empty},
{name: "none with key", mode: catalog.AuthModeNone, apiKey: &key, wantErr: "not supported"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
entry := catalog.Provider{ID: "test-provider", AuthMode: tt.mode}
err := validateProviderAPIKey(entry, tt.apiKey, tt.creating)
if tt.wantErr == "" {
require.NoError(t, err)
return
}
require.Error(t, err)
assert.Contains(t, err.Error(), tt.wantErr)
})
}
}

View File

@@ -9,12 +9,9 @@ import (
"strings"
"sync"
"time"
"unicode"
"unicode/utf8"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/catalog"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/labelgen"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
@@ -51,7 +48,6 @@ func ensureSessionKeys(p *types.Provider) error {
type Manager interface {
GetAllProviders(ctx context.Context, accountID, userID string) ([]*types.Provider, error)
GetProvider(ctx context.Context, accountID, userID, providerID string) (*types.Provider, error)
DiscoverProviderModels(ctx context.Context, accountID, userID, providerID string) (*types.ModelDiscoveryResult, error)
CreateProvider(ctx context.Context, userID string, provider *types.Provider, bootstrapCluster string) (*types.Provider, error)
UpdateProvider(ctx context.Context, userID string, provider *types.Provider) (*types.Provider, error)
DeleteProvider(ctx context.Context, accountID, userID, providerID string) error
@@ -132,27 +128,8 @@ type managerImpl struct {
// state; concurrent provider creates would otherwise race.
labelRngMu sync.Mutex
labelRng *rand.Rand
// discoveryAttempts provides lightweight per-provider admission control for
// the explicit endpoint probe. It prevents duplicate clicks from occupying
// multiple proxy control-stream requests at once and adds a short cooldown
// after each attempt.
discoveryMu sync.Mutex
discoveryAttempts map[string]modelDiscoveryAttempt
}
type modelDiscoveryAttempt struct {
inFlight bool
lastStarted time.Time
}
const (
modelDiscoveryTimeout = 10 * time.Second
modelDiscoveryCooldown = 2 * time.Second
maxDiscoveredModels = 500
maxDiscoveredModelLen = 512
)
// NewManager constructs the persistent Agent Network manager. The
// manager persists provider/policy/guardrail configuration and, on
// every mutation, reconciles the in-memory synthesised reverse-proxy
@@ -171,7 +148,6 @@ func NewManager(
proxyController: proxyController,
reconcileCache: make(map[string]map[string]*proto.ProxyMapping),
labelRng: rand.New(rand.NewSource(time.Now().UnixNano())),
discoveryAttempts: make(map[string]modelDiscoveryAttempt),
}
}
@@ -189,202 +165,6 @@ func (m *managerImpl) GetProvider(ctx context.Context, accountID, userID, provid
return m.store.GetAgentNetworkProviderByID(ctx, store.LockingStrengthNone, accountID, providerID)
}
// DiscoverProviderModels asks a capable proxy in the account's selected
// cluster to query the persisted provider endpoint. The browser supplies only
// the provider id: URL, TLS policy, and credential are loaded here so a caller
// cannot turn this operation into an arbitrary network probe.
func (m *managerImpl) DiscoverProviderModels(ctx context.Context, accountID, userID, providerID string) (*types.ModelDiscoveryResult, error) {
if err := m.requirePermission(ctx, accountID, userID, operations.Update); err != nil {
return nil, err
}
providerID = strings.TrimSpace(providerID)
if providerID == "" {
return nil, status.Errorf(status.InvalidArgument, "provider ID is required")
}
provider, err := m.store.GetAgentNetworkProviderByID(ctx, store.LockingStrengthNone, accountID, providerID)
if err != nil {
return nil, err
}
entry, ok := catalog.Lookup(provider.ProviderID)
if !ok {
return nil, status.Errorf(status.PreconditionFailed, "provider references an unknown catalog provider")
}
if entry.ModelDiscovery == nil {
return nil, status.Errorf(status.PreconditionFailed, "provider type %q does not support model discovery", provider.ProviderID)
}
if m.proxyController == nil {
return nil, status.Errorf(status.PreconditionFailed, "model discovery is unavailable")
}
settings, err := m.store.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, accountID)
if err != nil {
var statusErr *status.Error
if errors.As(err, &statusErr) && statusErr.Type() == status.NotFound {
return nil, status.Errorf(status.PreconditionFailed, "configure an Agent Network proxy cluster before discovering models")
}
return nil, err
}
cluster := strings.TrimSpace(settings.Cluster)
if cluster == "" {
return nil, status.Errorf(status.PreconditionFailed, "configure an Agent Network proxy cluster before discovering models")
}
authHeaderName, authHeaderValue, gcpKey, err := providerAuthHeader(provider)
if err != nil {
return nil, status.Errorf(status.PreconditionFailed, "provider authentication is not configured correctly")
}
if gcpKey != "" {
return nil, status.Errorf(status.PreconditionFailed, "provider authentication is not supported for model discovery")
}
attemptKey := accountID + "\x00" + provider.ID
if !m.beginModelDiscovery(attemptKey, time.Now()) {
return nil, status.Errorf(status.TooManyRequests, "model discovery is already running or was requested too recently")
}
defer m.finishModelDiscovery(attemptKey)
probeCtx, cancel := context.WithTimeout(ctx, modelDiscoveryTimeout)
defer cancel()
probeResult, err := m.proxyController.DiscoverModels(probeCtx, accountID, cluster, &proto.ModelDiscoveryRequest{
UpstreamUrl: strings.TrimSpace(provider.UpstreamURL),
AuthHeaderName: authHeaderName,
AuthHeaderValue: authHeaderValue,
SkipTlsVerify: provider.SkipTLSVerification,
OllamaFallback: entry.ModelDiscovery.OllamaFallback,
})
if err != nil {
if errors.Is(err, context.DeadlineExceeded) || errors.Is(probeCtx.Err(), context.DeadlineExceeded) {
return nil, status.Errorf(status.PreconditionFailed, "model discovery timed out")
}
if _, ok := status.FromError(err); ok {
return nil, err
}
return nil, status.Errorf(status.PreconditionFailed, "model discovery could not be completed")
}
if probeResult == nil {
return nil, status.Errorf(status.PreconditionFailed, "proxy returned an empty model discovery response")
}
if strings.TrimSpace(probeResult.Error) != "" {
message := safeModelDiscoveryText(probeResult.Error, 256)
if message == "" {
message = "proxy reported a discovery failure"
}
requestID := safeModelDiscoveryText(probeResult.RequestId, 128)
if requestID != "" {
return nil, status.Errorf(status.PreconditionFailed, "model discovery failed: %s (request_id: %s)", message, requestID)
}
return nil, status.Errorf(status.PreconditionFailed, "model discovery failed: %s", message)
}
requestID := safeModelDiscoveryText(probeResult.RequestId, 128)
if requestID == "" {
return nil, status.Errorf(status.PreconditionFailed, "proxy returned an uncorrelated model discovery response")
}
source := strings.TrimSpace(probeResult.Source)
switch source {
case "openai_v1_models", "ollama_api_tags":
default:
return nil, status.Errorf(status.PreconditionFailed, "proxy returned an unsupported model discovery response")
}
return &types.ModelDiscoveryResult{
RequestID: requestID,
Source: source,
ProxyCluster: cluster,
Models: normalizeDiscoveredModels(probeResult.Models),
}, nil
}
func (m *managerImpl) beginModelDiscovery(key string, now time.Time) bool {
m.discoveryMu.Lock()
defer m.discoveryMu.Unlock()
if m.discoveryAttempts == nil {
m.discoveryAttempts = make(map[string]modelDiscoveryAttempt)
}
attempt := m.discoveryAttempts[key]
if attempt.inFlight || (!attempt.lastStarted.IsZero() && now.Sub(attempt.lastStarted) < modelDiscoveryCooldown) {
return false
}
attempt.inFlight = true
attempt.lastStarted = now
m.discoveryAttempts[key] = attempt
return true
}
func (m *managerImpl) finishModelDiscovery(key string) {
m.discoveryMu.Lock()
attempt, ok := m.discoveryAttempts[key]
if !ok {
m.discoveryMu.Unlock()
return
}
attempt.inFlight = false
m.discoveryAttempts[key] = attempt
m.discoveryMu.Unlock()
remaining := time.Until(attempt.lastStarted.Add(modelDiscoveryCooldown))
if remaining <= 0 {
m.expireModelDiscoveryAttempt(key, attempt.lastStarted)
return
}
time.AfterFunc(remaining, func() {
m.expireModelDiscoveryAttempt(key, attempt.lastStarted)
})
}
func (m *managerImpl) expireModelDiscoveryAttempt(key string, lastStarted time.Time) {
m.discoveryMu.Lock()
defer m.discoveryMu.Unlock()
attempt, ok := m.discoveryAttempts[key]
if ok && !attempt.inFlight && attempt.lastStarted.Equal(lastStarted) {
delete(m.discoveryAttempts, key)
}
}
func normalizeDiscoveredModels(models []*proto.ModelDiscoveryModel) []types.DiscoveredModel {
out := make([]types.DiscoveredModel, 0, min(len(models), maxDiscoveredModels))
seen := make(map[string]struct{}, min(len(models), maxDiscoveredModels))
for _, model := range models {
if model == nil {
continue
}
id := safeModelDiscoveryText(model.Id, maxDiscoveredModelLen)
if id == "" || len(id) > maxDiscoveredModelLen {
continue
}
if _, ok := seen[id]; ok {
continue
}
label := safeModelDiscoveryText(model.Label, maxDiscoveredModelLen)
if label == "" || len(label) > maxDiscoveredModelLen {
label = id
}
seen[id] = struct{}{}
out = append(out, types.DiscoveredModel{ID: id, Label: label})
if len(out) == maxDiscoveredModels {
break
}
}
slices.SortFunc(out, func(a, b types.DiscoveredModel) int {
return strings.Compare(a.ID, b.ID)
})
return out
}
func safeModelDiscoveryText(value string, maxLen int) string {
value = strings.TrimSpace(value)
if value == "" || len(value) > maxLen || !utf8.ValidString(value) {
return ""
}
for _, r := range value {
if unicode.IsControl(r) {
return ""
}
}
return value
}
// CreateProvider persists a new provider for the account. bootstrapCluster
// is used only when the per-account agent-network Settings row hasn't
// been created yet; otherwise it is ignored (the cluster is pinned on
@@ -394,8 +174,11 @@ func (m *managerImpl) CreateProvider(ctx context.Context, userID string, provide
return nil, err
}
if err := prepareProviderAPIKey(provider, nil); err != nil {
return nil, err
// An empty api_key would silently produce a synthesised service
// that 401s on every upstream request. Surface the misconfiguration
// at create time instead.
if strings.TrimSpace(provider.APIKey) == "" {
return nil, status.Errorf(status.InvalidArgument, "api_key is required when creating an agent network provider")
}
if provider.ID == "" {
@@ -439,8 +222,13 @@ func (m *managerImpl) UpdateProvider(ctx context.Context, userID string, provide
return nil, fmt.Errorf("failed to get agent network provider: %w", err)
}
if err := prepareProviderAPIKey(provider, existing); err != nil {
return nil, err
// Preserve the API key if the caller didn't rotate it. A
// whitespace-only value is treated as "not rotated" rather than a
// real key, but it must not silently overwrite a valid stored key.
if provider.APIKey == "" {
provider.APIKey = existing.APIKey
} else if strings.TrimSpace(provider.APIKey) == "" {
return nil, status.Errorf(status.InvalidArgument, "api_key must be non-blank when rotating an agent network provider")
}
// Always preserve the session keypair across updates so existing
// session cookies stay valid. The keys are server-managed and
@@ -463,52 +251,6 @@ func (m *managerImpl) UpdateProvider(ctx context.Context, userID string, provide
return provider, nil
}
// prepareProviderAPIKey applies catalog-owned authentication semantics before
// a provider is persisted. existing is nil on create. On update, omission
// preserves a key only when the provider type is unchanged; switching types
// never carries an old provider's secret into the new upstream.
func prepareProviderAPIKey(provider, existing *types.Provider) error {
entry, ok := catalog.Lookup(provider.ProviderID)
if !ok {
return status.Errorf(status.InvalidArgument, "provider_id %q is not a known catalog provider", provider.ProviderID)
}
return prepareProviderAPIKeyForEntry(provider, existing, entry)
}
func prepareProviderAPIKeyForEntry(provider, existing *types.Provider, entry catalog.Provider) error {
keyProvided := provider.APIKeyProvided || provider.APIKey != ""
providerChanged := existing != nil && provider.ProviderID != existing.ProviderID
if existing != nil && !keyProvided && !providerChanged && entry.EffectiveAuthMode() != catalog.AuthModeNone {
provider.APIKey = existing.APIKey
}
// Normalize after a preserved value is restored so a legacy
// whitespace-only credential cannot pass manager validation and then fail
// later during synthesis.
if strings.TrimSpace(provider.APIKey) == "" {
provider.APIKey = ""
}
switch entry.EffectiveAuthMode() {
case catalog.AuthModeRequired:
if provider.APIKey == "" {
return status.Errorf(status.InvalidArgument, "api_key is required for provider_id %q", provider.ProviderID)
}
case catalog.AuthModeOptional:
// Empty is a valid credential state.
case catalog.AuthModeNone:
if keyProvided && provider.APIKey != "" {
return status.Errorf(status.InvalidArgument, "api_key is not supported for provider_id %q", provider.ProviderID)
}
// Clear any stale credential already stored for a provider whose
// catalog semantics changed to no authentication.
provider.APIKey = ""
default:
return status.Errorf(status.InvalidArgument, "provider_id %q has an invalid authentication mode", provider.ProviderID)
}
return nil
}
func (m *managerImpl) DeleteProvider(ctx context.Context, accountID, userID, providerID string) error {
if err := m.requirePermission(ctx, accountID, userID, operations.Delete); err != nil {
return err
@@ -1066,10 +808,6 @@ func (*mockManager) GetProvider(_ context.Context, _, _, _ string) (*types.Provi
return &types.Provider{}, nil
}
func (*mockManager) DiscoverProviderModels(_ context.Context, _, _, _ string) (*types.ModelDiscoveryResult, error) {
return nil, status.Errorf(status.PreconditionFailed, "model discovery is unavailable")
}
func (*mockManager) CreateProvider(_ context.Context, _ string, p *types.Provider, _ string) (*types.Provider, error) {
return p, nil
}

View File

@@ -1,248 +0,0 @@
package agentnetwork
import (
"context"
"strings"
"testing"
"time"
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
"github.com/netbirdio/netbird/management/server/permissions"
"github.com/netbirdio/netbird/management/server/permissions/modules"
"github.com/netbirdio/netbird/management/server/permissions/operations"
"github.com/netbirdio/netbird/management/server/store"
"github.com/netbirdio/netbird/shared/management/proto"
"github.com/netbirdio/netbird/shared/management/status"
)
func newModelDiscoveryManager(t *testing.T) (*managerImpl, *store.MockStore, *permissions.MockManager, *proxy.MockController) {
t.Helper()
ctrl := gomock.NewController(t)
mockStore := store.NewMockStore(ctrl)
mockPermissions := permissions.NewMockManager(ctrl)
mockProxy := proxy.NewMockController(ctrl)
return &managerImpl{
store: mockStore,
permissionsManager: mockPermissions,
proxyController: mockProxy,
discoveryAttempts: make(map[string]modelDiscoveryAttempt),
}, mockStore, mockPermissions, mockProxy
}
func allowModelDiscovery(mockPermissions *permissions.MockManager) {
mockPermissions.EXPECT().
ValidateUserPermissions(gomock.Any(), "account-1", "user-1", modules.AgentNetwork, operations.Update).
Return(true, context.Background(), nil)
}
func TestDiscoverProviderModelsUsesPersistedProviderAndCluster(t *testing.T) {
manager, mockStore, mockPermissions, mockProxy := newModelDiscoveryManager(t)
allowModelDiscovery(mockPermissions)
provider := &types.Provider{
ID: "provider-1",
AccountID: "account-1",
ProviderID: "ollama",
UpstreamURL: "http://ollama.internal:11434/base",
APIKey: "stored-key",
SkipTLSVerification: true,
}
mockStore.EXPECT().
GetAgentNetworkProviderByID(gomock.Any(), store.LockingStrengthNone, "account-1", "provider-1").
Return(provider, nil)
mockStore.EXPECT().
GetAgentNetworkSettings(gomock.Any(), store.LockingStrengthNone, "account-1").
Return(&types.Settings{AccountID: "account-1", Cluster: "private.example.com"}, nil)
mockProxy.EXPECT().
DiscoverModels(gomock.Any(), "account-1", "private.example.com", gomock.Any()).
DoAndReturn(func(_ context.Context, _, _ string, request *proto.ModelDiscoveryRequest) (*proto.ModelDiscoveryResult, error) {
assert.Empty(t, request.RequestId, "the control-plane server owns request IDs")
assert.Equal(t, provider.UpstreamURL, request.UpstreamUrl)
assert.Equal(t, "Authorization", request.AuthHeaderName)
assert.Equal(t, "Bearer stored-key", request.AuthHeaderValue)
assert.True(t, request.SkipTlsVerify)
assert.True(t, request.OllamaFallback)
return &proto.ModelDiscoveryResult{
RequestId: "probe-123",
Source: "openai_v1_models",
Models: []*proto.ModelDiscoveryModel{
{Id: "zeta", Label: ""},
{Id: "alpha", Label: "Alpha"},
{Id: "zeta", Label: "duplicate"},
{Id: "bad\x00model", Label: "bad"},
{Id: strings.Repeat("x", maxDiscoveredModelLen+1), Label: "too long"},
},
}, nil
})
result, err := manager.DiscoverProviderModels(context.Background(), "account-1", "user-1", "provider-1")
require.NoError(t, err)
assert.Equal(t, "probe-123", result.RequestID)
assert.Equal(t, "openai_v1_models", result.Source)
assert.Equal(t, "private.example.com", result.ProxyCluster)
assert.Equal(t, []types.DiscoveredModel{
{ID: "alpha", Label: "Alpha"},
{ID: "zeta", Label: "zeta"},
}, result.Models)
}
func TestDiscoverProviderModelsSupportsOllamaCloud(t *testing.T) {
manager, mockStore, mockPermissions, mockProxy := newModelDiscoveryManager(t)
allowModelDiscovery(mockPermissions)
provider := &types.Provider{
ID: "provider-cloud",
AccountID: "account-1",
ProviderID: "ollama_cloud",
UpstreamURL: "https://ollama.com",
APIKey: "ollama-cloud-key",
}
mockStore.EXPECT().
GetAgentNetworkProviderByID(gomock.Any(), store.LockingStrengthNone, "account-1", provider.ID).
Return(provider, nil)
mockStore.EXPECT().
GetAgentNetworkSettings(gomock.Any(), store.LockingStrengthNone, "account-1").
Return(&types.Settings{AccountID: "account-1", Cluster: "cloud-proxies.example.com"}, nil)
mockProxy.EXPECT().
DiscoverModels(gomock.Any(), "account-1", "cloud-proxies.example.com", gomock.Any()).
DoAndReturn(func(_ context.Context, _, _ string, request *proto.ModelDiscoveryRequest) (*proto.ModelDiscoveryResult, error) {
assert.Equal(t, "https://ollama.com", request.UpstreamUrl)
assert.Equal(t, "Authorization", request.AuthHeaderName)
assert.Equal(t, "Bearer ollama-cloud-key", request.AuthHeaderValue)
assert.False(t, request.SkipTlsVerify)
assert.True(t, request.OllamaFallback)
return &proto.ModelDiscoveryResult{
RequestId: "probe-cloud",
Source: "openai_v1_models",
Models: []*proto.ModelDiscoveryModel{
{Id: "gpt-oss:120b", Label: "gpt-oss:120b"},
},
}, nil
})
result, err := manager.DiscoverProviderModels(context.Background(), "account-1", "user-1", provider.ID)
require.NoError(t, err)
assert.Equal(t, "probe-cloud", result.RequestID)
assert.Equal(t, "cloud-proxies.example.com", result.ProxyCluster)
assert.Equal(t, []types.DiscoveredModel{
{ID: "gpt-oss:120b", Label: "gpt-oss:120b"},
}, result.Models)
}
func TestDiscoverProviderModelsRejectsOllamaCloudWithoutKey(t *testing.T) {
manager, mockStore, mockPermissions, _ := newModelDiscoveryManager(t)
allowModelDiscovery(mockPermissions)
provider := &types.Provider{
ID: "provider-cloud",
AccountID: "account-1",
ProviderID: "ollama_cloud",
UpstreamURL: "https://ollama.com",
}
mockStore.EXPECT().
GetAgentNetworkProviderByID(gomock.Any(), store.LockingStrengthNone, "account-1", provider.ID).
Return(provider, nil)
mockStore.EXPECT().
GetAgentNetworkSettings(gomock.Any(), store.LockingStrengthNone, "account-1").
Return(&types.Settings{AccountID: "account-1", Cluster: "cloud-proxies.example.com"}, nil)
_, err := manager.DiscoverProviderModels(context.Background(), "account-1", "user-1", provider.ID)
require.Error(t, err)
statusErr, ok := status.FromError(err)
require.True(t, ok)
assert.Equal(t, status.PreconditionFailed, statusErr.Type())
assert.Contains(t, err.Error(), "authentication is not configured correctly")
}
func TestDiscoverProviderModelsRejectsUnsupportedProvider(t *testing.T) {
manager, mockStore, mockPermissions, _ := newModelDiscoveryManager(t)
allowModelDiscovery(mockPermissions)
mockStore.EXPECT().
GetAgentNetworkProviderByID(gomock.Any(), store.LockingStrengthNone, "account-1", "provider-1").
Return(&types.Provider{
ID: "provider-1",
AccountID: "account-1",
ProviderID: "openai_api",
}, nil)
_, err := manager.DiscoverProviderModels(context.Background(), "account-1", "user-1", "provider-1")
require.Error(t, err)
statusErr, ok := status.FromError(err)
require.True(t, ok)
assert.Equal(t, status.PreconditionFailed, statusErr.Type())
assert.Contains(t, err.Error(), "does not support")
}
func TestDiscoverProviderModelsPreservesCorrelatedProxyError(t *testing.T) {
manager, mockStore, mockPermissions, mockProxy := newModelDiscoveryManager(t)
allowModelDiscovery(mockPermissions)
mockStore.EXPECT().
GetAgentNetworkProviderByID(gomock.Any(), store.LockingStrengthNone, "account-1", "provider-1").
Return(&types.Provider{
ID: "provider-1",
AccountID: "account-1",
ProviderID: "ollama",
UpstreamURL: "http://ollama.internal:11434",
}, nil)
mockStore.EXPECT().
GetAgentNetworkSettings(gomock.Any(), store.LockingStrengthNone, "account-1").
Return(&types.Settings{AccountID: "account-1", Cluster: "private.example.com"}, nil)
mockProxy.EXPECT().
DiscoverModels(gomock.Any(), "account-1", "private.example.com", gomock.Any()).
Return(&proto.ModelDiscoveryResult{
RequestId: "probe-failed",
Error: "upstream returned HTTP 401",
}, nil)
_, err := manager.DiscoverProviderModels(context.Background(), "account-1", "user-1", "provider-1")
require.Error(t, err)
assert.Contains(t, err.Error(), "upstream returned HTTP 401")
assert.Contains(t, err.Error(), "probe-failed")
}
func TestModelDiscoveryAdmissionControl(t *testing.T) {
manager := &managerImpl{}
start := time.Now()
require.True(t, manager.beginModelDiscovery("account/provider", start))
assert.False(t, manager.beginModelDiscovery("account/provider", start.Add(time.Second)))
manager.finishModelDiscovery("account/provider")
assert.False(t, manager.beginModelDiscovery("account/provider", start.Add(time.Second)))
assert.True(t, manager.beginModelDiscovery("account/provider", start.Add(modelDiscoveryCooldown)))
}
func TestModelDiscoveryAdmissionControlExpiresFinishedAttempts(t *testing.T) {
manager := &managerImpl{}
start := time.Now()
require.True(t, manager.beginModelDiscovery("account/provider", start))
manager.finishModelDiscovery("account/provider")
manager.expireModelDiscoveryAttempt("account/provider", start)
manager.discoveryMu.Lock()
_, retained := manager.discoveryAttempts["account/provider"]
manager.discoveryMu.Unlock()
assert.False(t, retained)
}
func TestModelDiscoveryAdmissionControlKeepsReplacementAttempt(t *testing.T) {
manager := &managerImpl{}
start := time.Now()
require.True(t, manager.beginModelDiscovery("account/provider", start))
manager.finishModelDiscovery("account/provider")
require.True(t, manager.beginModelDiscovery("account/provider", start.Add(modelDiscoveryCooldown)))
manager.expireModelDiscoveryAttempt("account/provider", start)
manager.discoveryMu.Lock()
attempt, retained := manager.discoveryAttempts["account/provider"]
manager.discoveryMu.Unlock()
require.True(t, retained)
assert.True(t, attempt.inFlight)
assert.Equal(t, start.Add(modelDiscoveryCooldown), attempt.lastStarted)
}

View File

@@ -1,124 +0,0 @@
package agentnetwork
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/catalog"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
)
func TestPrepareProviderAPIKey(t *testing.T) {
t.Run("required create rejects an empty key", func(t *testing.T) {
provider := &types.Provider{ProviderID: "openai_api"}
err := prepareProviderAPIKey(provider, nil)
require.Error(t, err)
assert.Contains(t, err.Error(), "api_key is required")
})
t.Run("Ollama Cloud requires a key", func(t *testing.T) {
provider := &types.Provider{ProviderID: "ollama_cloud"}
err := prepareProviderAPIKey(provider, nil)
require.Error(t, err)
assert.Contains(t, err.Error(), "api_key is required")
})
t.Run("Ollama Cloud accepts a key", func(t *testing.T) {
provider := &types.Provider{
ProviderID: "ollama_cloud",
APIKey: "ollama-cloud-key",
APIKeyProvided: true,
}
require.NoError(t, prepareProviderAPIKey(provider, nil))
assert.Equal(t, "ollama-cloud-key", provider.APIKey)
})
t.Run("optional create accepts an empty key", func(t *testing.T) {
provider := &types.Provider{ProviderID: "ollama"}
require.NoError(t, prepareProviderAPIKey(provider, nil))
assert.Empty(t, provider.APIKey)
})
t.Run("optional update omission preserves the existing key", func(t *testing.T) {
existing := &types.Provider{ProviderID: "ollama", APIKey: "existing"}
provider := &types.Provider{ProviderID: "ollama"}
require.NoError(t, prepareProviderAPIKey(provider, existing))
assert.Equal(t, "existing", provider.APIKey)
})
t.Run("optional update explicit empty clears the existing key", func(t *testing.T) {
existing := &types.Provider{ProviderID: "ollama", APIKey: "existing"}
provider := &types.Provider{ProviderID: "ollama", APIKeyProvided: true}
require.NoError(t, prepareProviderAPIKey(provider, existing))
assert.Empty(t, provider.APIKey)
})
t.Run("changing to optional without a key does not carry the old secret", func(t *testing.T) {
existing := &types.Provider{ProviderID: "openai_api", APIKey: "sk-old-provider"}
provider := &types.Provider{ProviderID: "ollama"}
require.NoError(t, prepareProviderAPIKey(provider, existing))
assert.Empty(t, provider.APIKey)
})
t.Run("changing to required without a key fails", func(t *testing.T) {
existing := &types.Provider{ProviderID: "ollama", APIKey: "old-optional-key"}
provider := &types.Provider{ProviderID: "openai_api"}
err := prepareProviderAPIKey(provider, existing)
require.Error(t, err)
assert.Contains(t, err.Error(), "api_key is required")
assert.Empty(t, provider.APIKey)
})
t.Run("required update omission preserves the existing key", func(t *testing.T) {
existing := &types.Provider{ProviderID: "openai_api", APIKey: "sk-existing"}
provider := &types.Provider{ProviderID: "openai_api"}
require.NoError(t, prepareProviderAPIKey(provider, existing))
assert.Equal(t, "sk-existing", provider.APIKey)
})
t.Run("required update rejects a preserved whitespace-only key", func(t *testing.T) {
existing := &types.Provider{ProviderID: "openai_api", APIKey: " "}
provider := &types.Provider{ProviderID: "openai_api"}
err := prepareProviderAPIKey(provider, existing)
require.Error(t, err)
assert.Contains(t, err.Error(), "api_key is required")
assert.Empty(t, provider.APIKey)
})
t.Run("required update explicit empty is rejected", func(t *testing.T) {
existing := &types.Provider{ProviderID: "openai_api", APIKey: "sk-existing"}
provider := &types.Provider{ProviderID: "openai_api", APIKeyProvided: true}
err := prepareProviderAPIKey(provider, existing)
require.Error(t, err)
assert.Contains(t, err.Error(), "api_key is required")
})
t.Run("unknown catalog provider is rejected", func(t *testing.T) {
provider := &types.Provider{ProviderID: "unknown", APIKey: "secret"}
err := prepareProviderAPIKey(provider, nil)
require.Error(t, err)
assert.Contains(t, err.Error(), "not a known catalog provider")
})
t.Run("none mode clears a stale key", func(t *testing.T) {
entry := catalog.Provider{ID: "authless", AuthMode: catalog.AuthModeNone}
existing := &types.Provider{ProviderID: "authless", APIKey: "stale-secret"}
provider := &types.Provider{ProviderID: "authless"}
require.NoError(t, prepareProviderAPIKeyForEntry(provider, existing, entry))
assert.Empty(t, provider.APIKey)
})
t.Run("none mode rejects a supplied key", func(t *testing.T) {
entry := catalog.Provider{ID: "authless", AuthMode: catalog.AuthModeNone}
provider := &types.Provider{
ProviderID: "authless",
APIKey: "unexpected-secret",
APIKeyProvided: true,
}
err := prepareProviderAPIKeyForEntry(provider, nil, entry)
require.Error(t, err)
assert.Contains(t, err.Error(), "not supported")
})
}

View File

@@ -906,33 +906,23 @@ const (
noopUpstreamPort = uint16(443)
)
// providerAuthHeader builds the upstream auth header pair for a provider from
// its catalog entry. Optional providers with no key and providers using no
// authentication return an empty pair. Otherwise, the synthesiser substitutes
// the decrypted API key into the catalog template and returns the pair the
// router injects after stripping inbound vendor credentials.
// providerAuthHeader builds the upstream auth header pair for a
// provider from its catalog entry. The catalog declares which header
// name and template a provider's API expects; the synthesiser
// substitutes the provider's decrypted API key into the template and
// returns the (name, value) pair the router middleware injects after
// stripping the inbound vendor auth headers.
func providerAuthHeader(p *types.Provider) (name, value, gcpSAKeyB64 string, err error) {
entry, ok := catalog.Lookup(p.ProviderID)
if !ok {
return "", "", "", fmt.Errorf("provider %s references unknown catalog id %q", p.ID, p.ProviderID)
}
switch entry.EffectiveAuthMode() {
case catalog.AuthModeNone:
return "", "", "", nil
case catalog.AuthModeOptional:
if strings.TrimSpace(p.APIKey) == "" {
return "", "", "", nil
}
case catalog.AuthModeRequired:
if strings.TrimSpace(p.APIKey) == "" {
return "", "", "", fmt.Errorf("provider %s has no api key", p.ID)
}
default:
return "", "", "", fmt.Errorf("catalog entry %q has invalid authentication mode %q", p.ProviderID, entry.AuthMode)
}
if entry.AuthHeaderName == "" || entry.AuthHeaderTemplate == "" {
return "", "", "", fmt.Errorf("catalog entry %q has no auth header configured", p.ProviderID)
}
if p.APIKey == "" {
return "", "", "", fmt.Errorf("provider %s has no api key", p.ID)
}
// A "keyfile::<base64 json>" api_key is a GCP service-account key, not a
// static bearer. The proxy mints + refreshes a short-lived OAuth token from
// it at request time, so carry the key material on the route and emit no

View File

@@ -1219,102 +1219,3 @@ func TestSynthesizeServices_EmptyAPIKey_FailsClosed(t *testing.T) {
require.Error(t, err, "synthesis must refuse a provider with no api key")
assert.Contains(t, err.Error(), "no api key", "error must surface the missing credential")
}
func TestSynthesizeServices_OllamaOptionalAPIKey(t *testing.T) {
tests := []struct {
name string
apiKey string
wantHeaderName string
wantHeaderValue string
}{
{
name: "no key emits no replacement auth header",
},
{
name: "configured key emits bearer auth",
apiKey: "protected-endpoint-token",
wantHeaderName: "Authorization",
wantHeaderValue: "Bearer protected-endpoint-token",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ctx := context.Background()
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockStore := store.NewMockStore(ctrl)
provider := newSynthTestProvider()
provider.ProviderID = "ollama"
provider.Name = "Ollama"
provider.UpstreamURL = "http://ollama.internal:11434"
provider.APIKey = tt.apiKey
provider.Models = []types.ProviderModel{}
policy := newSynthTestPolicy(provider.ID, "grp-eng", "")
expectSynthBaseInputs(mockStore, ctx, newSynthTestSettings(),
[]*types.Provider{provider},
[]*types.Policy{policy},
[]*types.Guardrail{})
services, err := SynthesizeServices(ctx, mockStore, testAccountID)
require.NoError(t, err)
require.Len(t, services, 1)
var routerCfg routerConfig
for _, middleware := range services[0].Targets[0].Options.Middlewares {
if middleware.ID == middlewareIDLLMRouter {
require.NoError(t, json.Unmarshal(middleware.ConfigJSON, &routerCfg))
break
}
}
require.Len(t, routerCfg.Providers, 1)
assert.Equal(t, tt.wantHeaderName, routerCfg.Providers[0].AuthHeaderName)
assert.Equal(t, tt.wantHeaderValue, routerCfg.Providers[0].AuthHeaderValue)
})
}
}
func TestSynthesizeServices_OllamaCloudRoute(t *testing.T) {
ctx := context.Background()
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockStore := store.NewMockStore(ctrl)
provider := newSynthTestProvider()
provider.ProviderID = "ollama_cloud"
provider.Name = "Ollama Cloud"
provider.UpstreamURL = "https://ollama.com"
provider.APIKey = "ollama-cloud-key"
provider.Models = []types.ProviderModel{{ID: "gpt-oss:120b"}}
policy := newSynthTestPolicy(provider.ID, "grp-eng", "")
expectSynthBaseInputs(mockStore, ctx, newSynthTestSettings(),
[]*types.Provider{provider},
[]*types.Policy{policy},
[]*types.Guardrail{})
services, err := SynthesizeServices(ctx, mockStore, testAccountID)
require.NoError(t, err)
require.Len(t, services, 1)
var routerCfg routerConfig
for _, middleware := range services[0].Targets[0].Options.Middlewares {
if middleware.ID == middlewareIDLLMRouter {
require.NoError(t, json.Unmarshal(middleware.ConfigJSON, &routerCfg))
break
}
}
require.Len(t, routerCfg.Providers, 1)
route := routerCfg.Providers[0]
assert.Empty(t, route.Vendor, "Ollama Cloud preserves the untagged Ollama/vLLM/custom routing behavior")
assert.Equal(t, []string{"gpt-oss:120b"}, route.Models)
assert.Equal(t, "https", route.UpstreamScheme)
assert.Equal(t, "ollama.com", route.UpstreamHost)
assert.Empty(t, route.UpstreamPath)
assert.Equal(t, "Authorization", route.AuthHeaderName)
assert.Equal(t, "Bearer ollama-cloud-key", route.AuthHeaderValue)
assert.False(t, route.SkipTLSVerify)
}

View File

@@ -1,19 +0,0 @@
package types
// DiscoveredModel is a normalized model identifier returned by a provider's
// persisted upstream endpoint. Discovery does not persist models; the caller
// must explicitly update the provider to opt into the returned list.
type DiscoveredModel struct {
ID string
Label string
}
// ModelDiscoveryResult describes a successful proxy-executed discovery probe.
// RequestID correlates the management request with the proxy control message,
// and ProxyCluster identifies the network vantage point that ran it.
type ModelDiscoveryResult struct {
RequestID string
Source string
ProxyCluster string
Models []DiscoveredModel
}

View File

@@ -33,10 +33,6 @@ type Provider struct {
// the operator selected.
UpstreamURL string `gorm:"column:upstream_url"`
APIKey string `gorm:"column:api_key"`
// APIKeyProvided records whether api_key was present in the request. It is
// transient and lets updates distinguish omission (preserve) from an
// explicit empty value (clear an optional credential).
APIKeyProvided bool `gorm:"-" json:"-"`
// ExtraValues holds operator-typed values for catalog-declared
// ExtraHeaders (see catalog.Provider.ExtraHeaders). Keyed by
// header name (e.g. "x-portkey-config"); a non-empty value is
@@ -101,16 +97,15 @@ func NewProvider(accountID string) *Provider {
}
}
// FromAPIRequest applies the request payload onto the receiver. APIKeyProvided
// preserves the distinction between an omitted api_key and an explicit empty
// value; the manager applies the catalog-specific preserve/clear semantics.
// FromAPIRequest applies the request payload onto the receiver. The api_key
// is only overwritten when the caller provided one — empty/nil leaves the
// existing key intact, so updates can omit it.
func (p *Provider) FromAPIRequest(req *api.AgentNetworkProviderRequest) {
p.ProviderID = req.ProviderId
p.Name = req.Name
p.UpstreamURL = req.UpstreamUrl
p.APIKeyProvided = req.ApiKey != nil
if req.ApiKey != nil {
p.APIKey = strings.TrimSpace(*req.ApiKey)
if req.ApiKey != nil && strings.TrimSpace(*req.ApiKey) != "" {
p.APIKey = *req.ApiKey
}
if req.ExtraValues != nil {
// Replace the whole map (rather than merge) so unsetting a
@@ -183,7 +178,6 @@ func (p *Provider) ToAPIResponse() *api.AgentNetworkProvider {
UpstreamUrl: p.UpstreamURL,
Models: models,
Enabled: p.Enabled,
HasApiKey: strings.TrimSpace(p.APIKey) != "",
SkipTlsVerification: p.SkipTLSVerification,
MetadataDisabled: p.MetadataDisabled,
CreatedAt: &created,

View File

@@ -77,45 +77,3 @@ func TestProvider_MetadataDisabled_RoundTrip(t *testing.T) {
assert.False(t, p.MetadataDisabled, "explicit false must clear metadata_disabled")
assert.False(t, p.ToAPIResponse().MetadataDisabled, "response must reflect the cleared value")
}
func TestProvider_APIKeyPresenceAndResponse(t *testing.T) {
base := func() *api.AgentNetworkProviderRequest {
return &api.AgentNetworkProviderRequest{
ProviderId: "ollama",
Name: "Ollama",
UpstreamUrl: "http://ollama.internal:11434",
}
}
p := NewProvider("acc-1")
p.FromAPIRequest(base())
assert.False(t, p.APIKeyProvided, "omission must remain distinguishable from an explicit clear")
assert.False(t, p.ToAPIResponse().HasApiKey)
key := "protected-endpoint-token"
withKey := base()
withKey.ApiKey = &key
p.FromAPIRequest(withKey)
assert.True(t, p.APIKeyProvided)
assert.Equal(t, key, p.APIKey)
assert.True(t, p.ToAPIResponse().HasApiKey)
paddedKey := " \tprotected-endpoint-token\n"
withPaddedKey := base()
withPaddedKey.ApiKey = &paddedKey
p.FromAPIRequest(withPaddedKey)
assert.True(t, p.APIKeyProvided)
assert.Equal(t, key, p.APIKey, "non-blank API keys must be normalized before storage")
assert.True(t, p.ToAPIResponse().HasApiKey, "the response must reflect the normalized stored key")
empty := ""
clearKey := base()
clearKey.ApiKey = &empty
p.FromAPIRequest(clearKey)
assert.True(t, p.APIKeyProvided, "explicit empty must be carried to the manager as a clear")
assert.Empty(t, p.APIKey)
assert.False(t, p.ToAPIResponse().HasApiKey)
p.FromAPIRequest(base())
assert.False(t, p.APIKeyProvided, "a later omitted value must reset transient request presence")
}

View File

@@ -23,6 +23,9 @@ type Domain struct {
// SupportsCrowdSec is populated at query time from proxy cluster capabilities.
// Not persisted.
SupportsCrowdSec *bool `gorm:"-"`
// SupportsAppSec is populated at query time from proxy cluster capabilities.
// Not persisted.
SupportsAppSec *bool `gorm:"-"`
// SupportsPrivate is populated at query time from proxy cluster capabilities. Not persisted.
SupportsPrivate *bool `gorm:"-"`
}

View File

@@ -49,6 +49,7 @@ func domainToApi(d *domain.Domain) api.ReverseProxyDomain {
SupportsCustomPorts: d.SupportsCustomPorts,
RequireSubdomain: d.RequireSubdomain,
SupportsCrowdsec: d.SupportsCrowdSec,
SupportsAppsec: d.SupportsAppSec,
SupportsPrivate: d.SupportsPrivate,
}
if d.TargetCluster != "" {

View File

@@ -35,6 +35,7 @@ type proxyManager interface {
ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
ClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool
ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
}
@@ -94,6 +95,7 @@ func (m Manager) GetDomains(ctx context.Context, accountID, userID string) ([]*d
d.SupportsCustomPorts = m.proxyManager.ClusterSupportsCustomPorts(ctx, cluster)
d.RequireSubdomain = m.proxyManager.ClusterRequireSubdomain(ctx, cluster)
d.SupportsCrowdSec = m.proxyManager.ClusterSupportsCrowdSec(ctx, cluster)
d.SupportsAppSec = m.proxyManager.ClusterSupportsAppSec(ctx, cluster)
d.SupportsPrivate = m.proxyManager.ClusterSupportsPrivate(ctx, cluster)
ret = append(ret, d)
}
@@ -111,6 +113,7 @@ func (m Manager) GetDomains(ctx context.Context, accountID, userID string) ([]*d
if d.TargetCluster != "" {
cd.SupportsCustomPorts = m.proxyManager.ClusterSupportsCustomPorts(ctx, d.TargetCluster)
cd.SupportsCrowdSec = m.proxyManager.ClusterSupportsCrowdSec(ctx, d.TargetCluster)
cd.SupportsAppSec = m.proxyManager.ClusterSupportsAppSec(ctx, d.TargetCluster)
cd.SupportsPrivate = m.proxyManager.ClusterSupportsPrivate(ctx, d.TargetCluster)
}
// Custom domains never require a subdomain by default since

View File

@@ -40,6 +40,10 @@ func (m *mockProxyManager) ClusterSupportsCrowdSec(_ context.Context, _ string)
return nil
}
func (m *mockProxyManager) ClusterSupportsAppSec(_ context.Context, _ string) *bool {
return nil
}
func (m *mockProxyManager) ClusterSupportsPrivate(_ context.Context, _ string) *bool {
return nil
}

View File

@@ -4,16 +4,11 @@ package proxy
import (
"context"
"errors"
"time"
"github.com/netbirdio/netbird/shared/management/proto"
)
// ErrModelDiscoveryUnavailable is returned when no connected proxy in the
// requested cluster can execute a model-discovery request.
var ErrModelDiscoveryUnavailable = errors.New("model discovery proxy unavailable")
// Manager defines the interface for proxy operations
type Manager interface {
Connect(ctx context.Context, proxyID, sessionID, clusterAddress, ipAddress string, accountID *string, capabilities *Capabilities) (*Proxy, error)
@@ -24,6 +19,7 @@ type Manager interface {
ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
ClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool
ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
CleanupStale(ctx context.Context, inactivityDuration time.Duration) error
GetAccountProxy(ctx context.Context, accountID string) (*Proxy, error)
@@ -43,7 +39,6 @@ type OIDCValidationConfig struct {
// Controller is responsible for managing proxy clusters and routing service updates.
type Controller interface {
SendServiceUpdateToCluster(ctx context.Context, accountID string, update *proto.ProxyMapping, clusterAddr string)
DiscoverModels(ctx context.Context, accountID, clusterAddr string, req *proto.ModelDiscoveryRequest) (*proto.ModelDiscoveryResult, error)
GetOIDCValidationConfig() OIDCValidationConfig
RegisterProxyToCluster(ctx context.Context, clusterAddr, proxyID string) error
UnregisterProxyFromCluster(ctx context.Context, clusterAddr, proxyID string) error

View File

@@ -39,12 +39,6 @@ func (c *GRPCController) SendServiceUpdateToCluster(ctx context.Context, account
c.metrics.IncrementServiceUpdateSendCount(clusterAddr)
}
// DiscoverModels executes a correlated model-discovery request on one capable
// proxy connected to the requested cluster.
func (c *GRPCController) DiscoverModels(ctx context.Context, accountID, clusterAddr string, req *proto.ModelDiscoveryRequest) (*proto.ModelDiscoveryResult, error) {
return c.proxyGRPCServer.DiscoverModels(ctx, accountID, clusterAddr, req)
}
// GetOIDCValidationConfig returns the OIDC validation configuration from the gRPC server.
func (c *GRPCController) GetOIDCValidationConfig() proxy.OIDCValidationConfig {
return c.proxyGRPCServer.GetOIDCValidationConfig()
@@ -56,12 +50,10 @@ func (c *GRPCController) RegisterProxyToCluster(ctx context.Context, clusterAddr
return nil
}
proxySet, _ := c.clusterProxies.LoadOrStore(clusterAddr, &sync.Map{})
_, alreadyRegistered := proxySet.(*sync.Map).LoadOrStore(proxyID, struct{}{})
proxySet.(*sync.Map).Store(proxyID, struct{}{})
log.WithContext(ctx).Debugf("Registered proxy %s to cluster %s", proxyID, clusterAddr)
if !alreadyRegistered {
c.metrics.IncrementProxyConnectionCount(clusterAddr)
}
c.metrics.IncrementProxyConnectionCount(clusterAddr)
return nil
}
@@ -72,10 +64,10 @@ func (c *GRPCController) UnregisterProxyFromCluster(ctx context.Context, cluster
return nil
}
if proxySet, ok := c.clusterProxies.Load(clusterAddr); ok {
if _, registered := proxySet.(*sync.Map).LoadAndDelete(proxyID); registered {
log.WithContext(ctx).Debugf("Unregistered proxy %s from cluster %s", proxyID, clusterAddr)
c.metrics.DecrementProxyConnectionCount(clusterAddr)
}
proxySet.(*sync.Map).Delete(proxyID)
log.WithContext(ctx).Debugf("Unregistered proxy %s from cluster %s", proxyID, clusterAddr)
c.metrics.DecrementProxyConnectionCount(clusterAddr)
}
return nil
}

View File

@@ -21,6 +21,7 @@ type store interface {
GetClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
GetClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
GetClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
GetClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool
GetClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
CleanupStaleProxies(ctx context.Context, inactivityDuration time.Duration) error
GetProxyByAccountID(ctx context.Context, accountID string) (*proxy.Proxy, error)
@@ -138,6 +139,13 @@ func (m Manager) ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string
return m.store.GetClusterSupportsCrowdSec(ctx, clusterAddr)
}
// ClusterSupportsAppSec returns whether all active proxies in the cluster have
// a CrowdSec AppSec endpoint configured (unanimous). Returns nil when no proxy
// has reported capabilities.
func (m Manager) ClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool {
return m.store.GetClusterSupportsAppSec(ctx, clusterAddr)
}
// ClusterSupportsPrivate reports whether any active proxy claims the private capability (nil = unreported).
func (m Manager) ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool {
return m.store.GetClusterSupportsPrivate(ctx, clusterAddr)

View File

@@ -99,6 +99,9 @@ func (m *mockStore) GetClusterRequireSubdomain(_ context.Context, _ string) *boo
func (m *mockStore) GetClusterSupportsCrowdSec(_ context.Context, _ string) *bool {
return nil
}
func (m *mockStore) GetClusterSupportsAppSec(_ context.Context, _ string) *bool {
return nil
}
func (m *mockStore) GetClusterSupportsPrivate(_ context.Context, _ string) *bool {
return nil
}

View File

@@ -50,20 +50,6 @@ func (mr *MockManagerMockRecorder) CleanupStale(ctx, inactivityDuration interfac
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CleanupStale", reflect.TypeOf((*MockManager)(nil).CleanupStale), ctx, inactivityDuration)
}
// ClusterSupportsCustomPorts mocks base method.
func (m *MockManager) ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ClusterSupportsCustomPorts", ctx, clusterAddr)
ret0, _ := ret[0].(*bool)
return ret0
}
// ClusterSupportsCustomPorts indicates an expected call of ClusterSupportsCustomPorts.
func (mr *MockManagerMockRecorder) ClusterSupportsCustomPorts(ctx, clusterAddr interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterSupportsCustomPorts", reflect.TypeOf((*MockManager)(nil).ClusterSupportsCustomPorts), ctx, clusterAddr)
}
// ClusterRequireSubdomain mocks base method.
func (m *MockManager) ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool {
m.ctrl.T.Helper()
@@ -78,6 +64,20 @@ func (mr *MockManagerMockRecorder) ClusterRequireSubdomain(ctx, clusterAddr inte
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterRequireSubdomain", reflect.TypeOf((*MockManager)(nil).ClusterRequireSubdomain), ctx, clusterAddr)
}
// ClusterSupportsAppSec mocks base method.
func (m *MockManager) ClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ClusterSupportsAppSec", ctx, clusterAddr)
ret0, _ := ret[0].(*bool)
return ret0
}
// ClusterSupportsAppSec indicates an expected call of ClusterSupportsAppSec.
func (mr *MockManagerMockRecorder) ClusterSupportsAppSec(ctx, clusterAddr interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterSupportsAppSec", reflect.TypeOf((*MockManager)(nil).ClusterSupportsAppSec), ctx, clusterAddr)
}
// ClusterSupportsCrowdSec mocks base method.
func (m *MockManager) ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool {
m.ctrl.T.Helper()
@@ -92,6 +92,20 @@ func (mr *MockManagerMockRecorder) ClusterSupportsCrowdSec(ctx, clusterAddr inte
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterSupportsCrowdSec", reflect.TypeOf((*MockManager)(nil).ClusterSupportsCrowdSec), ctx, clusterAddr)
}
// ClusterSupportsCustomPorts mocks base method.
func (m *MockManager) ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ClusterSupportsCustomPorts", ctx, clusterAddr)
ret0, _ := ret[0].(*bool)
return ret0
}
// ClusterSupportsCustomPorts indicates an expected call of ClusterSupportsCustomPorts.
func (mr *MockManagerMockRecorder) ClusterSupportsCustomPorts(ctx, clusterAddr interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterSupportsCustomPorts", reflect.TypeOf((*MockManager)(nil).ClusterSupportsCustomPorts), ctx, clusterAddr)
}
// ClusterSupportsPrivate mocks base method.
func (m *MockManager) ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool {
m.ctrl.T.Helper()
@@ -121,6 +135,35 @@ func (mr *MockManagerMockRecorder) Connect(ctx, proxyID, sessionID, clusterAddre
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Connect", reflect.TypeOf((*MockManager)(nil).Connect), ctx, proxyID, sessionID, clusterAddress, ipAddress, accountID, capabilities)
}
// CountAccountProxies mocks base method.
func (m *MockManager) CountAccountProxies(ctx context.Context, accountID string) (int64, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "CountAccountProxies", ctx, accountID)
ret0, _ := ret[0].(int64)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// CountAccountProxies indicates an expected call of CountAccountProxies.
func (mr *MockManagerMockRecorder) CountAccountProxies(ctx, accountID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CountAccountProxies", reflect.TypeOf((*MockManager)(nil).CountAccountProxies), ctx, accountID)
}
// DeleteAccountCluster mocks base method.
func (m *MockManager) DeleteAccountCluster(ctx context.Context, clusterAddress, accountID string) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "DeleteAccountCluster", ctx, clusterAddress, accountID)
ret0, _ := ret[0].(error)
return ret0
}
// DeleteAccountCluster indicates an expected call of DeleteAccountCluster.
func (mr *MockManagerMockRecorder) DeleteAccountCluster(ctx, clusterAddress, accountID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteAccountCluster", reflect.TypeOf((*MockManager)(nil).DeleteAccountCluster), ctx, clusterAddress, accountID)
}
// Disconnect mocks base method.
func (m *MockManager) Disconnect(ctx context.Context, proxyID, sessionID string) error {
m.ctrl.T.Helper()
@@ -135,6 +178,21 @@ func (mr *MockManagerMockRecorder) Disconnect(ctx, proxyID, sessionID interface{
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Disconnect", reflect.TypeOf((*MockManager)(nil).Disconnect), ctx, proxyID, sessionID)
}
// GetAccountProxy mocks base method.
func (m *MockManager) GetAccountProxy(ctx context.Context, accountID string) (*Proxy, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetAccountProxy", ctx, accountID)
ret0, _ := ret[0].(*Proxy)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetAccountProxy indicates an expected call of GetAccountProxy.
func (mr *MockManagerMockRecorder) GetAccountProxy(ctx, accountID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAccountProxy", reflect.TypeOf((*MockManager)(nil).GetAccountProxy), ctx, accountID)
}
// GetActiveClusterAddresses mocks base method.
func (m *MockManager) GetActiveClusterAddresses(ctx context.Context) ([]string, error) {
m.ctrl.T.Helper()
@@ -150,6 +208,7 @@ func (mr *MockManagerMockRecorder) GetActiveClusterAddresses(ctx interface{}) *g
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetActiveClusterAddresses", reflect.TypeOf((*MockManager)(nil).GetActiveClusterAddresses), ctx)
}
// GetActiveClusterAddressesForAccount mocks base method.
func (m *MockManager) GetActiveClusterAddressesForAccount(ctx context.Context, accountID string) ([]string, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetActiveClusterAddressesForAccount", ctx, accountID)
@@ -158,6 +217,7 @@ func (m *MockManager) GetActiveClusterAddressesForAccount(ctx context.Context, a
return ret0, ret1
}
// GetActiveClusterAddressesForAccount indicates an expected call of GetActiveClusterAddressesForAccount.
func (mr *MockManagerMockRecorder) GetActiveClusterAddressesForAccount(ctx, accountID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetActiveClusterAddressesForAccount", reflect.TypeOf((*MockManager)(nil).GetActiveClusterAddressesForAccount), ctx, accountID)
@@ -177,36 +237,6 @@ func (mr *MockManagerMockRecorder) Heartbeat(ctx, p interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Heartbeat", reflect.TypeOf((*MockManager)(nil).Heartbeat), ctx, p)
}
// GetAccountProxy mocks base method.
func (m *MockManager) GetAccountProxy(ctx context.Context, accountID string) (*Proxy, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetAccountProxy", ctx, accountID)
ret0, _ := ret[0].(*Proxy)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetAccountProxy indicates an expected call of GetAccountProxy.
func (mr *MockManagerMockRecorder) GetAccountProxy(ctx, accountID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAccountProxy", reflect.TypeOf((*MockManager)(nil).GetAccountProxy), ctx, accountID)
}
// CountAccountProxies mocks base method.
func (m *MockManager) CountAccountProxies(ctx context.Context, accountID string) (int64, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "CountAccountProxies", ctx, accountID)
ret0, _ := ret[0].(int64)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// CountAccountProxies indicates an expected call of CountAccountProxies.
func (mr *MockManagerMockRecorder) CountAccountProxies(ctx, accountID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CountAccountProxies", reflect.TypeOf((*MockManager)(nil).CountAccountProxies), ctx, accountID)
}
// IsClusterAddressAvailable mocks base method.
func (m *MockManager) IsClusterAddressAvailable(ctx context.Context, clusterAddress, accountID string) (bool, error) {
m.ctrl.T.Helper()
@@ -222,20 +252,6 @@ func (mr *MockManagerMockRecorder) IsClusterAddressAvailable(ctx, clusterAddress
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsClusterAddressAvailable", reflect.TypeOf((*MockManager)(nil).IsClusterAddressAvailable), ctx, clusterAddress, accountID)
}
// DeleteAccountCluster mocks base method.
func (m *MockManager) DeleteAccountCluster(ctx context.Context, clusterAddress, accountID string) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "DeleteAccountCluster", ctx, clusterAddress, accountID)
ret0, _ := ret[0].(error)
return ret0
}
// DeleteAccountCluster indicates an expected call of DeleteAccountCluster.
func (mr *MockManagerMockRecorder) DeleteAccountCluster(ctx, clusterAddress, accountID interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteAccountCluster", reflect.TypeOf((*MockManager)(nil).DeleteAccountCluster), ctx, clusterAddress, accountID)
}
// MockController is a mock of Controller interface.
type MockController struct {
ctrl *gomock.Controller
@@ -259,21 +275,6 @@ func (m *MockController) EXPECT() *MockControllerMockRecorder {
return m.recorder
}
// DiscoverModels mocks base method.
func (m *MockController) DiscoverModels(ctx context.Context, accountID, clusterAddr string, req *proto.ModelDiscoveryRequest) (*proto.ModelDiscoveryResult, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "DiscoverModels", ctx, accountID, clusterAddr, req)
ret0, _ := ret[0].(*proto.ModelDiscoveryResult)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// DiscoverModels indicates an expected call of DiscoverModels.
func (mr *MockControllerMockRecorder) DiscoverModels(ctx, accountID, clusterAddr, req interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DiscoverModels", reflect.TypeOf((*MockController)(nil).DiscoverModels), ctx, accountID, clusterAddr, req)
}
// GetOIDCValidationConfig mocks base method.
func (m *MockController) GetOIDCValidationConfig() OIDCValidationConfig {
m.ctrl.T.Helper()

View File

@@ -20,6 +20,9 @@ type Capabilities struct {
RequireSubdomain *bool
// SupportsCrowdsec indicates whether this proxy has CrowdSec configured.
SupportsCrowdsec *bool
// SupportsAppsec indicates whether this proxy has a CrowdSec AppSec (WAF)
// endpoint configured.
SupportsAppsec *bool
// Private indicates whether this proxy supports inbound access via Wireguard
// tunnel and netbird-only authentication policies
Private *bool
@@ -74,5 +77,6 @@ type Cluster struct {
SupportsCustomPorts *bool
RequireSubdomain *bool
SupportsCrowdSec *bool
SupportsAppSec *bool
Private *bool
}

View File

@@ -204,6 +204,7 @@ func (h *handler) getClusters(w http.ResponseWriter, r *http.Request) {
SupportsCustomPorts: c.SupportsCustomPorts,
RequireSubdomain: c.RequireSubdomain,
SupportsCrowdsec: c.SupportsCrowdSec,
SupportsAppsec: c.SupportsAppSec,
Private: c.Private,
})
}

View File

@@ -82,6 +82,7 @@ type CapabilityProvider interface {
ClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
ClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
ClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
ClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool
ClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
}
@@ -137,6 +138,7 @@ func (m *Manager) GetClusters(ctx context.Context, accountID, userID string) ([]
clusters[i].SupportsCustomPorts = m.capabilities.ClusterSupportsCustomPorts(ctx, clusters[i].Address)
clusters[i].RequireSubdomain = m.capabilities.ClusterRequireSubdomain(ctx, clusters[i].Address)
clusters[i].SupportsCrowdSec = m.capabilities.ClusterSupportsCrowdSec(ctx, clusters[i].Address)
clusters[i].SupportsAppSec = m.capabilities.ClusterSupportsAppSec(ctx, clusters[i].Address)
clusters[i].Private = m.capabilities.ClusterSupportsPrivate(ctx, clusters[i].Address)
}

View File

@@ -165,6 +165,18 @@ type AccessRestrictions struct {
AllowedCountries []string `json:"allowed_countries,omitempty" gorm:"serializer:json"`
BlockedCountries []string `json:"blocked_countries,omitempty" gorm:"serializer:json"`
CrowdSecMode string `json:"crowdsec_mode,omitempty" gorm:"serializer:json"`
// AppSecMode is the CrowdSec AppSec (WAF) request inspection mode: "",
// "off", "enforce", or "observe". HTTP services only.
AppSecMode string `json:"appsec_mode,omitempty" gorm:"serializer:json"`
}
// isEmpty reports whether no restriction is configured. Both conversions drop
// the object entirely in that case, so a field missing from this check is
// silently discarded on the way to the API and the proxy.
func (r AccessRestrictions) isEmpty() bool {
return len(r.AllowedCIDRs) == 0 && len(r.BlockedCIDRs) == 0 &&
len(r.AllowedCountries) == 0 && len(r.BlockedCountries) == 0 &&
r.CrowdSecMode == "" && r.AppSecMode == ""
}
// Copy returns a deep copy of the AccessRestrictions.
@@ -175,6 +187,7 @@ func (r AccessRestrictions) Copy() AccessRestrictions {
AllowedCountries: slices.Clone(r.AllowedCountries),
BlockedCountries: slices.Clone(r.BlockedCountries),
CrowdSecMode: r.CrowdSecMode,
AppSecMode: r.AppSecMode,
}
}
@@ -808,13 +821,17 @@ func restrictionsFromAPI(r *api.AccessRestrictions) (AccessRestrictions, error)
}
res.CrowdSecMode = string(*r.CrowdsecMode)
}
if r.AppsecMode != nil {
if !r.AppsecMode.Valid() {
return AccessRestrictions{}, fmt.Errorf("invalid appsec_mode %q", *r.AppsecMode)
}
res.AppSecMode = string(*r.AppsecMode)
}
return res, nil
}
func restrictionsToAPI(r AccessRestrictions) *api.AccessRestrictions {
if len(r.AllowedCIDRs) == 0 && len(r.BlockedCIDRs) == 0 &&
len(r.AllowedCountries) == 0 && len(r.BlockedCountries) == 0 &&
r.CrowdSecMode == "" {
if r.isEmpty() {
return nil
}
res := &api.AccessRestrictions{}
@@ -834,13 +851,15 @@ func restrictionsToAPI(r AccessRestrictions) *api.AccessRestrictions {
mode := api.AccessRestrictionsCrowdsecMode(r.CrowdSecMode)
res.CrowdsecMode = &mode
}
if r.AppSecMode != "" {
mode := api.AccessRestrictionsAppsecMode(r.AppSecMode)
res.AppsecMode = &mode
}
return res
}
func restrictionsToProto(r AccessRestrictions) *proto.AccessRestrictions {
if len(r.AllowedCIDRs) == 0 && len(r.BlockedCIDRs) == 0 &&
len(r.AllowedCountries) == 0 && len(r.BlockedCountries) == 0 &&
r.CrowdSecMode == "" {
if r.isEmpty() {
return nil
}
return &proto.AccessRestrictions{
@@ -849,6 +868,7 @@ func restrictionsToProto(r AccessRestrictions) *proto.AccessRestrictions {
AllowedCountries: r.AllowedCountries,
BlockedCountries: r.BlockedCountries,
CrowdsecMode: r.CrowdSecMode,
AppsecMode: r.AppSecMode,
}
}
@@ -874,6 +894,11 @@ func (s *Service) Validate() error {
if err := validateAccessRestrictions(&s.Restrictions); err != nil {
return err
}
// AppSec inspects HTTP requests, so it cannot apply to the L4 modes, which
// forward opaque byte streams.
if appSecEnabled(s.Restrictions.AppSecMode) && s.Mode != ModeHTTP {
return fmt.Errorf("appsec_mode is only supported for HTTP services, got mode %q", s.Mode)
}
if err := s.validatePrivateRequirements(); err != nil {
return err
}
@@ -1242,10 +1267,27 @@ func validateCrowdSecMode(mode string) error {
}
}
func validateAppSecMode(mode string) error {
switch mode {
case "", "off", "enforce", "observe":
return nil
default:
return fmt.Errorf("appsec_mode %q is invalid", mode)
}
}
// appSecEnabled reports whether the mode asks for request inspection.
func appSecEnabled(mode string) bool {
return mode == "enforce" || mode == "observe"
}
func validateAccessRestrictions(r *AccessRestrictions) error {
if err := validateCrowdSecMode(r.CrowdSecMode); err != nil {
return err
}
if err := validateAppSecMode(r.AppSecMode); err != nil {
return err
}
if len(r.AllowedCIDRs) > maxCIDREntries {
return fmt.Errorf("allowed_cidrs: exceeds maximum of %d entries", maxCIDREntries)

View File

@@ -26,6 +26,17 @@ func validProxy() *Service {
}
}
// validL4Proxy returns a service that passes validation in one of the L4 modes.
func validL4Proxy(mode string) *Service {
rp := validProxy()
rp.Mode = mode
rp.ListenPort = 9000
rp.Targets = []*Target{
{TargetId: "peer-1", TargetType: TargetTypePeer, Host: "10.0.0.1", Port: 5432, Protocol: mode, Enabled: true},
}
return rp
}
func TestValidate_Valid(t *testing.T) {
require.NoError(t, validProxy().Validate())
}
@@ -1315,3 +1326,68 @@ func TestValidate_Private_RejectsNonHTTPMode(t *testing.T) {
}}
assert.ErrorContains(t, rp.Validate(), "HTTP")
}
func TestRestrictions_AppSecMode_RoundTrip(t *testing.T) {
mode := api.AccessRestrictionsAppsecModeEnforce
apiIn := &api.AccessRestrictions{AppsecMode: &mode}
model, err := restrictionsFromAPI(apiIn)
require.NoError(t, err)
assert.Equal(t, "enforce", model.AppSecMode)
// appsec_mode alone must keep the restrictions object alive on both the API
// and proto legs: it is meaningful without any CIDR or country entry.
apiOut := restrictionsToAPI(model)
require.NotNil(t, apiOut, "appsec_mode alone must not collapse the restrictions to nil")
require.NotNil(t, apiOut.AppsecMode)
assert.Equal(t, api.AccessRestrictionsAppsecModeEnforce, *apiOut.AppsecMode)
protoOut := restrictionsToProto(model)
require.NotNil(t, protoOut, "appsec_mode alone must reach the proxy")
assert.Equal(t, "enforce", protoOut.AppsecMode)
}
func TestRestrictions_AppSecMode_EmptyIsOmitted(t *testing.T) {
model, err := restrictionsFromAPI(&api.AccessRestrictions{
AllowedCidrs: &[]string{"203.0.113.0/24"},
})
require.NoError(t, err)
assert.Empty(t, model.AppSecMode)
apiOut := restrictionsToAPI(model)
require.NotNil(t, apiOut)
assert.Nil(t, apiOut.AppsecMode, "empty appsec_mode is omitted from the API response")
}
func TestRestrictions_AppSecMode_CopyIsDeep(t *testing.T) {
original := AccessRestrictions{AppSecMode: "observe", CrowdSecMode: "enforce"}
assert.Equal(t, original, original.Copy(), "Copy must carry every mode field")
}
func TestValidate_RejectsInvalidAppSecMode(t *testing.T) {
rp := validProxy()
rp.Restrictions = AccessRestrictions{AppSecMode: "sometimes"}
assert.ErrorContains(t, rp.Validate(), "appsec_mode")
}
func TestValidate_RejectsAppSecOnL4Modes(t *testing.T) {
// AppSec inspects HTTP requests, so the L4 modes cannot honor it. Accepting
// the field there would report protection that never runs.
for _, mode := range []string{ModeTCP, ModeUDP, ModeTLS} {
t.Run(mode, func(t *testing.T) {
rp := validL4Proxy(mode)
rp.Restrictions = AccessRestrictions{AppSecMode: "enforce"}
assert.ErrorContains(t, rp.Validate(), "appsec_mode is only supported for HTTP services")
})
}
}
func TestValidate_AllowsAppSecOffOnL4Modes(t *testing.T) {
for _, mode := range []string{ModeTCP, ModeUDP, ModeTLS} {
t.Run(mode, func(t *testing.T) {
rp := validL4Proxy(mode)
rp.Restrictions = AccessRestrictions{AppSecMode: "off"}
require.NoError(t, rp.Validate(), "an explicit off must not be rejected on L4 services")
})
}
}

View File

@@ -1,404 +0,0 @@
package grpc
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
rpproxy "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
"github.com/netbirdio/netbird/shared/management/proto"
nbstatus "github.com/netbirdio/netbird/shared/management/status"
)
type modelDiscoveryCall struct {
result *proto.ModelDiscoveryResult
err error
}
func newModelDiscoveryTestServer(t *testing.T) (*ProxyServiceServer, *testProxyController) {
t.Helper()
controller := newTestProxyController()
server := &ProxyServiceServer{
proxyController: controller,
modelDiscoveryPending: make(map[string]*pendingModelDiscovery),
}
return server, controller
}
func addModelDiscoveryTestConnection(
t *testing.T,
server *ProxyServiceServer,
controller *testProxyController,
proxyID, sessionID, cluster string,
accountID *string,
supportsDiscovery bool,
) (*proxyConnection, context.CancelFunc) {
t.Helper()
ctx, cancel := context.WithCancel(context.Background())
ready := make(chan struct{})
close(ready)
conn := &proxyConnection{
proxyID: proxyID,
sessionID: sessionID,
address: cluster,
accountID: accountID,
capabilities: &proto.ProxyCapabilities{
SupportsModelDiscovery: &supportsDiscovery,
},
syncStream: &syncRecordingStream{},
modelDiscoveryChan: make(chan *proto.ModelDiscoveryRequest, modelDiscoveryQueueSize),
ready: ready,
ctx: ctx,
cancel: cancel,
}
server.connectedProxies.Store(proxyID, conn)
require.NoError(t, controller.RegisterProxyToCluster(context.Background(), cluster, proxyID))
t.Cleanup(func() {
cancel()
server.connectedProxies.Delete(proxyID)
_ = controller.UnregisterProxyFromCluster(context.Background(), cluster, proxyID)
})
return conn, cancel
}
func callDiscoverModels(
server *ProxyServiceServer,
ctx context.Context,
accountID, cluster string,
req *proto.ModelDiscoveryRequest,
) <-chan modelDiscoveryCall {
done := make(chan modelDiscoveryCall, 1)
go func() {
result, err := server.DiscoverModels(ctx, accountID, cluster, req)
done <- modelDiscoveryCall{result: result, err: err}
}()
return done
}
func TestDiscoverModels_CorrelatesResultAndCopiesSensitiveRequest(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
conn, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "session-a", "cluster.example.com", nil, true,
)
callerReq := &proto.ModelDiscoveryRequest{
RequestId: "caller-controlled",
UpstreamUrl: "http://ollama.internal:11434",
AuthHeaderName: "Authorization",
AuthHeaderValue: "Bearer secret",
SkipTlsVerify: true,
OllamaFallback: true,
}
done := callDiscoverModels(server, context.Background(), "account-a", "cluster.example.com", callerReq)
wireReq := <-conn.modelDiscoveryChan
assert.NotEmpty(t, wireReq.GetRequestId())
assert.NotEqual(t, callerReq.GetRequestId(), wireReq.GetRequestId(), "management must own correlation IDs")
assert.Equal(t, callerReq.GetUpstreamUrl(), wireReq.GetUpstreamUrl())
assert.Equal(t, callerReq.GetAuthHeaderName(), wireReq.GetAuthHeaderName())
assert.Equal(t, callerReq.GetAuthHeaderValue(), wireReq.GetAuthHeaderValue())
assert.Equal(t, callerReq.GetSkipTlsVerify(), wireReq.GetSkipTlsVerify())
assert.Equal(t, callerReq.GetOllamaFallback(), wireReq.GetOllamaFallback())
assert.Equal(t, "caller-controlled", callerReq.GetRequestId(), "caller request must not be mutated")
want := &proto.ModelDiscoveryResult{
RequestId: wireReq.GetRequestId(),
Models: []*proto.ModelDiscoveryModel{
{Id: "llama3.2:latest", Label: "llama3.2:latest"},
},
Source: "openai_v1_models",
}
server.completeModelDiscovery(conn, want)
got := <-done
require.NoError(t, got.err)
assert.Equal(t, want, got.result)
}
func TestDiscoverModels_IgnoresMismatchedCorrelation(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
conn, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "session-a", "cluster.example.com", nil, true,
)
done := callDiscoverModels(server, context.Background(), "account-a", "cluster.example.com", &proto.ModelDiscoveryRequest{
UpstreamUrl: "http://ollama.internal:11434",
})
wireReq := <-conn.modelDiscoveryChan
server.completeModelDiscovery(conn, &proto.ModelDiscoveryResult{RequestId: "wrong-request"})
select {
case got := <-done:
t.Fatalf("mismatched request completed discovery: %+v", got)
default:
}
wrongSession := *conn
wrongSession.sessionID = "session-b"
server.completeModelDiscovery(&wrongSession, &proto.ModelDiscoveryResult{RequestId: wireReq.GetRequestId()})
select {
case got := <-done:
t.Fatalf("mismatched proxy session completed discovery: %+v", got)
default:
}
server.completeModelDiscovery(conn, &proto.ModelDiscoveryResult{RequestId: wireReq.GetRequestId()})
got := <-done
require.NoError(t, got.err)
require.NotNil(t, got.result)
}
func TestDiscoverModels_ReturnsProxyReportedErrorForManagerContext(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
conn, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "session-a", "cluster.example.com", nil, true,
)
done := callDiscoverModels(server, context.Background(), "account-a", "cluster.example.com", &proto.ModelDiscoveryRequest{
UpstreamUrl: "http://ollama.internal:11434",
})
wireReq := <-conn.modelDiscoveryChan
server.completeModelDiscovery(conn, &proto.ModelDiscoveryResult{
RequestId: wireReq.GetRequestId(),
Error: "upstream returned HTTP 401",
})
got := <-done
require.NoError(t, got.err)
require.NotNil(t, got.result)
assert.Equal(t, "upstream returned HTTP 401", got.result.GetError())
assert.Equal(t, wireReq.GetRequestId(), got.result.GetRequestId())
}
func TestDiscoverModels_FiltersCapabilityAndAccountScope(t *testing.T) {
tests := []struct {
name string
connectionAccount *string
requestAccount string
supported bool
}{
{
name: "capability absent",
requestAccount: "account-a",
supported: false,
},
{
name: "BYOP account mismatch",
connectionAccount: stringPointer("account-b"),
requestAccount: "account-a",
supported: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "session-a", "cluster.example.com",
tt.connectionAccount, tt.supported,
)
result, err := server.DiscoverModels(
context.Background(),
tt.requestAccount,
"cluster.example.com",
&proto.ModelDiscoveryRequest{UpstreamUrl: "http://ollama.internal:11434"},
)
require.Nil(t, result)
requireStatusType(t, err, nbstatus.PreconditionFailed)
})
}
}
func TestDiscoverModels_RejectsStaleClusterMembershipAfterReconnect(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
conn, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "new-session", "new-cluster.example.com", nil, true,
)
// Simulate the stale membership left behind when this proxy ID previously
// connected to another cluster and that superseded session skipped cleanup.
require.NoError(t, controller.RegisterProxyToCluster(
context.Background(),
"old-cluster.example.com",
"proxy-a",
))
result, err := server.DiscoverModels(
context.Background(),
"account-a",
"old-cluster.example.com",
&proto.ModelDiscoveryRequest{UpstreamUrl: "http://ollama.internal:11434"},
)
require.Nil(t, result)
requireStatusType(t, err, nbstatus.PreconditionFailed)
select {
case request := <-conn.modelDiscoveryChan:
t.Fatalf("sent credentials to connection in a different cluster: %+v", request)
default:
}
}
func TestDiscoverModels_CancelAndDisconnectCleanPendingRequest(t *testing.T) {
t.Run("caller cancellation", func(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
conn, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "session-a", "cluster.example.com", nil, true,
)
ctx, cancel := context.WithCancel(context.Background())
done := callDiscoverModels(server, ctx, "account-a", "cluster.example.com", &proto.ModelDiscoveryRequest{
UpstreamUrl: "http://ollama.internal:11434",
})
<-conn.modelDiscoveryChan
cancel()
got := <-done
require.Nil(t, got.result)
requireStatusType(t, got.err, nbstatus.PreconditionFailed)
assert.Empty(t, server.modelDiscoveryPending)
})
t.Run("proxy disconnect", func(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
conn, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "session-a", "cluster.example.com", nil, true,
)
done := callDiscoverModels(server, context.Background(), "account-a", "cluster.example.com", &proto.ModelDiscoveryRequest{
UpstreamUrl: "http://ollama.internal:11434",
})
<-conn.modelDiscoveryChan
server.failModelDiscoveries(conn, rpproxy.ErrModelDiscoveryUnavailable)
got := <-done
require.Nil(t, got.result)
requireStatusType(t, got.err, nbstatus.PreconditionFailed)
assert.Empty(t, server.modelDiscoveryPending)
})
}
func TestDiscoverModels_UsesNextCapableProxyWhenFirstQueueIsFull(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
first, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "session-a", "cluster.example.com", nil, true,
)
second, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-b", "session-b", "cluster.example.com", nil, true,
)
first.modelDiscoveryChan = make(chan *proto.ModelDiscoveryRequest, 1)
first.modelDiscoveryChan <- &proto.ModelDiscoveryRequest{RequestId: "occupied"}
done := callDiscoverModels(server, context.Background(), "account-a", "cluster.example.com", &proto.ModelDiscoveryRequest{
UpstreamUrl: "http://ollama.internal:11434",
})
wireReq := <-second.modelDiscoveryChan
server.completeModelDiscovery(second, &proto.ModelDiscoveryResult{RequestId: wireReq.GetRequestId()})
got := <-done
require.NoError(t, got.err)
require.NotNil(t, got.result)
}
func TestSenderSkipsCanceledQueuedModelDiscovery(t *testing.T) {
server, controller := newModelDiscoveryTestServer(t)
conn, _ := addModelDiscoveryTestConnection(
t, server, controller, "proxy-a", "session-a", "cluster.example.com", nil, true,
)
stream := conn.syncStream.(*syncRecordingStream)
ctx, cancel := context.WithCancel(context.Background())
done := callDiscoverModels(server, ctx, "account-a", "cluster.example.com", &proto.ModelDiscoveryRequest{
UpstreamUrl: "http://ollama.internal:11434",
})
require.Eventually(t, func() bool {
return len(conn.modelDiscoveryChan) == 1
}, time.Second, time.Millisecond)
cancel()
got := <-done
require.Nil(t, got.result)
requireStatusType(t, got.err, nbstatus.PreconditionFailed)
liveRequest := &proto.ModelDiscoveryRequest{RequestId: "live-request"}
livePending := &pendingModelDiscovery{
proxyID: conn.proxyID,
sessionID: conn.sessionID,
done: make(chan modelDiscoveryCompletion, 1),
}
server.registerModelDiscovery(liveRequest.GetRequestId(), livePending)
defer server.removeModelDiscovery(liveRequest.GetRequestId(), livePending)
conn.modelDiscoveryChan <- liveRequest
errCh := make(chan error, 1)
go server.sender(conn, errCh)
require.Eventually(t, func() bool {
stream.mu.Lock()
defer stream.mu.Unlock()
return len(stream.sent) == 1
}, time.Second, time.Millisecond)
stream.mu.Lock()
sent := append([]*proto.SyncMappingsResponse(nil), stream.sent...)
stream.mu.Unlock()
require.Len(t, sent, 1)
assert.Equal(t, "live-request", sent[0].GetModelDiscoveryRequest().GetRequestId())
}
func TestDrainRecv_DispatchesModelDiscoveryResult(t *testing.T) {
server, _ := newModelDiscoveryTestServer(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
conn := &proxyConnection{
proxyID: "proxy-a",
sessionID: "session-a",
ctx: ctx,
cancel: cancel,
}
pending := &pendingModelDiscovery{
proxyID: conn.proxyID,
sessionID: conn.sessionID,
done: make(chan modelDiscoveryCompletion, 1),
}
server.registerModelDiscovery("request-a", pending)
stream := &syncRecordingStream{
recvMsgs: []*proto.SyncMappingsRequest{
{
Msg: &proto.SyncMappingsRequest_ModelDiscoveryResult{
ModelDiscoveryResult: &proto.ModelDiscoveryResult{RequestId: "request-a"},
},
},
},
}
errCh := make(chan error, 1)
go server.drainRecv(conn, stream, errCh)
completion := <-pending.done
require.NoError(t, completion.err)
require.NotNil(t, completion.result)
assert.Equal(t, "request-a", completion.result.GetRequestId())
}
func TestSendModelDiscoveryRequest_UsesOutOfBandSyncField(t *testing.T) {
stream := &syncRecordingStream{}
conn := &proxyConnection{syncStream: stream}
req := &proto.ModelDiscoveryRequest{RequestId: "request-a"}
require.NoError(t, conn.sendModelDiscoveryRequest(req))
require.Len(t, stream.sent, 1)
assert.Empty(t, stream.sent[0].GetMapping())
assert.Equal(t, req, stream.sent[0].GetModelDiscoveryRequest())
}
func stringPointer(value string) *string {
return &value
}
func requireStatusType(t *testing.T, err error, want nbstatus.Type) {
t.Helper()
require.Error(t, err)
statusErr, ok := nbstatus.FromError(err)
require.True(t, ok)
assert.Equal(t, want, statusErr.Type())
}

View File

@@ -15,7 +15,6 @@ import (
"net/http"
"net/url"
"os"
"sort"
"strconv"
"strings"
"sync"
@@ -85,10 +84,6 @@ type ProxyServiceServer struct {
// Map of connected proxies: proxy_id -> proxy connection
connectedProxies sync.Map
// modelDiscoveryPending correlates out-of-band model-discovery results
// received on SyncMappings with the management request waiting for them.
modelDiscoveryMu sync.Mutex
modelDiscoveryPending map[string]*pendingModelDiscovery
// Manager for access logs
accessLogManager accesslogs.Manager
@@ -148,8 +143,6 @@ const defaultProxyTokenTTL = 5 * time.Minute
const defaultSnapshotBatchSize = 500
const modelDiscoveryQueueSize = 16
func snapshotBatchSizeFromEnv() int {
if v := os.Getenv("NB_PROXY_SNAPSHOT_BATCH_SIZE"); v != "" {
if n, err := strconv.Atoi(v); err == nil && n > 0 {
@@ -178,26 +171,10 @@ type proxyConnection struct {
stream proto.ProxyService_GetMappingUpdateServer
// syncStream is set when the proxy connected via SyncMappings.
// When non-nil, the sender goroutine uses this instead of stream.
syncStream proto.ProxyService_SyncMappingsServer
sendChan chan *proto.GetMappingUpdateResponse
modelDiscoveryChan chan *proto.ModelDiscoveryRequest
// ready closes after the initial snapshot completes and the sender
// goroutine is about to start. Discovery never competes with snapshot
// delivery on the stream.
ready chan struct{}
ctx context.Context
cancel context.CancelFunc
}
type modelDiscoveryCompletion struct {
result *proto.ModelDiscoveryResult
err error
}
type pendingModelDiscovery struct {
proxyID string
sessionID string
done chan modelDiscoveryCompletion
syncStream proto.ProxyService_SyncMappingsServer
sendChan chan *proto.GetMappingUpdateResponse
ctx context.Context
cancel context.CancelFunc
}
func enforceAccountScope(ctx context.Context, requestAccountID string) error {
@@ -215,18 +192,17 @@ func enforceAccountScope(ctx context.Context, requestAccountID string) error {
func NewProxyServiceServer(accessLogMgr accesslogs.Manager, tokenStore *OneTimeTokenStore, pkceStore *PKCEVerifierStore, oidcConfig ProxyOIDCConfig, peersManager peers.Manager, usersManager users.Manager, idpManager idp.Manager, proxyMgr proxy.Manager, tokenChecker ProxyTokenChecker) *ProxyServiceServer {
ctx, cancel := context.WithCancel(context.Background())
s := &ProxyServiceServer{
accessLogManager: accessLogMgr,
oidcConfig: oidcConfig,
tokenStore: tokenStore,
pkceVerifierStore: pkceStore,
peersManager: peersManager,
usersManager: usersManager,
idpManager: idpManager,
proxyManager: proxyMgr,
tokenChecker: tokenChecker,
snapshotBatchSize: snapshotBatchSizeFromEnv(),
modelDiscoveryPending: make(map[string]*pendingModelDiscovery),
cancel: cancel,
accessLogManager: accessLogMgr,
oidcConfig: oidcConfig,
tokenStore: tokenStore,
pkceVerifierStore: pkceStore,
peersManager: peersManager,
usersManager: usersManager,
idpManager: idpManager,
proxyManager: proxyMgr,
tokenChecker: tokenChecker,
snapshotBatchSize: snapshotBatchSizeFromEnv(),
cancel: cancel,
}
go s.cleanupStaleProxies(ctx)
return s
@@ -416,7 +392,6 @@ func (s *ProxyServiceServer) GetMappingUpdate(req *proto.GetMappingUpdateRequest
return fmt.Errorf("send snapshot to proxy %s: %w", params.proxyID, err)
}
close(conn.ready)
errChan := make(chan error, 2)
go s.sender(conn, errChan)
@@ -450,10 +425,9 @@ func (s *ProxyServiceServer) SyncMappings(stream proto.ProxyService_SyncMappings
return fmt.Errorf("send snapshot to proxy %s: %w", params.proxyID, err)
}
close(conn.ready)
errChan := make(chan error, 2)
go s.sender(conn, errChan)
go s.drainRecv(conn, stream, errChan)
go s.drainRecv(stream, errChan)
return s.serveProxyConnection(conn, proxyRecord, errChan, true)
}
@@ -522,8 +496,6 @@ func (s *ProxyServiceServer) registerProxyConnection(ctx context.Context, params
connSeed.tokenID = tokenID
connSeed.capabilities = params.capabilities
connSeed.sendChan = make(chan *proto.GetMappingUpdateResponse, 100)
connSeed.modelDiscoveryChan = make(chan *proto.ModelDiscoveryRequest, modelDiscoveryQueueSize)
connSeed.ready = make(chan struct{})
connSeed.ctx = connCtx
connSeed.cancel = cancel
@@ -533,6 +505,7 @@ func (s *ProxyServiceServer) registerProxyConnection(ctx context.Context, params
SupportsCustomPorts: c.SupportsCustomPorts,
RequireSubdomain: c.RequireSubdomain,
SupportsCrowdsec: c.SupportsCrowdsec,
SupportsAppsec: c.SupportsAppsec,
Private: c.Private,
}
}
@@ -571,13 +544,10 @@ func (s *ProxyServiceServer) supersedePriorConnection(proxyID, newSessionID stri
// cleanupFailedSnapshot removes the connection from the cluster and store
// after a snapshot send failure.
func (s *ProxyServiceServer) cleanupFailedSnapshot(ctx context.Context, conn *proxyConnection) {
s.failModelDiscoveries(conn, proxy.ErrModelDiscoveryUnavailable)
if s.connectedProxies.CompareAndDelete(conn.proxyID, conn) {
if err := s.proxyController.UnregisterProxyFromCluster(context.Background(), conn.address, conn.proxyID); err != nil {
log.WithContext(ctx).Debugf("cleanup after snapshot failure for proxy %s: %v", conn.proxyID, err)
}
} else {
s.unregisterSupersededProxyCluster(ctx, conn)
}
conn.cancel()
if err := s.proxyManager.Disconnect(context.Background(), conn.proxyID, conn.sessionID); err != nil {
@@ -585,19 +555,15 @@ func (s *ProxyServiceServer) cleanupFailedSnapshot(ctx context.Context, conn *pr
}
}
// drainRecv consumes post-snapshot messages from a bidirectional stream.
// Incremental mapping acks need no action; model-discovery results are
// dispatched to the correlated management caller.
func (s *ProxyServiceServer) drainRecv(conn *proxyConnection, stream proto.ProxyService_SyncMappingsServer, errChan chan<- error) {
// drainRecv consumes and discards messages from a bidirectional stream.
// The proxy sends an ack for every incremental update; we don't need them
// after the snapshot phase. Recv errors are forwarded to errChan.
func (s *ProxyServiceServer) drainRecv(stream proto.ProxyService_SyncMappingsServer, errChan chan<- error) {
for {
msg, err := stream.Recv()
if err != nil {
if _, err := stream.Recv(); err != nil {
errChan <- err
return
}
if result := msg.GetModelDiscoveryResult(); result != nil {
s.completeModelDiscovery(conn, result)
}
}
}
@@ -634,9 +600,7 @@ func (s *ProxyServiceServer) serveProxyConnection(conn *proxyConnection, proxyRe
// disconnectProxy removes the connection from cluster and store, unless it
// has already been superseded by a newer connection.
func (s *ProxyServiceServer) disconnectProxy(conn *proxyConnection) {
s.failModelDiscoveries(conn, proxy.ErrModelDiscoveryUnavailable)
if !s.connectedProxies.CompareAndDelete(conn.proxyID, conn) {
s.unregisterSupersededProxyCluster(context.Background(), conn)
log.Infof("Proxy %s session %s: skipping cleanup, superseded by new connection", conn.proxyID, conn.sessionID)
conn.cancel()
return
@@ -653,26 +617,6 @@ func (s *ProxyServiceServer) disconnectProxy(conn *proxyConnection) {
log.Infof("Proxy %s session %s disconnected", conn.proxyID, conn.sessionID)
}
// unregisterSupersededProxyCluster removes membership owned only by an old
// session after the same proxy ID reconnects at a different address. When the
// address is unchanged, the membership is shared with the replacement and
// must remain registered.
func (s *ProxyServiceServer) unregisterSupersededProxyCluster(ctx context.Context, conn *proxyConnection) {
if conn == nil || s.proxyController == nil {
return
}
currentValue, ok := s.connectedProxies.Load(conn.proxyID)
if ok {
current := currentValue.(*proxyConnection)
if current == conn || current.address == conn.address {
return
}
}
if err := s.proxyController.UnregisterProxyFromCluster(ctx, conn.address, conn.proxyID); err != nil {
log.WithContext(ctx).Debugf("cleanup superseded cluster membership for proxy %s: %v", conn.proxyID, err)
}
}
// sendSnapshotSync sends the initial snapshot with back-pressure: it sends
// one batch, then waits for the proxy to ack before sending the next.
func (s *ProxyServiceServer) sendSnapshotSync(ctx context.Context, conn *proxyConnection, stream proto.ProxyService_SyncMappingsServer) error {
@@ -909,20 +853,6 @@ func (s *ProxyServiceServer) sender(conn *proxyConnection, errChan chan<- error)
return
}
log.WithContext(conn.ctx).Tracef("Send response to proxy %s", conn.proxyID)
case req := <-conn.modelDiscoveryChan:
// The API caller may have timed out while this request waited
// behind other stream work. Pending correlation is removed on
// cancellation, so skip stale probes before they reach the
// proxy and expose a stored credential unnecessarily.
if !s.isModelDiscoveryPendingFor(conn, req.GetRequestId()) {
continue
}
if err := conn.sendModelDiscoveryRequest(req); err != nil {
errChan <- err
log.WithContext(conn.ctx).Tracef("Failed to send model discovery request to proxy %s: %v", conn.proxyID, err)
return
}
log.WithContext(conn.ctx).Tracef("Sent model discovery request to proxy %s", conn.proxyID)
case <-conn.ctx.Done():
return
}
@@ -940,15 +870,6 @@ func (conn *proxyConnection) sendResponse(resp *proto.GetMappingUpdateResponse)
return conn.stream.Send(resp)
}
func (conn *proxyConnection) sendModelDiscoveryRequest(req *proto.ModelDiscoveryRequest) error {
if conn.syncStream == nil {
return proxy.ErrModelDiscoveryUnavailable
}
return conn.syncStream.Send(&proto.SyncMappingsResponse{
ModelDiscoveryRequest: req,
})
}
// SendAccessLog processes access log from proxy
func (s *ProxyServiceServer) SendAccessLog(ctx context.Context, req *proto.SendAccessLogRequest) (*proto.SendAccessLogResponse, error) {
accessLog := req.GetLog()
@@ -1097,181 +1018,6 @@ func (s *ProxyServiceServer) GetConnectedProxyURLs() []string {
return urls
}
// DiscoverModels sends one correlated model-discovery request to a capable
// proxy in clusterAddr and waits for its result. Only the stored connection's
// cluster/account scope is used for routing; no routing identity is carried in
// the request delivered to the proxy.
func (s *ProxyServiceServer) DiscoverModels(ctx context.Context, accountID, clusterAddr string, req *proto.ModelDiscoveryRequest) (*proto.ModelDiscoveryResult, error) {
if req == nil {
return nil, nbstatus.Errorf(nbstatus.InvalidArgument, "model discovery request is required")
}
if strings.TrimSpace(accountID) == "" {
return nil, nbstatus.Errorf(nbstatus.InvalidArgument, "account ID is required for model discovery")
}
if strings.TrimSpace(clusterAddr) == "" {
return nil, nbstatus.Errorf(nbstatus.PreconditionFailed, "agent network proxy cluster is not configured")
}
if s.proxyController == nil {
return nil, nbstatus.Errorf(nbstatus.PreconditionFailed, "model discovery is unavailable for the configured proxy cluster")
}
proxyIDs := s.proxyController.GetProxiesForCluster(clusterAddr)
sort.Strings(proxyIDs)
request := &proto.ModelDiscoveryRequest{
RequestId: uuid.NewString(),
UpstreamUrl: req.GetUpstreamUrl(),
AuthHeaderName: req.GetAuthHeaderName(),
AuthHeaderValue: req.GetAuthHeaderValue(),
SkipTlsVerify: req.GetSkipTlsVerify(),
OllamaFallback: req.GetOllamaFallback(),
}
for _, proxyID := range proxyIDs {
connVal, ok := s.connectedProxies.Load(proxyID)
if !ok {
continue
}
conn := connVal.(*proxyConnection)
// Cluster membership can briefly retain a proxy ID from a superseded
// session. The live connection is authoritative: never send provider
// credentials unless it is connected to the exact requested cluster.
if conn.address != clusterAddr {
continue
}
if !modelDiscoveryCapable(conn, accountID) {
continue
}
pending := &pendingModelDiscovery{
proxyID: conn.proxyID,
sessionID: conn.sessionID,
done: make(chan modelDiscoveryCompletion, 1),
}
s.registerModelDiscovery(request.GetRequestId(), pending)
queued := false
select {
case conn.modelDiscoveryChan <- request:
queued = true
case <-ctx.Done():
s.removeModelDiscovery(request.GetRequestId(), pending)
return nil, modelDiscoveryContextError(ctx.Err())
default:
s.removeModelDiscovery(request.GetRequestId(), pending)
}
if !queued {
continue
}
defer s.removeModelDiscovery(request.GetRequestId(), pending)
select {
case completion := <-pending.done:
if completion.err != nil {
return nil, nbstatus.Errorf(nbstatus.PreconditionFailed, "model discovery proxy disconnected")
}
return completion.result, nil
case <-ctx.Done():
return nil, modelDiscoveryContextError(ctx.Err())
case <-conn.ctx.Done():
return nil, nbstatus.Errorf(nbstatus.PreconditionFailed, "model discovery proxy disconnected")
}
}
return nil, nbstatus.Errorf(nbstatus.PreconditionFailed, "no connected proxy in the configured cluster supports model discovery")
}
func modelDiscoveryCapable(conn *proxyConnection, accountID string) bool {
if conn == nil || conn.syncStream == nil || conn.modelDiscoveryChan == nil || conn.ready == nil {
return false
}
if conn.ctx == nil || conn.ctx.Err() != nil {
return false
}
if conn.accountID != nil && *conn.accountID != accountID {
return false
}
if conn.capabilities == nil || !conn.capabilities.GetSupportsModelDiscovery() {
return false
}
select {
case <-conn.ready:
return true
default:
return false
}
}
func modelDiscoveryContextError(err error) error {
if errors.Is(err, context.DeadlineExceeded) {
return nbstatus.Errorf(nbstatus.PreconditionFailed, "model discovery timed out")
}
return nbstatus.Errorf(nbstatus.PreconditionFailed, "model discovery was canceled")
}
func (s *ProxyServiceServer) registerModelDiscovery(requestID string, pending *pendingModelDiscovery) {
s.modelDiscoveryMu.Lock()
defer s.modelDiscoveryMu.Unlock()
if s.modelDiscoveryPending == nil {
s.modelDiscoveryPending = make(map[string]*pendingModelDiscovery)
}
s.modelDiscoveryPending[requestID] = pending
}
func (s *ProxyServiceServer) removeModelDiscovery(requestID string, pending *pendingModelDiscovery) {
s.modelDiscoveryMu.Lock()
defer s.modelDiscoveryMu.Unlock()
if s.modelDiscoveryPending[requestID] == pending {
delete(s.modelDiscoveryPending, requestID)
}
}
func (s *ProxyServiceServer) isModelDiscoveryPendingFor(conn *proxyConnection, requestID string) bool {
if conn == nil || requestID == "" {
return false
}
s.modelDiscoveryMu.Lock()
defer s.modelDiscoveryMu.Unlock()
pending := s.modelDiscoveryPending[requestID]
return pending != nil && pending.proxyID == conn.proxyID && pending.sessionID == conn.sessionID
}
func (s *ProxyServiceServer) completeModelDiscovery(conn *proxyConnection, result *proto.ModelDiscoveryResult) {
if conn == nil || result == nil || result.GetRequestId() == "" {
return
}
s.modelDiscoveryMu.Lock()
pending := s.modelDiscoveryPending[result.GetRequestId()]
s.modelDiscoveryMu.Unlock()
if pending == nil || pending.proxyID != conn.proxyID || pending.sessionID != conn.sessionID {
return
}
select {
case pending.done <- modelDiscoveryCompletion{result: result}:
default:
}
}
func (s *ProxyServiceServer) failModelDiscoveries(conn *proxyConnection, err error) {
if conn == nil {
return
}
s.modelDiscoveryMu.Lock()
pending := make([]*pendingModelDiscovery, 0)
for _, item := range s.modelDiscoveryPending {
if item.proxyID == conn.proxyID && item.sessionID == conn.sessionID {
pending = append(pending, item)
}
}
s.modelDiscoveryMu.Unlock()
for _, item := range pending {
select {
case item.done <- modelDiscoveryCompletion{err: err}:
default:
}
}
}
// SendServiceUpdateToCluster sends a service update to all proxy servers in a specific cluster.
// If clusterAddr is empty, broadcasts to all connected proxy servers (backward compatibility).
// For create/update operations a unique one-time auth token is generated per
@@ -1304,12 +1050,6 @@ func (s *ProxyServiceServer) SendServiceUpdateToCluster(ctx context.Context, upd
continue
}
conn := connVal.(*proxyConnection)
// Membership can retain this proxy ID from a superseded session in a
// different cluster. The live connection address is authoritative,
// especially because Agent Network mappings can contain credentials.
if conn.address != clusterAddr {
continue
}
if conn.accountID != nil && update.AccountId != "" && *conn.accountID != update.AccountId {
continue
}

View File

@@ -42,10 +42,6 @@ func newTestProxyController() *testProxyController {
func (c *testProxyController) SendServiceUpdateToCluster(_ context.Context, _ string, _ *proto.ProxyMapping, _ string) {
}
func (c *testProxyController) DiscoverModels(_ context.Context, _, _ string, _ *proto.ModelDiscoveryRequest) (*proto.ModelDiscoveryResult, error) {
return nil, proxy.ErrModelDiscoveryUnavailable
}
func (c *testProxyController) GetOIDCValidationConfig() proxy.OIDCValidationConfig {
return proxy.OIDCValidationConfig{}
}
@@ -222,65 +218,6 @@ func TestSendServiceUpdateToCluster_DeleteNoToken(t *testing.T) {
assert.Empty(t, msg2.AuthToken)
}
func TestSendServiceUpdateToCluster_RejectsStaleClusterMembership(t *testing.T) {
ctx := context.Background()
s := &ProxyServiceServer{
tokenStore: NewOneTimeTokenStore(ctx, testCacheStore(t)),
}
controller := newTestProxyController()
s.SetProxyController(controller)
ch := registerFakeProxy(s, "proxy-a", "new-cluster.example.com")
require.NoError(t, controller.RegisterProxyToCluster(ctx, "old-cluster.example.com", "proxy-a"))
s.SendServiceUpdateToCluster(ctx, &proto.ProxyMapping{
Type: proto.ProxyMappingUpdateType_UPDATE_TYPE_CREATED,
Id: "agent-network-provider-1",
AccountId: "account-1",
Domain: "agent.example.com",
Path: []*proto.PathMapping{
{Path: "/", Target: "http://ollama.internal:11434/"},
},
}, "old-cluster.example.com")
assert.True(t, drainEmpty(ch), "stale membership must not route a mapping to a different live cluster")
}
func TestDisconnectProxy_RemovesSupersededMembershipFromOldCluster(t *testing.T) {
controller := newTestProxyController()
server := &ProxyServiceServer{proxyController: controller}
oldCtx, cancelOld := context.WithCancel(context.Background())
newCtx, cancelNew := context.WithCancel(context.Background())
t.Cleanup(cancelOld)
t.Cleanup(cancelNew)
oldConnection := &proxyConnection{
proxyID: "proxy-a",
sessionID: "old-session",
address: "old-cluster.example.com",
ctx: oldCtx,
cancel: cancelOld,
}
newConnection := &proxyConnection{
proxyID: "proxy-a",
sessionID: "new-session",
address: "new-cluster.example.com",
ctx: newCtx,
cancel: cancelNew,
}
require.NoError(t, controller.RegisterProxyToCluster(context.Background(), oldConnection.address, oldConnection.proxyID))
require.NoError(t, controller.RegisterProxyToCluster(context.Background(), newConnection.address, newConnection.proxyID))
server.connectedProxies.Store(newConnection.proxyID, newConnection)
server.disconnectProxy(oldConnection)
assert.Empty(t, controller.GetProxiesForCluster(oldConnection.address))
assert.Equal(t, []string{"proxy-a"}, controller.GetProxiesForCluster(newConnection.address))
current, ok := server.connectedProxies.Load(newConnection.proxyID)
require.True(t, ok)
assert.Same(t, newConnection, current)
}
func TestSendServiceUpdate_UniqueTokensPerProxy(t *testing.T) {
ctx := context.Background()
tokenStore := NewOneTimeTokenStore(ctx, testCacheStore(t))

View File

@@ -2259,7 +2259,7 @@ func (s *SqlStore) getPostureChecks(ctx context.Context, accountID string) ([]*p
}
func (s *SqlStore) getServices(ctx context.Context, accountID string) ([]*rpservice.Service, error) {
const serviceQuery = `SELECT id, account_id, name, domain, enabled, auth,
const serviceQuery = `SELECT id, account_id, name, domain, enabled, auth, restrictions,
meta_created_at, meta_certificate_issued_at, meta_status, proxy_cluster,
pass_host_header, rewrite_redirects, session_private_key, session_public_key,
mode, listen_port, port_auto_assigned, source, source_peer, terminated,
@@ -2278,6 +2278,7 @@ func (s *SqlStore) getServices(ctx context.Context, accountID string) ([]*rpserv
services, err := pgx.CollectRows(serviceRows, func(row pgx.CollectableRow) (*rpservice.Service, error) {
var s rpservice.Service
var auth []byte
var restrictions []byte
var accessGroups []byte
var createdAt, certIssuedAt sql.NullTime
var status, proxyCluster, sessionPrivateKey, sessionPublicKey sql.NullString
@@ -2291,6 +2292,7 @@ func (s *SqlStore) getServices(ctx context.Context, accountID string) ([]*rpserv
&s.Domain,
&s.Enabled,
&auth,
&restrictions,
&createdAt,
&certIssuedAt,
&status,
@@ -2318,6 +2320,12 @@ func (s *SqlStore) getServices(ctx context.Context, accountID string) ([]*rpserv
}
}
if len(restrictions) > 0 {
if err := json.Unmarshal(restrictions, &s.Restrictions); err != nil {
return nil, fmt.Errorf("unmarshal restrictions: %w", err)
}
}
if len(accessGroups) > 0 {
if err := json.Unmarshal(accessGroups, &s.AccessGroups); err != nil {
return nil, fmt.Errorf("unmarshal access_groups: %w", err)
@@ -6357,6 +6365,7 @@ var validCapabilityColumns = map[string]struct{}{
"supports_custom_ports": {},
"require_subdomain": {},
"supports_crowdsec": {},
"supports_appsec": {},
"private": {},
}
@@ -6387,6 +6396,14 @@ func (s *SqlStore) GetClusterSupportsCrowdSec(ctx context.Context, clusterAddr s
return s.getClusterUnanimousCapability(ctx, clusterAddr, "supports_crowdsec")
}
// GetClusterSupportsAppSec returns whether all active proxies in the cluster
// have a CrowdSec AppSec endpoint configured. Returns nil when no proxy
// reported the capability. Unanimous for the same reason as CrowdSec: a single
// proxy without AppSec would let requests through uninspected.
func (s *SqlStore) GetClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool {
return s.getClusterUnanimousCapability(ctx, clusterAddr, "supports_appsec")
}
// getClusterUnanimousCapability returns an aggregated boolean capability
// requiring all active proxies in the cluster to report true.
func (s *SqlStore) getClusterUnanimousCapability(ctx context.Context, clusterAddr, column string) *bool {

View File

@@ -0,0 +1,119 @@
package store
import (
"context"
"fmt"
"os"
"runtime"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
)
// Capabilities travel proxy → gRPC → embedded gorm columns → aggregation → API.
// A field dropped at any of those hops reads as "capability absent", which is
// indistinguishable from a proxy that never reported it: the dashboard simply
// hides the feature and nothing fails. These assertions cover the persistence
// and aggregation hops.
func TestSqlStore_ClusterCapabilityAggregation(t *testing.T) {
if os.Getenv("CI") == "true" && (runtime.GOOS == "darwin" || runtime.GOOS == "windows") {
t.Skip("skip CI tests on darwin and windows")
}
yes, no := true, false
tests := []struct {
name string
reported []*bool // one entry per connected proxy in the cluster
wantAppSec *bool
wantAssertion string
}{
{
name: "unreported stays unknown",
reported: []*bool{nil},
wantAppSec: nil,
wantAssertion: "an unreported capability must not read as false",
},
{
name: "single proxy reporting true",
reported: []*bool{&yes},
wantAppSec: &yes,
wantAssertion: "a reported capability must survive persistence",
},
{
name: "one proxy without it disables the cluster",
reported: []*bool{&yes, &no},
wantAppSec: &no,
wantAssertion: "capability must be unanimous, so a rolling upgrade cannot leave traffic uninspected",
},
{
name: "one proxy yet to report disables the cluster",
reported: []*bool{&yes, nil},
wantAppSec: &no,
wantAssertion: "a proxy that has not reported must not count as capable",
},
}
runTestForAllEngines(t, "", func(t *testing.T, store Store) {
ctx := context.Background()
for i, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cluster := fmt.Sprintf("cluster-%d.proxy.example", i)
for j, reported := range tt.reported {
require.NoError(t, store.SaveProxy(ctx, &proxy.Proxy{
ID: fmt.Sprintf("proxy-%d-%d", i, j),
ClusterAddress: cluster,
Status: proxy.StatusConnected,
LastSeen: time.Now(),
Capabilities: proxy.Capabilities{SupportsAppsec: reported},
}))
}
got := store.GetClusterSupportsAppSec(ctx, cluster)
if tt.wantAppSec == nil {
assert.Nil(t, got, tt.wantAssertion)
return
}
require.NotNil(t, got, tt.wantAssertion)
assert.Equal(t, *tt.wantAppSec, *got, tt.wantAssertion)
})
}
})
}
// AppSec and IP reputation are separate endpoints, so a cluster can have either
// without the other. Gating one on the other would silently disable a feature
// the operator configured.
func TestSqlStore_ClusterCapabilitiesAreIndependent(t *testing.T) {
if os.Getenv("CI") == "true" && (runtime.GOOS == "darwin" || runtime.GOOS == "windows") {
t.Skip("skip CI tests on darwin and windows")
}
runTestForAllEngines(t, "", func(t *testing.T, store Store) {
ctx := context.Background()
const cluster = "independent.proxy.example"
yes, no := true, false
require.NoError(t, store.SaveProxy(ctx, &proxy.Proxy{
ID: "proxy-independent",
ClusterAddress: cluster,
Status: proxy.StatusConnected,
LastSeen: time.Now(),
Capabilities: proxy.Capabilities{
SupportsAppsec: &yes,
SupportsCrowdsec: &no,
},
}))
appsec := store.GetClusterSupportsAppSec(ctx, cluster)
crowdsec := store.GetClusterSupportsCrowdSec(ctx, cluster)
require.NotNil(t, appsec)
require.NotNil(t, crowdsec)
assert.True(t, *appsec, "AppSec must not be gated on CrowdSec")
assert.False(t, *crowdsec, "CrowdSec must not be implied by AppSec")
})
}

View File

@@ -44,3 +44,42 @@ func TestSqlStore_GetAccount_PrivateServiceRoundtrip(t *testing.T) {
assert.Equal(t, []string{"grp-admins", "grp-ops"}, got.AccessGroups)
})
}
// Restrictions are stored as a JSON blob, and the Postgres read path lists
// columns by hand: a mode that is not read there is silently off on Postgres
// while working in SQLite dev.
func TestSqlStore_GetAccount_ServiceRestrictionsRoundtrip(t *testing.T) {
if os.Getenv("CI") == "true" && (runtime.GOOS == "darwin" || runtime.GOOS == "windows") {
t.Skip("skip CI tests on darwin and windows")
}
runTestForAllEngines(t, "", func(t *testing.T, store Store) {
ctx := context.Background()
account := newAccountWithId(ctx, "account_svc_restrictions", "testuser", "")
require.NoError(t, store.SaveAccount(ctx, account))
svc := &rpservice.Service{
ID: "svc-restrictions",
AccountID: account.Id,
Name: "restricted-svc",
Domain: "restricted.example",
Enabled: true,
Mode: rpservice.ModeHTTP,
Restrictions: rpservice.AccessRestrictions{
AllowedCIDRs: []string{"203.0.113.0/24"},
CrowdSecMode: "observe",
AppSecMode: "enforce",
},
}
require.NoError(t, store.CreateService(ctx, svc))
loaded, err := store.GetAccount(ctx, account.Id)
require.NoError(t, err)
require.Len(t, loaded.Services, 1)
got := loaded.Services[0].Restrictions
assert.Equal(t, []string{"203.0.113.0/24"}, got.AllowedCIDRs)
assert.Equal(t, "observe", got.CrowdSecMode)
assert.Equal(t, "enforce", got.AppSecMode)
})
}

View File

@@ -321,6 +321,7 @@ type Store interface {
GetClusterSupportsCustomPorts(ctx context.Context, clusterAddr string) *bool
GetClusterRequireSubdomain(ctx context.Context, clusterAddr string) *bool
GetClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool
GetClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool
GetClusterSupportsPrivate(ctx context.Context, clusterAddr string) *bool
CleanupStaleProxies(ctx context.Context, inactivityDuration time.Duration) error
GetAllProxies(ctx context.Context) ([]*proxy.Proxy, error)

View File

@@ -1835,6 +1835,20 @@ func (mr *MockStoreMockRecorder) GetClusterRequireSubdomain(ctx, clusterAddr int
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetClusterRequireSubdomain", reflect.TypeOf((*MockStore)(nil).GetClusterRequireSubdomain), ctx, clusterAddr)
}
// GetClusterSupportsAppSec mocks base method.
func (m *MockStore) GetClusterSupportsAppSec(ctx context.Context, clusterAddr string) *bool {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetClusterSupportsAppSec", ctx, clusterAddr)
ret0, _ := ret[0].(*bool)
return ret0
}
// GetClusterSupportsAppSec indicates an expected call of GetClusterSupportsAppSec.
func (mr *MockStoreMockRecorder) GetClusterSupportsAppSec(ctx, clusterAddr interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetClusterSupportsAppSec", reflect.TypeOf((*MockStore)(nil).GetClusterSupportsAppSec), ctx, clusterAddr)
}
// GetClusterSupportsCrowdSec mocks base method.
func (m *MockStore) GetClusterSupportsCrowdSec(ctx context.Context, clusterAddr string) *bool {
m.ctrl.T.Helper()

View File

@@ -79,6 +79,11 @@ var (
geoDataDir string
crowdsecAPIURL string
crowdsecAPIKey string
appsecURL string
appsecTimeout time.Duration
appsecMaxBodyBytes int64
captureBudgetBytes int64
appsecMaxConcurrent int
)
var rootCmd = &cobra.Command{
@@ -125,6 +130,11 @@ func init() {
rootCmd.Flags().StringVar(&geoDataDir, "geo-data-dir", envStringOrDefault("NB_PROXY_GEO_DATA_DIR", "/var/lib/netbird/geolocation"), "Directory for the GeoLite2 MMDB file (auto-downloaded if missing)")
rootCmd.Flags().StringVar(&crowdsecAPIURL, "crowdsec-api-url", envStringOrDefault("NB_PROXY_CROWDSEC_API_URL", ""), "CrowdSec LAPI URL for IP reputation checks")
rootCmd.Flags().StringVar(&crowdsecAPIKey, "crowdsec-api-key", envStringOrDefault("NB_PROXY_CROWDSEC_API_KEY", ""), "CrowdSec bouncer API key")
rootCmd.Flags().StringVar(&appsecURL, "crowdsec-appsec-url", envStringOrDefault("NB_PROXY_CROWDSEC_APPSEC_URL", ""), "CrowdSec AppSec (WAF) endpoint for HTTP request inspection, e.g. http://127.0.0.1:7422/ (reuses the bouncer API key)")
rootCmd.Flags().DurationVar(&appsecTimeout, "crowdsec-appsec-timeout", envDurationOrDefault("NB_PROXY_CROWDSEC_APPSEC_TIMEOUT", 0), "Timeout for a single AppSec inspection call (0 = 200ms)")
rootCmd.Flags().IntVar(&appsecMaxConcurrent, "crowdsec-appsec-max-concurrent", int(envInt64OrDefault("NB_PROXY_CROWDSEC_APPSEC_MAX_CONCURRENT", 0)), "Cap on AppSec inspections in flight; further requests are denied in enforce mode rather than queued (0 = 256, negative = no cap)")
rootCmd.Flags().Int64Var(&captureBudgetBytes, "capture-budget-bytes", envInt64OrDefault("NB_PROXY_CAPTURE_BUDGET_BYTES", 0), "Total in-flight request-body buffering across the proxy, shared by AppSec inspection and agent-network capture (0 = 256MiB)")
rootCmd.Flags().Int64Var(&appsecMaxBodyBytes, "crowdsec-appsec-max-body-bytes", envInt64OrDefault("NB_PROXY_CROWDSEC_APPSEC_MAX_BODY_BYTES", 0), "Cap on the request body mirrored to AppSec (0 = 64KiB, negative = headers and URI only)")
}
// Execute runs the root command.
@@ -218,47 +228,59 @@ func runServer(cmd *cobra.Command, args []string) error {
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
defer stop()
srv := proxy.New(ctx, proxy.Config{
ListenAddr: addr,
Logger: logger,
Version: Version,
ManagementAddress: mgmtAddr,
ProxyURL: proxyDomain,
ProxyToken: proxyToken,
CertificateDirectory: certDir,
CertificateFile: certFile,
CertificateKeyFile: certKeyFile,
GenerateACMECertificates: acmeCerts,
ACMEChallengeAddress: acmeAddr,
ACMEDirectory: acmeDir,
ACMEEABKID: acmeEABKID,
ACMEEABHMACKey: acmeEABHMACKey,
ACMEChallengeType: acmeChallengeType,
DebugEndpointEnabled: debugEndpoint,
DebugEndpointAddress: debugEndpointAddr,
HealthAddr: healthAddr,
ForwardedProto: forwardedProto,
TrustedProxies: parsedTrustedProxies,
CertLockMethod: nbacme.CertLockMethod(certLockMethod),
WildcardCertDir: wildcardCertDir,
WireguardPort: wgPort,
Performance: perf,
ProxyProtocol: proxyProtocol,
PreSharedKey: preSharedKey,
SupportsCustomPorts: supportsCustomPorts,
RequireSubdomain: requireSubdomain,
Private: private,
MaxDialTimeout: maxDialTimeout,
MaxSessionIdleTimeout: maxSessionIdleTimeout,
MappingBatchWatchdog: envDurationOrDefault("NB_PROXY_MAPPING_BATCH_WATCHDOG", 0),
GeoDataDir: geoDataDir,
CrowdSecAPIURL: crowdsecAPIURL,
CrowdSecAPIKey: crowdsecAPIKey,
})
srv := proxy.New(ctx, serverConfig(logger, proxyToken, parsedTrustedProxies, perf))
return srv.ListenAndServe(ctx, addr)
}
// serverConfig maps the parsed flags and environment onto the proxy config.
// Kept separate from runServer so registering a new flag does not grow the
// startup path.
func serverConfig(logger *log.Logger, proxyToken string, trustedProxyList *trustedproxy.List, perf embed.Performance) proxy.Config {
return proxy.Config{
ListenAddr: addr,
Logger: logger,
Version: Version,
ManagementAddress: mgmtAddr,
ProxyURL: proxyDomain,
ProxyToken: proxyToken,
CertificateDirectory: certDir,
CertificateFile: certFile,
CertificateKeyFile: certKeyFile,
GenerateACMECertificates: acmeCerts,
ACMEChallengeAddress: acmeAddr,
ACMEDirectory: acmeDir,
ACMEEABKID: acmeEABKID,
ACMEEABHMACKey: acmeEABHMACKey,
ACMEChallengeType: acmeChallengeType,
DebugEndpointEnabled: debugEndpoint,
DebugEndpointAddress: debugEndpointAddr,
HealthAddr: healthAddr,
ForwardedProto: forwardedProto,
TrustedProxies: trustedProxyList,
CertLockMethod: nbacme.CertLockMethod(certLockMethod),
WildcardCertDir: wildcardCertDir,
WireguardPort: wgPort,
Performance: perf,
ProxyProtocol: proxyProtocol,
PreSharedKey: preSharedKey,
SupportsCustomPorts: supportsCustomPorts,
RequireSubdomain: requireSubdomain,
Private: private,
MaxDialTimeout: maxDialTimeout,
MaxSessionIdleTimeout: maxSessionIdleTimeout,
MappingBatchWatchdog: envDurationOrDefault("NB_PROXY_MAPPING_BATCH_WATCHDOG", 0),
GeoDataDir: geoDataDir,
CrowdSecAPIURL: crowdsecAPIURL,
CrowdSecAPIKey: crowdsecAPIKey,
CrowdSecAppSecURL: appsecURL,
CrowdSecAppSecTimeout: appsecTimeout,
CrowdSecAppSecMaxBodyBytes: appsecMaxBodyBytes,
CrowdSecAppSecMaxConcurrent: appsecMaxConcurrent,
MiddlewareCaptureBudgetBytes: captureBudgetBytes,
}
}
func envBoolOrDefault(key string, def bool) bool {
v, exists := os.LookupEnv(key)
if !exists {
@@ -293,6 +315,19 @@ func envUint16OrDefault(key string, def uint16) uint16 {
return uint16(parsed)
}
func envInt64OrDefault(key string, def int64) int64 {
v, exists := os.LookupEnv(key)
if !exists {
return def
}
parsed, err := strconv.ParseInt(v, 10, 64)
if err != nil {
log.Warnf("parse %s=%q: %v, using default %d", key, v, err, def)
return def
}
return parsed
}
func envDurationOrDefault(key string, def time.Duration) time.Duration {
v, exists := os.LookupEnv(key)
if !exists {

View File

@@ -0,0 +1,163 @@
package appsec
import (
"bytes"
"errors"
"io"
"mime"
"net/http"
"net/url"
"slices"
"strings"
)
// bufferBody reads up to limit+1 bytes from r.Body and always restores r.Body so
// the request stays forwardable. oversize reports that the body exceeded limit, in
// which case the returned prefix must not be used for inspection: the bytes are
// only read so they can be replayed to the backend.
func bufferBody(r *http.Request, limit int64) (body []byte, oversize bool, err error) {
original := r.Body
buf, readErr := io.ReadAll(io.LimitReader(original, limit+1))
if readErr != nil && !errors.Is(readErr, io.EOF) {
// Restore what was read so a downstream retry sees a consistent stream,
// then surface the failure.
r.Body = replay(buf, original)
return nil, false, readErr
}
if int64(len(buf)) > limit {
r.Body = replay(buf, original)
return nil, true, nil
}
// The whole body is buffered, so the original is drained and can be closed.
// A close error on a drained read-only body does not invalidate the bytes.
_ = original.Close()
r.Body = io.NopCloser(bytes.NewReader(buf))
// Framing is deliberately left as the client sent it. Rewriting a chunked
// request to a fixed Content-Length here would be invisible to the client
// but not to the rest of the chain: a later body capture with a smaller cap
// sees a known length over its cap and skips capture entirely, where an
// unknown length would have given it a truncated prefix. Inspecting a
// request must not change what any other layer gets to inspect.
return buf, false, nil
}
// replay returns a ReadCloser that yields the already-read prefix followed by
// the remainder of the original stream, and closes the original.
func replay(prefix []byte, rest io.ReadCloser) io.ReadCloser {
return struct {
io.Reader
io.Closer
}{
Reader: io.MultiReader(bytes.NewReader(prefix), rest),
Closer: rest,
}
}
// redactedPlaceholder replaces a credential value in the mirrored body. It is
// inert for rule matching, and its fixed length leaks nothing about the secret.
const redactedPlaceholder = "redacted"
// redactFormFields returns the body to mirror for a URL-encoded form, with the
// values of the named fields replaced. The proxy's own password / PIN login
// form posts to the service path itself, so without this the plaintext
// credential would reach the Security Engine.
//
// Only the credential values are removed, never the whole body: dropping the
// body outright would let a caller exempt any payload from inspection just by
// appending a field named "password". Everything else in the form stays
// inspectable, which is the point.
//
// Returns body unchanged when it is not a URL-encoded form or carries none of
// the fields.
//
// Substitution happens on the raw bytes rather than by re-encoding parsed
// values. Re-encoding would drop pairs that url.ParseQuery rejects, so a
// payload hidden in a malformed pair alongside a credential-named field would
// never be inspected while a tolerant backend parser still acted on it. Working
// byte-wise also avoids reordering keys and normalizing escapes, so the engine
// sees the same bytes the backend will.
//
// Field names match case-sensitively, on purpose: the caller passes the exact
// names the login handler reads via r.FormValue, and that lookup is itself
// case-sensitive. A "Password" field is therefore never a credential as far as
// the proxy is concerned, and redacting it would only blind the WAF to a value
// the proxy does not own.
func redactFormFields(contentType string, body []byte, fields []string) []byte {
if len(fields) == 0 || len(body) == 0 {
return body
}
media, _, err := mime.ParseMediaType(contentType)
if err != nil || media != "application/x-www-form-urlencoded" {
return body
}
return redactURLEncoded(body, fields)
}
// redactURLEncoded replaces the values of the named keys in a URL-encoded
// key/value sequence, the shared syntax of a query string and a form body.
func redactURLEncoded(raw []byte, fields []string) []byte {
// Split on "&" only, matching how Go's form parser delimits pairs.
segments := bytes.Split(raw, []byte("&"))
redacted := false
for i, segment := range segments {
rawKey, _, hasValue := bytes.Cut(segment, []byte("="))
if !hasValue {
continue
}
// Compare the decoded name, so an escaped spelling of the field
// ("pass%77ord") is redacted too: the reader decodes before looking it
// up. A key that fails to decode never reaches that reader either,
// since the parser drops the pair.
name, err := url.QueryUnescape(string(rawKey))
if err != nil || !slices.Contains(fields, name) {
continue
}
// Keep the key bytes as sent and replace only the value. Assigning a
// fresh slice leaves raw untouched, which matters: the caller restored
// the request body from the same buffer.
segments[i] = []byte(string(rawKey) + "=" + redactedPlaceholder)
redacted = true
}
if !redacted {
return raw
}
return bytes.Join(segments, []byte("&"))
}
// redactQuery replaces the values of the named query parameters in a raw query
// string, leaving every other byte as sent.
func redactQuery(rawQuery string, params []string) string {
if len(params) == 0 || rawQuery == "" {
return rawQuery
}
return string(redactURLEncoded([]byte(rawQuery), params))
}
// redactCookieHeader replaces the values of the named cookies in a Cookie
// header, keeping the others intact: cookies are a zone WAF rules match on, so
// dropping the whole header would cost real coverage.
func redactCookieHeader(value string, names []string) string {
if len(names) == 0 || value == "" {
return value
}
parts := strings.Split(value, ";")
redacted := false
for i, part := range parts {
name, _, hasValue := strings.Cut(part, "=")
if !hasValue {
continue
}
// Cookie names are case-sensitive and are not percent-decoded.
if !slices.Contains(names, strings.TrimSpace(name)) {
continue
}
parts[i] = name + "=" + redactedPlaceholder
redacted = true
}
if !redacted {
return value
}
return strings.Join(parts, ";")
}

View File

@@ -0,0 +1,571 @@
// Package appsec implements the CrowdSec AppSec (WAF) side of the remediation
// component protocol: each inspected HTTP request is mirrored to the Security
// Engine's AppSec endpoint, which replies with an allow / ban / captcha verdict
// for that request.
//
// This is a separate endpoint from the LAPI decision stream used by the
// crowdsec package: LAPI answers "is this IP known bad", AppSec answers "is
// this request an attack". The two are configured and enabled independently.
package appsec
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/netip"
"net/url"
"strings"
"sync"
"time"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/proxy/internal/netutil"
"github.com/netbirdio/netbird/proxy/internal/restrict"
)
// Header names the AppSec component reads off the mirrored request. IP, URI and
// Verb are mandatory: the engine answers 500 when any of them is missing.
const (
headerAPIKey = "X-Crowdsec-Appsec-Api-Key" //nolint:gosec // G101: a header name, not a credential
headerIP = "X-Crowdsec-Appsec-Ip"
headerURI = "X-Crowdsec-Appsec-Uri"
headerVerb = "X-Crowdsec-Appsec-Verb"
headerHost = "X-Crowdsec-Appsec-Host"
headerUserAgent = "X-Crowdsec-Appsec-User-Agent"
headerHTTPVersion = "X-Crowdsec-Appsec-Http-Version"
headerTransactionID = "X-Crowdsec-Appsec-Transaction-Id"
)
// headerPrefix covers every protocol header. Any client-supplied header in this
// namespace is dropped before forwarding so a caller cannot influence the
// engine's view of its own address, or replay an API key.
const headerPrefix = "X-Crowdsec-Appsec-"
// Remediation actions the engine can return.
const (
actionAllow = "allow"
actionBan = "ban"
actionCaptcha = "captcha"
)
const (
// DefaultTimeout matches the 200ms budget CrowdSec's remediation component
// spec sets for the blocking AppSec call.
DefaultTimeout = 200 * time.Millisecond
// MinTimeout and MaxTimeout bound the configured inspection timeout.
// Inspection is synchronous, so the upper bound is what keeps a
// mis-set value from parking every request to an inspected service on a
// slow engine; the lower bound keeps the call from timing out before the
// engine can realistically answer. Mirrors the per-middleware bounds the
// proxy already applies to in-path calls.
MinTimeout = 10 * time.Millisecond
MaxTimeout = 5 * time.Second
// DefaultMaxBodyBytes caps the request body mirrored to the engine.
// Requests with a larger body are inspected on headers and URI only.
DefaultMaxBodyBytes int64 = 64 << 10
// DefaultMaxConcurrent bounds inspections in flight toward the engine. The
// point is to fail fast instead of parking a goroutine per request for the
// whole timeout once the engine is saturated: a slow engine otherwise turns
// a traffic burst into a pile of waiters that all time out anyway. Sized so
// a healthy engine (single-digit milliseconds per call) never reaches it.
DefaultMaxConcurrent = 256
// MaxConcurrentLimit is the ceiling for that bound.
MaxConcurrentLimit = 4096
// MaxBodyBytesLimit is the ceiling for that cap. A single request can hold
// this much in memory; the shared Budget is what bounds the total across
// concurrent requests. Matches the proxy-wide body-capture ceiling.
MaxBodyBytesLimit int64 = 8 << 20
// maxResponseBytes bounds how much of a verdict response is read. The
// engine answers with a two-field JSON object, so anything beyond this is
// not a response we can act on.
maxResponseBytes int64 = 4 << 10
)
// Reasons the request body was not mirrored. Reported so an access-log reader
// can distinguish "inspected and clean" from "never inspected", and so an
// oversize opt-out is visible rather than silent.
const (
BypassOversize = "oversize"
BypassUpgrade = "upgrade"
BypassDisabled = "disabled"
BypassBudget = "budget_exhausted"
)
// ErrUnavailable reports that the engine could not produce a verdict: the call
// failed, timed out, or the engine rejected it (401 bad key, 500 malformed).
// Distinguished from a block verdict so the caller can apply the per-service
// mode: enforce fails closed, observe allows.
var ErrUnavailable = errors.New("appsec engine unavailable")
// Config configures a Client.
type Config struct {
// URL is the AppSec endpoint, e.g. http://127.0.0.1:7422/.
URL string
// APIKey is the CrowdSec bouncer API key. The AppSec component validates it
// against LAPI, so the same key used for the decision stream works here.
APIKey string
// Timeout bounds a single inspection call. Zero means DefaultTimeout.
Timeout time.Duration
// MaxBodyBytes caps the mirrored request body. Zero means
// DefaultMaxBodyBytes; negative disables body forwarding entirely.
MaxBodyBytes int64
// MaxConcurrent bounds inspections in flight toward the engine. Zero means
// DefaultMaxConcurrent; negative disables the bound.
MaxConcurrent int
// Budget bounds the total body buffering in flight across all inspected
// requests. Nil disables that ceiling, which leaves the worst case at
// MaxBodyBytes times the concurrent request count; callers serving
// untrusted traffic should share the proxy-wide capture budget here.
Budget Budget
Logger *log.Entry
}
// Budget is the shared allowance for in-flight body buffering. Acquire reports
// whether n bytes could be reserved; every successful Acquire is matched by a
// Release of the same n. Satisfied by the proxy's capture budget, so AppSec and
// the middleware body tap draw down one pool rather than two independent ones.
type Budget interface {
Acquire(n int64) bool
Release(n int64)
}
// Client mirrors HTTP requests to a CrowdSec AppSec endpoint. It holds no
// per-service state and is safe for concurrent use.
type Client struct {
url string
apiKey string
maxBodyBytes int64
// sem bounds in-flight inspections. Nil when the bound is disabled.
sem chan struct{}
budget Budget
http *http.Client
logger *log.Entry
}
// New validates the config and returns a Client. The endpoint is not contacted
// here: the engine may come up after the proxy.
func New(cfg Config) (*Client, error) {
if cfg.URL == "" {
return nil, errors.New("appsec url is empty")
}
if cfg.APIKey == "" {
return nil, errors.New("appsec api key is empty")
}
parsed, err := url.Parse(cfg.URL)
if err != nil {
return nil, fmt.Errorf("parse appsec url: %w", err)
}
if parsed.Scheme != "http" && parsed.Scheme != "https" {
return nil, fmt.Errorf("appsec url scheme %q is not http(s)", parsed.Scheme)
}
if parsed.Host == "" {
return nil, errors.New("appsec url has no host")
}
logger := cfg.Logger
if logger == nil {
logger = log.NewEntry(log.StandardLogger())
}
timeout := cfg.Timeout
switch {
case timeout <= 0:
timeout = DefaultTimeout
case timeout < MinTimeout:
logger.Warnf("appsec timeout %s is below the minimum, using %s", timeout, MinTimeout)
timeout = MinTimeout
case timeout > MaxTimeout:
logger.Warnf("appsec timeout %s exceeds the maximum, using %s", timeout, MaxTimeout)
timeout = MaxTimeout
}
// A negative cap is meaningful: forward no body at all.
maxBody := cfg.MaxBodyBytes
switch {
case maxBody == 0:
maxBody = DefaultMaxBodyBytes
case maxBody > MaxBodyBytesLimit:
logger.Warnf("appsec max body %d exceeds the maximum, using %d", maxBody, MaxBodyBytesLimit)
maxBody = MaxBodyBytesLimit
}
maxConcurrent := cfg.MaxConcurrent
switch {
case maxConcurrent == 0:
maxConcurrent = DefaultMaxConcurrent
case maxConcurrent > MaxConcurrentLimit:
logger.Warnf("appsec max concurrent %d exceeds the maximum, using %d", maxConcurrent, MaxConcurrentLimit)
maxConcurrent = MaxConcurrentLimit
}
var sem chan struct{}
if maxConcurrent > 0 {
sem = make(chan struct{}, maxConcurrent)
}
return &Client{
url: cfg.URL,
apiKey: cfg.APIKey,
maxBodyBytes: maxBody,
sem: sem,
budget: cfg.Budget,
logger: logger,
http: &http.Client{
Timeout: timeout,
Transport: &http.Transport{
MaxIdleConns: 100,
MaxIdleConnsPerHost: 32,
IdleConnTimeout: 90 * time.Second,
},
},
}, nil
}
// Request is one inspection request.
type Request struct {
// HTTP is the in-flight client request. Inspect buffers and restores its
// body, so the request stays forwardable afterwards.
HTTP *http.Request
// ClientIP is the resolved client address (after trusted-proxy handling).
ClientIP netip.Addr
// TransactionID correlates the engine's alert with the proxy's access log
// entry. Empty lets the engine generate its own UUID.
TransactionID string
// RedactBodyFields lists form fields whose values are replaced before the
// body is mirrored. Used to keep credentials submitted to the proxy's own
// login form out of the engine while still inspecting the rest.
RedactBodyFields []string
// RedactHeaders, RedactCookies and RedactQueryParams name the credentials
// the proxy already withholds from backends: the header-auth values, its
// session cookie, and the OIDC session token. The engine logs and alerts on
// what it inspects, so mirroring them there would reintroduce the leak the
// upstream strippers exist to prevent. Only the values are replaced, so the
// surrounding headers, cookies and query stay inspectable.
RedactHeaders []string
RedactCookies []string
RedactQueryParams []string
}
// Result is the outcome of an inspection.
type Result struct {
Verdict restrict.Verdict
// BodyBypass names why the request body was not mirrored, empty when it
// was (or when the request had none). The engine still saw the headers and
// URI, so this is a coverage note, not a failure.
BodyBypass string
// Release returns the buffered body's budget reservation. Never nil, so it
// is always safe to defer. It must run only once the request has been
// served, not when Inspect returns: the buffer stays alive as r.Body for
// the backend to read, so releasing earlier would let the budget admit
// buffering that is still resident.
Release func()
}
// noopRelease is the Release for inspections that reserved no budget.
func noopRelease() {}
// Inspect mirrors r to the AppSec engine and returns its verdict. A nil error
// with restrict.Allow means the request passed. On failure it returns
// DenyAppSecUnavailable wrapped with ErrUnavailable; the caller decides whether
// that blocks, based on the per-service mode.
func (c *Client) Inspect(ctx context.Context, req Request) (Result, error) {
if c == nil {
return Result{Verdict: restrict.DenyAppSecUnavailable, Release: noopRelease}, ErrUnavailable
}
if req.HTTP == nil {
return Result{Verdict: restrict.DenyAppSecUnavailable, Release: noopRelease}, fmt.Errorf("%w: nil request", ErrUnavailable)
}
// release is carried out to the caller rather than deferred here: the
// buffered body outlives this call as r.Body.
if !c.acquireSlot() {
// Deny rather than wave through: a flood must not be a way to switch
// inspection off. Enforce blocks, observe logs and allows, exactly as
// for an unreachable engine.
return Result{Verdict: restrict.DenyAppSecUnavailable, Release: noopRelease},
fmt.Errorf("%w: %d inspections already in flight", ErrUnavailable, cap(c.sem))
}
defer c.releaseSlot()
body, bypass, release, err := c.readBody(req)
if err != nil {
return Result{Verdict: restrict.DenyAppSecUnavailable, Release: release}, fmt.Errorf("%w: read body: %w", ErrUnavailable, err)
}
outbound, err := c.buildRequest(ctx, req, body)
if err != nil {
return Result{Verdict: restrict.DenyAppSecUnavailable, BodyBypass: bypass, Release: release}, fmt.Errorf("%w: %w", ErrUnavailable, err)
}
resp, err := c.http.Do(outbound)
if err != nil {
return Result{Verdict: restrict.DenyAppSecUnavailable, BodyBypass: bypass, Release: release}, fmt.Errorf("%w: %w", ErrUnavailable, err)
}
defer func() {
// Drain before closing. net/http only returns a connection to the idle
// pool once its body is read to EOF; closing with bytes outstanding
// discards it. Every verdict carries a JSON body, so skipping this
// would cost a fresh handshake per inspected request, inside the
// timeout budget.
if _, err := io.Copy(io.Discard, io.LimitReader(resp.Body, maxResponseBytes)); err != nil {
c.logger.Tracef("drain appsec response body: %v", err)
}
if err := resp.Body.Close(); err != nil {
c.logger.Tracef("close appsec response body: %v", err)
}
}()
verdict, err := c.verdict(resp)
return Result{Verdict: verdict, BodyBypass: bypass, Release: release}, err
}
// acquireSlot takes an in-flight slot without blocking, reporting false when
// the engine is already at capacity.
func (c *Client) acquireSlot() bool {
if c.sem == nil {
return true
}
select {
case c.sem <- struct{}{}:
return true
default:
return false
}
}
// releaseSlot returns the slot. Scoped to the engine call, not the request: the
// buffered body outlives the call but the engine's attention does not.
func (c *Client) releaseSlot() {
if c.sem == nil {
return
}
select {
case <-c.sem:
default:
}
}
// readBody buffers the body so it can be mirrored, always restoring it on the
// original request. Returns nil when there is no body to forward: no body at
// all, an upgrade request, or a body over the cap. A login form is forwarded
// with its credential values redacted rather than suppressed.
// release is never nil; the caller invokes it once the request has been served.
func (c *Client) readBody(req Request) (body []byte, bypass string, release func(), err error) {
r := req.HTTP
if r.Body == nil || r.Body == http.NoBody {
return nil, "", noopRelease, nil
}
if c.maxBodyBytes < 0 {
return nil, BypassDisabled, noopRelease, nil
}
// A genuine upgrade request carries no body to inspect (net/http hands us
// http.NoBody, caught above); the hijacked stream is reached through
// Hijacker, never r.Body. The test has to be the forwarder's own, because a
// looser one would skip inspection for requests the forwarder still
// delivers to the backend with their body intact.
if netutil.IsUpgradeRequest(r.Header) {
return nil, BypassUpgrade, noopRelease, nil
}
// A Content-Length over the cap is known to be too large before reading.
if r.ContentLength > c.maxBodyBytes {
return nil, BypassOversize, noopRelease, nil
}
// Reserve the whole cap rather than the eventual length: the reservation
// has to be made before the body is read, and until then the only bound
// known is the cap. Skipping inspection when the pool is drained keeps a
// burst of large bodies from being an out-of-memory lever; the bypass is
// recorded so the gap in coverage is visible.
release = noopRelease
if c.budget != nil {
if !c.budget.Acquire(c.maxBodyBytes) {
c.logger.Debugf("appsec buffer budget exhausted, inspecting headers and URI only")
return nil, BypassBudget, noopRelease, nil
}
var once sync.Once
release = func() { once.Do(func() { c.budget.Release(c.maxBodyBytes) }) }
}
buffered, oversize, err := bufferBody(r, c.maxBodyBytes)
if err != nil {
// bufferBody restored r.Body from the bytes it did read, so the
// reservation stays held until the caller releases it.
return nil, "", release, err
}
// An oversize body was only partially read: a truncated prefix changes the
// engine's verdict in both directions, so inspect headers and URI only.
if oversize {
return nil, BypassOversize, release, nil
}
return redactFormFields(r.Header.Get("Content-Type"), buffered, req.RedactBodyFields), "", release, nil
}
// buildRequest assembles the mirrored request. Per the protocol it is a GET
// when there is no body and a POST otherwise; bytes.Reader gives the outbound
// request an accurate Content-Length, which the engine relies on to read the
// body at all.
func (c *Client) buildRequest(ctx context.Context, req Request, body []byte) (*http.Request, error) {
method := http.MethodGet
var payload io.Reader
if len(body) > 0 {
method = http.MethodPost
payload = bytes.NewReader(body)
}
outbound, err := http.NewRequestWithContext(ctx, method, c.url, payload)
if err != nil {
return nil, fmt.Errorf("build appsec request: %w", err)
}
r := req.HTTP
copyInspectableHeaders(outbound.Header, r.Header)
redactSecrets(outbound.Header, req)
outbound.Header.Set(headerAPIKey, c.apiKey)
outbound.Header.Set(headerIP, req.ClientIP.Unmap().String())
outbound.Header.Set(headerURI, mirroredURI(r.URL, req.RedactQueryParams))
outbound.Header.Set(headerVerb, r.Method)
outbound.Header.Set(headerHost, r.Host)
if ua := r.UserAgent(); ua != "" {
outbound.Header.Set(headerUserAgent, ua)
}
outbound.Header.Set(headerHTTPVersion, httpVersion(r))
if req.TransactionID != "" {
outbound.Header.Set(headerTransactionID, req.TransactionID)
}
return outbound, nil
}
// verdict maps the engine's response to a restrict.Verdict. 200 is a pass and
// 401/500 are engine-side failures; every other status carries a remediation in
// the body. The blocked status code is operator-configurable
// (blocked_http_code), so the action field decides, not the status.
func (c *Client) verdict(resp *http.Response) (restrict.Verdict, error) {
switch resp.StatusCode {
case http.StatusUnauthorized:
return restrict.DenyAppSecUnavailable, fmt.Errorf("%w: rejected api key", ErrUnavailable)
case http.StatusInternalServerError:
return restrict.DenyAppSecUnavailable, fmt.Errorf("%w: engine error", ErrUnavailable)
}
// Every status, 200 included, has to carry a decodable remediation. Taking a
// bare 200 as a pass would mean a URL pointing at anything that answers 200
// (a health endpoint, a load balancer's default page) silently allows every
// request while the service reports itself as enforcing.
var decoded struct {
Action string `json:"action"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, maxResponseBytes)).Decode(&decoded); err != nil {
// Every remediation carries a decodable action, so a response without
// one is not a verdict: most often the URL points at something that is
// not the AppSec endpoint, which answers 404 with HTML. Reported as
// unavailable rather than a ban so the access log names the real fault
// instead of sending an operator hunting for a rule that never fired.
// Enforce still blocks either way; only the recorded reason differs.
return restrict.DenyAppSecUnavailable, fmt.Errorf("%w: undecodable response (status %d): %w", ErrUnavailable, resp.StatusCode, err)
}
switch decoded.Action {
case actionAllow:
return restrict.Allow, nil
case actionCaptcha:
return restrict.DenyAppSecCaptcha, nil
case actionBan:
return restrict.DenyAppSecBan, nil
case "":
// Decodable JSON without a remediation is not a verdict either: the
// endpoint answered, but not as the engine. Same reasoning as an
// undecodable body, and the same reason to point at configuration.
return restrict.DenyAppSecUnavailable,
fmt.Errorf("%w: response carried no remediation (status %d)", ErrUnavailable, resp.StatusCode)
default:
// A remediation we do not implement still means the engine flagged the
// request, so deny.
c.logger.Debugf("unknown appsec action %q (status %d), treating as ban", decoded.Action, resp.StatusCode)
return restrict.DenyAppSecBan, nil
}
}
// copyInspectableHeaders copies the client's headers, which are what the WAF
// rules actually match on, dropping hop-by-hop headers that describe the
// proxy-to-engine connection rather than the client request, and any header in
// the AppSec protocol namespace.
func copyInspectableHeaders(dst, src http.Header) {
for name, values := range src {
if hopByHopHeaders[http.CanonicalHeaderKey(name)] {
continue
}
if strings.HasPrefix(http.CanonicalHeaderKey(name), headerPrefix) {
continue
}
dst[http.CanonicalHeaderKey(name)] = append([]string(nil), values...)
}
// Content-Length describes the mirrored payload, not the client's: net/http
// sets it from the body we actually attach. Content-Type is kept either way
// so rules matching on it still fire when the body was not forwarded.
dst.Del("Content-Length")
}
// redactSecrets replaces the credential values the proxy withholds from
// backends, so the mirrored copy does not carry them either.
func redactSecrets(dst http.Header, req Request) {
for _, name := range req.RedactHeaders {
// Presence, not Get: a header whose first value is empty still carries
// its later values to the engine, while the upstream strip deletes the
// name outright. Set collapses every value into the placeholder.
if len(dst.Values(name)) > 0 {
dst.Set(name, redactedPlaceholder)
}
}
// Every Cookie line, not just the first: a client may send several, and Get
// would leave the session cookie in any later one mirrored in the clear.
if cookies := dst.Values("Cookie"); len(cookies) > 0 {
redacted := make([]string, len(cookies))
for i, cookie := range cookies {
redacted[i] = redactCookieHeader(cookie, req.RedactCookies)
}
dst["Cookie"] = redacted
}
}
// mirroredURI renders the request target for the URI header, with the named
// query parameter values replaced.
func mirroredURI(u *url.URL, redactParams []string) string {
uri := u.RequestURI()
if u.RawQuery == "" || len(redactParams) == 0 {
return uri
}
redacted := redactQuery(u.RawQuery, redactParams)
if redacted == u.RawQuery {
return uri
}
// RequestURI is path + "?" + RawQuery; swap only the query part so the
// path keeps its original encoding.
return strings.TrimSuffix(uri, u.RawQuery) + redacted
}
var hopByHopHeaders = map[string]bool{
"Connection": true,
"Keep-Alive": true,
"Proxy-Authenticate": true,
"Proxy-Authorization": true,
"Proxy-Connection": true,
"Te": true,
"Trailer": true,
"Transfer-Encoding": true,
"Upgrade": true,
}
// httpVersion renders the two-digit form the engine parses ("11", "20").
func httpVersion(r *http.Request) string {
major, minor := r.ProtoMajor, r.ProtoMinor
if major < 0 || major > 9 || minor < 0 || minor > 9 {
return ""
}
return fmt.Sprintf("%d%d", major, minor)
}

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,306 @@
package auth
import (
"crypto/ed25519"
"encoding/base64"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/proxy/internal/appsec"
"github.com/netbirdio/netbird/proxy/internal/proxy"
"github.com/netbirdio/netbird/proxy/internal/restrict"
"github.com/netbirdio/netbird/proxy/internal/types"
)
// appsecEngine is a stub AppSec component returning a fixed remediation.
func appsecEngine(t *testing.T, status int, body string) *httptest.Server {
t.Helper()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(status)
if body != "" {
_, _ = w.Write([]byte(body))
}
}))
t.Cleanup(srv.Close)
return srv
}
// serveWithAppSec runs a request through the middleware for a domain configured
// with the given AppSec mode, returning the response and the captured metadata.
func serveWithAppSec(t *testing.T, mode restrict.AppSecMode, client *appsec.Client, r *http.Request) (*httptest.ResponseRecorder, map[string]string, bool) {
t.Helper()
mw := NewMiddleware(log.StandardLogger(), nil, nil)
mw.SetAppSec(client)
require.NoError(t, mw.AddDomain("svc.example.com", DomainSettings{
AccountID: types.AccountID("acct-1"),
ServiceID: types.ServiceID("svc-1"),
AppSecMode: mode,
}))
reached := false
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
reached = true
w.WriteHeader(http.StatusOK)
}))
cd := proxy.NewCapturedData("req-1")
r = r.WithContext(proxy.WithCapturedData(r.Context(), cd))
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, r)
return rec, cd.GetMetadata(), reached
}
func appsecRequest() *http.Request {
r := httptest.NewRequest(http.MethodGet, "http://svc.example.com/?x=/etc/passwd", nil)
r.Host = "svc.example.com"
r.RemoteAddr = "203.0.113.7:44444"
return r
}
func TestCheckAppSec_EnforceBlocksBannedRequest(t *testing.T) {
srv := appsecEngine(t, http.StatusForbidden, `{"action":"ban","http_status":403}`)
client, err := appsec.New(appsec.Config{URL: srv.URL, APIKey: "k"})
require.NoError(t, err)
rec, meta, reached := serveWithAppSec(t, restrict.AppSecEnforce, client, appsecRequest())
assert.Equal(t, http.StatusForbidden, rec.Code)
assert.False(t, reached, "a banned request must not reach the backend")
assert.Equal(t, "appsec_ban", meta["appsec_verdict"])
assert.NotContains(t, meta, "appsec_mode", "enforce is the default, only observe is annotated")
}
func TestCheckAppSec_ObserveAllowsAndRecordsVerdict(t *testing.T) {
srv := appsecEngine(t, http.StatusForbidden, `{"action":"ban","http_status":403}`)
client, err := appsec.New(appsec.Config{URL: srv.URL, APIKey: "k"})
require.NoError(t, err)
rec, meta, reached := serveWithAppSec(t, restrict.AppSecObserve, client, appsecRequest())
assert.Equal(t, http.StatusOK, rec.Code)
assert.True(t, reached, "observe mode must not block")
assert.Equal(t, "appsec_ban", meta["appsec_verdict"])
assert.Equal(t, "observe", meta["appsec_mode"])
}
func TestCheckAppSec_AllowedRequestPasses(t *testing.T) {
srv := appsecEngine(t, http.StatusOK, `{"action":"allow","http_status":200}`)
client, err := appsec.New(appsec.Config{URL: srv.URL, APIKey: "k"})
require.NoError(t, err)
rec, meta, reached := serveWithAppSec(t, restrict.AppSecEnforce, client, appsecRequest())
assert.Equal(t, http.StatusOK, rec.Code)
assert.True(t, reached)
assert.NotContains(t, meta, "appsec_verdict", "a clean request records no verdict")
}
func TestCheckAppSec_OffSkipsInspection(t *testing.T) {
// An engine that would ban everything; the mode must keep us away from it.
srv := appsecEngine(t, http.StatusForbidden, `{"action":"ban"}`)
client, err := appsec.New(appsec.Config{URL: srv.URL, APIKey: "k"})
require.NoError(t, err)
rec, meta, reached := serveWithAppSec(t, restrict.AppSecOff, client, appsecRequest())
assert.Equal(t, http.StatusOK, rec.Code)
assert.True(t, reached)
assert.Empty(t, meta)
}
func TestCheckAppSec_EnforceFailsClosedWithoutClient(t *testing.T) {
rec, meta, reached := serveWithAppSec(t, restrict.AppSecEnforce, nil, appsecRequest())
assert.Equal(t, http.StatusForbidden, rec.Code,
"enforce with no configured endpoint must deny rather than pass traffic uninspected")
assert.False(t, reached)
assert.Equal(t, "appsec_unavailable", meta["appsec_verdict"])
}
func TestCheckAppSec_ObserveAllowsWithoutClient(t *testing.T) {
rec, meta, reached := serveWithAppSec(t, restrict.AppSecObserve, nil, appsecRequest())
assert.Equal(t, http.StatusOK, rec.Code)
assert.True(t, reached)
assert.Equal(t, "appsec_unavailable", meta["appsec_verdict"])
assert.Equal(t, "observe", meta["appsec_mode"])
}
func TestCheckAppSec_EnforceFailsClosedWhenEngineUnreachable(t *testing.T) {
client, err := appsec.New(appsec.Config{URL: "http://127.0.0.1:1/", APIKey: "k"})
require.NoError(t, err)
rec, meta, reached := serveWithAppSec(t, restrict.AppSecEnforce, client, appsecRequest())
assert.Equal(t, http.StatusForbidden, rec.Code)
assert.False(t, reached)
assert.Equal(t, "appsec_unavailable", meta["appsec_verdict"])
}
func TestCheckAppSec_InspectsOverlayTraffic(t *testing.T) {
srv := appsecEngine(t, http.StatusForbidden, `{"action":"ban"}`)
client, err := appsec.New(appsec.Config{URL: srv.URL, APIKey: "k"})
require.NoError(t, err)
// Requests arriving over the WireGuard overlay skip the geo and IP-reputation
// checks, but request content is just as inspectable.
r := appsecRequest()
r = r.WithContext(types.WithOverlayOrigin(r.Context()))
rec, meta, reached := serveWithAppSec(t, restrict.AppSecEnforce, client, r)
assert.Equal(t, http.StatusForbidden, rec.Code, "overlay traffic must still be inspected")
assert.False(t, reached)
assert.Equal(t, "appsec_ban", meta["appsec_verdict"])
}
func TestCheckAppSec_UnresolvableClientIPFailsClosed(t *testing.T) {
srv := appsecEngine(t, http.StatusOK, `{"action":"allow"}`)
client, err := appsec.New(appsec.Config{URL: srv.URL, APIKey: "k"})
require.NoError(t, err)
r := appsecRequest()
r.RemoteAddr = "not-an-address"
rec, meta, reached := serveWithAppSec(t, restrict.AppSecEnforce, client, r)
assert.Equal(t, http.StatusForbidden, rec.Code,
"the engine requires a client address; a request we cannot attribute must not pass")
assert.False(t, reached)
assert.Equal(t, "appsec_unavailable", meta["appsec_verdict"])
}
// The redaction sets are resolved from the domain's schemes at registration, so
// what AppSec withholds cannot drift from what those schemes actually accept.
func TestAddDomain_ResolvesRedactionSetsFromSchemes(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
require.NoError(t, mw.AddDomain("svc.example.com", DomainSettings{
Schemes: []Scheme{
NewPassword(nil, "svc-1", "acct-1"),
NewHeader(nil, "svc-1", "acct-1", "X-Api-Key"),
},
SessionPublicKey: base64.StdEncoding.EncodeToString(make([]byte, ed25519.PublicKeySize)),
SessionExpiration: time.Hour,
AppSecMode: restrict.AppSecEnforce,
}))
mw.domainsMux.RLock()
config := mw.domains["svc.example.com"]
mw.domainsMux.RUnlock()
assert.Equal(t, []string{"password"}, config.redactBodyFields)
assert.Equal(t, []string{"X-Api-Key"}, config.redactHeaders)
// r.FormValue merges the query into the form, so a credential passed there
// authenticates and must be redacted alongside the OIDC session token.
assert.Equal(t, []string{"session_token", "password"}, config.redactQueryParams)
}
// countingBudget records reservations so a test can observe when the
// middleware hands them back.
type countingBudget struct {
mu sync.Mutex
total int64
used int64
maxAtOnce int64
}
func (b *countingBudget) Acquire(n int64) bool {
b.mu.Lock()
defer b.mu.Unlock()
if b.used+n > b.total {
return false
}
b.used += n
if b.used > b.maxAtOnce {
b.maxAtOnce = b.used
}
return true
}
func (b *countingBudget) Release(n int64) {
b.mu.Lock()
defer b.mu.Unlock()
b.used -= n
}
func (b *countingBudget) inUse() int64 {
b.mu.Lock()
defer b.mu.Unlock()
return b.used
}
// The buffered body stays alive as r.Body until the backend has read it, so
// Protect must hold the reservation for the whole request and return it only
// once the handler chain has unwound. Releasing inside Inspect would let the
// budget admit buffering that is still resident.
func TestProtect_AppSecBudgetHeldForRequestAndReleasedAfter(t *testing.T) {
srv := appsecEngine(t, http.StatusOK, `{"action":"allow"}`)
budget := &countingBudget{total: 1 << 20}
client, err := appsec.New(appsec.Config{
URL: srv.URL,
APIKey: "k",
MaxBodyBytes: 4096,
Budget: budget,
})
require.NoError(t, err)
mw := NewMiddleware(log.StandardLogger(), nil, nil)
mw.SetAppSec(client)
require.NoError(t, mw.AddDomain("svc.example.com", DomainSettings{
AccountID: types.AccountID("acct-1"),
ServiceID: types.ServiceID("svc-1"),
AppSecMode: restrict.AppSecEnforce,
}))
var inHandler int64
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// The backend reads the buffered body here, so the reservation must
// still be held at this point.
inHandler = budget.inUse()
_, _ = io.ReadAll(r.Body)
w.WriteHeader(http.StatusOK)
}))
r := httptest.NewRequest(http.MethodPost, "http://svc.example.com/", strings.NewReader("payload"))
r.Host = "svc.example.com"
cd := proxy.NewCapturedData("req-1")
r = r.WithContext(proxy.WithCapturedData(r.Context(), cd))
handler.ServeHTTP(httptest.NewRecorder(), r)
assert.Equal(t, int64(4096), inHandler, "the reservation must be held while the backend reads the body")
assert.Equal(t, int64(0), budget.inUse(), "Protect must release the reservation once the request is served")
}
// A denied request never reaches the backend, but Protect still has to hand the
// reservation back or the pool leaks one cap per blocked request.
func TestProtect_AppSecBudgetReleasedOnDeny(t *testing.T) {
srv := appsecEngine(t, http.StatusForbidden, `{"action":"ban"}`)
budget := &countingBudget{total: 1 << 20}
client, err := appsec.New(appsec.Config{
URL: srv.URL,
APIKey: "k",
MaxBodyBytes: 4096,
Budget: budget,
})
require.NoError(t, err)
r := httptest.NewRequest(http.MethodPost, "http://svc.example.com/", strings.NewReader("payload"))
r.Host = "svc.example.com"
rec, _, reached := serveWithAppSec(t, restrict.AppSecEnforce, client, r)
assert.False(t, reached, "a banned request must not reach the backend")
assert.Equal(t, http.StatusForbidden, rec.Code)
assert.Equal(t, int64(0), budget.inUse(), "a blocked request must not leak its reservation")
}

View File

@@ -39,6 +39,11 @@ func (Header) Type() auth.Method {
return auth.MethodHeader
}
// HeaderName returns the request header this scheme reads its credential from.
func (h Header) HeaderName() string {
return h.headerName
}
// Authenticate checks for the configured header in the request. If absent,
// returns empty (unauthenticated). If present, validates via gRPC.
func (h Header) Authenticate(r *http.Request) (string, string, error) {

View File

@@ -18,6 +18,7 @@ import (
"google.golang.org/grpc"
"github.com/netbirdio/netbird/proxy/auth"
"github.com/netbirdio/netbird/proxy/internal/appsec"
"github.com/netbirdio/netbird/proxy/internal/proxy"
"github.com/netbirdio/netbird/proxy/internal/restrict"
"github.com/netbirdio/netbird/proxy/internal/types"
@@ -25,6 +26,11 @@ import (
"github.com/netbirdio/netbird/shared/management/proto"
)
// sessionCookieNames is the cookie set AppSec redacts before mirroring: the
// proxy's session cookie is a bearer credential for the service, and the
// reverse proxy already strips it before forwarding upstream.
var sessionCookieNames = []string{auth.SessionCookieName}
// errValidationUnavailable indicates that session validation failed due to
// an infrastructure error (e.g. gRPC unavailable), not an invalid token.
var errValidationUnavailable = errors.New("session validation unavailable")
@@ -59,6 +65,14 @@ type DomainConfig struct {
IPRestrictions *restrict.Filter
// Private routes the domain through ValidateTunnelPeer; failure → 403.
Private bool
// AppSecMode enables CrowdSec AppSec request inspection for this domain.
AppSecMode restrict.AppSecMode
// redact* name the credentials this domain's schemes accept, resolved once
// at registration. AppSec replaces their values before mirroring a request,
// matching what the reverse proxy strips before forwarding upstream.
redactBodyFields []string
redactHeaders []string
redactQueryParams []string
}
type validationResult struct {
@@ -82,6 +96,9 @@ type Middleware struct {
sessionValidator SessionValidator
geo restrict.GeoResolver
tunnelCache *tunnelValidationCache
// 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
}
// NewMiddleware creates a new authentication middleware. The sessionValidator is
@@ -99,6 +116,12 @@ func NewMiddleware(logger *log.Logger, sessionValidator SessionValidator, geo re
}
}
// SetAppSec installs the shared CrowdSec AppSec client. Must be called during
// startup, before the middleware serves any request.
func (mw *Middleware) SetAppSec(client *appsec.Client) {
mw.appsec = client
}
// Protect wraps next with per-domain authentication and IP restriction checks.
// Requests whose Host is not registered pass through unchanged.
func (mw *Middleware) Protect(next http.Handler) http.Handler {
@@ -123,6 +146,14 @@ func (mw *Middleware) Protect(next http.Handler) http.Handler {
return
}
// Deferred, not released here: the inspected body stays alive as r.Body
// until the backend has read it, which happens inside next.ServeHTTP.
appSecAllowed, releaseAppSec := mw.checkAppSec(w, r, config)
defer releaseAppSec()
if !appSecAllowed {
return
}
// Private services bypass operator schemes and gate on tunnel peer.
if config.Private {
if mw.forwardWithTunnelPeer(w, r, host, config, next) {
@@ -262,6 +293,134 @@ func (mw *Middleware) checkIPRestrictions(w http.ResponseWriter, r *http.Request
return false
}
// checkAppSec mirrors the request to the CrowdSec AppSec engine when the domain
// enables inspection. Returns false when the request was blocked and a response
// has been written.
//
// The returned release frees the body-buffering budget the inspection reserved
// and is never nil. It must run only after the request has been served, since
// the buffered body stays alive as r.Body for the backend to read.
//
// Every non-allow remediation blocks with 403, captcha included: the proxy has
// no challenge flow to serve. The distinct verdict is still recorded so the
// access log shows which remediation the engine actually chose.
//
// Unlike the geo and IP-reputation checks, this runs for overlay traffic too:
// AppSec inspects request content, which is just as meaningful when the caller
// reached the proxy through the WireGuard tunnel.
func (mw *Middleware) checkAppSec(w http.ResponseWriter, r *http.Request, config DomainConfig) (bool, func()) {
if !config.AppSecMode.Enabled() {
return true, func() {}
}
verdict, release := mw.inspectAppSec(r, config)
if verdict == restrict.Allow {
return true, release
}
observe := config.AppSecMode == restrict.AppSecObserve
if cd := proxy.CapturedDataFromContext(r.Context()); cd != nil {
cd.SetMetadata("appsec_verdict", verdict.String())
if observe {
cd.SetMetadata("appsec_mode", "observe")
}
}
if observe {
mw.logger.Debugf("AppSec observe: would block %s for %s (%s)", r.RemoteAddr, r.Host, verdict)
return true, release
}
mw.markDenied(r, verdict.String())
mw.logger.Debugf("AppSec: %s for %s %s", verdict, r.Host, r.RemoteAddr)
http.Error(w, "Forbidden", http.StatusForbidden)
return false, release
}
// inspectAppSec runs the AppSec call and returns its verdict. Failures come
// back as DenyAppSecUnavailable regardless of mode so observe mode still
// records that inspection did not happen; the caller decides what blocks. The
// returned release is never nil.
func (mw *Middleware) inspectAppSec(r *http.Request, config DomainConfig) (restrict.Verdict, func()) {
// Mode requested but the proxy has no AppSec endpoint configured. Management
// gates this on the cluster capability; a stale mapping can still arrive.
if mw.appsec == nil {
mw.logger.Debugf("AppSec mode %q requested for %s but no AppSec endpoint is configured", config.AppSecMode, r.Host)
return restrict.DenyAppSecUnavailable, func() {}
}
clientIP := mw.resolveClientIP(r)
if !clientIP.IsValid() {
// The engine requires a client address, and a request whose source we
// cannot establish is exactly the kind we must not wave through.
mw.logger.Debugf("AppSec: cannot resolve client address for %q", r.RemoteAddr)
return restrict.DenyAppSecUnavailable, func() {}
}
req := appsec.Request{
HTTP: r,
ClientIP: clientIP,
RedactBodyFields: config.redactBodyFields,
RedactHeaders: config.redactHeaders,
RedactCookies: sessionCookieNames,
RedactQueryParams: config.redactQueryParams,
}
if cd := proxy.CapturedDataFromContext(r.Context()); cd != nil {
req.TransactionID = cd.GetRequestID()
}
result, err := mw.appsec.Inspect(r.Context(), req)
if err != nil {
mw.logger.Debugf("AppSec inspection failed for %s: %v", r.Host, err)
}
// Record when the body went uninspected: headers and URI were still
// checked, but an operator reading the log should not read a clean verdict
// as "the payload was examined". Oversize is reachable by padding, so its
// absence from the log would hide a deliberate opt-out.
if result.BodyBypass != "" {
if cd := proxy.CapturedDataFromContext(r.Context()); cd != nil {
cd.SetMetadata("appsec_body_bypass", result.BodyBypass)
}
}
return result.Verdict, result.Release
}
// credentialFormFields lists the login form fields whose values are redacted
// from the mirrored body, so a password or PIN submitted to the proxy's own
// login form never reaches the Security Engine.
func credentialFormFields(schemes []Scheme) []string {
var fields []string
for _, s := range schemes {
switch s.Type() {
case auth.MethodPassword:
fields = append(fields, passwordFormId)
case auth.MethodPIN:
fields = append(fields, pinFormId)
}
}
return fields
}
// credentialHeaders lists the request headers whose values are redacted from
// the mirrored request. A header-auth scheme carries a session token the proxy
// validates and never forwards upstream, so the engine must not see it either.
func credentialHeaders(schemes []Scheme) []string {
var names []string
for _, s := range schemes {
// Structural, not a concrete Header assertion: if the scheme is ever
// registered as a pointer, a type assertion would quietly stop matching
// and the header would start reaching the engine again.
named, ok := s.(interface{ HeaderName() string })
if !ok {
continue
}
if name := named.HeaderName(); name != "" {
names = append(names, name)
}
}
return names
}
// 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 {
@@ -281,12 +440,18 @@ func (mw *Middleware) resolveClientIP(r *http.Request) netip.Addr {
return addr.Unmap()
}
// blockIPRestriction sets captured data fields for an IP-restriction block event.
func (mw *Middleware) blockIPRestriction(r *http.Request, reason string) {
// markDenied records the deny reason on the captured data so the access log
// attributes the response to the proxy rather than the backend.
func (mw *Middleware) markDenied(r *http.Request, reason string) {
if cd := proxy.CapturedDataFromContext(r.Context()); cd != nil {
cd.SetOrigin(proxy.OriginAuth)
cd.SetAuthMethod(reason)
}
}
// blockIPRestriction sets captured data fields for an IP-restriction block event.
func (mw *Middleware) blockIPRestriction(r *http.Request, reason string) {
mw.markDenied(r, reason)
mw.logger.Debugf("IP restriction: %s for %s", reason, r.RemoteAddr)
}
@@ -637,45 +802,61 @@ func wasCredentialSubmitted(r *http.Request, method auth.Method) bool {
case auth.MethodPassword:
return r.FormValue("password") != ""
case auth.MethodOIDC:
return r.URL.Query().Get("session_token") != ""
return r.URL.Query().Get(sessionTokenParam) != ""
}
return false
}
// AddDomain registers authentication schemes for the given domain. With schemes a valid session public key is required.
// private=true forces ValidateTunnelPeer enforcement (403 on failure) regardless of the schemes list.
func (mw *Middleware) AddDomain(domain string, schemes []Scheme, publicKeyB64 string, expiration time.Duration, accountID types.AccountID, serviceID types.ServiceID, ipRestrictions *restrict.Filter, private bool) error {
if len(schemes) == 0 {
mw.domainsMux.Lock()
defer mw.domainsMux.Unlock()
mw.domains[domain] = DomainConfig{
AccountID: accountID,
ServiceID: serviceID,
IPRestrictions: ipRestrictions,
Private: private,
}
return nil
// DomainSettings is the per-domain configuration AddDomain applies.
type DomainSettings struct {
Schemes []Scheme
// SessionPublicKey is the base64-encoded ed25519 key used to verify session
// cookies. Required when Schemes is non-empty.
SessionPublicKey string
SessionExpiration time.Duration
AccountID types.AccountID
ServiceID types.ServiceID
IPRestrictions *restrict.Filter
// Private forces ValidateTunnelPeer enforcement (403 on failure) regardless
// of the schemes list.
Private bool
AppSecMode restrict.AppSecMode
}
// AddDomain registers authentication schemes for the given domain. With schemes
// a valid session public key is required.
func (mw *Middleware) AddDomain(domain string, settings DomainSettings) error {
credentialFields := credentialFormFields(settings.Schemes)
config := DomainConfig{
AccountID: settings.AccountID,
ServiceID: settings.ServiceID,
IPRestrictions: settings.IPRestrictions,
Private: settings.Private,
AppSecMode: settings.AppSecMode,
redactBodyFields: credentialFields,
redactHeaders: credentialHeaders(settings.Schemes),
// A credential can arrive in the query too: r.FormValue merges the URL
// query into the form, so "?password=..." authenticates just as a form
// post does and must not be mirrored in the clear either.
redactQueryParams: append([]string{sessionTokenParam}, credentialFields...),
}
pubKeyBytes, err := base64.StdEncoding.DecodeString(publicKeyB64)
if err != nil {
return fmt.Errorf("decode session public key for domain %s: %w", domain, err)
}
if len(pubKeyBytes) != ed25519.PublicKeySize {
return fmt.Errorf("invalid session public key size for domain %s: got %d, want %d", domain, len(pubKeyBytes), ed25519.PublicKeySize)
if len(settings.Schemes) > 0 {
pubKeyBytes, err := base64.StdEncoding.DecodeString(settings.SessionPublicKey)
if err != nil {
return fmt.Errorf("decode session public key for domain %s: %w", domain, err)
}
if len(pubKeyBytes) != ed25519.PublicKeySize {
return fmt.Errorf("invalid session public key size for domain %s: got %d, want %d", domain, len(pubKeyBytes), ed25519.PublicKeySize)
}
config.Schemes = settings.Schemes
config.SessionPublicKey = pubKeyBytes
config.SessionExpiration = settings.SessionExpiration
}
mw.domainsMux.Lock()
defer mw.domainsMux.Unlock()
mw.domains[domain] = DomainConfig{
Schemes: schemes,
SessionPublicKey: pubKeyBytes,
SessionExpiration: expiration,
AccountID: accountID,
ServiceID: serviceID,
IPRestrictions: ipRestrictions,
Private: private,
}
mw.domains[domain] = config
return nil
}
@@ -730,10 +911,10 @@ func (mw *Middleware) validateSessionToken(ctx context.Context, host, token stri
// parameter removed so it doesn't linger in the browser's address bar or history.
func stripSessionTokenParam(u *url.URL) string {
q := u.Query()
if !q.Has("session_token") {
if !q.Has(sessionTokenParam) {
return u.RequestURI()
}
q.Del("session_token")
q.Del(sessionTokenParam)
clean := *u
clean.RawQuery = q.Encode()
return clean.RequestURI()

View File

@@ -64,7 +64,7 @@ func TestAddDomain_ValidKey(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
err := mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false)
err := mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour})
require.NoError(t, err)
mw.domainsMux.RLock()
@@ -81,7 +81,7 @@ func TestAddDomain_EmptyKey(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
err := mw.AddDomain("example.com", []Scheme{scheme}, "", time.Hour, "", "", nil, false)
err := mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionExpiration: time.Hour})
require.Error(t, err)
assert.Contains(t, err.Error(), "invalid session public key size")
@@ -95,7 +95,7 @@ func TestAddDomain_InvalidBase64(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
err := mw.AddDomain("example.com", []Scheme{scheme}, "not-valid-base64!!!", time.Hour, "", "", nil, false)
err := mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: "not-valid-base64!!!", SessionExpiration: time.Hour})
require.Error(t, err)
assert.Contains(t, err.Error(), "decode session public key")
@@ -110,7 +110,7 @@ func TestAddDomain_WrongKeySize(t *testing.T) {
shortKey := base64.StdEncoding.EncodeToString([]byte("tooshort"))
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
err := mw.AddDomain("example.com", []Scheme{scheme}, shortKey, time.Hour, "", "", nil, false)
err := mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: shortKey, SessionExpiration: time.Hour})
require.Error(t, err)
assert.Contains(t, err.Error(), "invalid session public key size")
@@ -123,7 +123,7 @@ func TestAddDomain_WrongKeySize(t *testing.T) {
func TestAddDomain_NoSchemes_NoKeyRequired(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
err := mw.AddDomain("example.com", nil, "", time.Hour, "", "", nil, false)
err := mw.AddDomain("example.com", DomainSettings{SessionExpiration: time.Hour})
require.NoError(t, err, "domains with no auth schemes should not require a key")
mw.domainsMux.RLock()
@@ -139,8 +139,8 @@ func TestAddDomain_OverwritesPreviousConfig(t *testing.T) {
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp1.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp2.PublicKey, 2*time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp1.PublicKey, SessionExpiration: time.Hour}))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp2.PublicKey, SessionExpiration: 2 * time.Hour}))
mw.domainsMux.RLock()
config := mw.domains["example.com"]
@@ -156,7 +156,7 @@ func TestRemoveDomain(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
mw.RemoveDomain("example.com")
@@ -180,7 +180,7 @@ func TestProtect_UnknownDomainPassesThrough(t *testing.T) {
func TestProtect_DomainWithNoSchemesPassesThrough(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
require.NoError(t, mw.AddDomain("example.com", nil, "", time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -197,7 +197,7 @@ func TestProtect_UnauthenticatedRequestIsBlocked(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
var backendCalled bool
backend := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -218,7 +218,7 @@ func TestProtect_HostWithPortIsMatched(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
var backendCalled bool
backend := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -239,7 +239,7 @@ func TestProtect_ValidSessionCookiePassesThrough(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
token, err := sessionkey.SignToken(kp.PrivateKey, "test-user", "", "example.com", auth.MethodPIN, nil, nil, time.Hour)
require.NoError(t, err)
@@ -272,7 +272,7 @@ func TestProtect_SessionCookieGroupsPropagate(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
groups := []string{"engineering", "sre"}
token, err := sessionkey.SignToken(kp.PrivateKey, "test-user", "", "example.com", auth.MethodPIN, groups, nil, time.Hour)
@@ -337,7 +337,7 @@ func TestProtect_PrivateService_TunnelPeerGroupsPropagate(t *testing.T) {
kp := generateTestKeyPair(t)
// Private service: no operator schemes — auth gates solely on the tunnel peer.
require.NoError(t, mw.AddDomain("agent.example.com", nil, kp.PublicKey, time.Hour, "acct-1", "svc-1", nil, true))
require.NoError(t, mw.AddDomain("agent.example.com", DomainSettings{SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acct-1", ServiceID: "svc-1", Private: true}))
cd := proxy.NewCapturedData("")
cd.SetClientIP(netip.MustParseAddr("100.90.1.14")) // CGNAT tunnel source
@@ -377,7 +377,7 @@ func TestProtect_PrivateService_TunnelPeerDenied(t *testing.T) {
}}
mw := NewMiddleware(log.StandardLogger(), validator, nil)
kp := generateTestKeyPair(t)
require.NoError(t, mw.AddDomain("agent.example.com", nil, kp.PublicKey, time.Hour, "acct-1", "svc-1", nil, true))
require.NoError(t, mw.AddDomain("agent.example.com", DomainSettings{SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acct-1", ServiceID: "svc-1", Private: true}))
cd := proxy.NewCapturedData("")
cd.SetClientIP(netip.MustParseAddr("100.90.1.14"))
@@ -405,7 +405,7 @@ func TestProtect_ExpiredSessionCookieIsRejected(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
// Sign a token that expired 1 second ago.
token, err := sessionkey.SignToken(kp.PrivateKey, "test-user", "", "example.com", auth.MethodPIN, nil, nil, -time.Second)
@@ -431,7 +431,7 @@ func TestProtect_WrongDomainCookieIsRejected(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
// Token signed for a different domain audience.
token, err := sessionkey.SignToken(kp.PrivateKey, "test-user", "", "other.com", auth.MethodPIN, nil, nil, time.Hour)
@@ -458,7 +458,7 @@ func TestProtect_WrongKeyCookieIsRejected(t *testing.T) {
kp2 := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp1.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp1.PublicKey, SessionExpiration: time.Hour}))
// Token signed with a different private key.
token, err := sessionkey.SignToken(kp2.PrivateKey, "test-user", "", "example.com", auth.MethodPIN, nil, nil, time.Hour)
@@ -495,7 +495,7 @@ func TestProtect_SchemeAuthRedirectsWithCookie(t *testing.T) {
return "", "pin", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
var backendCalled bool
backend := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -548,7 +548,7 @@ func TestProtect_FailedAuthDoesNotSetCookie(t *testing.T) {
return "", "pin", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -584,7 +584,7 @@ func TestProtect_MultipleSchemes(t *testing.T) {
return "", "password", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{pinScheme, passwordScheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{pinScheme, passwordScheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
var backendCalled bool
backend := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -614,7 +614,7 @@ func TestProtect_InvalidTokenFromSchemeReturns400(t *testing.T) {
return "invalid-jwt-token", "", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -638,7 +638,7 @@ func TestAddDomain_RandomBytes32NotEd25519(t *testing.T) {
key := base64.StdEncoding.EncodeToString(randomBytes)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
err = mw.AddDomain("example.com", []Scheme{scheme}, key, time.Hour, "", "", nil, false)
err = mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: key, SessionExpiration: time.Hour})
require.NoError(t, err, "any 32-byte key should be accepted at registration time")
}
@@ -647,10 +647,10 @@ func TestAddDomain_InvalidKeyDoesNotCorruptExistingConfig(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
// Attempt to overwrite with an invalid key.
err := mw.AddDomain("example.com", []Scheme{scheme}, "bad", time.Hour, "", "", nil, false)
err := mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: "bad", SessionExpiration: time.Hour})
require.Error(t, err)
// The original valid config should still be intact.
@@ -674,7 +674,7 @@ func TestProtect_FailedPinAuthCapturesAuthMethod(t *testing.T) {
return "", "pin", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
capturedData := proxy.NewCapturedData("")
handler := mw.Protect(newPassthroughHandler())
@@ -701,7 +701,7 @@ func TestProtect_FailedPasswordAuthCapturesAuthMethod(t *testing.T) {
return "", "password", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
capturedData := proxy.NewCapturedData("")
handler := mw.Protect(newPassthroughHandler())
@@ -728,7 +728,7 @@ func TestProtect_NoCredentialsDoesNotCaptureAuthMethod(t *testing.T) {
return "", "pin", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
capturedData := proxy.NewCapturedData("")
handler := mw.Protect(newPassthroughHandler())
@@ -815,8 +815,7 @@ func TestWasCredentialSubmitted(t *testing.T) {
func TestCheckIPRestrictions_UnparseableAddress(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
err := mw.AddDomain("example.com", nil, "", 0, "acc1", "svc1",
restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"10.0.0.0/8"}}), false)
err := mw.AddDomain("example.com", DomainSettings{AccountID: "acc1", ServiceID: "svc1", IPRestrictions: restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"10.0.0.0/8"}})})
require.NoError(t, err)
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -851,8 +850,7 @@ func TestCheckIPRestrictions_UsesCapturedDataClientIP(t *testing.T) {
// trusted proxies), checkIPRestrictions should use that IP, not RemoteAddr.
mw := NewMiddleware(log.StandardLogger(), nil, nil)
err := mw.AddDomain("example.com", nil, "", 0, "acc1", "svc1",
restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"203.0.113.0/24"}}), false)
err := mw.AddDomain("example.com", DomainSettings{AccountID: "acc1", ServiceID: "svc1", IPRestrictions: restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"203.0.113.0/24"}})})
require.NoError(t, err)
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -892,8 +890,7 @@ func TestCheckIPRestrictions_NilGeoWithCountryRules(t *testing.T) {
// Geo is nil, country restrictions are configured: must deny (fail-close).
mw := NewMiddleware(log.StandardLogger(), nil, nil)
err := mw.AddDomain("example.com", nil, "", 0, "acc1", "svc1",
restrict.ParseFilter(restrict.FilterConfig{AllowedCountries: []string{"US"}}), false)
err := mw.AddDomain("example.com", DomainSettings{AccountID: "acc1", ServiceID: "svc1", IPRestrictions: restrict.ParseFilter(restrict.FilterConfig{AllowedCountries: []string{"US"}})})
require.NoError(t, err)
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -916,11 +913,10 @@ func TestCheckIPRestrictions_NilGeoWithCountryRules(t *testing.T) {
func TestCheckIPRestrictions_OverlayOriginSkipsCountryRules(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
err := mw.AddDomain("example.com", nil, "", 0, "acc1", "svc1",
restrict.ParseFilter(restrict.FilterConfig{
AllowedCIDRs: []string{"100.64.0.0/10"},
AllowedCountries: []string{"US"},
}), false)
err := mw.AddDomain("example.com", DomainSettings{AccountID: "acc1", ServiceID: "svc1", IPRestrictions: restrict.ParseFilter(restrict.FilterConfig{
AllowedCIDRs: []string{"100.64.0.0/10"},
AllowedCountries: []string{"US"},
})})
require.NoError(t, err)
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -953,8 +949,7 @@ func TestCheckIPRestrictions_OverlayOriginSkipsCountryRules(t *testing.T) {
func TestCheckIPRestrictions_OverlayOriginRespectsCIDR(t *testing.T) {
mw := NewMiddleware(log.StandardLogger(), nil, nil)
err := mw.AddDomain("example.com", nil, "", 0, "acc1", "svc1",
restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"100.64.0.0/16"}}), false)
err := mw.AddDomain("example.com", DomainSettings{AccountID: "acc1", ServiceID: "svc1", IPRestrictions: restrict.ParseFilter(restrict.FilterConfig{AllowedCIDRs: []string{"100.64.0.0/16"}})})
require.NoError(t, err)
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -982,7 +977,7 @@ func TestProtect_OIDCOnlyRedirectsDirectly(t *testing.T) {
return "", oidcURL, nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -1011,7 +1006,7 @@ func TestProtect_OIDCWithOtherMethodShowsLoginPage(t *testing.T) {
return "", "pin", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{oidcScheme, pinScheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{oidcScheme, pinScheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -1055,7 +1050,7 @@ func TestProtect_HeaderAuth_ForwardsOnSuccess(t *testing.T) {
kp := generateTestKeyPair(t)
hdr := newHeaderSchemeWithToken(t, kp, "X-API-Key", "secret-key")
require.NoError(t, mw.AddDomain("example.com", []Scheme{hdr}, kp.PublicKey, time.Hour, "acc1", "svc1", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{hdr}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acc1", ServiceID: "svc1"}))
var backendCalled bool
capturedData := proxy.NewCapturedData("")
@@ -1098,7 +1093,7 @@ func TestProtect_HeaderAuth_MissingHeaderFallsThrough(t *testing.T) {
hdr := newHeaderSchemeWithToken(t, kp, "X-API-Key", "secret-key")
// Also add a PIN scheme so we can verify fallthrough behavior.
pinScheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{hdr, pinScheme}, kp.PublicKey, time.Hour, "acc1", "svc1", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{hdr, pinScheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acc1", ServiceID: "svc1"}))
handler := mw.Protect(newPassthroughHandler())
@@ -1118,7 +1113,7 @@ func TestProtect_HeaderAuth_WrongValueReturns401(t *testing.T) {
return &proto.AuthenticateResponse{Success: false}, nil
}}
hdr := NewHeader(mock, "svc1", "acc1", "X-API-Key")
require.NoError(t, mw.AddDomain("example.com", []Scheme{hdr}, kp.PublicKey, time.Hour, "acc1", "svc1", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{hdr}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acc1", ServiceID: "svc1"}))
capturedData := proxy.NewCapturedData("")
handler := mw.Protect(newPassthroughHandler())
@@ -1141,7 +1136,7 @@ func TestProtect_HeaderAuth_InfraErrorReturns502(t *testing.T) {
return nil, errors.New("gRPC unavailable")
}}
hdr := NewHeader(mock, "svc1", "acc1", "X-API-Key")
require.NoError(t, mw.AddDomain("example.com", []Scheme{hdr}, kp.PublicKey, time.Hour, "acc1", "svc1", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{hdr}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acc1", ServiceID: "svc1"}))
handler := mw.Protect(newPassthroughHandler())
@@ -1158,7 +1153,7 @@ func TestProtect_HeaderAuth_SubsequentRequestUsesSessionCookie(t *testing.T) {
kp := generateTestKeyPair(t)
hdr := newHeaderSchemeWithToken(t, kp, "X-API-Key", "secret-key")
require.NoError(t, mw.AddDomain("example.com", []Scheme{hdr}, kp.PublicKey, time.Hour, "acc1", "svc1", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{hdr}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acc1", ServiceID: "svc1"}))
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
@@ -1218,7 +1213,7 @@ func TestProtect_HeaderAuth_MultipleValuesSameHeader(t *testing.T) {
// Single Header scheme (as if one entry existed), but the mock checks both values.
hdr := NewHeader(mock, "svc1", "acc1", "Authorization")
require.NoError(t, mw.AddDomain("example.com", []Scheme{hdr}, kp.PublicKey, time.Hour, "acc1", "svc1", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{hdr}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour, AccountID: "acc1", ServiceID: "svc1"}))
var backendCalled bool
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -1276,7 +1271,7 @@ func TestProtect_OIDCOnPlainHTTP_BlockedWith400(t *testing.T) {
return "", "https://idp.example.com/authorize", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -1300,7 +1295,7 @@ func TestProtect_OIDCOverTLS_NotBlocked(t *testing.T) {
return "", "https://idp.example.com/authorize", nil
},
}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -1320,7 +1315,7 @@ func TestProtect_NonOIDCSchemes_PlainHTTP_NotBlocked(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -1350,7 +1345,7 @@ func TestProtect_TunnelPeerFastPath_RequiresInboundMarker(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())
@@ -1385,7 +1380,7 @@ func TestProtect_TunnelPeerFastPath_TakesPathWithInboundMarker(t *testing.T) {
kp := generateTestKeyPair(t)
scheme := &stubScheme{method: auth.MethodPIN, promptID: "pin"}
require.NoError(t, mw.AddDomain("example.com", []Scheme{scheme}, kp.PublicKey, time.Hour, "", "", nil, false))
require.NoError(t, mw.AddDomain("example.com", DomainSettings{Schemes: []Scheme{scheme}, SessionPublicKey: kp.PublicKey, SessionExpiration: time.Hour}))
handler := mw.Protect(newPassthroughHandler())

View File

@@ -13,6 +13,10 @@ import (
"github.com/netbirdio/netbird/shared/management/proto"
)
// sessionTokenParam is the query parameter the management server uses to hand
// the minted session token back to the proxy after an OIDC login.
const sessionTokenParam = "session_token"
type urlGenerator interface {
GetOIDCURL(context.Context, *proto.GetOIDCURLRequest, ...grpc.CallOption) (*proto.GetOIDCURLResponse, error)
}
@@ -43,7 +47,7 @@ func (o OIDC) Authenticate(r *http.Request) (string, string, error) {
// Check for the session_token query param (from OIDC redirects).
// The management server passes the token in the URL because it cannot set
// cookies for the proxy's domain (cookies are domain-scoped per RFC 6265).
if token := r.URL.Query().Get("session_token"); token != "" {
if token := r.URL.Query().Get(sessionTokenParam); token != "" {
return token, "", nil
}

View File

@@ -44,7 +44,7 @@ func (s *stubSessionValidator) ValidateTunnelPeer(_ context.Context, in *proto.V
func newTunnelMiddleware(t *testing.T, validator SessionValidator) *Middleware {
t.Helper()
mw := NewMiddleware(log.New(), validator, nil)
require.NoError(t, mw.AddDomain("svc.example", nil, "", 0, "acct-1", "svc-1", nil, false))
require.NoError(t, mw.AddDomain("svc.example", DomainSettings{AccountID: "acct-1", ServiceID: "svc-1"}))
return mw
}
@@ -235,8 +235,8 @@ func TestForwardWithTunnelPeer_RoutesAccountIDIntoCacheKey(t *testing.T) {
}
mw := NewMiddleware(log.New(), validator, nil)
require.NoError(t, mw.AddDomain("svc-a.example", nil, "", 0, "acct-a", "svc-a", nil, false))
require.NoError(t, mw.AddDomain("svc-b.example", nil, "", 0, "acct-b", "svc-b", nil, false))
require.NoError(t, mw.AddDomain("svc-a.example", DomainSettings{AccountID: "acct-a", ServiceID: "svc-a"}))
require.NoError(t, mw.AddDomain("svc-b.example", DomainSettings{AccountID: "acct-b", ServiceID: "svc-b"}))
// The fast-path requires the inbound-listener marker on the context.
// The peerstore lookup itself is account-agnostic at this level
@@ -299,7 +299,7 @@ func TestForwardWithTunnelPeer_LocalLookupShortCircuitDoesNotPopulateCache(t *te
func TestPrivateService_FailsClosedOnTunnelPeerFailure(t *testing.T) {
mw := NewMiddleware(log.New(), nil, nil)
require.NoError(t, mw.AddDomain("private.svc", nil, "", 0, "acct-1", "svc-1", nil, true))
require.NoError(t, mw.AddDomain("private.svc", DomainSettings{AccountID: "acct-1", ServiceID: "svc-1", Private: true}))
called := false
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -328,7 +328,7 @@ func TestPrivateService_ForwardsOnTunnelPeerSuccess(t *testing.T) {
},
}
mw := NewMiddleware(log.New(), validator, nil)
require.NoError(t, mw.AddDomain("private.svc", nil, "", 0, "acct-1", "svc-1", nil, true))
require.NoError(t, mw.AddDomain("private.svc", DomainSettings{AccountID: "acct-1", ServiceID: "svc-1", Private: true}))
called := false
handler := mw.Protect(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {

View File

@@ -21,6 +21,8 @@ import (
"strconv"
"strings"
"sync"
"github.com/netbirdio/netbird/proxy/internal/netutil"
)
// MaxRoutingScanBytes bounds how far ScanRoutingFields will read into a
@@ -34,7 +36,6 @@ const MaxRoutingScanBytes int64 = 32 << 20
// metadata key by the chain when a request body is not surfaced.
const (
BypassUpgradeHeader = "upgrade_header"
BypassConnectionUpgrd = "connection_upgrade"
BypassContentType = "content_type_not_allowed"
BypassBudget = "capture_budget_exhausted"
BypassNoConfig = "no_capture_config"
@@ -125,12 +126,13 @@ func CaptureRequest(r *http.Request, cfg *Config, b Budget) (body []byte, trunca
if cfg.MaxRequestBytes <= 0 {
return nil, false, 0, BypassCapZero, release, nil
}
if r.Header.Get("Upgrade") != "" {
// The predicate has to be the forwarder's own: a looser one (either header
// on its own) skips capture for requests the forwarder still delivers to
// the upstream with their body intact, which hides them from every
// deny-capable middleware in the chain.
if netutil.IsUpgradeRequest(r.Header) {
return nil, false, 0, BypassUpgradeHeader, release, nil
}
if strings.EqualFold(r.Header.Get("Connection"), "upgrade") {
return nil, false, 0, BypassConnectionUpgrd, release, nil
}
if !contentTypeAllowed(r.Header.Get("Content-Type"), cfg.ContentTypes) {
return nil, false, 0, BypassContentType, release, nil
}

View File

@@ -1,337 +0,0 @@
// Package modeldiscovery fetches and normalizes model catalogs from
// Agent Network provider endpoints. Discovery always runs on the proxy so
// it observes the same network path as inference traffic.
package modeldiscovery
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"path"
"sort"
"strings"
"time"
"unicode"
"unicode/utf8"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/proxy/internal/roundtrip"
)
const (
SourceOpenAIV1Models = "openai_v1_models"
SourceOllamaAPITags = "ollama_api_tags"
defaultTimeout = 5 * time.Second
maxResponseBytes = 1 << 20 // 1 MiB, after HTTP decompression.
maxUpstreamURLBytes = 4096
maxHeaderNameBytes = 256
maxHeaderValueBytes = 64 << 10
maxModels = 500
maxModelIDBytes = 512
)
// Request contains the provider-owned values management resolved from the
// persisted provider record. Callers must not populate these fields from a
// dashboard-supplied URL or credential.
type Request struct {
UpstreamURL string
AuthHeaderName string
AuthHeaderValue string
SkipTLSVerify bool
AllowOllamaFallback bool
}
// Model is the deliberately small response surface returned to management.
// Arbitrary fields supplied by an upstream never cross the control channel.
type Model struct {
ID string
Label string
}
// Result is a normalized model catalog and the endpoint shape that supplied
// it.
type Result struct {
Models []Model
Source string
}
// Discoverer owns the HTTP client used for provider probes.
type Discoverer struct {
client *http.Client
timeout time.Duration
bodyLimit int64
}
// New constructs a direct-upstream discoverer. The transport is the same
// host-network transport family used by Agent Network inference routes.
func New(logger *log.Logger) *Discoverer {
return newWithTransport(roundtrip.NewDirectOnly(logger))
}
func newWithTransport(transport http.RoundTripper) *Discoverer {
return &Discoverer{
client: &http.Client{
Transport: transport,
// Redirects could move a credentialed request away from the
// persisted provider origin. Discovery never follows them.
CheckRedirect: func(_ *http.Request, _ []*http.Request) error {
return http.ErrUseLastResponse
},
},
timeout: defaultTimeout,
bodyLimit: maxResponseBytes,
}
}
// Discover queries the OpenAI-compatible model-list endpoint. Ollama's native
// tags endpoint is attempted only when explicitly enabled and the primary
// endpoint reports that the route does not exist.
func (d *Discoverer) Discover(ctx context.Context, in Request) (Result, error) {
if d == nil || d.client == nil {
return Result{}, errors.New("model discovery client is unavailable")
}
if err := validateRequest(in); err != nil {
return Result{}, err
}
timeout := d.timeout
if timeout <= 0 || timeout > defaultTimeout {
timeout = defaultTimeout
}
probeCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
models, err := d.fetchOpenAIModels(probeCtx, in)
if err == nil {
return Result{Models: models, Source: SourceOpenAIV1Models}, nil
}
if !in.AllowOllamaFallback || !isMissingEndpoint(err) {
return Result{}, err
}
models, err = d.fetchOllamaTags(probeCtx, in)
if err != nil {
return Result{}, err
}
return Result{Models: models, Source: SourceOllamaAPITags}, nil
}
func validateRequest(in Request) error {
rawURL := strings.TrimSpace(in.UpstreamURL)
if rawURL == "" {
return errors.New("model discovery upstream URL is required")
}
if len(rawURL) > maxUpstreamURLBytes {
return errors.New("model discovery upstream URL is too long")
}
if len(in.AuthHeaderName) > maxHeaderNameBytes || len(in.AuthHeaderValue) > maxHeaderValueBytes {
return errors.New("model discovery authentication header is too large")
}
if (in.AuthHeaderName == "") != (in.AuthHeaderValue == "") {
return errors.New("model discovery authentication header is incomplete")
}
if strings.ContainsAny(in.AuthHeaderName, "\r\n") || strings.ContainsAny(in.AuthHeaderValue, "\r\n") {
return errors.New("model discovery authentication header is invalid")
}
if in.AuthHeaderName != "" && !strings.EqualFold(in.AuthHeaderName, "Authorization") {
return errors.New("model discovery authentication header is unsupported")
}
return nil
}
func (d *Discoverer) fetchOpenAIModels(ctx context.Context, in Request) ([]Model, error) {
body, err := d.fetch(ctx, in, "v1/models")
if err != nil {
return nil, err
}
var payload struct {
Data *[]struct {
ID string `json:"id"`
} `json:"data"`
}
if err := json.Unmarshal(body, &payload); err != nil || payload.Data == nil {
return nil, errors.New("upstream returned invalid OpenAI model-list JSON")
}
ids := make([]string, 0, len(*payload.Data))
for _, model := range *payload.Data {
ids = append(ids, model.ID)
}
return normalize(ids)
}
func (d *Discoverer) fetchOllamaTags(ctx context.Context, in Request) ([]Model, error) {
body, err := d.fetch(ctx, in, "api/tags")
if err != nil {
return nil, err
}
var payload struct {
Models *[]struct {
Name string `json:"name"`
Model string `json:"model"`
} `json:"models"`
}
if err := json.Unmarshal(body, &payload); err != nil || payload.Models == nil {
return nil, errors.New("upstream returned invalid Ollama tags JSON")
}
ids := make([]string, 0, len(*payload.Models))
for _, model := range *payload.Models {
id := model.Model
if strings.TrimSpace(id) == "" {
id = model.Name
}
ids = append(ids, id)
}
return normalize(ids)
}
func (d *Discoverer) fetch(ctx context.Context, in Request, endpointPath string) ([]byte, error) {
endpoint, err := buildEndpointURL(in.UpstreamURL, endpointPath)
if err != nil {
return nil, err
}
reqCtx := roundtrip.WithDirectUpstream(ctx)
if in.SkipTLSVerify {
reqCtx = roundtrip.WithSkipTLSVerify(reqCtx)
}
req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, endpoint.String(), nil)
if err != nil {
return nil, errors.New("create model discovery request")
}
req.Header.Set("Accept", "application/json")
if in.AuthHeaderName != "" {
// Discovery is currently enabled only for Ollama-compatible providers.
// Canonicalizing the sole catalog-owned credential header keeps the
// control message from becoming a generic arbitrary-header primitive.
req.Header.Set("Authorization", in.AuthHeaderValue)
}
resp, err := d.client.Do(req)
if err != nil {
if errors.Is(ctx.Err(), context.DeadlineExceeded) {
return nil, errors.New("model discovery timed out")
}
if errors.Is(ctx.Err(), context.Canceled) {
return nil, context.Canceled
}
// Deliberately omit the underlying error: net/http errors include the
// internal URL, which should not be reflected through the public API.
return nil, errors.New("model discovery request failed")
}
defer func() {
_ = resp.Body.Close()
}()
if resp.StatusCode != http.StatusOK {
return nil, &upstreamStatusError{statusCode: resp.StatusCode}
}
limit := d.bodyLimit
if limit <= 0 || limit > maxResponseBytes {
limit = maxResponseBytes
}
if resp.ContentLength > limit {
return nil, errors.New("model discovery response is too large")
}
body, err := io.ReadAll(io.LimitReader(resp.Body, limit+1))
if err != nil {
return nil, errors.New("read model discovery response")
}
if int64(len(body)) > limit {
return nil, errors.New("model discovery response is too large")
}
return body, nil
}
func buildEndpointURL(rawURL, endpointPath string) (*url.URL, error) {
parsed, err := url.Parse(strings.TrimSpace(rawURL))
if err != nil || parsed.Host == "" || parsed.Hostname() == "" || parsed.Opaque != "" {
return nil, errors.New("model discovery upstream URL is invalid")
}
switch strings.ToLower(parsed.Scheme) {
case "http":
parsed.Scheme = "http"
case "https":
parsed.Scheme = "https"
default:
return nil, errors.New("model discovery upstream URL must use http or https")
}
if parsed.User != nil {
return nil, errors.New("model discovery upstream URL must not contain credentials")
}
// Match Agent Network routing semantics: the static discovery path is
// appended to any persisted base path. Queries and fragments on a provider
// URL are not forwarded to inference and are likewise excluded here.
parsed.Path = "/" + strings.TrimPrefix(path.Join(parsed.Path, endpointPath), "/")
parsed.RawPath = ""
parsed.RawQuery = ""
parsed.ForceQuery = false
parsed.Fragment = ""
return parsed, nil
}
func normalize(ids []string) ([]Model, error) {
if len(ids) > maxModels {
return nil, errors.New("upstream returned too many models")
}
seen := make(map[string]struct{}, len(ids))
normalized := make([]string, 0, len(ids))
for _, raw := range ids {
id := strings.TrimSpace(raw)
if !validModelID(id) {
continue
}
if _, ok := seen[id]; ok {
continue
}
seen[id] = struct{}{}
normalized = append(normalized, id)
}
sort.Strings(normalized)
models := make([]Model, 0, len(normalized))
for _, id := range normalized {
models = append(models, Model{ID: id, Label: id})
}
return models, nil
}
func validModelID(id string) bool {
if id == "" || len(id) > maxModelIDBytes || !utf8.ValidString(id) {
return false
}
for _, r := range id {
if unicode.IsControl(r) {
return false
}
}
return true
}
type upstreamStatusError struct {
statusCode int
}
func (e *upstreamStatusError) Error() string {
return fmt.Sprintf("upstream returned HTTP %d", e.statusCode)
}
func isMissingEndpoint(err error) bool {
var statusErr *upstreamStatusError
if !errors.As(err, &statusErr) {
return false
}
return statusErr.statusCode == http.StatusNotFound || statusErr.statusCode == http.StatusMethodNotAllowed
}

View File

@@ -1,315 +0,0 @@
package modeldiscovery
import (
"context"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestDiscoverOpenAIModels(t *testing.T) {
t.Parallel()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
assert.Equal(t, http.MethodGet, r.Method)
assert.Equal(t, "/gateway/v1/models", r.URL.Path)
assert.Empty(t, r.URL.RawQuery)
assert.Equal(t, "application/json", r.Header.Get("Accept"))
assert.Equal(t, "Bearer secret", r.Header.Get("Authorization"))
_, _ = io.WriteString(w, `{
"object": "list",
"data": [
{"id": " qwen2.5:latest ", "owned_by": "ignored"},
{"id": "llama3.2:latest"},
{"id": "llama3.2:latest"},
{"id": ""},
{"id": "bad\u0000id"}
]
}`)
}))
defer server.Close()
discoverer := newWithTransport(http.DefaultTransport)
result, err := discoverer.Discover(context.Background(), Request{
UpstreamURL: server.URL + "/gateway/?ignored=true#fragment",
AuthHeaderName: "authorization",
AuthHeaderValue: "Bearer secret",
})
require.NoError(t, err)
assert.Equal(t, SourceOpenAIV1Models, result.Source)
assert.Equal(t, []Model{
{ID: "llama3.2:latest", Label: "llama3.2:latest"},
{ID: "qwen2.5:latest", Label: "qwen2.5:latest"},
}, result.Models)
}
func TestDiscoverOllamaFallback(t *testing.T) {
t.Parallel()
var primaryCalls atomic.Int32
var fallbackCalls atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/v1/models":
primaryCalls.Add(1)
http.NotFound(w, r)
case "/api/tags":
fallbackCalls.Add(1)
_, _ = io.WriteString(w, `{
"models": [
{"name": "ignored-name", "model": "gemma3:latest"},
{"name": "llama3.2:latest", "model": ""}
]
}`)
default:
http.Error(w, "unexpected path", http.StatusInternalServerError)
}
}))
defer server.Close()
result, err := newWithTransport(http.DefaultTransport).Discover(context.Background(), Request{
UpstreamURL: server.URL,
AllowOllamaFallback: true,
})
require.NoError(t, err)
assert.Equal(t, int32(1), primaryCalls.Load())
assert.Equal(t, int32(1), fallbackCalls.Load())
assert.Equal(t, SourceOllamaAPITags, result.Source)
assert.Equal(t, []Model{
{ID: "gemma3:latest", Label: "gemma3:latest"},
{ID: "llama3.2:latest", Label: "llama3.2:latest"},
}, result.Models)
}
func TestDiscoverDoesNotFallbackOnAuthenticationFailure(t *testing.T) {
t.Parallel()
var fallbackCalls atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/api/tags" {
fallbackCalls.Add(1)
}
http.Error(w, "secret upstream body", http.StatusUnauthorized)
}))
defer server.Close()
_, err := newWithTransport(http.DefaultTransport).Discover(context.Background(), Request{
UpstreamURL: server.URL,
AllowOllamaFallback: true,
})
require.EqualError(t, err, "upstream returned HTTP 401")
assert.Zero(t, fallbackCalls.Load())
assert.NotContains(t, err.Error(), "secret upstream body")
assert.NotContains(t, err.Error(), server.URL)
}
func TestDiscoverDoesNotFollowRedirectsOrLeakAuth(t *testing.T) {
t.Parallel()
var redirectedCalls atomic.Int32
redirected := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
redirectedCalls.Add(1)
assert.Empty(t, r.Header.Get("Authorization"))
_, _ = io.WriteString(w, `{"data":[]}`)
}))
defer redirected.Close()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, redirected.URL+"/captured", http.StatusFound)
}))
defer server.Close()
_, err := newWithTransport(http.DefaultTransport).Discover(context.Background(), Request{
UpstreamURL: server.URL,
AuthHeaderName: "Authorization",
AuthHeaderValue: "Bearer must-not-leak",
})
require.EqualError(t, err, "upstream returned HTTP 302")
assert.Zero(t, redirectedCalls.Load())
}
func TestDiscoverEnforcesResponseLimit(t *testing.T) {
t.Parallel()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_, _ = io.WriteString(w, `{"data":[{"id":"llama3.2:latest"}]}`)
}))
defer server.Close()
discoverer := newWithTransport(http.DefaultTransport)
discoverer.bodyLimit = 16
_, err := discoverer.Discover(context.Background(), Request{UpstreamURL: server.URL})
require.EqualError(t, err, "model discovery response is too large")
}
func TestDiscoverEnforcesTimeout(t *testing.T) {
t.Parallel()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
select {
case <-r.Context().Done():
case <-time.After(time.Second):
_, _ = io.WriteString(w, `{"data":[]}`)
}
}))
defer server.Close()
discoverer := newWithTransport(http.DefaultTransport)
discoverer.timeout = 30 * time.Millisecond
_, err := discoverer.Discover(context.Background(), Request{UpstreamURL: server.URL})
require.EqualError(t, err, "model discovery timed out")
}
func TestDiscoverHonorsSkipTLSVerification(t *testing.T) {
t.Parallel()
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_, _ = io.WriteString(w, `{"data":[]}`)
}))
defer server.Close()
discoverer := New(nil)
_, err := discoverer.Discover(context.Background(), Request{UpstreamURL: server.URL})
require.EqualError(t, err, "model discovery request failed")
result, err := discoverer.Discover(context.Background(), Request{
UpstreamURL: server.URL,
SkipTLSVerify: true,
})
require.NoError(t, err)
assert.Empty(t, result.Models)
}
func TestDiscoverRejectsUnsafeRequestValues(t *testing.T) {
t.Parallel()
tests := []struct {
name string
request Request
wantErr string
}{
{
name: "unsupported scheme",
request: Request{UpstreamURL: "file:///etc/passwd"},
wantErr: "model discovery upstream URL is invalid",
},
{
name: "URL credentials",
request: Request{UpstreamURL: "http://user:pass@example.com"},
wantErr: "model discovery upstream URL must not contain credentials",
},
{
name: "missing hostname",
request: Request{UpstreamURL: "http://:11434"},
wantErr: "model discovery upstream URL is invalid",
},
{
name: "incomplete auth header",
request: Request{
UpstreamURL: "http://example.com",
AuthHeaderName: "Authorization",
},
wantErr: "model discovery authentication header is incomplete",
},
{
name: "header injection",
request: Request{
UpstreamURL: "http://example.com",
AuthHeaderName: "Authorization",
AuthHeaderValue: "Bearer safe\r\nX-Evil: yes",
},
wantErr: "model discovery authentication header is invalid",
},
{
name: "unsupported auth header",
request: Request{
UpstreamURL: "http://example.com",
AuthHeaderName: "X-API-Key",
AuthHeaderValue: "secret",
},
wantErr: "model discovery authentication header is unsupported",
},
{
name: "oversized URL",
request: Request{
UpstreamURL: "http://example.com/" + strings.Repeat("a", maxUpstreamURLBytes),
},
wantErr: "model discovery upstream URL is too long",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
t.Parallel()
_, err := newWithTransport(http.DefaultTransport).Discover(context.Background(), test.request)
require.EqualError(t, err, test.wantErr)
})
}
}
func TestDiscoverRejectsInvalidPayloads(t *testing.T) {
t.Parallel()
tests := []struct {
name string
status int
body string
wantErr string
}{
{
name: "malformed JSON",
status: http.StatusOK,
body: `{"data":`,
wantErr: "upstream returned invalid OpenAI model-list JSON",
},
{
name: "missing data",
status: http.StatusOK,
body: `{}`,
wantErr: "upstream returned invalid OpenAI model-list JSON",
},
{
name: "too many models",
status: http.StatusOK,
body: modelListJSON(maxModels + 1),
wantErr: "upstream returned too many models",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
t.Parallel()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(test.status)
_, _ = io.WriteString(w, test.body)
}))
defer server.Close()
_, err := newWithTransport(http.DefaultTransport).Discover(context.Background(), Request{
UpstreamURL: server.URL,
})
require.EqualError(t, err, test.wantErr)
})
}
}
func modelListJSON(count int) string {
var body strings.Builder
body.WriteString(`{"data":[`)
for i := range count {
if i > 0 {
body.WriteByte(',')
}
_, _ = fmt.Fprintf(&body, `{"id":"model-%d"}`, i)
}
body.WriteString(`]}`)
return body.String()
}

View File

@@ -0,0 +1,23 @@
package netutil
import (
"net/http"
"golang.org/x/net/http/httpguts"
)
// IsUpgradeRequest reports whether r is a protocol-upgrade request, using the
// same predicate httputil.ReverseProxy applies when it decides to hand the
// connection over instead of proxying normally.
//
// Matching the forwarder exactly matters for anything that inspects a request
// before it is proxied: a looser test (an Upgrade header on its own, say) marks
// a request as an upgrade and skips inspection, while the forwarder still
// delivers it to the backend as an ordinary request with its body intact. That
// gap is a body-inspection bypass reachable by adding one header.
func IsUpgradeRequest(h http.Header) bool {
if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") {
return false
}
return h.Get("Upgrade") != ""
}

View File

@@ -50,6 +50,37 @@ const (
CrowdSecObserve CrowdSecMode = "observe"
)
// AppSecMode is the per-service CrowdSec AppSec (WAF) enforcement mode.
type AppSecMode string
const (
// AppSecOff disables request inspection.
AppSecOff AppSecMode = ""
// AppSecEnforce blocks requests the engine flags, and fails closed when the
// engine is unreachable.
AppSecEnforce AppSecMode = "enforce"
// AppSecObserve records the verdict without blocking.
AppSecObserve AppSecMode = "observe"
)
// ParseAppSecMode maps a wire value to an AppSecMode. Unrecognized values map
// to AppSecOff so a typo never turns inspection into an unintended block.
func ParseAppSecMode(s string) AppSecMode {
switch AppSecMode(s) {
case AppSecEnforce:
return AppSecEnforce
case AppSecObserve:
return AppSecObserve
default:
return AppSecOff
}
}
// Enabled reports whether the mode asks for request inspection.
func (m AppSecMode) Enabled() bool {
return m == AppSecEnforce || m == AppSecObserve
}
// Filter evaluates IP restrictions. CIDR checks are performed first
// (cheap), followed by country lookups (more expensive) only when needed.
type Filter struct {
@@ -146,6 +177,13 @@ const (
// DenyCrowdSecUnavailable indicates enforce mode but the bouncer has not
// completed its initial sync.
DenyCrowdSecUnavailable
// DenyAppSecBan indicates a CrowdSec AppSec "ban" remediation.
DenyAppSecBan
// DenyAppSecCaptcha indicates a CrowdSec AppSec "captcha" remediation.
DenyAppSecCaptcha
// DenyAppSecUnavailable indicates enforce mode but the AppSec engine could
// not produce a verdict (unreachable, timed out, or it rejected the call).
DenyAppSecUnavailable
)
// String returns the deny reason string matching the HTTP auth mechanism names.
@@ -167,6 +205,12 @@ func (v Verdict) String() string {
return "crowdsec_throttle"
case DenyCrowdSecUnavailable:
return "crowdsec_unavailable"
case DenyAppSecBan:
return "appsec_ban"
case DenyAppSecCaptcha:
return "appsec_captcha"
case DenyAppSecUnavailable:
return "appsec_unavailable"
default:
return "unknown"
}
@@ -182,6 +226,16 @@ func (v Verdict) IsCrowdSec() bool {
}
}
// IsAppSec returns true when the verdict originates from an AppSec inspection.
func (v Verdict) IsAppSec() bool {
switch v {
case DenyAppSecBan, DenyAppSecCaptcha, DenyAppSecUnavailable:
return true
default:
return false
}
}
// IsObserveOnly returns true when v is a CrowdSec verdict and the filter is in
// observe mode. Callers should log the verdict but not block the request.
func (f *Filter) IsObserveOnly(v Verdict) bool {

View File

@@ -126,6 +126,23 @@ type Config struct {
// CrowdSecAPIKey is the CrowdSec bouncer API key. Empty disables
// CrowdSec.
CrowdSecAPIKey string
// CrowdSecAppSecURL is the CrowdSec AppSec (WAF) endpoint. Empty disables
// HTTP request inspection.
CrowdSecAppSecURL string
// CrowdSecAppSecTimeout bounds a single AppSec inspection call. Zero falls
// back to the internal default.
CrowdSecAppSecTimeout time.Duration
// CrowdSecAppSecMaxBodyBytes caps the request body mirrored to AppSec.
// Zero falls back to the internal default; negative forwards no body.
CrowdSecAppSecMaxBodyBytes int64
// CrowdSecAppSecMaxConcurrent bounds AppSec inspections in flight toward
// the engine. Zero falls back to the internal default; negative removes
// the bound.
CrowdSecAppSecMaxConcurrent int
// MiddlewareCaptureBudgetBytes bounds the total request-body buffering in
// flight across the proxy, shared by AppSec inspection and the
// agent-network capture. Zero falls back to the internal default.
MiddlewareCaptureBudgetBytes int64
}
// New builds a Server from cfg without performing any I/O. No goroutines
@@ -135,42 +152,47 @@ type Config struct {
// directly) byte-for-byte equivalent.
func New(ctx context.Context, cfg Config) *Server {
return &Server{
ctx: ctx,
ListenAddr: cfg.ListenAddr,
ID: cfg.ID,
Logger: cfg.Logger,
Version: cfg.Version,
ProxyURL: cfg.ProxyURL,
ManagementAddress: cfg.ManagementAddress,
ProxyToken: cfg.ProxyToken,
CertificateDirectory: cfg.CertificateDirectory,
CertificateFile: cfg.CertificateFile,
CertificateKeyFile: cfg.CertificateKeyFile,
GenerateACMECertificates: cfg.GenerateACMECertificates,
ACMEChallengeAddress: cfg.ACMEChallengeAddress,
ACMEDirectory: cfg.ACMEDirectory,
ACMEEABKID: cfg.ACMEEABKID,
ACMEEABHMACKey: cfg.ACMEEABHMACKey,
ACMEChallengeType: cfg.ACMEChallengeType,
CertLockMethod: cfg.CertLockMethod,
WildcardCertDir: cfg.WildcardCertDir,
DebugEndpointEnabled: cfg.DebugEndpointEnabled,
DebugEndpointAddress: cfg.DebugEndpointAddress,
HealthAddress: cfg.HealthAddr,
ForwardedProto: cfg.ForwardedProto,
TrustedProxies: cfg.TrustedProxies,
WireguardPort: cfg.WireguardPort,
ProxyProtocol: cfg.ProxyProtocol,
PreSharedKey: cfg.PreSharedKey,
Performance: cfg.Performance,
SupportsCustomPorts: cfg.SupportsCustomPorts,
RequireSubdomain: cfg.RequireSubdomain,
Private: cfg.Private,
MaxDialTimeout: cfg.MaxDialTimeout,
MaxSessionIdleTimeout: cfg.MaxSessionIdleTimeout,
MappingBatchWatchdog: cfg.MappingBatchWatchdog,
GeoDataDir: cfg.GeoDataDir,
CrowdSecAPIURL: cfg.CrowdSecAPIURL,
CrowdSecAPIKey: cfg.CrowdSecAPIKey,
ctx: ctx,
ListenAddr: cfg.ListenAddr,
ID: cfg.ID,
Logger: cfg.Logger,
Version: cfg.Version,
ProxyURL: cfg.ProxyURL,
ManagementAddress: cfg.ManagementAddress,
ProxyToken: cfg.ProxyToken,
CertificateDirectory: cfg.CertificateDirectory,
CertificateFile: cfg.CertificateFile,
CertificateKeyFile: cfg.CertificateKeyFile,
GenerateACMECertificates: cfg.GenerateACMECertificates,
ACMEChallengeAddress: cfg.ACMEChallengeAddress,
ACMEDirectory: cfg.ACMEDirectory,
ACMEEABKID: cfg.ACMEEABKID,
ACMEEABHMACKey: cfg.ACMEEABHMACKey,
ACMEChallengeType: cfg.ACMEChallengeType,
CertLockMethod: cfg.CertLockMethod,
WildcardCertDir: cfg.WildcardCertDir,
DebugEndpointEnabled: cfg.DebugEndpointEnabled,
DebugEndpointAddress: cfg.DebugEndpointAddress,
HealthAddress: cfg.HealthAddr,
ForwardedProto: cfg.ForwardedProto,
TrustedProxies: cfg.TrustedProxies,
WireguardPort: cfg.WireguardPort,
ProxyProtocol: cfg.ProxyProtocol,
PreSharedKey: cfg.PreSharedKey,
Performance: cfg.Performance,
SupportsCustomPorts: cfg.SupportsCustomPorts,
RequireSubdomain: cfg.RequireSubdomain,
Private: cfg.Private,
MaxDialTimeout: cfg.MaxDialTimeout,
MaxSessionIdleTimeout: cfg.MaxSessionIdleTimeout,
MappingBatchWatchdog: cfg.MappingBatchWatchdog,
GeoDataDir: cfg.GeoDataDir,
CrowdSecAPIURL: cfg.CrowdSecAPIURL,
CrowdSecAPIKey: cfg.CrowdSecAPIKey,
CrowdSecAppSecURL: cfg.CrowdSecAppSecURL,
CrowdSecAppSecTimeout: cfg.CrowdSecAppSecTimeout,
CrowdSecAppSecMaxBodyBytes: cfg.CrowdSecAppSecMaxBodyBytes,
CrowdSecAppSecMaxConcurrent: cfg.CrowdSecAppSecMaxConcurrent,
MiddlewareCaptureBudgetBytes: cfg.MiddlewareCaptureBudgetBytes,
}
}

45
proxy/lifecycle_test.go Normal file
View File

@@ -0,0 +1,45 @@
package proxy
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// New maps Config onto Server field by field, and a field left out of that
// literal still compiles: the knob is simply parsed and then dropped, so an
// operator setting it sees the default with no error anywhere. These assertions
// are the only thing standing between a new setting and that silent no-op.
func TestNew_ForwardsOperatorTuning(t *testing.T) {
cfg := Config{
ManagementAddress: "http://localhost:8080",
ProxyToken: "token",
CrowdSecAPIURL: "http://crowdsec:8080/",
CrowdSecAPIKey: "key",
CrowdSecAppSecURL: "http://crowdsec:7422/",
CrowdSecAppSecTimeout: 321 * time.Millisecond,
CrowdSecAppSecMaxBodyBytes: 4321,
CrowdSecAppSecMaxConcurrent: 17,
MiddlewareCaptureBudgetBytes: 5 << 20,
MaxDialTimeout: 7 * time.Second,
MaxSessionIdleTimeout: 11 * time.Second,
GeoDataDir: "/var/lib/geo",
}
srv := New(context.Background(), cfg)
require.NotNil(t, srv)
assert.Equal(t, cfg.CrowdSecAPIURL, srv.CrowdSecAPIURL)
assert.Equal(t, cfg.CrowdSecAPIKey, srv.CrowdSecAPIKey)
assert.Equal(t, cfg.CrowdSecAppSecURL, srv.CrowdSecAppSecURL)
assert.Equal(t, cfg.CrowdSecAppSecTimeout, srv.CrowdSecAppSecTimeout)
assert.Equal(t, cfg.CrowdSecAppSecMaxBodyBytes, srv.CrowdSecAppSecMaxBodyBytes)
assert.Equal(t, cfg.CrowdSecAppSecMaxConcurrent, srv.CrowdSecAppSecMaxConcurrent)
assert.Equal(t, cfg.MiddlewareCaptureBudgetBytes, srv.MiddlewareCaptureBudgetBytes)
assert.Equal(t, cfg.MaxDialTimeout, srv.MaxDialTimeout)
assert.Equal(t, cfg.MaxSessionIdleTimeout, srv.MaxSessionIdleTimeout)
assert.Equal(t, cfg.GeoDataDir, srv.GeoDataDir)
}

View File

@@ -240,6 +240,10 @@ func (m *testProxyManager) ClusterSupportsCrowdSec(_ context.Context, _ string)
return nil
}
func (m *testProxyManager) ClusterSupportsAppSec(_ context.Context, _ string) *bool {
return nil
}
func (m *testProxyManager) ClusterSupportsPrivate(_ context.Context, _ string) *bool {
return nil
}
@@ -271,10 +275,6 @@ func (c *testProxyController) SendServiceUpdateToCluster(_ context.Context, _ st
// noop
}
func (c *testProxyController) DiscoverModels(_ context.Context, _, _ string, _ *proto.ModelDiscoveryRequest) (*proto.ModelDiscoveryResult, error) {
return nil, nbproxy.ErrModelDiscoveryUnavailable
}
func (c *testProxyController) GetOIDCValidationConfig() nbproxy.OIDCValidationConfig {
return nbproxy.OIDCValidationConfig{}
}
@@ -566,16 +566,11 @@ func TestIntegration_ProxyConnection_ReconnectDoesNotDuplicateState(t *testing.T
addMappingCalls.Add(1)
// Apply to real auth middleware (idempotent)
err := authMw.AddDomain(
mapping.GetDomain(),
nil,
"",
0,
proxytypes.AccountID(mapping.GetAccountId()),
proxytypes.ServiceID(mapping.GetId()),
nil,
mapping.GetPrivate(),
)
err := authMw.AddDomain(mapping.GetDomain(), auth.DomainSettings{
AccountID: proxytypes.AccountID(mapping.GetAccountId()),
ServiceID: proxytypes.ServiceID(mapping.GetId()),
Private: mapping.GetPrivate(),
})
require.NoError(t, err)
// Apply to real proxy (idempotent)

View File

@@ -1,349 +0,0 @@
package proxy
import (
"context"
"errors"
"fmt"
"io"
"sync/atomic"
"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/crowdsec"
"github.com/netbirdio/netbird/proxy/internal/modeldiscovery"
"github.com/netbirdio/netbird/shared/management/proto"
)
type stubModelDiscoverer struct {
started chan modeldiscovery.Request
release <-chan struct{}
result modeldiscovery.Result
err error
}
func (s *stubModelDiscoverer) Discover(ctx context.Context, request modeldiscovery.Request) (modeldiscovery.Result, error) {
if s.started != nil {
select {
case s.started <- request:
case <-ctx.Done():
return modeldiscovery.Result{}, ctx.Err()
}
}
if s.release != nil {
select {
case <-s.release:
case <-ctx.Done():
return modeldiscovery.Result{}, ctx.Err()
}
}
return s.result, s.err
}
type modelDiscoverySyncStream struct {
grpc.ClientStream
ctx context.Context
recv chan *proto.SyncMappingsResponse
sent chan *proto.SyncMappingsRequest
sendWait time.Duration
sending atomic.Int32
overlap atomic.Bool
}
func newModelDiscoverySyncStream(ctx context.Context) *modelDiscoverySyncStream {
return &modelDiscoverySyncStream{
ctx: ctx,
recv: make(chan *proto.SyncMappingsResponse, 16),
sent: make(chan *proto.SyncMappingsRequest, 16),
}
}
func (s *modelDiscoverySyncStream) Send(message *proto.SyncMappingsRequest) error {
if s.sending.Add(1) != 1 {
s.overlap.Store(true)
}
defer s.sending.Add(-1)
if s.sendWait > 0 {
time.Sleep(s.sendWait)
}
select {
case s.sent <- message:
return nil
case <-s.ctx.Done():
return s.ctx.Err()
}
}
func (s *modelDiscoverySyncStream) Recv() (*proto.SyncMappingsResponse, error) {
select {
case message, ok := <-s.recv:
if !ok {
return nil, io.EOF
}
return message, nil
case <-s.ctx.Done():
return nil, s.ctx.Err()
}
}
func (s *modelDiscoverySyncStream) Context() context.Context {
return s.ctx
}
func completeModelDiscoveryInitialSync(t *testing.T, stream *modelDiscoverySyncStream) {
t.Helper()
stream.recv <- &proto.SyncMappingsResponse{InitialSyncComplete: true}
select {
case sent := <-stream.sent:
require.NotNil(t, sent.GetAck())
case <-time.After(time.Second):
t.Fatal("initial snapshot was not acknowledged")
}
}
func TestProxyCapabilitiesAdvertiseModelDiscovery(t *testing.T) {
t.Parallel()
server := &Server{
crowdsecRegistry: crowdsec.NewRegistry("", "", log.New().WithField("test", true)),
}
capabilities := server.proxyCapabilities()
require.NotNil(t, capabilities.SupportsModelDiscovery)
assert.True(t, capabilities.GetSupportsModelDiscovery())
}
func TestExecuteModelDiscoveryMapsControlShape(t *testing.T) {
t.Parallel()
discoverer := &stubModelDiscoverer{
result: modeldiscovery.Result{
Source: modeldiscovery.SourceOpenAIV1Models,
Models: []modeldiscovery.Model{
{ID: "llama3.2:latest", Label: "Llama 3.2"},
},
},
}
request := &proto.ModelDiscoveryRequest{
RequestId: "request-1",
UpstreamUrl: "http://ollama.internal:11434",
AuthHeaderName: "Authorization",
AuthHeaderValue: "Bearer secret",
SkipTlsVerify: true,
OllamaFallback: true,
}
result := executeModelDiscovery(context.Background(), discoverer, request)
assert.Equal(t, "request-1", result.GetRequestId())
assert.Equal(t, modeldiscovery.SourceOpenAIV1Models, result.GetSource())
require.Len(t, result.GetModels(), 1)
assert.Equal(t, "llama3.2:latest", result.GetModels()[0].GetId())
assert.Equal(t, "Llama 3.2", result.GetModels()[0].GetLabel())
discoverer.started = make(chan modeldiscovery.Request, 1)
_ = executeModelDiscovery(context.Background(), discoverer, request)
received := <-discoverer.started
assert.Equal(t, request.GetUpstreamUrl(), received.UpstreamURL)
assert.Equal(t, request.GetAuthHeaderName(), received.AuthHeaderName)
assert.Equal(t, request.GetAuthHeaderValue(), received.AuthHeaderValue)
assert.True(t, received.SkipTLSVerify)
assert.True(t, received.AllowOllamaFallback)
}
func TestExecuteModelDiscoveryReturnsSanitizedError(t *testing.T) {
t.Parallel()
result := executeModelDiscovery(context.Background(), &stubModelDiscoverer{
err: errors.New("model discovery request failed"),
}, &proto.ModelDiscoveryRequest{RequestId: "request-error"})
assert.Equal(t, "request-error", result.GetRequestId())
assert.Equal(t, "model discovery request failed", result.GetError())
assert.Empty(t, result.GetModels())
assert.Empty(t, result.GetSource())
}
func TestHandleSyncMappingsStreamRunsDiscoveryOutOfBand(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
release := make(chan struct{})
started := make(chan modeldiscovery.Request, 1)
server := &Server{
Logger: log.New(),
routerReady: closedChan(),
modelDiscoverer: &stubModelDiscoverer{
started: started,
release: release,
result: modeldiscovery.Result{
Source: modeldiscovery.SourceOpenAIV1Models,
Models: []modeldiscovery.Model{{ID: "model-a", Label: "model-a"}},
},
},
}
stream := newModelDiscoverySyncStream(ctx)
stream.sendWait = 10 * time.Millisecond
done := make(chan error, 1)
initialSyncDone := false
go func() {
done <- server.handleSyncMappingsStream(ctx, stream, &initialSyncDone, time.Time{})
}()
completeModelDiscoveryInitialSync(t, stream)
stream.recv <- &proto.SyncMappingsResponse{
ModelDiscoveryRequest: &proto.ModelDiscoveryRequest{
RequestId: "request-1",
UpstreamUrl: "http://ollama.internal:11434",
},
}
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("model discovery did not start")
}
// A normal mapping batch must still be acknowledged while the HTTP probe
// is in flight.
stream.recv <- &proto.SyncMappingsResponse{}
select {
case sent := <-stream.sent:
assert.NotNil(t, sent.GetAck())
assert.Nil(t, sent.GetModelDiscoveryResult())
case <-time.After(time.Second):
t.Fatal("mapping ack was blocked by model discovery")
}
close(release)
select {
case sent := <-stream.sent:
result := sent.GetModelDiscoveryResult()
require.NotNil(t, result)
assert.Equal(t, "request-1", result.GetRequestId())
assert.Equal(t, modeldiscovery.SourceOpenAIV1Models, result.GetSource())
assert.Nil(t, sent.GetAck())
case <-time.After(time.Second):
t.Fatal("model discovery result was not sent")
}
assert.False(t, stream.overlap.Load(), "acks and discovery results must use one serialized sender")
close(stream.recv)
require.NoError(t, <-done)
}
func TestHandleSyncMappingsStreamBoundsConcurrentDiscovery(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
release := make(chan struct{})
started := make(chan modeldiscovery.Request, 8)
server := &Server{
Logger: log.New(),
routerReady: closedChan(),
modelDiscoverer: &stubModelDiscoverer{
started: started,
release: release,
},
}
stream := newModelDiscoverySyncStream(ctx)
done := make(chan error, 1)
initialSyncDone := false
go func() {
done <- server.handleSyncMappingsStream(ctx, stream, &initialSyncDone, time.Time{})
}()
completeModelDiscoveryInitialSync(t, stream)
for i := range 5 {
stream.recv <- &proto.SyncMappingsResponse{
ModelDiscoveryRequest: &proto.ModelDiscoveryRequest{
RequestId: fmt.Sprintf("request-%d", i),
UpstreamUrl: "http://ollama.internal:11434",
},
}
}
for range 4 {
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("expected four concurrent model discoveries")
}
}
select {
case sent := <-stream.sent:
result := sent.GetModelDiscoveryResult()
require.NotNil(t, result)
assert.Equal(t, "model discovery is busy", result.GetError())
case <-time.After(time.Second):
t.Fatal("fifth discovery did not fail fast")
}
close(release)
for range 4 {
select {
case sent := <-stream.sent:
require.NotNil(t, sent.GetModelDiscoveryResult())
case <-time.After(time.Second):
t.Fatal("in-flight model discovery did not complete")
}
}
close(stream.recv)
require.NoError(t, <-done)
}
func TestHandleSyncMappingsStreamRejectsMixedDiscoveryMessage(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
server := &Server{
Logger: log.New(),
routerReady: closedChan(),
modelDiscoverer: &stubModelDiscoverer{},
}
stream := newModelDiscoverySyncStream(ctx)
stream.recv <- &proto.SyncMappingsResponse{
Mapping: []*proto.ProxyMapping{{Id: "mapping-1"}},
InitialSyncComplete: true,
ModelDiscoveryRequest: &proto.ModelDiscoveryRequest{
RequestId: "request-1",
},
}
close(stream.recv)
initialSyncDone := true
err := server.handleSyncMappingsStream(ctx, stream, &initialSyncDone, time.Time{})
require.EqualError(t, err, "model discovery message must not include mapping data or set initial_sync_complete")
}
func TestHandleSyncMappingsStreamRejectsDiscoveryBeforeInitialSnapshot(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
server := &Server{
Logger: log.New(),
routerReady: closedChan(),
modelDiscoverer: &stubModelDiscoverer{},
}
stream := newModelDiscoverySyncStream(ctx)
stream.recv <- &proto.SyncMappingsResponse{
ModelDiscoveryRequest: &proto.ModelDiscoveryRequest{
RequestId: "request-1",
},
}
close(stream.recv)
initialSyncDone := true
err := server.handleSyncMappingsStream(ctx, stream, &initialSyncDone, time.Time{})
require.EqualError(t, err, "model discovery request received before initial sync completed")
}

View File

@@ -45,6 +45,7 @@ import (
"github.com/netbirdio/netbird/client/embed"
"github.com/netbirdio/netbird/proxy/internal/accesslog"
"github.com/netbirdio/netbird/proxy/internal/acme"
"github.com/netbirdio/netbird/proxy/internal/appsec"
"github.com/netbirdio/netbird/proxy/internal/auth"
"github.com/netbirdio/netbird/proxy/internal/certwatch"
"github.com/netbirdio/netbird/proxy/internal/conntrack"
@@ -57,7 +58,6 @@ import (
proxymetrics "github.com/netbirdio/netbird/proxy/internal/metrics"
"github.com/netbirdio/netbird/proxy/internal/middleware"
mwbuiltin "github.com/netbirdio/netbird/proxy/internal/middleware/builtin"
"github.com/netbirdio/netbird/proxy/internal/modeldiscovery"
"github.com/netbirdio/netbird/proxy/internal/netutil"
"github.com/netbirdio/netbird/proxy/internal/proxy"
"github.com/netbirdio/netbird/proxy/internal/restrict"
@@ -79,10 +79,6 @@ type portRouter struct {
cancel context.CancelFunc
}
type providerModelDiscoverer interface {
Discover(context.Context, modeldiscovery.Request) (modeldiscovery.Result, error)
}
type Server struct {
ctx context.Context
mgmtClient proto.ProxyServiceClient
@@ -105,20 +101,16 @@ type Server struct {
// middlewareRegistry is the source of registered middleware factories.
// Concrete middlewares register themselves through init().
middlewareRegistry *middleware.Registry
// modelDiscoverer executes explicit provider model-list probes on the
// proxy's host network. Lazily constructed for normal servers; injectable
// in focused control-stream tests.
modelDiscoverer providerModelDiscoverer
mainRouter *nbtcp.Router
mainPort uint16
udpMu sync.Mutex
udpRelays map[types.ServiceID]*udprelay.Relay
udpRelayWg sync.WaitGroup
portMu sync.RWMutex
portRouters map[uint16]*portRouter
svcPorts map[types.ServiceID][]uint16
lastMappings map[types.ServiceID]*proto.ProxyMapping
portRouterWg sync.WaitGroup
mainRouter *nbtcp.Router
mainPort uint16
udpMu sync.Mutex
udpRelays map[types.ServiceID]*udprelay.Relay
udpRelayWg sync.WaitGroup
portMu sync.RWMutex
portRouters map[uint16]*portRouter
svcPorts map[types.ServiceID][]uint16
lastMappings map[types.ServiceID]*proto.ProxyMapping
portRouterWg sync.WaitGroup
// hijackTracker tracks hijacked connections (e.g. WebSocket upgrades)
// so they can be closed during graceful shutdown, since http.Server.Shutdown
@@ -135,6 +127,10 @@ type Server struct {
crowdsecMu sync.Mutex
crowdsecServices map[types.ServiceID]bool
// appsecClient is the shared CrowdSec AppSec client, nil when no AppSec
// endpoint is configured. Stateless, so it needs no per-service lifecycle.
appsecClient *appsec.Client
// routerReady is closed once mainRouter is fully initialized.
// The mapping worker waits on this before processing updates.
routerReady chan struct{}
@@ -247,6 +243,20 @@ type Server struct {
CrowdSecAPIURL string
// CrowdSecAPIKey is the CrowdSec bouncer API key. Empty disables CrowdSec.
CrowdSecAPIKey string
// CrowdSecAppSecURL is the CrowdSec AppSec (WAF) endpoint, e.g.
// http://127.0.0.1:7422/. Empty disables request inspection. Requires
// CrowdSecAPIKey, which the AppSec component validates against LAPI.
CrowdSecAppSecURL string
// CrowdSecAppSecTimeout bounds a single AppSec inspection call.
// Zero means appsec.DefaultTimeout.
CrowdSecAppSecTimeout time.Duration
// CrowdSecAppSecMaxBodyBytes caps the request body mirrored to the AppSec
// engine. Zero means appsec.DefaultMaxBodyBytes; negative disables body
// forwarding, leaving header and URI inspection.
CrowdSecAppSecMaxBodyBytes int64
// CrowdSecAppSecMaxConcurrent bounds AppSec inspections in flight. Zero
// means appsec.DefaultMaxConcurrent; negative removes the bound.
CrowdSecAppSecMaxConcurrent int
// MaxSessionIdleTimeout caps the per-service session idle timeout.
// Zero means no cap (the proxy honors whatever management sends).
// Set via NB_PROXY_MAX_SESSION_IDLE_TIMEOUT for shared deployments.
@@ -393,6 +403,13 @@ func (s *Server) Start(ctx context.Context) error {
s.crowdsecRegistry = crowdsec.NewRegistry(s.CrowdSecAPIURL, s.CrowdSecAPIKey, log.NewEntry(s.Logger))
s.crowdsecServices = make(map[types.ServiceID]bool)
// Must precede the mapping worker: the worker opens the management stream
// and reports proxyCapabilities, which reads appsecClient. Building it
// afterwards would both race the read and, when the worker won, advertise
// the proxy as AppSec-incapable for the lifetime of that stream.
if err := s.initAppSec(); err != nil {
return err
}
go s.newManagementMappingWorker(runCtx, s.mgmtClient)
@@ -420,6 +437,7 @@ func (s *Server) Start(ctx context.Context) error {
}()
s.auth = auth.NewMiddleware(s.Logger, s.mgmtClient, s.geo)
s.auth.SetAppSec(s.appsecClient)
s.accessLog = accesslog.NewLogger(s.mgmtClient, s.Logger, s.TrustedProxies)
s.startDebugEndpoint()
@@ -1283,19 +1301,17 @@ func (s *Server) newManagementMappingWorker(ctx context.Context, client proto.Pr
func (s *Server) proxyCapabilities() *proto.ProxyCapabilities {
supportsCrowdSec := s.crowdsecRegistry.Available()
supportsAppSec := s.appsecClient != nil
privateCapability := s.Private
// Always true: this build enforces ProxyMapping.private via the auth middleware.
supportsPrivateService := true
// Model discovery is handled only on SyncMappings. Management also gates
// dispatch on the live connection using this capability.
supportsModelDiscovery := true
return &proto.ProxyCapabilities{
SupportsCustomPorts: &s.SupportsCustomPorts,
RequireSubdomain: &s.RequireSubdomain,
SupportsCrowdsec: &supportsCrowdSec,
SupportsAppsec: &supportsAppSec,
Private: &privateCapability,
SupportsPrivateService: &supportsPrivateService,
SupportsModelDiscovery: &supportsModelDiscovery,
}
}
@@ -1362,8 +1378,7 @@ func isSyncUnimplemented(err error) bool {
// handleSyncMappingsStream consumes batches from a bidirectional SyncMappings
// stream, sending an ack after each batch is fully processed. Management waits
// for the ack before sending the next batch, providing application-level
// back-pressure. Model discovery commands are out-of-band: they run with
// bounded concurrency and return a correlated result instead of an ack.
// back-pressure.
func (s *Server) handleSyncMappingsStream(ctx context.Context, stream proto.ProxyService_SyncMappingsClient, initialSyncDone *bool, connectTime time.Time) error {
select {
case <-s.routerReady:
@@ -1372,30 +1387,6 @@ func (s *Server) handleSyncMappingsStream(ctx context.Context, stream proto.Prox
}
tracker := s.newSnapshotTracker(initialSyncDone, connectTime)
initialSnapshotComplete := false
discoverer := s.modelDiscoverer
if discoverer == nil {
discoverer = modeldiscovery.New(s.Logger)
}
const maxConcurrentDiscoveries = 4
discoverySlots := make(chan struct{}, maxConcurrentDiscoveries)
discoveryCtx, cancelDiscoveries := context.WithCancel(ctx)
var discoveryWG sync.WaitGroup
var sendMu sync.Mutex
defer func() {
cancelDiscoveries()
discoveryWG.Wait()
}()
// gRPC permits one concurrent sender and one concurrent receiver, but not
// multiple senders. Mapping acks and asynchronous discovery results share
// this serialized send path.
send := func(message *proto.SyncMappingsRequest) error {
sendMu.Lock()
defer sendMu.Unlock()
return stream.Send(message)
}
for {
select {
@@ -1410,62 +1401,15 @@ func (s *Server) handleSyncMappingsStream(ctx context.Context, stream proto.Prox
return fmt.Errorf("receive msg: %w", err)
}
if discovery := msg.GetModelDiscoveryRequest(); discovery != nil {
if len(msg.GetMapping()) != 0 || msg.GetInitialSyncComplete() {
return errors.New("model discovery message must not include mapping data or set initial_sync_complete")
}
if !initialSnapshotComplete {
return errors.New("model discovery request received before initial sync completed")
}
select {
case discoverySlots <- struct{}{}:
discoveryWG.Add(1)
go func() {
defer discoveryWG.Done()
defer func() { <-discoverySlots }()
result := executeModelDiscovery(discoveryCtx, discoverer, discovery)
if discoveryCtx.Err() != nil {
return
}
if err := send(&proto.SyncMappingsRequest{
Msg: &proto.SyncMappingsRequest_ModelDiscoveryResult{
ModelDiscoveryResult: result,
},
}); err != nil {
s.Logger.WithError(err).Debug("failed to send model discovery result")
}
}()
default:
result := &proto.ModelDiscoveryResult{
RequestId: discovery.GetRequestId(),
Error: "model discovery is busy",
}
if err := send(&proto.SyncMappingsRequest{
Msg: &proto.SyncMappingsRequest_ModelDiscoveryResult{
ModelDiscoveryResult: result,
},
}); err != nil {
return fmt.Errorf("send model discovery busy result: %w", err)
}
}
continue
}
batchStart := time.Now()
s.Logger.Debug("Received mapping update, starting processing")
if err := s.processMappingsGuarded(ctx, msg.GetMapping()); err != nil {
return err
}
s.Logger.Debug("Processing mapping update completed")
syncComplete := msg.GetInitialSyncComplete()
tracker.recordBatch(ctx, s, msg.GetMapping(), syncComplete, batchStart)
if syncComplete {
initialSnapshotComplete = true
}
tracker.recordBatch(ctx, s, msg.GetMapping(), msg.GetInitialSyncComplete(), batchStart)
if err := send(&proto.SyncMappingsRequest{
if err := stream.Send(&proto.SyncMappingsRequest{
Msg: &proto.SyncMappingsRequest_Ack{Ack: &proto.SyncMappingsAck{}},
}); err != nil {
return fmt.Errorf("send ack: %w", err)
@@ -1474,31 +1418,6 @@ func (s *Server) handleSyncMappingsStream(ctx context.Context, stream proto.Prox
}
}
func executeModelDiscovery(ctx context.Context, discoverer providerModelDiscoverer, request *proto.ModelDiscoveryRequest) *proto.ModelDiscoveryResult {
result := &proto.ModelDiscoveryResult{RequestId: request.GetRequestId()}
discovered, err := discoverer.Discover(ctx, modeldiscovery.Request{
UpstreamURL: request.GetUpstreamUrl(),
AuthHeaderName: request.GetAuthHeaderName(),
AuthHeaderValue: request.GetAuthHeaderValue(),
SkipTLSVerify: request.GetSkipTlsVerify(),
AllowOllamaFallback: request.GetOllamaFallback(),
})
if err != nil {
result.Error = err.Error()
return result
}
result.Source = discovered.Source
result.Models = make([]*proto.ModelDiscoveryModel, 0, len(discovered.Models))
for _, model := range discovered.Models {
result.Models = append(result.Models, &proto.ModelDiscoveryModel{
Id: model.ID,
Label: model.Label,
})
}
return result
}
// snapshotTracker accumulates service IDs during the initial snapshot and
// finalises sync state when the complete flag arrives. Used by both
// handleMappingStream and handleSyncMappingsStream so metric emission and
@@ -2018,6 +1937,67 @@ func (s *Server) parseRestrictions(mapping *proto.ProxyMapping) *restrict.Filter
})
}
// initAppSec builds the shared AppSec client when an endpoint is configured.
// A configured-but-invalid endpoint is a startup error rather than a silent
// downgrade: services asking for enforce would otherwise fail closed on every
// request with no indication why.
//
// Runs before the management stream opens so the reported capability is stable;
// the auth middleware picks the client up separately once it exists.
func (s *Server) initAppSec() error {
if s.CrowdSecAppSecURL == "" {
return nil
}
if s.CrowdSecAPIKey == "" {
return errors.New("crowdsec appsec url is set but the crowdsec api key is empty")
}
// Share the middleware capture budget rather than opening a second pool:
// AppSec buffers before authentication, so its ceiling has to count against
// the same proxy-wide allowance the body tap draws from.
var budget appsec.Budget
if s.middlewareManager != nil {
budget = s.middlewareManager.Budget()
}
client, err := appsec.New(appsec.Config{
URL: s.CrowdSecAppSecURL,
APIKey: s.CrowdSecAPIKey,
Timeout: s.CrowdSecAppSecTimeout,
MaxBodyBytes: s.CrowdSecAppSecMaxBodyBytes,
MaxConcurrent: s.CrowdSecAppSecMaxConcurrent,
Budget: budget,
Logger: log.NewEntry(s.Logger),
})
if err != nil {
return fmt.Errorf("init crowdsec appsec: %w", err)
}
s.appsecClient = client
s.Logger.Infof("CrowdSec AppSec inspection available at %s", s.CrowdSecAppSecURL)
return nil
}
// appSecMode resolves the per-service AppSec mode. A service asking for
// inspection on a proxy with no AppSec endpoint keeps its mode so the auth
// middleware fails closed for enforce, mirroring the CrowdSec behavior.
func (s *Server) appSecMode(mapping *proto.ProxyMapping) restrict.AppSecMode {
raw := mapping.GetAccessRestrictions().GetAppsecMode()
mode := restrict.ParseAppSecMode(raw)
// An unrecognized value disables inspection, which is the safe default but a
// silent one: with a newer management and an older proxy, a mode this build
// does not know would look identical to "off" on a service the operator set
// to enforce. Say so rather than leaving it to be discovered.
if mode == restrict.AppSecOff && raw != "" && raw != "off" {
s.Logger.Warnf("service %s requests unrecognized AppSec mode %q; this build supports %q and %q, so inspection is disabled",
mapping.GetId(), raw, restrict.AppSecEnforce, restrict.AppSecObserve)
}
if mode.Enabled() && s.appsecClient == nil {
s.Logger.Warnf("service %s requests AppSec mode %q but proxy has no AppSec endpoint configured", mapping.GetId(), mode)
}
return mode
}
// releaseCrowdSec releases the CrowdSec bouncer reference for the given
// service if it had one.
func (s *Server) releaseCrowdSec(svcID types.ServiceID) {
@@ -2184,7 +2164,17 @@ func (s *Server) updateMapping(ctx context.Context, mapping *proto.ProxyMapping)
s.warnIfGeoUnavailable(mapping.GetDomain(), mapping.GetAccessRestrictions())
maxSessionAge := time.Duration(mapping.GetAuth().GetMaxSessionAgeSeconds()) * time.Second
if err := s.auth.AddDomain(mapping.GetDomain(), schemes, mapping.GetAuth().GetSessionKey(), maxSessionAge, accountID, svcID, ipRestrictions, mapping.GetPrivate()); err != nil {
settings := auth.DomainSettings{
Schemes: schemes,
SessionPublicKey: mapping.GetAuth().GetSessionKey(),
SessionExpiration: maxSessionAge,
AccountID: accountID,
ServiceID: svcID,
IPRestrictions: ipRestrictions,
Private: mapping.GetPrivate(),
AppSecMode: s.appSecMode(mapping),
}
if err := s.auth.AddDomain(mapping.GetDomain(), settings); err != nil {
return fmt.Errorf("auth setup for domain %s: %w", mapping.GetDomain(), err)
}
m := s.protoToMapping(ctx, mapping)

View File

@@ -3379,6 +3379,18 @@ components:
- "observe"
default: "off"
description: CrowdSec IP reputation mode. Only available when the proxy cluster supports CrowdSec.
appsec_mode:
type: string
enum:
- "off"
- "enforce"
- "observe"
default: "off"
description: >-
CrowdSec AppSec (WAF) request inspection mode. Only available when
the proxy cluster supports AppSec, and only applied to HTTP
services. "enforce" blocks requests the WAF flags; "observe" records
the verdict in the access log without blocking.
PasswordAuthConfig:
type: object
properties:
@@ -3515,6 +3527,10 @@ components:
type: boolean
description: Whether all active proxies in the cluster have CrowdSec configured
example: false
supports_appsec:
type: boolean
description: Whether all active proxies in the cluster have a CrowdSec AppSec (WAF) endpoint configured
example: false
private:
type: boolean
description: True when at least one connected proxy in this cluster is running embedded in a netbird client (`netbird proxy`) and serving over a WireGuard tunnel. Lets the dashboard distinguish per-peer / private clusters from centralised ones.
@@ -3574,6 +3590,10 @@ components:
type: boolean
description: Whether the proxy cluster has CrowdSec configured
example: false
supports_appsec:
type: boolean
description: Whether the proxy cluster has a CrowdSec AppSec (WAF) endpoint configured
example: false
supports_private:
type: boolean
description: Whether the proxy cluster supports private (NetBird-only) services. True when at least one connected proxy in the cluster runs embedded in a netbird client.
@@ -5138,9 +5158,6 @@ components:
description: Models exposed through this endpoint, with the operator's per-1k input/output prices. Empty means all catalog models are allowed at catalog prices.
items:
$ref: '#/components/schemas/AgentNetworkProviderModel'
has_api_key:
type: boolean
description: Whether an upstream API key is currently stored. The key itself is never returned.
extra_values:
type: object
description: |
@@ -5189,7 +5206,6 @@ components:
- name
- upstream_url
- models
- has_api_key
- enabled
- skip_tls_verification
- metadata_disabled
@@ -5216,8 +5232,7 @@ components:
example: "eu.proxy.netbird.io"
api_key:
type: string
description: |
Upstream provider API key. Sealed at rest on the management server and never returned in responses. Whether a key is accepted or required is declared by the selected catalog provider's `auth_mode`. On update, omit this field to preserve the existing key or send an empty string to clear an optional key.
description: Upstream provider API key. Sealed at rest on the management server and never returned in responses. Required on create; optional on update (omit to keep the existing key).
example: "sk-..."
models:
type: array
@@ -5280,51 +5295,6 @@ components:
- id
- input_per_1k
- output_per_1k
AgentNetworkDiscoveredModel:
type: object
description: A model reported by the provider's persisted upstream endpoint.
properties:
id:
type: string
description: Exact model identifier accepted by the upstream endpoint.
example: "llama3.2:latest"
label:
type: string
description: Human-friendly label reported by the endpoint. Falls back to the model identifier.
example: "llama3.2:latest"
required:
- id
- label
AgentNetworkModelDiscoveryResponse:
type: object
description: |
Result of an explicit model-discovery probe executed by a connected
proxy in the account's selected Agent Network cluster. Discovered
models are not persisted until the provider is updated.
properties:
models:
type: array
description: Normalized, deduplicated models reported by the upstream.
items:
$ref: '#/components/schemas/AgentNetworkDiscoveredModel'
source:
type: string
description: Upstream API shape used to obtain the result.
enum: [openai_v1_models, ollama_api_tags]
example: "openai_v1_models"
proxy_cluster:
type: string
description: Proxy cluster from which the endpoint was queried.
example: "eu.proxy.netbird.io"
request_id:
type: string
description: Correlation identifier for the management-to-proxy probe.
example: "b7ef7da0-78cc-4583-8c4f-55f2ce72015f"
required:
- models
- source
- proxy_cluster
- request_id
AgentNetworkCatalogModel:
type: object
properties:
@@ -5375,11 +5345,6 @@ components:
type: string
description: Default upstream host suggested when adding a provider of this type.
example: "api.openai.com"
auth_mode:
type: string
description: Whether this provider requires, optionally accepts, or does not support an upstream API key.
enum: [required, optional, none]
example: "required"
auth_header_template:
type: string
description: Template the proxy uses to inject the API key (the literal string ${API_KEY} is replaced at request time).
@@ -5401,10 +5366,6 @@ components:
"custom" — generic OpenAI-compatible self-hosted endpoint catch-all.
enum: [provider, gateway, custom]
example: "provider"
supports_model_discovery:
type: boolean
description: Whether models can be loaded from a persisted endpoint through the selected proxy cluster.
example: false
extra_headers:
type: array
description: |
@@ -5423,12 +5384,10 @@ components:
- name
- description
- default_host
- auth_mode
- auth_header_template
- default_content_type
- brand_color
- kind
- supports_model_discovery
- models
AgentNetworkCatalogIdentityInjection:
type: object
@@ -14089,48 +14048,6 @@ paths:
"$ref": "#/components/responses/not_found"
'500':
"$ref": "#/components/responses/internal_error"
/api/agent-network/providers/{providerId}/discover-models:
post:
summary: Discover models from an Agent Network provider
description: |
Executes a bounded, read-only model-list probe from a connected proxy
in the account's selected Agent Network cluster. The probe uses the
provider's persisted endpoint, TLS setting, and stored credential.
Returned models are not persisted by this operation.
tags: [ Agent Network ]
security:
- BearerAuth: [ ]
- TokenAuth: [ ]
parameters:
- in: path
name: providerId
required: true
schema:
type: string
description: The unique identifier of a persisted Agent Network provider
responses:
'200':
description: Models discovered successfully
content:
application/json:
schema:
$ref: '#/components/schemas/AgentNetworkModelDiscoveryResponse'
'400':
"$ref": "#/components/responses/bad_request"
'401':
"$ref": "#/components/responses/requires_authentication"
'403':
"$ref": "#/components/responses/forbidden"
'404':
"$ref": "#/components/responses/not_found"
'412':
description: The provider cannot be queried or no capable proxy is connected
content: { }
'429':
description: A discovery probe is already running or was requested too recently
content: { }
'500':
"$ref": "#/components/responses/internal_error"
/api/agent-network/policies:
get:
summary: List all Agent Network Policies

View File

@@ -17,6 +17,27 @@ const (
TokenAuthScopes tokenAuthContextKey = "TokenAuth.Scopes"
)
// Defines values for AccessRestrictionsAppsecMode.
const (
AccessRestrictionsAppsecModeEnforce AccessRestrictionsAppsecMode = "enforce"
AccessRestrictionsAppsecModeObserve AccessRestrictionsAppsecMode = "observe"
AccessRestrictionsAppsecModeOff AccessRestrictionsAppsecMode = "off"
)
// Valid indicates whether the value is a known member of the AccessRestrictionsAppsecMode enum.
func (e AccessRestrictionsAppsecMode) Valid() bool {
switch e {
case AccessRestrictionsAppsecModeEnforce:
return true
case AccessRestrictionsAppsecModeObserve:
return true
case AccessRestrictionsAppsecModeOff:
return true
default:
return false
}
}
// Defines values for AccessRestrictionsCrowdsecMode.
const (
AccessRestrictionsCrowdsecModeEnforce AccessRestrictionsCrowdsecMode = "enforce"
@@ -38,27 +59,6 @@ func (e AccessRestrictionsCrowdsecMode) Valid() bool {
}
}
// Defines values for AgentNetworkCatalogProviderAuthMode.
const (
AgentNetworkCatalogProviderAuthModeNone AgentNetworkCatalogProviderAuthMode = "none"
AgentNetworkCatalogProviderAuthModeOptional AgentNetworkCatalogProviderAuthMode = "optional"
AgentNetworkCatalogProviderAuthModeRequired AgentNetworkCatalogProviderAuthMode = "required"
)
// Valid indicates whether the value is a known member of the AgentNetworkCatalogProviderAuthMode enum.
func (e AgentNetworkCatalogProviderAuthMode) Valid() bool {
switch e {
case AgentNetworkCatalogProviderAuthModeNone:
return true
case AgentNetworkCatalogProviderAuthModeOptional:
return true
case AgentNetworkCatalogProviderAuthModeRequired:
return true
default:
return false
}
}
// Defines values for AgentNetworkCatalogProviderKind.
const (
AgentNetworkCatalogProviderKindCustom AgentNetworkCatalogProviderKind = "custom"
@@ -98,24 +98,6 @@ func (e AgentNetworkConsumptionDimensionKind) Valid() bool {
}
}
// Defines values for AgentNetworkModelDiscoveryResponseSource.
const (
AgentNetworkModelDiscoveryResponseSourceOllamaApiTags AgentNetworkModelDiscoveryResponseSource = "ollama_api_tags"
AgentNetworkModelDiscoveryResponseSourceOpenaiV1Models AgentNetworkModelDiscoveryResponseSource = "openai_v1_models"
)
// Valid indicates whether the value is a known member of the AgentNetworkModelDiscoveryResponseSource enum.
func (e AgentNetworkModelDiscoveryResponseSource) Valid() bool {
switch e {
case AgentNetworkModelDiscoveryResponseSourceOllamaApiTags:
return true
case AgentNetworkModelDiscoveryResponseSourceOpenaiV1Models:
return true
default:
return false
}
}
// Defines values for CreateAzureIntegrationRequestHost.
const (
CreateAzureIntegrationRequestHostMicrosoftCom CreateAzureIntegrationRequestHost = "microsoft.com"
@@ -1579,6 +1561,9 @@ type AccessRestrictions struct {
// AllowedCountries ISO 3166-1 alpha-2 country codes to allow. If non-empty, only these countries are permitted.
AllowedCountries *[]string `json:"allowed_countries,omitempty"`
// AppsecMode CrowdSec AppSec (WAF) request inspection mode. Only available when the proxy cluster supports AppSec, and only applied to HTTP services. "enforce" blocks requests the WAF flags; "observe" records the verdict in the access log without blocking.
AppsecMode *AccessRestrictionsAppsecMode `json:"appsec_mode,omitempty"`
// BlockedCidrs CIDR blocklist. Connections from these CIDRs are rejected. Evaluated after allowed_cidrs.
BlockedCidrs *[]string `json:"blocked_cidrs,omitempty"`
@@ -1589,6 +1574,9 @@ type AccessRestrictions struct {
CrowdsecMode *AccessRestrictionsCrowdsecMode `json:"crowdsec_mode,omitempty"`
}
// AccessRestrictionsAppsecMode CrowdSec AppSec (WAF) request inspection mode. Only available when the proxy cluster supports AppSec, and only applied to HTTP services. "enforce" blocks requests the WAF flags; "observe" records the verdict in the access log without blocking.
type AccessRestrictionsAppsecMode string
// AccessRestrictionsCrowdsecMode CrowdSec IP reputation mode. Only available when the proxy cluster supports CrowdSec.
type AccessRestrictionsCrowdsecMode string
@@ -2077,9 +2065,6 @@ type AgentNetworkCatalogProvider struct {
// AuthHeaderTemplate Template the proxy uses to inject the API key (the literal string ${API_KEY} is replaced at request time).
AuthHeaderTemplate string `json:"auth_header_template"`
// AuthMode Whether this provider requires, optionally accepts, or does not support an upstream API key.
AuthMode AgentNetworkCatalogProviderAuthMode `json:"auth_mode"`
// BrandColor Hex brand color used to render the provider badge in the dashboard.
BrandColor string `json:"brand_color"`
@@ -2112,14 +2097,8 @@ type AgentNetworkCatalogProvider struct {
// Name Display name for the provider.
Name string `json:"name"`
// SupportsModelDiscovery Whether models can be loaded from a persisted endpoint through the selected proxy cluster.
SupportsModelDiscovery bool `json:"supports_model_discovery"`
}
// AgentNetworkCatalogProviderAuthMode Whether this provider requires, optionally accepts, or does not support an upstream API key.
type AgentNetworkCatalogProviderAuthMode string
// AgentNetworkCatalogProviderKind Presentation grouping for the provider Select on the dashboard.
// "provider" — first-party vendor API (OpenAI, Anthropic, …); the upstream is the model itself.
// "gateway" — routing/aggregation layer in front of multiple providers (LiteLLM, Portkey, …); typically pairs with NetBird identity stamping.
@@ -2156,15 +2135,6 @@ type AgentNetworkConsumption struct {
// AgentNetworkConsumptionDimensionKind Whether this row counts a single end user or a single source group across every member.
type AgentNetworkConsumptionDimensionKind string
// AgentNetworkDiscoveredModel A model reported by the provider's persisted upstream endpoint.
type AgentNetworkDiscoveredModel struct {
// Id Exact model identifier accepted by the upstream endpoint.
Id string `json:"id"`
// Label Human-friendly label reported by the endpoint. Falls back to the model identifier.
Label string `json:"label"`
}
// AgentNetworkGuardrail defines model for AgentNetworkGuardrail.
type AgentNetworkGuardrail struct {
// Checks Guardrail check parameters. Each entry has an `enabled` flag plus per-check configuration; disabled entries are inert.
@@ -2212,26 +2182,6 @@ type AgentNetworkGuardrailRequest struct {
Name string `json:"name"`
}
// AgentNetworkModelDiscoveryResponse Result of an explicit model-discovery probe executed by a connected
// proxy in the account's selected Agent Network cluster. Discovered
// models are not persisted until the provider is updated.
type AgentNetworkModelDiscoveryResponse struct {
// Models Normalized, deduplicated models reported by the upstream.
Models []AgentNetworkDiscoveredModel `json:"models"`
// ProxyCluster Proxy cluster from which the endpoint was queried.
ProxyCluster string `json:"proxy_cluster"`
// RequestId Correlation identifier for the management-to-proxy probe.
RequestId string `json:"request_id"`
// Source Upstream API shape used to obtain the result.
Source AgentNetworkModelDiscoveryResponseSource `json:"source"`
}
// AgentNetworkModelDiscoveryResponseSource Upstream API shape used to obtain the result.
type AgentNetworkModelDiscoveryResponseSource string
// AgentNetworkPolicy defines model for AgentNetworkPolicy.
type AgentNetworkPolicy struct {
// CreatedAt Timestamp when the policy was created.
@@ -2337,9 +2287,6 @@ type AgentNetworkProvider struct {
// ExtraValues Operator-typed values for catalog-declared extra headers. Keys are wire header names (e.g. `x-portkey-config`); values are the strings the proxy stamps on every upstream request to this provider. Catalog (AgentNetworkCatalogProvider.extra_headers) declares which keys are accepted; values not declared by the catalog are ignored at synth time. Empty / missing values mean no header stamped.
ExtraValues *map[string]string `json:"extra_values,omitempty"`
// HasApiKey Whether an upstream API key is currently stored. The key itself is never returned.
HasApiKey bool `json:"has_api_key"`
// Id Provider ID
Id string `json:"id"`
@@ -2385,7 +2332,7 @@ type AgentNetworkProviderModel struct {
// AgentNetworkProviderRequest defines model for AgentNetworkProviderRequest.
type AgentNetworkProviderRequest struct {
// ApiKey Upstream provider API key. Sealed at rest on the management server and never returned in responses. Whether a key is accepted or required is declared by the selected catalog provider's `auth_mode`. On update, omit this field to preserve the existing key or send an empty string to clear an optional key.
// ApiKey Upstream provider API key. Sealed at rest on the management server and never returned in responses. Required on create; optional on update (omit to keep the existing key).
ApiKey *string `json:"api_key,omitempty"`
// BootstrapCluster Proxy cluster used to bootstrap the per-account agent-network endpoint when the first provider is created. Ignored on subsequent creates and on updates because the cluster is pinned on the account-level Settings row.
@@ -4808,6 +4755,9 @@ type ProxyCluster struct {
// RequireSubdomain Whether services on this cluster must include a subdomain label
RequireSubdomain *bool `json:"require_subdomain,omitempty"`
// SupportsAppsec Whether all active proxies in the cluster have a CrowdSec AppSec (WAF) endpoint configured
SupportsAppsec *bool `json:"supports_appsec,omitempty"`
// SupportsCrowdsec Whether all active proxies in the cluster have CrowdSec configured
SupportsCrowdsec *bool `json:"supports_crowdsec,omitempty"`
@@ -4876,6 +4826,9 @@ type ReverseProxyDomain struct {
// RequireSubdomain Whether a subdomain label is required in front of this domain. When true, the domain cannot be used bare.
RequireSubdomain *bool `json:"require_subdomain,omitempty"`
// SupportsAppsec Whether the proxy cluster has a CrowdSec AppSec (WAF) endpoint configured
SupportsAppsec *bool `json:"supports_appsec,omitempty"`
// SupportsCrowdsec Whether the proxy cluster has CrowdSec configured
SupportsCrowdsec *bool `json:"supports_crowdsec,omitempty"`

File diff suppressed because it is too large Load Diff

View File

@@ -73,10 +73,10 @@ message ProxyCapabilities {
optional bool private = 4;
// Whether the proxy enforces ProxyMapping.private (fails closed on ValidateTunnelPeer failure). Management MUST NOT stream private mappings to proxies that don't claim this.
optional bool supports_private_service = 5;
// Whether the proxy can execute correlated Agent Network model-discovery
// requests delivered over SyncMappings. Management must not send discovery
// requests to proxies that omit or disable this capability.
optional bool supports_model_discovery = 6;
// Whether the proxy has a CrowdSec AppSec (WAF) endpoint configured and can
// inspect HTTP requests. Independent of supports_crowdsec: AppSec is a
// separate endpoint on the Security Engine and applies to HTTP services only.
optional bool supports_appsec = 6;
}
// GetMappingUpdateRequest is sent to initialise a mapping stream.
@@ -207,6 +207,10 @@ message AccessRestrictions {
repeated string blocked_countries = 4;
// CrowdSec IP reputation mode: "", "off", "enforce", or "observe".
string crowdsec_mode = 5;
// CrowdSec AppSec (WAF) request inspection mode: "", "off", "enforce", or
// "observe". HTTP services only: "enforce" and "observe" are rejected at
// validation for TCP/UDP/TLS services, which carry no requests to inspect.
string appsec_mode = 7;
}
message ProxyMapping {
@@ -424,8 +428,6 @@ message SyncMappingsRequest {
oneof msg {
SyncMappingsInit init = 1;
SyncMappingsAck ack = 2;
// Correlated response to a ModelDiscoveryRequest received from management.
ModelDiscoveryResult model_discovery_result = 3;
}
}
@@ -449,9 +451,6 @@ message SyncMappingsResponse {
repeated ProxyMapping mapping = 1;
// initial_sync_complete is set on the last message of the initial snapshot.
bool initial_sync_complete = 2;
// An out-of-band model-discovery operation. This is sent only after the
// initial snapshot and only to proxies advertising supports_model_discovery.
ModelDiscoveryRequest model_discovery_request = 3;
}
// CheckLLMPolicyLimitsRequest carries the resolved caller identity and the
@@ -510,33 +509,3 @@ message RecordLLMUsageRequest {
message RecordLLMUsageResponse {
}
// ModelDiscoveryRequest asks a proxy to list models from a persisted provider
// endpoint using the proxy host's network path. request_id is assigned by
// management and echoed verbatim in ModelDiscoveryResult.
message ModelDiscoveryRequest {
string request_id = 1;
string upstream_url = 2;
string auth_header_name = 3;
string auth_header_value = 4;
bool skip_tls_verify = 5;
// When true, the proxy may fall back from the OpenAI-compatible
// GET /v1/models endpoint to Ollama's native GET /api/tags endpoint.
bool ollama_fallback = 6;
}
// ModelDiscoveryModel is the normalized model shape returned to management.
// Arbitrary upstream fields never cross the proxy control channel.
message ModelDiscoveryModel {
string id = 1;
string label = 2;
}
// ModelDiscoveryResult completes one ModelDiscoveryRequest. error is empty on
// success and contains only a sanitized diagnostic on failure.
message ModelDiscoveryResult {
string request_id = 1;
repeated ModelDiscoveryModel models = 2;
// Stable source label, currently openai_v1_models or ollama_api_tags.
string source = 3;
string error = 4;
}