Compare commits

..

5 Commits

Author SHA1 Message Date
Brandon Hopkins
fea131101d API-key normalization and model-discovery snapshot validation 2026-07-26 18:58:39 -07:00
Brandon Hopkins
0463b30152 Ollama Cloud provider integration and backend test coverage 2026-07-26 18:19:38 -07:00
Brandon Hopkins
d3909e4faf endpoint model discovery and proxy integration 2026-07-26 16:38:01 -07:00
Brandon Hopkins
a92cdb7dcd authentication semantics, API contracts, and tests 2026-07-26 15:37:41 -07:00
Brandon Hopkins
564abb660c Add self-hosted Ollama 2026-07-26 15:05:01 -07:00
68 changed files with 4623 additions and 3243 deletions

View File

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

View File

@@ -7,7 +7,6 @@ import (
"fmt"
"os"
"slices"
"strings"
"sync"
"time"
@@ -300,13 +299,6 @@ func (c *Client) SetInfoLogLevel() {
// PeersList return with the list of the PeerInfos
func (c *Client) PeersList() *PeerInfoArray {
// The recorder only caches transfer counters and handshake times; nothing
// refreshes them on its own, so without this they read as zero. The desktop
// daemon does the same before serving a full peer status.
if err := c.recorder.RefreshWireGuardStats(); err != nil {
log.Debugf("failed to refresh WireGuard stats: %v", err)
}
fullStatus := c.recorder.GetFullStatus()
peerInfos := make([]PeerInfo, len(fullStatus.Peers))
@@ -317,20 +309,6 @@ func (c *Client) PeersList() *PeerInfoArray {
FQDN: p.FQDN,
ConnStatus: int(p.ConnStatus),
Routes: PeerRoutes{routes: maps.Keys(p.GetRoutes())},
PubKey: p.PubKey,
Latency: formatDuration(p.Latency),
LatencyMs: p.Latency.Milliseconds(),
BytesRx: p.BytesRx,
BytesTx: p.BytesTx,
ConnStatusUpdate: formatTime(p.ConnStatusUpdate),
Relayed: p.Relayed,
RosenpassEnabled: p.RosenpassEnabled,
LastWireguardHandshake: formatTime(p.LastWireguardHandshake),
LocalIceCandidateType: p.LocalIceCandidateType,
RemoteIceCandidateType: p.RemoteIceCandidateType,
LocalIceCandidateEndpoint: p.LocalIceCandidateEndpoint,
RemoteIceCandidateEndpoint: p.RemoteIceCandidateEndpoint,
}
peerInfos[n] = pi
}
@@ -461,6 +439,10 @@ func (c *Client) RemoveConnectionListener() {
c.recorder.RemoveConnectionListener()
}
func (c *Client) toggleRoute(command routeCommand) error {
return command.toggleRoute()
}
func (c *Client) getRouteManager() (routemanager.Manager, error) {
client := c.getConnectClient()
if client == nil {
@@ -480,22 +462,22 @@ func (c *Client) getRouteManager() (routemanager.Manager, error) {
return manager, nil
}
func (c *Client) SelectRoute(id string) error {
func (c *Client) SelectRoute(route string) error {
manager, err := c.getRouteManager()
if err != nil {
return err
}
return manager.SelectRoutes([]route.NetID{route.NetID(id)}, true)
return c.toggleRoute(selectRouteCommand{route: route, manager: manager})
}
func (c *Client) DeselectRoute(id string) error {
func (c *Client) DeselectRoute(route string) error {
manager, err := c.getRouteManager()
if err != nil {
return err
}
return manager.DeselectRoutes([]route.NetID{route.NetID(id)})
return c.toggleRoute(deselectRouteCommand{route: route, manager: manager})
}
// getNetworkDomainsFromRoute extracts domains from a route and enriches each domain
@@ -530,28 +512,3 @@ func exportEnvList(list *EnvList) {
}
}
}
// formatDuration renders a duration for display, trimming the fractional part
// to two digits so latencies read as "12.34ms" rather than "12.345678ms".
func formatDuration(d time.Duration) string {
ds := d.String()
dotIndex := strings.Index(ds, ".")
if dotIndex == -1 {
return ds
}
endIndex := min(dotIndex+3, len(ds))
// Skip the remaining digits so only the unit suffix is appended back.
unitStart := endIndex
for unitStart < len(ds) && ds[unitStart] >= '0' && ds[unitStart] <= '9' {
unitStart++
}
return ds[:endIndex] + ds[unitStart:]
}
// formatTime renders a timestamp in UTC using a fixed layout. The zero time is
// passed through as-is so the UI can recognise it and show "never" instead.
func formatTime(t time.Time) string {
return t.UTC().Format("2006-01-02 15:04:05")
}

View File

@@ -12,30 +12,12 @@ const (
)
// PeerInfo describe information about the peers. It designed for the UI usage
//
// The fields below ConnStatus back the peer detail screen. Durations and times
// are pre-formatted into strings so the UI does not have to know Go's layouts;
// Latency is additionally exposed as LatencyMs for colour coding.
type PeerInfo struct {
IP string
IPv6 string
FQDN string
ConnStatus int
Routes PeerRoutes
PubKey string
Latency string
LatencyMs int64
BytesRx int64
BytesTx int64
ConnStatusUpdate string
Relayed bool
RosenpassEnabled bool
LastWireguardHandshake string
LocalIceCandidateType string
RemoteIceCandidateType string
LocalIceCandidateEndpoint string
RemoteIceCandidateEndpoint string
}
func (p *PeerInfo) GetPeerRoutes() *PeerRoutes {

View File

@@ -189,19 +189,6 @@ func (pm *ProfileManager) LogoutProfile(id string) error {
return nil
}
// RenameProfile changes a profile's display name. The profile ID, and therefore
// its on-disk filename, is left untouched: only the "name" field of the config
// is rewritten. This works for the default profile too, whose config lives in
// netbird.cfg rather than under profiles/.
func (pm *ProfileManager) RenameProfile(id string, newName string) error {
if err := pm.serviceMgr.RenameProfile(profilemanager.ID(id), androidUsername, newName); err != nil {
return fmt.Errorf("failed to rename profile: %w", err)
}
log.Infof("renamed profile %s to: %s", id, newName)
return nil
}
// RemoveProfile deletes a profile
func (pm *ProfileManager) RemoveProfile(id string) error {
// Use ServiceManager (removes profile from profiles/ directory)

View File

@@ -0,0 +1,70 @@
//go:build android
package android
import (
"fmt"
log "github.com/sirupsen/logrus"
"golang.org/x/exp/maps"
"github.com/netbirdio/netbird/client/internal/routemanager"
"github.com/netbirdio/netbird/route"
)
func executeRouteToggle(id string, manager routemanager.Manager,
operationName string,
routeOperation func(routes []route.NetID, allRoutes []route.NetID) error) error {
netID := route.NetID(id)
routes := []route.NetID{netID}
routesMap := manager.GetClientRoutesWithNetID()
routes = route.ExpandV6ExitPairs(routes, routesMap)
log.Debugf("%s with ids: %v", operationName, routes)
if err := routeOperation(routes, maps.Keys(routesMap)); err != nil {
log.Debugf("error when %s: %s", operationName, err)
return fmt.Errorf("error %s: %w", operationName, err)
}
manager.TriggerSelection(manager.GetClientRoutes())
return nil
}
type routeCommand interface {
toggleRoute() error
}
type selectRouteCommand struct {
route string
manager routemanager.Manager
}
func (s selectRouteCommand) toggleRoute() error {
routeSelector := s.manager.GetRouteSelector()
if routeSelector == nil {
return fmt.Errorf("no route selector available")
}
routeOperation := func(routes []route.NetID, allRoutes []route.NetID) error {
return routeSelector.SelectRoutes(routes, true, allRoutes)
}
return executeRouteToggle(s.route, s.manager, "selecting route", routeOperation)
}
type deselectRouteCommand struct {
route string
manager routemanager.Manager
}
func (d deselectRouteCommand) toggleRoute() error {
routeSelector := d.manager.GetRouteSelector()
if routeSelector == nil {
return fmt.Errorf("no route selector available")
}
return executeRouteToggle(d.route, d.manager, "deselecting route", routeSelector.DeselectRoutes)
}

View File

@@ -52,10 +52,6 @@ type Manager interface {
UpdateRoutes(updateSerial uint64, serverRoutes map[route.ID]*route.Route, clientRoutes route.HAMap, useNewDNSRoute bool) error
ClassifyRoutes(newRoutes []*route.Route) (map[route.ID]*route.Route, route.HAMap)
TriggerSelection(route.HAMap)
SelectRoutes(ids []route.NetID, appendRoute bool) error
DeselectRoutes(ids []route.NetID) error
SelectAllRoutes()
DeselectAllRoutes()
GetRouteSelector() *routeselector.RouteSelector
GetClientRoutes() route.HAMap
GetSelectedClientRoutes() route.HAMap
@@ -804,7 +800,7 @@ func (m *DefaultManager) collectExitNodeInfo(clientRoutes route.HAMap) exitNodeI
var info exitNodeInfo
for haID, routes := range clientRoutes {
if !isExitNodeRoutes(routes) {
if !m.isExitNodeRoute(routes) {
continue
}
@@ -824,6 +820,13 @@ func (m *DefaultManager) collectExitNodeInfo(clientRoutes route.HAMap) exitNodeI
return info
}
func (m *DefaultManager) isExitNodeRoute(routes []*route.Route) bool {
if len(routes) == 0 {
return false
}
return route.IsV4DefaultRoute(routes[0].Network) || route.IsV6DefaultRoute(routes[0].Network)
}
func (m *DefaultManager) categorizeUserSelection(netID route.NetID, info *exitNodeInfo) {
if m.routeSelector.IsSelected(netID) {
info.userSelected = append(info.userSelected, netID)

View File

@@ -16,8 +16,6 @@ type MockManager struct {
ClassifyRoutesFunc func(routes []*route.Route) (map[route.ID]*route.Route, route.HAMap)
UpdateRoutesFunc func(updateSerial uint64, serverRoutes map[route.ID]*route.Route, clientRoutes route.HAMap, useNewDNSRoute bool) error
TriggerSelectionFunc func(haMap route.HAMap)
SelectRoutesFunc func(ids []route.NetID, appendRoute bool) error
DeselectRoutesFunc func(ids []route.NetID) error
GetRouteSelectorFunc func() *routeselector.RouteSelector
GetClientRoutesFunc func() route.HAMap
GetSelectedClientRoutesFunc func() route.HAMap
@@ -57,30 +55,6 @@ func (m *MockManager) TriggerSelection(networks route.HAMap) {
}
}
// SelectRoutes mock implementation of SelectRoutes from Manager interface
func (m *MockManager) SelectRoutes(ids []route.NetID, appendRoute bool) error {
if m.SelectRoutesFunc != nil {
return m.SelectRoutesFunc(ids, appendRoute)
}
return nil
}
// DeselectRoutes mock implementation of DeselectRoutes from Manager interface
func (m *MockManager) DeselectRoutes(ids []route.NetID) error {
if m.DeselectRoutesFunc != nil {
return m.DeselectRoutesFunc(ids)
}
return nil
}
// SelectAllRoutes mock implementation of SelectAllRoutes from Manager interface
func (m *MockManager) SelectAllRoutes() {
}
// DeselectAllRoutes mock implementation of DeselectAllRoutes from Manager interface
func (m *MockManager) DeselectAllRoutes() {
}
// GetRouteSelector mock implementation of GetRouteSelector from Manager interface
func (m *MockManager) GetRouteSelector() *routeselector.RouteSelector {
if m.GetRouteSelectorFunc != nil {

View File

@@ -1,138 +0,0 @@
package routemanager
import (
"fmt"
"slices"
"github.com/hashicorp/go-multierror"
log "github.com/sirupsen/logrus"
"golang.org/x/exp/maps"
nberrors "github.com/netbirdio/netbird/client/errors"
"github.com/netbirdio/netbird/route"
)
// SelectRoutes selects the routes with the given network IDs and applies the
// new selection. V4/v6 exit-node pairs are expanded automatically. Exit nodes
// are mutually exclusive: if the selection activates an exit node, every other
// available exit node is deselected so two can't be active at once. With
// appendRoute=false the previous selection is replaced instead of extended.
func (m *DefaultManager) SelectRoutes(ids []route.NetID, appendRoute bool) error {
if err := m.selectRoutes(ids, appendRoute); err != nil {
return err
}
m.TriggerSelection(m.GetClientRoutes())
return nil
}
// DeselectRoutes removes the routes with the given network IDs from the
// selection and applies the change. V4/v6 exit-node pairs are expanded
// automatically.
func (m *DefaultManager) DeselectRoutes(ids []route.NetID) error {
if err := m.deselectRoutes(ids); err != nil {
return err
}
m.TriggerSelection(m.GetClientRoutes())
return nil
}
func (m *DefaultManager) deselectRoutes(ids []route.NetID) error {
routesMap := m.GetClientRoutesWithNetID()
routes := route.ExpandV6ExitPairs(slices.Clone(ids), routesMap)
log.Debugf("deselecting routes with ids: %v", routes)
if err := m.routeSelector.DeselectRoutes(routes, maps.Keys(routesMap)); err != nil {
return fmt.Errorf("deselect routes: %w", err)
}
return nil
}
// SelectAllRoutes selects every available route and applies the selection.
// Exit nodes stay mutually exclusive: at most one remains active.
func (m *DefaultManager) SelectAllRoutes() {
m.selectAllRoutes()
m.TriggerSelection(m.GetClientRoutes())
}
func (m *DefaultManager) selectAllRoutes() {
m.routeSelector.SelectAllRoutes()
// Select-all wipes every explicit selection, so exit nodes fall back to
// management's auto-apply flags — which may mark several at once.
// Reconcile immediately so at most one exit node stays active instead of
// waiting for the next network map to enforce it.
m.mux.Lock()
defer m.mux.Unlock()
m.updateRouteSelectorFromManagement(m.clientRoutes)
}
// DeselectAllRoutes deselects every route and applies the change.
func (m *DefaultManager) DeselectAllRoutes() {
m.routeSelector.DeselectAllRoutes()
m.TriggerSelection(m.GetClientRoutes())
}
func (m *DefaultManager) selectRoutes(ids []route.NetID, appendRoute bool) error {
routesMap := m.GetClientRoutesWithNetID()
routes := route.ExpandV6ExitPairs(slices.Clone(ids), routesMap)
allIDs := maps.Keys(routesMap)
log.Debugf("selecting routes with ids: %v", routes)
// A partial failure (e.g. an unknown ID in the request) still selects the
// valid routes, so exclusivity below must run regardless of the error.
var merr *multierror.Error
if err := m.routeSelector.SelectRoutes(routes, appendRoute, allIDs); err != nil {
merr = multierror.Append(merr, fmt.Errorf("select routes: %w", err))
}
// Exit nodes are mutually exclusive: if this selection activates an
// exit node, deselect every other available exit node so two can't be
// selected at once. Non-exit route selections are left untouched.
if requestActivatesExitNode(routes, routesMap) {
if others := otherExitNodeIDs(routesMap, routes); len(others) > 0 {
if err := m.routeSelector.DeselectRoutes(others, allIDs); err != nil {
merr = multierror.Append(merr, fmt.Errorf("deselect sibling exit nodes: %w", err))
}
}
}
return nberrors.FormatErrorOrNil(merr)
}
func isExitNodeRoutes(routes []*route.Route) bool {
return len(routes) > 0 && (route.IsV4DefaultRoute(routes[0].Network) || route.IsV6DefaultRoute(routes[0].Network))
}
// requestActivatesExitNode reports whether any requested NetID maps to an exit
// node (default route) in the current route table.
func requestActivatesExitNode(requested []route.NetID, routesMap map[route.NetID][]*route.Route) bool {
for _, id := range requested {
if isExitNodeRoutes(routesMap[id]) {
return true
}
}
return false
}
// otherExitNodeIDs returns every available exit-node NetID that is not in the
// requested set — the siblings to deselect so a single exit node stays active.
func otherExitNodeIDs(routesMap map[route.NetID][]*route.Route, requested []route.NetID) []route.NetID {
keep := make(map[route.NetID]struct{}, len(requested))
for _, id := range requested {
keep[id] = struct{}{}
}
var others []route.NetID
for id, routes := range routesMap {
if !isExitNodeRoutes(routes) {
continue
}
if _, ok := keep[id]; ok {
continue
}
others = append(others, id)
}
return others
}

View File

@@ -1,129 +0,0 @@
package routemanager
import (
"net/netip"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/client/internal/routeselector"
"github.com/netbirdio/netbird/route"
)
func v6ExitRoute(netID, peer string) *route.Route {
return &route.Route{
NetID: route.NetID(netID),
Network: netip.MustParsePrefix("::/0"),
Peer: peer,
}
}
func newSelectionTestManager() *DefaultManager {
return &DefaultManager{
routeSelector: routeselector.NewRouteSelector(),
clientRoutes: route.HAMap{
"exitA|0.0.0.0/0": {exitRoute("exitA", "p1", true)},
"exitA-v6|::/0": {v6ExitRoute("exitA-v6", "p1")},
"exitB|0.0.0.0/0": {exitRoute("exitB", "p2", true)},
"lan|192.168.1.0/24": {{NetID: "lan", Network: netip.MustParsePrefix("192.168.1.0/24"), Peer: "p3"}},
},
}
}
func TestSelectRoutes_ExitNodeExclusivity(t *testing.T) {
m := newSelectionTestManager()
// Selecting an exit node selects its v6 pair and deselects the sibling.
require.NoError(t, m.selectRoutes([]route.NetID{"exitA"}, true))
assert.True(t, m.routeSelector.IsSelected("exitA"), "exitA should be selected")
assert.True(t, m.routeSelector.IsSelected("exitA-v6"), "the v6 pair follows its v4 base")
assert.False(t, m.routeSelector.IsSelected("exitB"), "the sibling exit node must be deselected")
// Switching to the sibling deselects the previous exit node and its v6 pair.
require.NoError(t, m.selectRoutes([]route.NetID{"exitB"}, true))
assert.True(t, m.routeSelector.IsSelected("exitB"), "exitB should now be selected")
assert.False(t, m.routeSelector.IsSelected("exitA"), "the previous exit node must be deselected")
assert.False(t, m.routeSelector.IsSelected("exitA-v6"), "the previous exit node's v6 pair must be deselected")
assert.True(t, m.routeSelector.IsSelected("lan"), "non-exit route selection is untouched")
// Selecting a non-exit route leaves the active exit node alone.
require.NoError(t, m.selectRoutes([]route.NetID{"lan"}, true))
assert.True(t, m.routeSelector.IsSelected("exitB"), "selecting a non-exit route keeps the exit node")
// Deselecting the active exit node turns every exit node off.
require.NoError(t, m.deselectRoutes([]route.NetID{"exitB"}))
assert.False(t, m.routeSelector.IsSelected("exitB"), "exitB should be deselected")
assert.False(t, m.routeSelector.IsSelected("exitA"), "exitA stays deselected")
assert.True(t, m.routeSelector.IsSelected("lan"), "non-exit route selection is untouched")
}
func TestSelectRoutes_PartialErrorStillEnforcesExclusivity(t *testing.T) {
// The unknown ID must be reported, but the valid exit node in the same
// request is still selected — so its sibling must still be deselected.
// Both orderings are covered: processing must continue past the invalid
// ID wherever it sits in the request.
requests := map[string][]route.NetID{
"invalid id first": {"missing", "exitB"},
"invalid id last": {"exitB", "missing"},
}
for name, ids := range requests {
t.Run(name, func(t *testing.T) {
m := newSelectionTestManager()
require.NoError(t, m.selectRoutes([]route.NetID{"exitA"}, true))
err := m.selectRoutes(ids, true)
assert.Error(t, err, "unknown id must be reported")
assert.True(t, m.routeSelector.IsSelected("exitB"), "valid exit node from the request is selected")
assert.False(t, m.routeSelector.IsSelected("exitA"), "sibling exit node must be deselected despite the error")
assert.False(t, m.routeSelector.IsSelected("exitA-v6"), "sibling's v6 pair must be deselected too")
})
}
}
func TestSelectAllRoutes_KeepsSingleExitNode(t *testing.T) {
// Both exit nodes are marked for auto-apply by management
// (SkipAutoApply=false), the state where select-all could turn on two at
// once without the immediate reconciliation.
m := &DefaultManager{
routeSelector: routeselector.NewRouteSelector(),
clientRoutes: route.HAMap{
"exitA|0.0.0.0/0": {exitRoute("exitA", "p1", false)},
"exitB|0.0.0.0/0": {exitRoute("exitB", "p2", false)},
"lan|192.168.1.0/24": {{NetID: "lan", Network: netip.MustParsePrefix("192.168.1.0/24"), Peer: "p3"}},
},
}
require.NoError(t, m.selectRoutes([]route.NetID{"exitB"}, true))
m.selectAllRoutes()
assert.True(t, m.routeSelector.IsSelected("lan"), "non-exit routes are all selected")
assert.True(t, m.routeSelector.IsSelected("exitA"), "the deterministic management pick stays active")
assert.False(t, m.routeSelector.IsSelected("exitB"), "select-all must not leave a second exit node active")
}
func TestSelectRoutes_UnknownRoute(t *testing.T) {
m := newSelectionTestManager()
assert.Error(t, m.selectRoutes([]route.NetID{"missing"}, true), "selecting an unavailable route must fail")
assert.Error(t, m.deselectRoutes([]route.NetID{"missing"}), "deselecting an unavailable route must fail")
}
func TestExitNodeSelectionHelpers(t *testing.T) {
routesMap := map[route.NetID][]*route.Route{
"exitA": {{Network: netip.MustParsePrefix("0.0.0.0/0")}},
"exitB": {{Network: netip.MustParsePrefix("::/0")}},
"lan": {{Network: netip.MustParsePrefix("192.168.0.0/16")}},
}
assert.True(t, requestActivatesExitNode([]route.NetID{"exitA"}, routesMap), "v4 default route is an exit node")
assert.True(t, requestActivatesExitNode([]route.NetID{"exitB"}, routesMap), "v6 default route is an exit node")
assert.False(t, requestActivatesExitNode([]route.NetID{"lan"}, routesMap), "lan route is not an exit node")
assert.False(t, requestActivatesExitNode([]route.NetID{"missing"}, routesMap), "unknown id is not an exit node")
others := otherExitNodeIDs(routesMap, []route.NetID{"exitB"})
assert.ElementsMatch(t, []route.NetID{"exitA"}, others, "only the other exit node is a sibling; the lan route is ignored")
}

View File

@@ -13,6 +13,7 @@ import (
"time"
log "github.com/sirupsen/logrus"
"golang.org/x/exp/maps"
"github.com/netbirdio/netbird/client/internal"
"github.com/netbirdio/netbird/client/internal/auth"
@@ -636,18 +637,23 @@ func (c *Client) SelectRoute(id string) error {
}
routeManager := engine.GetRouteManager()
routeSelector := routeManager.GetRouteSelector()
if id == "All" {
log.Debugf("select all routes")
routeManager.SelectAllRoutes()
return nil
}
log.Debugf("select route with id: %s", id)
if err := routeManager.SelectRoutes(toNetIDs([]string{id}), true); err != nil {
log.Debugf("error when selecting routes: %s", err)
return err
routeSelector.SelectAllRoutes()
} else {
log.Debugf("select route with id: %s", id)
routes := toNetIDs([]string{id})
routesMap := routeManager.GetClientRoutesWithNetID()
routes = route.ExpandV6ExitPairs(routes, routesMap)
if err := routeSelector.SelectRoutes(routes, true, maps.Keys(routesMap)); err != nil {
log.Debugf("error when selecting routes: %s", err)
return fmt.Errorf("select routes: %w", err)
}
}
routeManager.TriggerSelection(routeManager.GetClientRoutes())
return nil
}
func (c *Client) DeselectRoute(id string) error {
@@ -661,17 +667,21 @@ func (c *Client) DeselectRoute(id string) error {
}
routeManager := engine.GetRouteManager()
routeSelector := routeManager.GetRouteSelector()
if id == "All" {
log.Debugf("deselect all routes")
routeManager.DeselectAllRoutes()
return nil
}
log.Debugf("deselect route with id: %s", id)
if err := routeManager.DeselectRoutes(toNetIDs([]string{id})); err != nil {
log.Debugf("error when deselecting routes: %s", err)
return err
routeSelector.DeselectAllRoutes()
} else {
log.Debugf("deselect route with id: %s", id)
routes := toNetIDs([]string{id})
routesMap := routeManager.GetClientRoutesWithNetID()
routes = route.ExpandV6ExitPairs(routes, routesMap)
if err := routeSelector.DeselectRoutes(routes, maps.Keys(routesMap)); err != nil {
log.Debugf("error when deselecting routes: %s", err)
return fmt.Errorf("deselect routes: %w", err)
}
}
routeManager.TriggerSelection(routeManager.GetClientRoutes())
return nil
}

View File

@@ -1,12 +0,0 @@
//go:build ios
package NetBirdSDK
import "github.com/netbirdio/netbird/version"
// GoClientVersion returns the NetBird Go client version that was baked into
// the framework at compile time via
// -ldflags "-X github.com/netbirdio/netbird/version.version=<version>".
func GoClientVersion() string {
return version.NetbirdVersion()
}

View File

@@ -8,6 +8,7 @@ import (
"sort"
"strings"
"golang.org/x/exp/maps"
"google.golang.org/grpc/codes"
gstatus "google.golang.org/grpc/status"
@@ -160,11 +161,30 @@ func (s *Server) SelectNetworks(_ context.Context, req *proto.SelectNetworksRequ
return nil, fmt.Errorf("no route manager")
}
routeSelector := routeManager.GetRouteSelector()
if req.GetAll() {
routeManager.SelectAllRoutes()
} else if err := routeManager.SelectRoutes(toNetIDs(req.GetNetworkIDs()), req.GetAppend()); err != nil {
return nil, err
routeSelector.SelectAllRoutes()
} else {
routes := toNetIDs(req.GetNetworkIDs())
routesMap := routeManager.GetClientRoutesWithNetID()
routes = route.ExpandV6ExitPairs(routes, routesMap)
netIdRoutes := maps.Keys(routesMap)
if err := routeSelector.SelectRoutes(routes, req.GetAppend(), netIdRoutes); err != nil {
return nil, fmt.Errorf("select routes: %w", err)
}
// Exit nodes are mutually exclusive: if this selection activates an
// exit node, deselect every other available exit node so two can't be
// selected at once. Non-exit route selections are left untouched.
if requestActivatesExitNode(routes, routesMap) {
if others := otherExitNodeIDs(routesMap, routes); len(others) > 0 {
if err := routeSelector.DeselectRoutes(others, netIdRoutes); err != nil {
return nil, fmt.Errorf("deselect sibling exit nodes: %w", err)
}
}
}
}
routeManager.TriggerSelection(routeManager.GetClientRoutes())
s.statusRecorder.PublishEvent(
proto.SystemEvent_INFO,
@@ -204,11 +224,19 @@ func (s *Server) DeselectNetworks(_ context.Context, req *proto.SelectNetworksRe
return nil, fmt.Errorf("no route manager")
}
routeSelector := routeManager.GetRouteSelector()
if req.GetAll() {
routeManager.DeselectAllRoutes()
} else if err := routeManager.DeselectRoutes(toNetIDs(req.GetNetworkIDs())); err != nil {
return nil, err
routeSelector.DeselectAllRoutes()
} else {
routes := toNetIDs(req.GetNetworkIDs())
routesMap := routeManager.GetClientRoutesWithNetID()
routes = route.ExpandV6ExitPairs(routes, routesMap)
netIdRoutes := maps.Keys(routesMap)
if err := routeSelector.DeselectRoutes(routes, netIdRoutes); err != nil {
return nil, fmt.Errorf("deselect routes: %w", err)
}
}
routeManager.TriggerSelection(routeManager.GetClientRoutes())
s.statusRecorder.PublishEvent(
proto.SystemEvent_INFO,
@@ -233,3 +261,37 @@ func toNetIDs(routes []string) []route.NetID {
return netIDs
}
func isExitNodeRoutes(routes []*route.Route) bool {
return len(routes) > 0 && (route.IsV4DefaultRoute(routes[0].Network) || route.IsV6DefaultRoute(routes[0].Network))
}
// requestActivatesExitNode reports whether any requested NetID maps to an exit
// node (default route) in the current route table.
func requestActivatesExitNode(requested []route.NetID, routesMap map[route.NetID][]*route.Route) bool {
for _, id := range requested {
if isExitNodeRoutes(routesMap[id]) {
return true
}
}
return false
}
// otherExitNodeIDs returns every available exit-node NetID that is not in the
// requested set — the siblings to deselect so a single exit node stays active.
func otherExitNodeIDs(routesMap map[route.NetID][]*route.Route, requested []route.NetID) []route.NetID {
keep := make(map[route.NetID]struct{}, len(requested))
for _, id := range requested {
keep[id] = struct{}{}
}
var others []route.NetID
for id, routes := range routesMap {
if !isExitNodeRoutes(routes) {
continue
}
if _, ok := keep[id]; ok {
continue
}
others = append(others, id)
}
return others
}

View File

@@ -0,0 +1,26 @@
package server
import (
"net/netip"
"testing"
"github.com/stretchr/testify/assert"
"github.com/netbirdio/netbird/route"
)
func TestExitNodeSelectionHelpers(t *testing.T) {
routesMap := map[route.NetID][]*route.Route{
"exitA": {{Network: netip.MustParsePrefix("0.0.0.0/0")}},
"exitB": {{Network: netip.MustParsePrefix("::/0")}},
"lan": {{Network: netip.MustParsePrefix("192.168.0.0/16")}},
}
assert.True(t, requestActivatesExitNode([]route.NetID{"exitA"}, routesMap), "v4 default route is an exit node")
assert.True(t, requestActivatesExitNode([]route.NetID{"exitB"}, routesMap), "v6 default route is an exit node")
assert.False(t, requestActivatesExitNode([]route.NetID{"lan"}, routesMap), "lan route is not an exit node")
assert.False(t, requestActivatesExitNode([]route.NetID{"missing"}, routesMap), "unknown id is not an exit node")
others := otherExitNodeIDs(routesMap, []route.NetID{"exitB"})
assert.ElementsMatch(t, []route.NetID{"exitA"}, others, "only the other exit node is a sibling; the lan route is ignored")
}

View File

@@ -2,7 +2,6 @@ import { type ReactNode } from "react";
import { useTranslation } from "react-i18next";
import { Browser } from "@wailsio/runtime";
import { DownloadIcon, NotepadText } from "lucide-react";
import { Update as UpdateSvc } from "@bindings/services";
import { Button } from "@/components/buttons/Button";
import { useClientVersion } from "@/contexts/ClientVersionContext";
import { cn } from "@/lib/cn";
@@ -15,12 +14,6 @@ function openUrl(url: string) {
});
}
function openInstallerDownload() {
UpdateSvc.DownloadURL()
.then(openUrl)
.catch(() => openUrl(GITHUB_RELEASES));
}
export function UpdateVersionCard() {
const { t } = useTranslation();
const { updateVersion, enforced, triggerUpdate } = useClientVersion();
@@ -44,7 +37,11 @@ export function UpdateVersionCard() {
{t("update.card.installNow")}
</Button>
) : (
<Button variant={"primary"} size={"xs"} onClick={openInstallerDownload}>
<Button
variant={"primary"}
size={"xs"}
onClick={() => openUrl(GITHUB_RELEASES)}
>
<DownloadIcon size={14} />
{t("update.card.getInstaller")}
</Button>

View File

@@ -10,7 +10,6 @@ import (
"github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/client/ui/updater"
"github.com/netbirdio/netbird/version"
)
// UpdateResult mirrors TriggerUpdateResponse.
@@ -34,12 +33,6 @@ func (s *Update) GetState() updater.State {
return s.holder.Get()
}
// DownloadURL returns the platform-appropriate installer download link for
// manual (non-enforced) updates.
func (s *Update) DownloadURL() string {
return version.DownloadUrl()
}
// Quit exits the app. Scheduled off the calling goroutine so the JS caller's
// response returns before the runtime tears down.
func (s *Update) Quit() {

View File

@@ -32,8 +32,9 @@ const (
quitDownTimeout = 5 * time.Second
urlGitHubRepo = "https://github.com/netbirdio/netbird"
urlDocs = "https://docs.netbird.io"
urlGitHubRepo = "https://github.com/netbirdio/netbird"
urlGitHubReleases = "https://github.com/netbirdio/netbird/releases/latest"
urlDocs = "https://docs.netbird.io"
)
// TrayServices bundles the services the tray menu needs, grouped so NewTray

View File

@@ -13,7 +13,6 @@ import (
"github.com/netbirdio/netbird/client/ui/services"
"github.com/netbirdio/netbird/client/ui/updater"
"github.com/netbirdio/netbird/version"
)
// trayUpdater owns the tray UI that reacts to auto-update. Composed inside Tray.
@@ -77,15 +76,15 @@ func (u *trayUpdater) applyLanguage() {
u.refreshMenuItem(state)
}
// handleClick opens the installer download link when not Enforced, otherwise
// shows the progress page and asks the daemon to start the installer.
// handleClick opens the GitHub releases page when not Enforced, otherwise shows
// the progress page and asks the daemon to start the installer.
func (u *trayUpdater) handleClick() {
u.mu.Lock()
state := u.state
u.mu.Unlock()
if !state.Enforced {
_ = u.app.Browser.OpenURL(version.DownloadUrl())
_ = u.app.Browser.OpenURL(urlGitHubReleases)
return
}

View File

@@ -109,7 +109,7 @@ sequenceDiagram
Chk->>Inj: continue
Inj->>Inj: inject NetBird identity headers per provider config
Inj->>Grd: continue
Grd->>Grd: enforce per-provider allowlist (fail-closed backstop)
Grd->>Grd: enforce model allowlist
Grd->>Up: forward (over WireGuard)
Up-->>Resp: response (JSON or SSE stream)
Resp->>Resp: parse usage tokens, completion
@@ -135,21 +135,6 @@ sequenceDiagram
(`redact_pii = settings.RedactPii`). Phones, emails, credit cards,
PII names — see `redact.go` for the full set. See
[`modules/31-proxy-middleware-builtin.md`](modules/31-proxy-middleware-builtin.md).
- The model allowlist is enforced in TWO places. `CheckLLMPolicyLimits`
is authoritative: it resolves the policy that governs this
(provider, caller-groups) and denies (`deny_code = llm_policy.model_blocked`)
when no applicable policy permits the model — so an allowlist scoped to
one group/provider never leaks to another, and an un-guardrailed policy
is genuinely unrestricted. `llm_guardrail` is a per-provider fail-closed
backstop: it only carries an allowlist for a provider every authorising
policy restricts, and blocks unknown/undetermined models even when
management is unreachable. Because that backstop allowlist is the UNION
of every restricting policy's models, per-group narrowing lives only in
the authoritative check: during a `CheckLLMPolicyLimits` outage
`llm_limit_check` fails open, so a caller can reach any model in the
provider's union — a group scoped to model A could reach model B if
another group restricts the same provider to B. This is the documented
fail-open trade-off; a future flag may switch it to fail-closed.
- SSE streaming requires special handling on the response side; the
parser must handle partial chunks without buffering the whole
stream. See [`modules/32-proxy-llm-parsers.md`](modules/32-proxy-llm-parsers.md).

View File

@@ -122,7 +122,7 @@ At request time the path is independent: the proxy calls `SelectPolicyForRequest
| on_request | 1 | `llm_router` | `{"providers":[{id, models[], upstream_*, auth_header_*, allowed_group_ids[]}]}` | **true** |
| on_request | 2 | `llm_limit_check` | `{}` | |
| on_request | 3 | `llm_identity_inject` | `{"providers":[{provider_id, header_pair?, json_metadata?, extra_headers?}]}` | **true** |
| on_request | 4 | `llm_guardrail` | `{"provider_allowlists"?: {providerID: []model}, "prompt_capture":{enabled,redact_pii}}` | |
| on_request | 4 | `llm_guardrail` | `{"model_allowlist"?, "prompt_capture":{enabled,redact_pii}}` | |
| on_response | 5 | `llm_limit_record` | `{}` (runs LAST at runtime) | |
| on_response | 6 | `cost_meter` | `{}` | |
| on_response | 7 | `llm_response_parser` | `{"capture_completion": <bool>, "redact_pii"?: true}` | |

View File

@@ -244,7 +244,7 @@ no mocks. Tests: `TestChain_AllowPath_StampsAttributionAndRecordsCounter`
| `llm_router` | `{providers: [{id, models, upstream_scheme, upstream_host, upstream_path?, auth_header_name, auth_header_value, allowed_group_ids}]}` |
| `llm_limit_check` | `{}` — pulls `MgmtClient` from `FactoryContext` |
| `llm_identity_inject` | `{providers: [{provider_id, header_pair?|json_metadata?, extra_headers?}]}` |
| `llm_guardrail` | `{provider_allowlists: {providerID: []string}, prompt_capture: {enabled, redact_pii}}` — allowlist keyed by resolved provider id; a provider absent from the map is unrestricted (fail-closed backstop; authoritative per-policy/group check is management's `CheckLLMPolicyLimits`) |
| `llm_guardrail` | `{model_allowlist: []string, prompt_capture: {enabled, redact_pii}}` |
| `llm_response_parser` | `{redact_pii?, capture_completion?: *bool}` |
| `cost_meter` | `{pricing_path?}` (basename inside data-dir; defaults `pricing.yaml`) |
| `llm_limit_record` | `{}` — same pattern as `llm_limit_check` |

View File

@@ -1,209 +0,0 @@
//go:build e2e
package agentnetwork
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/e2e/harness"
"github.com/netbirdio/netbird/shared/management/http/api"
)
// pathRoutedGuardrailCase is one provider's self-contained scenario: its own
// provider, its own guardrail whose allowlist holds ONLY that provider's
// allowed model, and its own policy. Each case runs in isolation (its own
// proxy + client), so the guardrail the proxy enforces contains exactly this
// provider's model — never a mixed cross-provider list.
type pathRoutedGuardrailCase struct {
name string
catalogID string // agent-network catalog provider id
wire string // harness.WireVertex | harness.WireBedrock
allowEntry string // the single model id put on the guardrail allowlist
allowModel string // model id sent that MUST be served (200)
blockModel string // model id sent that MUST be denied (403 model_blocked)
}
// TestGuardrailBlocksUnselectedModel_PathRouted is the end-to-end regression
// guard for the customer report that a model-allowlist guardrail attached to a
// policy has no effect for PATH-ROUTED providers — where the model travels in
// the URL, not the JSON body: Google Vertex (…/models/{model}:rawPredict) and
// AWS Bedrock (/model/{id}/invoke).
//
// Each provider is tested in isolation with a guardrail allowlisting a single
// model of its own: the allowed model (in the URL path) is served (200) and an
// unselected model (in the URL path) is denied 403 by the guardrail
// (llm_policy.model_blocked) before the upstream. The Vertex case mirrors the
// customer verbatim — allow Sonnet, and the unselected model is the exact
// claude-opus-4-6 they reported reaching the model unblocked. The Bedrock case
// sends a region-prefixed, versioned inference-profile id so URL-path model
// normalization is exercised too.
//
// The provider is catch-all (no models), so the router forwards any model and a
// 403 can only come from the guardrail, never model_not_routable. Only the
// upstream LLM is mocked (the vLLM nginx answers any path with 200); management
// synth/reconcile, the proxy middleware chain (URL-path model extraction,
// router, guardrail) and the tunnel are all real, and the guardrail denies
// before the upstream is dialed so the mock cannot influence the block. A
// static bearer api key is used so the router injects a static Authorization
// header instead of minting a GCP token — the only reason path-routed providers
// normally need live credentials — so the test runs with none and is always on.
func TestGuardrailBlocksUnselectedModel_PathRouted(t *testing.T) {
cases := []pathRoutedGuardrailCase{
{
name: "vertex",
catalogID: "vertex_ai_api",
wire: harness.WireVertex,
allowEntry: "claude-sonnet-4-5",
allowModel: "claude-sonnet-4-5",
blockModel: "claude-opus-4-6", // the customer-reported model
},
{
name: "bedrock",
catalogID: "bedrock_api",
wire: harness.WireBedrock,
allowEntry: "anthropic.claude-sonnet-4-5", // normalized catalog id
allowModel: "us.anthropic.claude-sonnet-4-5-v1:0",
blockModel: "us.anthropic.claude-opus-4-8-v1:0",
},
}
for _, tc := range cases {
tc := tc
t.Run(tc.name, func(t *testing.T) {
runPathRoutedGuardrailCase(t, tc)
})
}
}
func runPathRoutedGuardrailCase(t *testing.T, tc pathRoutedGuardrailCase) {
t.Helper()
const (
vertexProject = "e2e-project"
vertexRegion = "global"
)
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
defer cancel()
vllm, err := harness.StartVLLM(ctx, srv)
require.NoError(t, err, "start mock upstream")
t.Cleanup(func() { _ = vllm.Terminate(context.Background()) })
grp, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-guardrail-" + tc.name})
require.NoError(t, err, "create group")
t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grp.Id) })
ephemeral := false
sk, err := srv.API().SetupKeys.Create(ctx, api.PostApiSetupKeysJSONRequestBody{
Name: "e2e-guardrail-" + tc.name + "-client",
Type: "reusable",
ExpiresIn: 86400,
UsageLimit: 0,
AutoGroups: []string{grp.Id},
Ephemeral: &ephemeral,
})
require.NoError(t, err, "mint setup key")
require.NotEmpty(t, sk.Key, "setup key plaintext")
// Catch-all provider (no models) so the router forwards any model; a static
// bearer key means the router injects a static auth header instead of minting
// a GCP token. Bootstraps the cluster if it isn't already.
staticKey := "static-e2e-token"
prov, err := srv.CreateProvider(ctx, api.AgentNetworkProviderRequest{
Name: tc.name,
ProviderId: tc.catalogID,
UpstreamUrl: vllm.URL,
ApiKey: &staticKey,
Enabled: ptr(true),
BootstrapCluster: ptr(harness.AgentNetworkCluster),
})
require.NoError(t, err, "create %s provider", tc.name)
t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), prov.Id) })
// Guardrail allowlisting ONLY this provider's allowed model.
var gr api.AgentNetworkGuardrailRequest
gr.Name = "e2e-guardrail-" + tc.name
gr.Checks.ModelAllowlist.Enabled = true
gr.Checks.ModelAllowlist.Models = []string{tc.allowEntry}
guard, err := srv.CreateGuardrail(ctx, gr)
require.NoError(t, err, "create guardrail")
t.Cleanup(func() { _ = srv.DeleteGuardrail(context.Background(), guard.Id) })
enabled := true
pol, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-guardrail-" + tc.name,
Enabled: &enabled,
SourceGroups: []string{grp.Id},
DestinationProviderIds: []string{prov.Id},
GuardrailIds: &[]string{guard.Id},
})
require.NoError(t, err, "create policy")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), pol.Id) })
settings, err := srv.GetSettings(ctx)
require.NoError(t, err, "read settings")
require.NotEmpty(t, settings.Endpoint, "endpoint must be assigned")
proxyToken, err := srv.CreateProxyTokenCLI(ctx, "e2e-guardrail-"+tc.name+"-proxy")
require.NoError(t, err, "mint proxy token")
px, err := harness.StartProxy(ctx, srv, proxyToken)
require.NoError(t, err, "start proxy")
t.Cleanup(func() { _ = px.Terminate(context.Background()) })
cl, err := harness.StartClient(ctx, srv, sk.Key)
require.NoError(t, err, "start client")
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
// Probe first: the GET resolves the endpoint and its first packet wakes the
// lazy proxy peer, so WaitProxyPeer then observes it connected.
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
send := func(model string) (int, string) {
var code int
var body string
var cerr error
switch tc.wire {
case harness.WireVertex:
code, body, cerr = cl.Vertex(ctx, settings.Endpoint, proxyIP, vertexProject, vertexRegion, model, "Reply with exactly: pong", "")
case harness.WireBedrock:
code, body, cerr = cl.Bedrock(ctx, settings.Endpoint, proxyIP, model, "Reply with exactly: pong", "")
default:
t.Fatalf("unsupported wire %q", tc.wire)
}
require.NoError(t, cerr, "request must reach the proxy for %s", tc.name)
return code, body
}
// Allowed model (in the URL path) is served. Retry to absorb tunnel/DNS
// jitter on the first call over the freshly warmed tunnel.
var code int
var body string
deadline := time.Now().Add(90 * time.Second)
for time.Now().Before(deadline) {
code, body = send(tc.allowModel)
if code == 200 {
break
}
time.Sleep(5 * time.Second)
}
assert.Equal(t, 200, code,
"allowed %s model (URL path) must be served; body: %s\n=== proxy logs ===\n%s", tc.name, body, px.Logs(context.Background()))
// Unselected model (in the URL path) must be blocked by the guardrail.
code, body = send(tc.blockModel)
assert.Equal(t, 403, code,
"unselected %s model (URL path) must be denied, not served; body: %s\n=== proxy logs ===\n%s", tc.name, body, px.Logs(context.Background()))
assert.Contains(t, body, "llm_policy.model_blocked",
"%s denial must come from the guardrail allowlist, not routing; body: %s", tc.name, body)
}

View File

@@ -1,205 +0,0 @@
//go:build e2e
package agentnetwork
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/e2e/harness"
"github.com/netbirdio/netbird/shared/management/http/api"
)
// TestGuardrailGroupSwitchTakesEffectAfterTTL proves that moving a peer between
// groups flips its model-allowlist decision once the proxy's tunnel-peer cache
// expires. The peer's groups reach the guardrail via ValidateTunnelPeer, which
// the proxy caches; the switch is invisible until that cache expires. The proxy
// runs with a short NB_PROXY_TUNNEL_CACHE_TTL so the flip happens in seconds
// instead of the 5-minute default.
//
// Setup: one catch-all provider declaring modelA + modelB; polA (grpA -> allow
// modelA) and polB (grpB -> allow modelB). The client starts in grpA. modelA is
// served and modelB denied; after switching the client grpA -> grpB, modelB is
// served and modelA denied. The cross-group deny comes from management's
// per-policy/group CheckLLMPolicyLimits (the proxy backstop carries the union).
func TestGuardrailGroupSwitchTakesEffectAfterTTL(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
defer cancel()
const (
modelA = "e2e-model-a"
modelB = "e2e-model-b"
// Short tunnel-cache TTL so a group switch propagates in seconds.
// Exercises the NB_PROXY_TUNNEL_CACHE_TTL override.
cacheTTL = 3 * time.Second
)
vllm, err := harness.StartVLLM(ctx, srv)
require.NoError(t, err, "start mock upstream")
t.Cleanup(func() { _ = vllm.Terminate(context.Background()) })
grpA, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-gswitch-a"})
require.NoError(t, err, "create group A")
t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grpA.Id) })
grpB, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-gswitch-b"})
require.NoError(t, err, "create group B")
t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grpB.Id) })
ephemeral := false
sk, err := srv.API().SetupKeys.Create(ctx, api.PostApiSetupKeysJSONRequestBody{
Name: "e2e-gswitch-client",
Type: "reusable",
ExpiresIn: 86400,
UsageLimit: 0,
AutoGroups: []string{grpA.Id}, // client starts in group A
Ephemeral: &ephemeral,
})
require.NoError(t, err, "mint setup key")
require.NotEmpty(t, sk.Key, "setup key plaintext")
staticKey := "static-e2e-token"
prov, err := srv.CreateProvider(ctx, api.AgentNetworkProviderRequest{
Name: "gswitch",
ProviderId: "openai_api",
UpstreamUrl: vllm.URL,
ApiKey: &staticKey,
Enabled: ptr(true),
Models: &[]api.AgentNetworkProviderModel{
{Id: modelA, InputPer1k: 0.001, OutputPer1k: 0.001},
{Id: modelB, InputPer1k: 0.001, OutputPer1k: 0.001},
},
BootstrapCluster: ptr(harness.AgentNetworkCluster),
})
require.NoError(t, err, "create provider")
t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), prov.Id) })
mkGuard := func(name, model string) api.AgentNetworkGuardrail {
var gr api.AgentNetworkGuardrailRequest
gr.Name = name
gr.Checks.ModelAllowlist.Enabled = true
gr.Checks.ModelAllowlist.Models = []string{model}
g, gerr := srv.CreateGuardrail(ctx, gr)
require.NoError(t, gerr, "create guardrail %s", name)
t.Cleanup(func() { _ = srv.DeleteGuardrail(context.Background(), g.Id) })
return g
}
gA := mkGuard("e2e-gswitch-a", modelA)
gB := mkGuard("e2e-gswitch-b", modelB)
enabled := true
polA, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-gswitch-a",
Enabled: &enabled,
SourceGroups: []string{grpA.Id},
DestinationProviderIds: []string{prov.Id},
GuardrailIds: &[]string{gA.Id},
})
require.NoError(t, err, "create policy A")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polA.Id) })
polB, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-gswitch-b",
Enabled: &enabled,
SourceGroups: []string{grpB.Id},
DestinationProviderIds: []string{prov.Id},
GuardrailIds: &[]string{gB.Id},
})
require.NoError(t, err, "create policy B")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polB.Id) })
settings, err := srv.GetSettings(ctx)
require.NoError(t, err, "read settings")
require.NotEmpty(t, settings.Endpoint, "endpoint must be assigned")
proxyToken, err := srv.CreateProxyTokenCLI(ctx, "e2e-gswitch-proxy")
require.NoError(t, err, "mint proxy token")
px, err := harness.StartProxy(ctx, srv, proxyToken, map[string]string{
"NB_PROXY_TUNNEL_CACHE_TTL": cacheTTL.String(),
})
require.NoError(t, err, "start proxy")
t.Cleanup(func() { _ = px.Terminate(context.Background()) })
cl, err := harness.StartClient(ctx, srv, sk.Key)
require.NoError(t, err, "start client")
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
send := func(model string) (int, string) {
code, body, cerr := cl.Chat(ctx, settings.Endpoint, proxyIP, harness.WireChat, model, "Reply with exactly: pong", "")
require.NoError(t, cerr, "request must reach the proxy")
return code, body
}
sendUntil := func(model string, want int, timeout time.Duration) (int, string) {
var code int
var body string
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
code, body = send(model)
if code == want {
return code, body
}
time.Sleep(2 * time.Second)
}
return code, body
}
// Phase 1 — client is in group A: modelA served, modelB denied.
code, body := sendUntil(modelA, 200, 90*time.Second)
assert.Equal(t, 200, code,
"group-A model must be served while the client is in group A; body: %s\n=== proxy logs ===\n%s", body, px.Logs(context.Background()))
code, body = send(modelB)
assert.Equal(t, 403, code,
"group-B model must be denied while the client is in group A; body: %s", body)
assert.Contains(t, body, "llm_policy.model_blocked")
// Switch the client peer from group A to group B.
peerID := clientPeerInGroup(t, ctx, grpA.Id)
_, err = srv.API().Groups.Update(ctx, grpB.Id, api.PutApiGroupsGroupIdJSONRequestBody{
Name: grpB.Name,
Peers: &[]string{peerID},
})
require.NoError(t, err, "add peer to group B")
_, err = srv.API().Groups.Update(ctx, grpA.Id, api.PutApiGroupsGroupIdJSONRequestBody{
Name: grpA.Name,
Peers: &[]string{},
})
require.NoError(t, err, "remove peer from group A")
// Phase 2 — after the short TTL expires the proxy re-validates the peer,
// sees group B, and the decision flips. Poll to absorb TTL + re-validation.
code, body = sendUntil(modelB, 200, 60*time.Second)
assert.Equal(t, 200, code,
"after the group switch + TTL, the group-B model must be served; body: %s\n=== proxy logs ===\n%s", body, px.Logs(context.Background()))
code, body = send(modelA)
assert.Equal(t, 403, code,
"after the switch, the old group-A model must be denied; body: %s", body)
assert.Contains(t, body, "llm_policy.model_blocked")
}
// clientPeerInGroup returns the id of the single peer that is a member of the
// given group — the test client. The proxy peer is never added to test groups.
func clientPeerInGroup(t *testing.T, ctx context.Context, groupID string) string {
t.Helper()
peers, err := srv.API().Peers.List(ctx)
require.NoError(t, err, "list peers")
for _, p := range peers {
for _, g := range p.Groups {
if g.Id == groupID {
return p.Id
}
}
}
t.Fatalf("no peer found in group %s", groupID)
return ""
}

View File

@@ -1,201 +0,0 @@
//go:build e2e
package agentnetwork
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/e2e/harness"
"github.com/netbirdio/netbird/shared/management/http/api"
)
// TestGuardrailMultiPolicyModelAllowlist: modelSelected served (200), grpOther's
// modelOther denied for the grpMain client (403 model_blocked, no cross-group
// leak), and openModel on the un-guardrailed policy's provider served (200).
func TestGuardrailMultiPolicyModelAllowlist(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
defer cancel()
const (
modelSelected = "e2e-selected"
modelOther = "e2e-other"
openModel = "e2e-open"
)
vllm, err := harness.StartVLLM(ctx, srv)
require.NoError(t, err, "start mock upstream")
t.Cleanup(func() { _ = vllm.Terminate(context.Background()) })
grpMain, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-guardrail-mp-main"})
require.NoError(t, err, "create main group")
t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grpMain.Id) })
grpOther, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-guardrail-mp-other"})
require.NoError(t, err, "create other group")
t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grpOther.Id) })
ephemeral := false
sk, err := srv.API().SetupKeys.Create(ctx, api.PostApiSetupKeysJSONRequestBody{
Name: "e2e-guardrail-mp-client",
Type: "reusable",
ExpiresIn: 86400,
UsageLimit: 0,
AutoGroups: []string{grpMain.Id}, // client joins grpMain only
Ephemeral: &ephemeral,
})
require.NoError(t, err, "mint setup key")
require.NotEmpty(t, sk.Key, "setup key plaintext")
staticKey := "static-e2e-token"
models := func(ids ...string) *[]api.AgentNetworkProviderModel {
out := make([]api.AgentNetworkProviderModel, 0, len(ids))
for _, id := range ids {
out = append(out, api.AgentNetworkProviderModel{Id: id, InputPer1k: 0.001, OutputPer1k: 0.001})
}
return &out
}
// pRestricted declares the two guardrailed models so routing is deterministic
// (model -> provider). Created first, so it carries the bootstrap cluster.
pRestricted, err := srv.CreateProvider(ctx, api.AgentNetworkProviderRequest{
Name: "restricted",
ProviderId: "openai_api",
UpstreamUrl: vllm.URL,
ApiKey: &staticKey,
Enabled: ptr(true),
Models: models(modelSelected, modelOther),
BootstrapCluster: ptr(harness.AgentNetworkCluster),
})
require.NoError(t, err, "create restricted provider")
t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), pRestricted.Id) })
pOpen, err := srv.CreateProvider(ctx, api.AgentNetworkProviderRequest{
Name: "open",
ProviderId: "openai_api",
UpstreamUrl: vllm.URL,
ApiKey: &staticKey,
Enabled: ptr(true),
Models: models(openModel),
})
require.NoError(t, err, "create open provider")
t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), pOpen.Id) })
mkGuardrail := func(name, model string) api.AgentNetworkGuardrail {
var gr api.AgentNetworkGuardrailRequest
gr.Name = name
gr.Checks.ModelAllowlist.Enabled = true
gr.Checks.ModelAllowlist.Models = []string{model}
g, gerr := srv.CreateGuardrail(ctx, gr)
require.NoError(t, gerr, "create guardrail %s", name)
t.Cleanup(func() { _ = srv.DeleteGuardrail(context.Background(), g.Id) })
return g
}
gMain := mkGuardrail("e2e-guardrail-mp-main", modelSelected)
gOther := mkGuardrail("e2e-guardrail-mp-other", modelOther)
enabled := true
// polMain: grpMain restricted to modelSelected on pRestricted.
polMain, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-guardrail-mp-main",
Enabled: &enabled,
SourceGroups: []string{grpMain.Id},
DestinationProviderIds: []string{pRestricted.Id},
GuardrailIds: &[]string{gMain.Id},
})
require.NoError(t, err, "create main policy")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polMain.Id) })
// polOther: grpOther restricted to modelOther on the SAME provider. The
// client is not in grpOther, so modelOther must never be usable by it.
polOther, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-guardrail-mp-other",
Enabled: &enabled,
SourceGroups: []string{grpOther.Id},
DestinationProviderIds: []string{pRestricted.Id},
GuardrailIds: &[]string{gOther.Id},
})
require.NoError(t, err, "create other policy")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polOther.Id) })
// polOpen: grpMain on pOpen with NO guardrail — unrestricted.
polOpen, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-guardrail-mp-open",
Enabled: &enabled,
SourceGroups: []string{grpMain.Id},
DestinationProviderIds: []string{pOpen.Id},
})
require.NoError(t, err, "create open policy")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polOpen.Id) })
settings, err := srv.GetSettings(ctx)
require.NoError(t, err, "read settings")
require.NotEmpty(t, settings.Endpoint, "endpoint must be assigned")
proxyToken, err := srv.CreateProxyTokenCLI(ctx, "e2e-guardrail-mp-proxy")
require.NoError(t, err, "mint proxy token")
px, err := harness.StartProxy(ctx, srv, proxyToken)
require.NoError(t, err, "start proxy")
t.Cleanup(func() { _ = px.Terminate(context.Background()) })
cl, err := harness.StartClient(ctx, srv, sk.Key)
require.NoError(t, err, "start client")
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
send := func(model string) (int, string) {
code, body, cerr := cl.Chat(ctx, settings.Endpoint, proxyIP, harness.WireChat, model, "Reply with exactly: pong", "")
require.NoError(t, cerr, "request must reach the proxy")
return code, body
}
// sendUntil200 absorbs first-call tunnel/DNS jitter on the freshly warmed tunnel.
sendUntil200 := func(model string) (int, string) {
var code int
var body string
deadline := time.Now().Add(90 * time.Second)
for time.Now().Before(deadline) {
code, body = send(model)
if code == 200 {
break
}
time.Sleep(5 * time.Second)
}
return code, body
}
t.Run("selected model allowed for its group", func(t *testing.T) {
code, body := sendUntil200(modelSelected)
assert.Equal(t, 200, code,
"grpMain's allowlisted model must be served; body: %s\n=== proxy logs ===\n%s", body, px.Logs(context.Background()))
})
t.Run("other group's model does not leak", func(t *testing.T) {
// modelOther is allowlisted only for grpOther. The grpMain client must be
// denied by management's per-policy/group check — not waved through by an
// account-wide union. This is the security-critical wrong-ALLOW guard.
code, body := send(modelOther)
assert.Equal(t, 403, code,
"another group's allowlisted model must be denied for this caller; body: %s\n=== proxy logs ===\n%s", body, px.Logs(context.Background()))
assert.Contains(t, body, "llm_policy.model_blocked",
"denial must be a model-allowlist decision; body: %s", body)
})
t.Run("unguarded policy leaves its provider unrestricted", func(t *testing.T) {
// polOpen carries no guardrail, so pOpen is unrestricted for grpMain. The
// old account-wide union would have blocked openModel (it is on no
// allowlist); it must now be served — the false-DENY guard.
code, body := sendUntil200(openModel)
assert.Equal(t, 200, code,
"an un-guardrailed policy's provider must not be blocked by another policy's allowlist; body: %s\n=== proxy logs ===\n%s", body, px.Logs(context.Background()))
})
}

View File

@@ -1,422 +0,0 @@
//go:build e2e
package agentnetwork
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/e2e/harness"
"github.com/netbirdio/netbird/shared/management/http/api"
)
// pergroupCase describes one provider surface for the per-group allowlist matrix.
// selectedReq/otherReq are the model identifiers as they travel in the request
// (URL path for Bedrock/Vertex, body "model" for chat/messages). selectedAllow/
// otherAllow are the (normalized) forms the guardrail allowlist holds — for
// Bedrock these differ from the request form so path normalization is exercised.
type pergroupCase struct {
name string
catalogID string
wire string // "chat", "messages", "vertex", "bedrock"
models *[]api.AgentNetworkProviderModel
selectedReq string
selectedAllow string
otherReq string
otherAllow string
providerID string // filled during setup
}
// TestGuardrailPerGroupAllowlist_AllProviders proves the per-policy/group model
// allowlist end to end across every always-on provider surface, including the
// path-routed ones (Vertex, Bedrock) where the model travels in the URL.
//
// For each provider two policies target it: grpMain (the client) is allowed only
// selectedReq; grpOther (which the client is NOT in) is allowed only otherReq.
// The client must get selectedReq served (200) and otherReq denied (403,
// llm_policy.model_blocked) — the cross-group no-leak property. The deny is the
// authoritative per-policy/group decision from management (the proxy per-provider
// backstop carries the union of both models), so this also confirms management
// receives the correct normalized model for path-routed providers.
func TestGuardrailPerGroupAllowlist_AllProviders(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute)
defer cancel()
const (
vertexProject = "e2e-project"
vertexRegion = "global"
)
priced := func(ids ...string) *[]api.AgentNetworkProviderModel {
out := make([]api.AgentNetworkProviderModel, 0, len(ids))
for _, id := range ids {
out = append(out, api.AgentNetworkProviderModel{Id: id, InputPer1k: 0.001, OutputPer1k: 0.001})
}
return &out
}
cases := []*pergroupCase{
{
name: "openai", catalogID: "openai_api", wire: harness.WireChat,
models: priced("oai-model-a", "oai-model-b"),
selectedReq: "oai-model-a", selectedAllow: "oai-model-a",
otherReq: "oai-model-b", otherAllow: "oai-model-b",
},
{
name: "anthropic", catalogID: "anthropic_api", wire: harness.WireMessages,
models: priced("ant-model-a", "ant-model-b"),
selectedReq: "ant-model-a", selectedAllow: "ant-model-a",
otherReq: "ant-model-b", otherAllow: "ant-model-b",
},
{
// Vertex catalog ids travel bare in the rawPredict path.
name: "vertex", catalogID: "vertex_ai_api", wire: "vertex",
selectedReq: "claude-sonnet-4-5", selectedAllow: "claude-sonnet-4-5",
otherReq: "claude-opus-4-6", otherAllow: "claude-opus-4-6",
},
{
// Bedrock request ids are region-prefixed/versioned; the parser
// normalizes them to the catalog key the allowlist holds.
name: "bedrock", catalogID: "bedrock_api", wire: "bedrock",
selectedReq: "us.anthropic.claude-sonnet-4-5-v1:0", selectedAllow: "anthropic.claude-sonnet-4-5",
otherReq: "us.anthropic.claude-opus-4-8-v1:0", otherAllow: "anthropic.claude-opus-4-8",
},
}
vllm, err := harness.StartVLLM(ctx, srv)
require.NoError(t, err, "start mock upstream")
t.Cleanup(func() { _ = vllm.Terminate(context.Background()) })
grpMain, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-pergroup-main"})
require.NoError(t, err, "create main group")
t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grpMain.Id) })
grpOther, err := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: "e2e-pergroup-other"})
require.NoError(t, err, "create other group")
t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), grpOther.Id) })
ephemeral := false
sk, err := srv.API().SetupKeys.Create(ctx, api.PostApiSetupKeysJSONRequestBody{
Name: "e2e-pergroup-client",
Type: "reusable",
ExpiresIn: 86400,
UsageLimit: 0,
AutoGroups: []string{grpMain.Id},
Ephemeral: &ephemeral,
})
require.NoError(t, err, "mint setup key")
require.NotEmpty(t, sk.Key, "setup key plaintext")
staticKey := "static-e2e-token"
enabled := true
for i, c := range cases {
req := api.AgentNetworkProviderRequest{
Name: "e2e-pergroup-" + c.name,
ProviderId: c.catalogID,
UpstreamUrl: vllm.URL,
ApiKey: &staticKey,
Enabled: ptr(true),
Models: c.models,
}
if i == 0 {
req.BootstrapCluster = ptr(harness.AgentNetworkCluster)
}
prov, perr := srv.CreateProvider(ctx, req)
require.NoError(t, perr, "create provider %s", c.name)
c.providerID = prov.Id
t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), prov.Id) })
gSel := mkAllowGuardrail(t, ctx, "e2e-pergroup-"+c.name+"-sel", c.selectedAllow)
gOth := mkAllowGuardrail(t, ctx, "e2e-pergroup-"+c.name+"-oth", c.otherAllow)
polMain, merr := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-pergroup-" + c.name + "-main",
Enabled: &enabled,
SourceGroups: []string{grpMain.Id},
DestinationProviderIds: []string{prov.Id},
GuardrailIds: &[]string{gSel.Id},
})
require.NoError(t, merr, "create main policy %s", c.name)
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polMain.Id) })
polOther, oerr := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-pergroup-" + c.name + "-other",
Enabled: &enabled,
SourceGroups: []string{grpOther.Id},
DestinationProviderIds: []string{prov.Id},
GuardrailIds: &[]string{gOth.Id},
})
require.NoError(t, oerr, "create other policy %s", c.name)
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polOther.Id) })
}
settings, err := srv.GetSettings(ctx)
require.NoError(t, err, "read settings")
require.NotEmpty(t, settings.Endpoint, "endpoint must be assigned")
proxyToken, err := srv.CreateProxyTokenCLI(ctx, "e2e-pergroup-proxy")
require.NoError(t, err, "mint proxy token")
px, err := harness.StartProxy(ctx, srv, proxyToken)
require.NoError(t, err, "start proxy")
t.Cleanup(func() { _ = px.Terminate(context.Background()) })
cl, err := harness.StartClient(ctx, srv, sk.Key)
require.NoError(t, err, "start client")
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
send := func(c *pergroupCase, model string) (int, string) {
var code int
var body string
var cerr error
switch c.wire {
case "vertex":
code, body, cerr = cl.Vertex(ctx, settings.Endpoint, proxyIP, vertexProject, vertexRegion, model, "Reply with exactly: pong", "")
case "bedrock":
code, body, cerr = cl.Bedrock(ctx, settings.Endpoint, proxyIP, model, "Reply with exactly: pong", "")
default:
code, body, cerr = cl.Chat(ctx, settings.Endpoint, proxyIP, c.wire, model, "Reply with exactly: pong", "")
}
require.NoError(t, cerr, "request must reach the proxy for %s", c.name)
return code, body
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
// grpMain's own model is served. Retry to absorb tunnel/DNS jitter on
// the first call over the freshly warmed tunnel.
var code int
var body string
deadline := time.Now().Add(90 * time.Second)
for time.Now().Before(deadline) {
code, body = send(c, c.selectedReq)
if code == 200 {
break
}
time.Sleep(5 * time.Second)
}
assert.Equal(t, 200, code,
"%s: grpMain's allowlisted model must be served; body: %s\n=== proxy logs ===\n%s", c.name, body, px.Logs(context.Background()))
// grpOther's model must NOT leak to the grpMain client.
code, body = send(c, c.otherReq)
assert.Equal(t, 403, code,
"%s: another group's allowlisted model must be denied for this caller; body: %s\n=== proxy logs ===\n%s", c.name, body, px.Logs(context.Background()))
assert.Contains(t, body, "llm_policy.model_blocked",
"%s: denial must be a model-allowlist decision, not routing; body: %s", c.name, body)
})
}
}
// TestGuardrailMultiGroupUser proves the per-policy/group decision for a caller
// that belongs to MULTIPLE groups at once. Two scenarios, one shared stack:
//
// - union across the user's groups: the client is in gUX and gUY, each with
// its own policy+guardrail on provider P1 (gUX->union-a, gUY->union-b). The
// client may use BOTH models (the union of its groups' allowlists) while a
// third, un-allowlisted model is denied.
// - an un-guardrailed group lifts the restriction: the client is in gMP and
// gMQ on provider P2, where gMP restricts to mix-a but gMQ's policy carries
// NO guardrail. Because one applicable policy is unrestricted, the client may
// use a model on no allowlist (mix-z) as well as mix-a.
func TestGuardrailMultiGroupUser(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
defer cancel()
const (
unionA = "mg-union-a"
unionB = "mg-union-b"
unionC = "mg-union-c" // allowlisted by neither group
mixA = "mg-mix-a"
mixZ = "mg-mix-z" // on no allowlist; reachable only via the un-guardrailed policy
)
priced := func(ids ...string) *[]api.AgentNetworkProviderModel {
out := make([]api.AgentNetworkProviderModel, 0, len(ids))
for _, id := range ids {
out = append(out, api.AgentNetworkProviderModel{Id: id, InputPer1k: 0.001, OutputPer1k: 0.001})
}
return &out
}
vllm, err := harness.StartVLLM(ctx, srv)
require.NoError(t, err, "start mock upstream")
t.Cleanup(func() { _ = vllm.Terminate(context.Background()) })
mkGroup := func(name string) *api.Group {
g, gerr := srv.API().Groups.Create(ctx, api.PostApiGroupsJSONRequestBody{Name: name})
require.NoError(t, gerr, "create group %s", name)
t.Cleanup(func() { _ = srv.API().Groups.Delete(context.Background(), g.Id) })
return g
}
gUX := mkGroup("e2e-mg-union-x")
gUY := mkGroup("e2e-mg-union-y")
gMP := mkGroup("e2e-mg-mix-p")
gMQ := mkGroup("e2e-mg-mix-q")
ephemeral := false
sk, err := srv.API().SetupKeys.Create(ctx, api.PostApiSetupKeysJSONRequestBody{
Name: "e2e-mg-client",
Type: "reusable",
ExpiresIn: 86400,
UsageLimit: 0,
AutoGroups: []string{gUX.Id, gUY.Id, gMP.Id, gMQ.Id}, // client in all four groups
Ephemeral: &ephemeral,
})
require.NoError(t, err, "mint setup key")
require.NotEmpty(t, sk.Key, "setup key plaintext")
staticKey := "static-e2e-token"
enabled := true
// P1 — union scenario: two restricting policies, one per group.
p1, err := srv.CreateProvider(ctx, api.AgentNetworkProviderRequest{
Name: "e2e-mg-union",
ProviderId: "openai_api",
UpstreamUrl: vllm.URL,
ApiKey: &staticKey,
Enabled: ptr(true),
Models: priced(unionA, unionB, unionC),
BootstrapCluster: ptr(harness.AgentNetworkCluster),
})
require.NoError(t, err, "create union provider")
t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), p1.Id) })
polUX, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-mg-union-x",
Enabled: &enabled,
SourceGroups: []string{gUX.Id},
DestinationProviderIds: []string{p1.Id},
GuardrailIds: &[]string{mkAllowGuardrail(t, ctx, "e2e-mg-union-x", unionA).Id},
})
require.NoError(t, err, "create union policy X")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polUX.Id) })
polUY, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-mg-union-y",
Enabled: &enabled,
SourceGroups: []string{gUY.Id},
DestinationProviderIds: []string{p1.Id},
GuardrailIds: &[]string{mkAllowGuardrail(t, ctx, "e2e-mg-union-y", unionB).Id},
})
require.NoError(t, err, "create union policy Y")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polUY.Id) })
// P2 — mixed scenario: one restricting policy + one un-guardrailed policy.
p2, err := srv.CreateProvider(ctx, api.AgentNetworkProviderRequest{
Name: "e2e-mg-mix",
ProviderId: "openai_api",
UpstreamUrl: vllm.URL,
ApiKey: &staticKey,
Enabled: ptr(true),
Models: priced(mixA, mixZ),
})
require.NoError(t, err, "create mix provider")
t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), p2.Id) })
polMP, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-mg-mix-p",
Enabled: &enabled,
SourceGroups: []string{gMP.Id},
DestinationProviderIds: []string{p2.Id},
GuardrailIds: &[]string{mkAllowGuardrail(t, ctx, "e2e-mg-mix-p", mixA).Id},
})
require.NoError(t, err, "create mix policy P")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polMP.Id) })
polMQ, err := srv.CreatePolicy(ctx, api.AgentNetworkPolicyRequest{
Name: "e2e-mg-mix-q",
Enabled: &enabled,
SourceGroups: []string{gMQ.Id},
DestinationProviderIds: []string{p2.Id}, // NO guardrail -> unrestricted
})
require.NoError(t, err, "create mix policy Q")
t.Cleanup(func() { _ = srv.DeletePolicy(context.Background(), polMQ.Id) })
settings, err := srv.GetSettings(ctx)
require.NoError(t, err, "read settings")
require.NotEmpty(t, settings.Endpoint, "endpoint must be assigned")
proxyToken, err := srv.CreateProxyTokenCLI(ctx, "e2e-mg-proxy")
require.NoError(t, err, "mint proxy token")
px, err := harness.StartProxy(ctx, srv, proxyToken)
require.NoError(t, err, "start proxy")
t.Cleanup(func() { _ = px.Terminate(context.Background()) })
cl, err := harness.StartClient(ctx, srv, sk.Key)
require.NoError(t, err, "start client")
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
send := func(model string) (int, string) {
code, body, cerr := cl.Chat(ctx, settings.Endpoint, proxyIP, harness.WireChat, model, "Reply with exactly: pong", "")
require.NoError(t, cerr, "request must reach the proxy")
return code, body
}
sendUntil200 := func(model string) (int, string) {
var code int
var body string
deadline := time.Now().Add(90 * time.Second)
for time.Now().Before(deadline) {
code, body = send(model)
if code == 200 {
break
}
time.Sleep(5 * time.Second)
}
return code, body
}
t.Run("union across the user's groups", func(t *testing.T) {
code, body := sendUntil200(unionA)
assert.Equal(t, 200, code, "model allowed by group X must be served; body: %s\n=== proxy logs ===\n%s", body, px.Logs(context.Background()))
code, body = sendUntil200(unionB)
assert.Equal(t, 200, code, "model allowed by group Y must also be served (union across the user's groups); body: %s", body)
code, body = send(unionC)
assert.Equal(t, 403, code, "a model on neither group's allowlist must be denied; body: %s", body)
assert.Contains(t, body, "llm_policy.model_blocked")
})
t.Run("an un-guardrailed group lifts the restriction", func(t *testing.T) {
code, body := sendUntil200(mixA)
assert.Equal(t, 200, code, "the restricted group's model must be served; body: %s\n=== proxy logs ===\n%s", body, px.Logs(context.Background()))
code, body = sendUntil200(mixZ)
assert.Equal(t, 200, code,
"a non-allowlisted model must be served because the user is also in a group whose policy has no guardrail; body: %s\n=== proxy logs ===\n%s", body, px.Logs(context.Background()))
})
}
// mkAllowGuardrail creates a guardrail whose model allowlist is enabled and holds
// exactly the given model, registering cleanup.
func mkAllowGuardrail(t *testing.T, ctx context.Context, name, model string) api.AgentNetworkGuardrail {
t.Helper()
var gr api.AgentNetworkGuardrailRequest
gr.Name = name
gr.Checks.ModelAllowlist.Enabled = true
gr.Checks.ModelAllowlist.Models = []string{model}
g, err := srv.CreateGuardrail(ctx, gr)
require.NoError(t, err, "create guardrail %s", name)
t.Cleanup(func() { _ = srv.DeleteGuardrail(context.Background(), g.Id) })
return g
}

View File

@@ -38,11 +38,7 @@ type Proxy struct {
// network, registered via the given account proxy token and serving the
// AgentNetworkCluster over a self-signed wildcard cert. It does not wait for
// peer connectivity — callers poll management for the proxy peer.
// StartProxy launches the reverse-proxy container. Optional envOverrides are
// merged into the container environment after the defaults, so callers can set
// or override any NB_PROXY_* var (e.g. NB_PROXY_TUNNEL_CACHE_TTL for tests that
// need a short authorization-cache window).
func StartProxy(ctx context.Context, c *Combined, proxyToken string, envOverrides ...map[string]string) (*Proxy, error) {
func StartProxy(ctx context.Context, c *Combined, proxyToken string) (*Proxy, error) {
root, err := repoRoot()
if err != nil {
return nil, err
@@ -97,12 +93,6 @@ func StartProxy(ctx context.Context, c *Combined, proxyToken string, envOverride
WaitingFor: wait.ForLog("Initial mapping sync complete").WithStartupTimeout(90 * time.Second),
}
for _, ov := range envOverrides {
for k, v := range ov {
req.Env[k] = v
}
}
ctr, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
ContainerRequest: req,
Started: true,

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: the catch-all "OpenAI-compatible self-hosted endpoint"
// entry (vLLM, Ollama, custom inference servers).
// - KindCustom: named and generic OpenAI-compatible self-hosted
// endpoints (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,6 +40,18 @@ 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
@@ -48,6 +60,9 @@ 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
@@ -86,6 +101,17 @@ 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
@@ -461,6 +487,27 @@ 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,
@@ -701,11 +748,31 @@ 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: "OpenAI-compatible endpoint (vLLM, Ollama, …)",
Description: "Other OpenAI-compatible endpoint",
DefaultHost: "",
AuthHeaderName: "Authorization",
AuthHeaderTemplate: "Bearer ${API_KEY}",
@@ -738,6 +805,15 @@ 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})
@@ -774,15 +850,17 @@ 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,
AuthHeaderTemplate: p.AuthHeaderTemplate,
DefaultContentType: p.DefaultContentType,
BrandColor: p.BrandColor,
Models: models,
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,
}
if len(p.ExtraHeaders) > 0 {
extras := make([]api.AgentNetworkCatalogExtraHeader, 0, len(p.ExtraHeaders))

View File

@@ -0,0 +1,96 @@
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

@@ -0,0 +1,72 @@
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,6 +35,7 @@ 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)
@@ -98,6 +99,41 @@ 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 {
@@ -193,11 +229,12 @@ func (h *handler) deleteProvider(w http.ResponseWriter, r *http.Request) {
util.WriteJSONObject(r.Context(), w, util.EmptyObject{})
}
func validate(req *api.AgentNetworkProviderRequest, requireAPIKey bool) error {
func validate(req *api.AgentNetworkProviderRequest, creating bool) error {
if strings.TrimSpace(req.ProviderId) == "" {
return status.Errorf(status.InvalidArgument, "provider_id is required")
}
if !catalog.IsKnown(req.ProviderId) {
entry, ok := catalog.Lookup(req.ProviderId)
if !ok {
return status.Errorf(status.InvalidArgument, "provider_id %q is not a known catalog provider", req.ProviderId)
}
if strings.TrimSpace(req.Name) == "" {
@@ -210,8 +247,27 @@ func validate(req *api.AgentNetworkProviderRequest, requireAPIKey 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")
}
if requireAPIKey && (req.ApiKey == nil || strings.TrimSpace(*req.ApiKey) == "") {
return status.Errorf(status.InvalidArgument, "api_key is required")
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)
}
return nil
}

View File

@@ -0,0 +1,47 @@
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,9 +9,12 @@ 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"
@@ -48,6 +51,7 @@ 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
@@ -86,17 +90,12 @@ type Manager interface {
// PolicySelectionInput is the per-request selection envelope. The
// proxy populates it from CapturedData (account, user, groups) plus
// the provider llm_router resolved and the model it extracted.
// the provider llm_router resolved.
type PolicySelectionInput struct {
AccountID string
UserID string
GroupIDs []string
ProviderID string
// Model is the already-normalised upstream model id the proxy extracted
// (parser strips Bedrock region/version, Vertex @version), so a
// case-insensitive compare suffices. Empty = undetermined → not permitted
// (fail closed).
Model string
}
// PolicySelectionResult names the policy that "pays" for this request
@@ -133,8 +132,27 @@ 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
@@ -153,6 +171,7 @@ 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),
}
}
@@ -170,6 +189,202 @@ 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
@@ -179,11 +394,8 @@ func (m *managerImpl) CreateProvider(ctx context.Context, userID string, provide
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 err := prepareProviderAPIKey(provider, nil); err != nil {
return nil, err
}
if provider.ID == "" {
@@ -227,13 +439,8 @@ func (m *managerImpl) UpdateProvider(ctx context.Context, userID string, provide
return nil, fmt.Errorf("failed to get agent network provider: %w", 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")
if err := prepareProviderAPIKey(provider, existing); err != nil {
return nil, err
}
// Always preserve the session keypair across updates so existing
// session cookies stay valid. The keys are server-managed and
@@ -256,6 +463,52 @@ 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
@@ -813,6 +1066,10 @@ 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

@@ -0,0 +1,248 @@
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

@@ -5,7 +5,6 @@ import (
"fmt"
"math"
"sort"
"strings"
"time"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
@@ -36,10 +35,6 @@ const (
denyCodeAccountTokenCapExceeded = "llm_account.token_cap_exceeded"
//nolint:gosec // account deny code label, not a credential
denyCodeAccountBudgetCapExceeded = "llm_account.budget_cap_exceeded"
// denyCodeModelBlocked is returned when policies govern the request's
// (provider, caller-groups) but none permits the model. Matches the proxy
// guardrail's code so both layers surface the same label.
denyCodeModelBlocked = "llm_policy.model_blocked"
)
// consumptionCache holds the consumption counters prefetched for one
@@ -164,25 +159,6 @@ func (m *managerImpl) SelectPolicyForRequest(ctx context.Context, in PolicySelec
}
candidates := filterApplicablePolicies(policies, in)
// Model-allowlist gate scoped to the matched policies: keep candidates whose
// guardrails permit the model (none enabled = unrestricted), deny when
// policies apply but none permits it. Skip the load when none has a guardrail.
if len(candidates) > 0 && anyPolicyHasGuardrails(candidates) {
guardrailsByID, gErr := m.loadGuardrailsByID(ctx, in.AccountID)
if gErr != nil {
return nil, gErr
}
permitted := filterModelPermittedPolicies(candidates, guardrailsByID, in.Model)
if len(permitted) == 0 {
return &PolicySelectionResult{
Allow: false,
DenyCode: denyCodeModelBlocked,
DenyReason: modelBlockedReason(in.Model),
}, nil
}
candidates = permitted
}
// Prefetch every consumption counter the ceiling + candidate policies will
// read, in a single store round-trip, then score against the cache.
cache, err := m.prefetchConsumption(ctx, in, rules, candidates, now)
@@ -274,90 +250,6 @@ func filterApplicablePolicies(policies []*types.Policy, in PolicySelectionInput)
return out
}
// anyPolicyHasGuardrails reports whether any policy references at least one
// guardrail, so the selector can skip loading guardrails when none do.
func anyPolicyHasGuardrails(policies []*types.Policy) bool {
for _, p := range policies {
if p != nil && len(p.GuardrailIDs) > 0 {
return true
}
}
return false
}
// loadGuardrailsByID loads the account's guardrails indexed by ID. Used by the
// model-allowlist gate to resolve each candidate policy's attached guardrails.
func (m *managerImpl) loadGuardrailsByID(ctx context.Context, accountID string) (map[string]*types.Guardrail, error) {
guardrails, err := m.store.GetAccountAgentNetworkGuardrails(ctx, store.LockingStrengthNone, accountID)
if err != nil {
return nil, fmt.Errorf("list account guardrails: %w", err)
}
byID := make(map[string]*types.Guardrail, len(guardrails))
for _, g := range guardrails {
if g != nil {
byID[g.ID] = g
}
}
return byID, nil
}
// filterModelPermittedPolicies returns the subset of policies whose guardrails
// permit the model. Order is preserved so downstream scoring is unaffected.
func filterModelPermittedPolicies(policies []*types.Policy, byID map[string]*types.Guardrail, model string) []*types.Policy {
out := make([]*types.Policy, 0, len(policies))
for _, p := range policies {
if policyPermitsModel(p, byID, model) {
out = append(out, p)
}
}
return out
}
// policyPermitsModel reports whether a policy permits the model. No
// allowlist-enabled guardrail = unrestricted (permits any, incl. empty);
// otherwise the model must be in the union of its allowlists, so an
// empty/undetermined model fails closed.
func policyPermitsModel(p *types.Policy, byID map[string]*types.Guardrail, model string) bool {
if p == nil {
return false
}
wanted := normaliseModelID(model)
restricted := false
for _, gID := range p.GuardrailIDs {
g, ok := byID[gID]
if !ok || g == nil || !g.Checks.ModelAllowlist.Enabled {
continue
}
restricted = true
if wanted == "" {
continue
}
for _, allowed := range g.Checks.ModelAllowlist.Models {
if normaliseModelID(allowed) == wanted {
return true
}
}
}
return !restricted
}
// normaliseModelID lowercases and trims a model identifier so the allowlist
// compare is case-insensitive and trim-tolerant. Mirrors the proxy guardrail's
// normaliseModel so both layers agree on what "same model" means.
func normaliseModelID(model string) string {
return strings.ToLower(strings.TrimSpace(model))
}
// modelBlockedReason builds the human-readable deny reason for a model-allowlist
// rejection. The model is quoted when known; an undetermined model is reported
// as such so the access log distinguishes "wrong model" from "no model".
func modelBlockedReason(model string) string {
if normaliseModelID(model) == "" {
return "request model could not be determined for the policy allowlist"
}
return fmt.Sprintf("model %q is not permitted by any applicable policy allowlist", model)
}
// candidate is the per-policy intermediate the selector ranks. A
// policy that's been exhausted on any enabled cap never makes it
// into this slice; the selector's deny envelope carries the latest

View File

@@ -1,329 +0,0 @@
package agentnetwork
import (
"context"
"errors"
"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/server/store"
)
// guardedPolicy builds an enabled, uncapped policy that authorises sourceGroups
// to reach providerID under the given guardrails. Uncapped keeps the selector's
// headroom scoring trivial so these tests isolate the model-allowlist gate.
func guardedPolicy(id, account string, sourceGroups []string, providerID string, guardrailIDs ...string) *types.Policy {
return &types.Policy{
ID: id,
AccountID: account,
Enabled: true,
SourceGroups: sourceGroups,
DestinationProviderIDs: []string{providerID},
GuardrailIDs: guardrailIDs,
CreatedAt: time.Now().UTC(),
}
}
// allowlistGuardrail builds a guardrail whose model allowlist is enabled and
// carries the given models.
func allowlistGuardrail(id, account string, models ...string) *types.Guardrail {
return &types.Guardrail{
ID: id,
AccountID: account,
Checks: types.GuardrailChecks{
ModelAllowlist: types.GuardrailModelAllowlist{Enabled: true, Models: models},
},
}
}
func expectPolicies(mockStore *store.MockStore, account string, policies ...*types.Policy) {
mockStore.EXPECT().
GetAccountAgentNetworkPolicies(gomock.Any(), gomock.Any(), account).
Return(policies, nil)
}
func expectGuardrails(mockStore *store.MockStore, account string, guardrails ...*types.Guardrail) {
mockStore.EXPECT().
GetAccountAgentNetworkGuardrails(gomock.Any(), gomock.Any(), account).
Return(guardrails, nil)
}
// TestSelectPolicy_ModelBlockedByAllowlist proves the authoritative allowlist
// decision: a policy authorises the (provider, group) but restricts the model,
// and the requested model isn't on the list, so the request is denied.
func TestSelectPolicy_ModelBlockedByAllowlist(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
policy := guardedPolicy("pol-A", "acc-1", []string{"grp-eng"}, "prov-1", "g-1")
expectPolicies(mockStore, "acc-1", policy)
expectGuardrails(mockStore, "acc-1", allowlistGuardrail("g-1", "acc-1", "gpt-4o"))
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
UserID: "user-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "claude-opus-4",
})
require.NoError(t, err)
assert.False(t, res.Allow, "a model outside the only applicable policy's allowlist must be denied")
assert.Equal(t, denyCodeModelBlocked, res.DenyCode, "deny code must be model_blocked")
assert.NotEmpty(t, res.DenyReason, "deny reason must be populated")
}
// TestSelectPolicy_ModelAllowedByAllowlist is the allow counterpart: the model
// is on the applicable policy's allowlist, so selection proceeds normally.
func TestSelectPolicy_ModelAllowedByAllowlist(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
policy := guardedPolicy("pol-A", "acc-1", []string{"grp-eng"}, "prov-1", "g-1")
expectPolicies(mockStore, "acc-1", policy)
expectGuardrails(mockStore, "acc-1", allowlistGuardrail("g-1", "acc-1", "gpt-4o", "claude-opus-4"))
expectConsumptionBatch(mockStore, nil)
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
UserID: "user-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "claude-opus-4",
})
require.NoError(t, err)
assert.True(t, res.Allow, "a model on the applicable policy's allowlist must be allowed")
assert.Equal(t, "pol-A", res.SelectedPolicyID)
}
// TestSelectPolicy_CaseInsensitiveModelMatch proves the compare tolerates case
// and surrounding whitespace, matching the proxy guardrail's normalisation.
func TestSelectPolicy_CaseInsensitiveModelMatch(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
policy := guardedPolicy("pol-A", "acc-1", []string{"grp-eng"}, "prov-1", "g-1")
expectPolicies(mockStore, "acc-1", policy)
expectGuardrails(mockStore, "acc-1", allowlistGuardrail("g-1", "acc-1", " GPT-4o "))
expectConsumptionBatch(mockStore, nil)
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "gpt-4o",
})
require.NoError(t, err)
assert.True(t, res.Allow, "case/whitespace variants must match the allowlist entry")
}
// TestSelectPolicy_UnguardedPolicyIsUnrestricted is the false-deny fix: when two
// policies authorise the same (provider, group) and one has no guardrail, that
// policy makes the request unrestricted — not caught by the other's allowlist.
func TestSelectPolicy_UnguardedPolicyIsUnrestricted(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
restricted := guardedPolicy("pol-restricted", "acc-1", []string{"grp-eng"}, "prov-1", "g-1")
open := guardedPolicy("pol-open", "acc-1", []string{"grp-eng"}, "prov-1") // no guardrail
expectPolicies(mockStore, "acc-1", restricted, open)
expectGuardrails(mockStore, "acc-1", allowlistGuardrail("g-1", "acc-1", "gpt-4o"))
expectConsumptionBatch(mockStore, nil)
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "claude-opus-4",
})
require.NoError(t, err)
assert.True(t, res.Allow, "an un-guardrailed policy for the same (provider, group) must leave the request unrestricted")
assert.Equal(t, "pol-open", res.SelectedPolicyID, "the unrestricted policy must be the one that pays")
}
// TestSelectPolicy_AllowlistDoesNotLeakAcrossGroups is the false-allow fix: a
// model allowlisted only for grp-b must not be usable by a grp-a caller. The
// selector considers only policies applicable to the caller's groups.
func TestSelectPolicy_AllowlistDoesNotLeakAcrossGroups(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
polA := guardedPolicy("pol-a", "acc-1", []string{"grp-a"}, "prov-1", "g-a")
polB := guardedPolicy("pol-b", "acc-1", []string{"grp-b"}, "prov-1", "g-b")
expectPolicies(mockStore, "acc-1", polA, polB)
expectGuardrails(mockStore, "acc-1",
allowlistGuardrail("g-a", "acc-1", "gpt-4o"),
allowlistGuardrail("g-b", "acc-1", "claude-opus-4"),
)
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-a"},
ProviderID: "prov-1",
Model: "claude-opus-4", // only allowed for grp-b
})
require.NoError(t, err)
assert.False(t, res.Allow, "grp-b's allowlisted model must not leak to a grp-a caller")
assert.Equal(t, denyCodeModelBlocked, res.DenyCode)
}
// TestSelectPolicy_UndeterminedModelFailsClosed proves the fail-closed contract
// mirrors the proxy: with a restricted applicable policy and an empty model
// (e.g. a path-routed shape the parser couldn't map), the request is denied.
func TestSelectPolicy_UndeterminedModelFailsClosed(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
policy := guardedPolicy("pol-A", "acc-1", []string{"grp-eng"}, "prov-1", "g-1")
expectPolicies(mockStore, "acc-1", policy)
expectGuardrails(mockStore, "acc-1", allowlistGuardrail("g-1", "acc-1", "gpt-4o"))
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "", // undetermined
})
require.NoError(t, err)
assert.False(t, res.Allow, "an undetermined model must fail closed against a restricted policy")
assert.Equal(t, denyCodeModelBlocked, res.DenyCode)
}
// TestSelectPolicy_DisabledAllowlistDoesNotRestrict proves a guardrail whose
// model allowlist is disabled imposes no model restriction, even though the
// policy references it.
func TestSelectPolicy_DisabledAllowlistDoesNotRestrict(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
policy := guardedPolicy("pol-A", "acc-1", []string{"grp-eng"}, "prov-1", "g-1")
disabled := &types.Guardrail{
ID: "g-1",
AccountID: "acc-1",
Checks: types.GuardrailChecks{
ModelAllowlist: types.GuardrailModelAllowlist{Enabled: false, Models: []string{"gpt-4o"}},
},
}
expectPolicies(mockStore, "acc-1", policy)
expectGuardrails(mockStore, "acc-1", disabled)
expectConsumptionBatch(mockStore, nil)
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "anything-goes",
})
require.NoError(t, err)
assert.True(t, res.Allow, "a disabled allowlist must not restrict the model")
assert.Equal(t, "pol-A", res.SelectedPolicyID)
}
// TestSelectPolicy_UnionAcrossPolicyGuardrails proves a policy with multiple
// allowlist guardrails permits the union of their models (not just the first).
func TestSelectPolicy_UnionAcrossPolicyGuardrails(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
policy := guardedPolicy("pol-A", "acc-1", []string{"grp-eng"}, "prov-1", "g-1", "g-2")
expectPolicies(mockStore, "acc-1", policy)
expectGuardrails(mockStore, "acc-1",
allowlistGuardrail("g-1", "acc-1", "gpt-4o"),
allowlistGuardrail("g-2", "acc-1", "claude-opus-4"),
)
expectConsumptionBatch(mockStore, nil)
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "claude-opus-4", // only in the second guardrail's list
})
require.NoError(t, err)
assert.True(t, res.Allow, "a model in any of the policy's allowlist guardrails must be permitted")
assert.Equal(t, "pol-A", res.SelectedPolicyID)
}
// TestSelectPolicy_GuardrailLookupErrorPropagates proves a store failure while
// resolving the candidate policies' guardrails surfaces as an error, not a
// silent allow/deny.
func TestSelectPolicy_GuardrailLookupErrorPropagates(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
policy := guardedPolicy("pol-A", "acc-1", []string{"grp-eng"}, "prov-1", "g-1")
expectPolicies(mockStore, "acc-1", policy)
mockStore.EXPECT().
GetAccountAgentNetworkGuardrails(gomock.Any(), gomock.Any(), "acc-1").
Return(nil, errors.New("store unavailable"))
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "gpt-4o",
})
require.Error(t, err, "a guardrail-lookup failure must surface as an error")
assert.Nil(t, res)
}
// TestSelectPolicy_MissingGuardrailReferenceTreatedAsUnrestricted proves a
// policy referencing a guardrail ID absent from the account's set (a stale
// reference) imposes no model restriction — same as no guardrail.
func TestSelectPolicy_MissingGuardrailReferenceTreatedAsUnrestricted(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
policy := guardedPolicy("pol-A", "acc-1", []string{"grp-eng"}, "prov-1", "g-missing")
expectPolicies(mockStore, "acc-1", policy)
expectGuardrails(mockStore, "acc-1")
expectConsumptionBatch(mockStore, nil)
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "anything-goes",
})
require.NoError(t, err)
assert.True(t, res.Allow, "an orphaned guardrail reference must not restrict the model")
assert.Equal(t, "pol-A", res.SelectedPolicyID)
}
// TestSelectPolicy_PartialCandidatesPermittedAfterModelFilter proves the model
// gate narrows candidates before cap scoring: the permitting policy is selected
// even though the blocked one has a larger, more attractive cap.
func TestSelectPolicy_PartialCandidatesPermittedAfterModelFilter(t *testing.T) {
ctrl := gomock.NewController(t)
mgr, mockStore := newSelectorMgr(t, ctrl)
polBig := guardedPolicy("pol-big", "acc-1", []string{"grp-eng"}, "prov-1", "g-restrict")
polBig.Limits = types.PolicyLimits{
TokenLimit: types.PolicyTokenLimit{Enabled: true, GroupCap: 1_000_000, WindowSeconds: 3600},
}
polSmall := guardedPolicy("pol-small", "acc-1", []string{"grp-eng"}, "prov-1", "g-permit")
polSmall.Limits = types.PolicyLimits{
TokenLimit: types.PolicyTokenLimit{Enabled: true, GroupCap: 100, WindowSeconds: 3600},
}
expectPolicies(mockStore, "acc-1", polBig, polSmall)
expectGuardrails(mockStore, "acc-1",
allowlistGuardrail("g-restrict", "acc-1", "gpt-4o"),
allowlistGuardrail("g-permit", "acc-1", "claude-opus-4"),
)
expectConsumptionBatch(mockStore, nil)
res, err := mgr.SelectPolicyForRequest(context.Background(), PolicySelectionInput{
AccountID: "acc-1",
GroupIDs: []string{"grp-eng"},
ProviderID: "prov-1",
Model: "claude-opus-4", // only pol-small's guardrail permits this
})
require.NoError(t, err)
assert.True(t, res.Allow)
assert.Equal(t, "pol-small", res.SelectedPolicyID,
"the model filter must exclude pol-big before cap scoring")
}

View File

@@ -0,0 +1,124 @@
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

@@ -235,12 +235,7 @@ func SynthesizeServices(ctx context.Context, s store.Store, accountID string) ([
mergedGuardrails := mergeGuardrails(enabledPolicies, guardrailsByID)
applyAccountCollectionControls(&mergedGuardrails, settings)
// The proxy guardrail is a per-provider fail-closed backstop; the
// authoritative per-policy/group decision is management's
// SelectPolicyForRequest. A provider lands in this map only when every
// authorising policy restricts models.
providerAllowlists := buildProviderAllowlists(enabledPolicies, guardrailsByID)
guardrailJSON, err := marshalGuardrailConfig(providerAllowlists, mergedGuardrails.PromptCapture)
guardrailJSON, err := marshalGuardrailConfig(mergedGuardrails)
if err != nil {
return nil, err
}
@@ -785,12 +780,10 @@ func buildMiddlewareChain(routerCfgJSON, identityInjectJSON, guardrailJSON []byt
// guardrailConfig is the JSON shape the proxy-side llm_guardrail
// middleware expects. Mirrors the proxy registration documented in
// the management→proxy contract. provider_allowlists is keyed by the
// resolved provider id llm_router stamps; a provider absent from the map is
// unrestricted at the proxy layer.
// the management→proxy contract.
type guardrailConfig struct {
ProviderAllowlists map[string][]string `json:"provider_allowlists,omitempty"`
PromptCapture guardrailPromptCapture `json:"prompt_capture"`
ModelAllowlist []string `json:"model_allowlist,omitempty"`
PromptCapture guardrailPromptCapture `json:"prompt_capture"`
}
type guardrailPromptCapture struct {
@@ -835,10 +828,13 @@ func applyAccountCollectionControls(merged *MergedGuardrails, settings *types.Se
merged.PromptCapture.RedactPii = settings.RedactPii || merged.PromptCapture.RedactPii
}
func marshalGuardrailConfig(providerAllowlists map[string][]string, capture MergedPromptCapture) ([]byte, error) {
func marshalGuardrailConfig(merged MergedGuardrails) ([]byte, error) {
cfg := guardrailConfig{
ProviderAllowlists: providerAllowlists,
PromptCapture: guardrailPromptCapture(capture),
ModelAllowlist: merged.ModelAllowlist,
PromptCapture: guardrailPromptCapture{
Enabled: merged.PromptCapture.Enabled,
RedactPii: merged.PromptCapture.RedactPii,
},
}
out, err := json.Marshal(cfg)
if err != nil {
@@ -847,74 +843,6 @@ func marshalGuardrailConfig(providerAllowlists map[string][]string, capture Merg
return out, nil
}
// buildProviderAllowlists returns the proxy's per-provider backstop: a provider
// is included only when every authorising policy restricts models (their union);
// if any leaves it unrestricted it is omitted, so management decides per group.
func buildProviderAllowlists(policies []*types.Policy, byID map[string]*types.Guardrail) map[string][]string {
type providerAcc struct {
models map[string]struct{}
anyUnrestricted bool
}
accs := make(map[string]*providerAcc)
for _, p := range policies {
if p == nil {
continue
}
restricted, models := policyModelAllowlist(p, byID)
for _, providerID := range p.DestinationProviderIDs {
if providerID == "" {
continue
}
acc, ok := accs[providerID]
if !ok {
acc = &providerAcc{models: make(map[string]struct{})}
accs[providerID] = acc
}
if !restricted {
acc.anyUnrestricted = true
continue
}
for _, m := range models {
acc.models[m] = struct{}{}
}
}
}
out := make(map[string][]string, len(accs))
for providerID, acc := range accs {
if acc.anyUnrestricted {
continue
}
models := make([]string, 0, len(acc.models))
for m := range acc.models {
models = append(models, m)
}
sort.Strings(models)
out[providerID] = models
}
return out
}
// policyModelAllowlist reports whether a policy restricts models (has an
// allowlist-enabled guardrail) and the union of allowed models. Models are
// verbatim; the proxy factory lowercases/trims them at decode time.
func policyModelAllowlist(p *types.Policy, byID map[string]*types.Guardrail) (bool, []string) {
restricted := false
var models []string
for _, gID := range p.GuardrailIDs {
g, ok := byID[gID]
if !ok || g == nil || !g.Checks.ModelAllowlist.Enabled {
continue
}
restricted = true
for _, m := range g.Checks.ModelAllowlist.Models {
if m != "" {
models = append(models, m)
}
}
}
return restricted, models
}
// buildAccountService composes the per-account gateway Service. The
// target carries the noop placeholder URL — the router middleware
// rewrites every request to the matched provider's upstream before the
@@ -978,23 +906,33 @@ const (
noopUpstreamPort = uint16(443)
)
// 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.
// 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.
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
@@ -1058,11 +996,38 @@ func unionSourceGroups(policies []*types.Policy) []string {
return out
}
// MergedGuardrails is the synthesiser's fold target. Only prompt capture is
// merged here — the model allowlist is emitted per-provider, and
// token/budget/retention moved onto Policy.Limits and account Settings.
// MergedGuardrails is the JSON shape passed to the proxy via the
// guardrail middleware's config_json. Mirrors the proxy-side
// expectations and is intentionally distinct from
// types.GuardrailChecks so we can evolve either side independently.
type MergedGuardrails struct {
PromptCapture MergedPromptCapture
ModelAllowlist []string `json:"model_allowlist,omitempty"`
TokenLimits MergedTokenLimits `json:"token_limits"`
Budget MergedBudget `json:"budget"`
PromptCapture MergedPromptCapture `json:"prompt_capture"`
Retention MergedRetention `json:"retention"`
}
type MergedTokenLimits struct {
Hourly *MergedTokenWindow `json:"hourly,omitempty"`
Daily *MergedTokenWindow `json:"daily,omitempty"`
Monthly *MergedTokenWindow `json:"monthly,omitempty"`
}
type MergedTokenWindow struct {
MaxInputTokens int `json:"max_input_tokens,omitempty"`
MaxOutputTokens int `json:"max_output_tokens,omitempty"`
}
type MergedBudget struct {
Hourly *MergedBudgetWindow `json:"hourly,omitempty"`
Daily *MergedBudgetWindow `json:"daily,omitempty"`
Monthly *MergedBudgetWindow `json:"monthly,omitempty"`
}
type MergedBudgetWindow struct {
SoftCapUSD float64 `json:"soft_cap_usd,omitempty"`
HardCapUSD float64 `json:"hard_cap_usd,omitempty"`
}
type MergedPromptCapture struct {
@@ -1070,31 +1035,64 @@ type MergedPromptCapture struct {
RedactPii bool `json:"redact_pii"`
}
// mergeGuardrails folds the referencing policies' guardrails into the
// prompt-capture decision only. The model allowlist is enforced per-policy/group
// in management and shipped per-provider; token/budget/retention live off
// guardrails now.
type MergedRetention struct {
Enabled bool `json:"enabled"`
Days int `json:"days"`
}
// mergeGuardrails computes the effective guardrail spec applied at the
// proxy, given the referencing policies and the account's guardrail
// catalogue. Policy enabled-ness is the caller's responsibility — only
// enabled policies should be passed in.
//
// Merge rule — prompt capture: enabled if any policy enables it; redact_pii
// sticks if any enabling policy turns it on.
// Merge rules:
// - Model allowlist: union of allowlists across policies that enable it.
// - Token / Budget: most-restrictive (min of non-zero caps) per window.
// - Prompt capture: enabled if any policy enables it; redact_pii sticks
// if any enabling policy turns it on.
// - Retention: enabled if any enables it; smallest non-zero days wins.
func mergeGuardrails(policies []*types.Policy, byID map[string]*types.Guardrail) MergedGuardrails {
merged := MergedGuardrails{}
allowlist := make(map[string]struct{})
allowlistEnabled := false
for _, policy := range policies {
for _, gID := range policy.GuardrailIDs {
g, ok := byID[gID]
if !ok || g == nil {
continue
}
mergeGuardrail(g, &merged)
mergeGuardrail(g, &merged, allowlist, &allowlistEnabled)
}
}
if allowlistEnabled {
merged.ModelAllowlist = make([]string, 0, len(allowlist))
for m := range allowlist {
merged.ModelAllowlist = append(merged.ModelAllowlist, m)
}
sort.Strings(merged.ModelAllowlist)
}
return merged
}
// mergeGuardrail folds a single guardrail's prompt-capture settings into the
// running merge: enabled / redact-pii stick once any enabling guardrail turns
// them on.
func mergeGuardrail(g *types.Guardrail, merged *MergedGuardrails) {
// mergeGuardrail folds a single guardrail's enabled checks into the
// running merge: model-allowlist models join the shared set (and flip
// allowlistEnabled), and prompt-capture / redact-pii stick once any
// enabling guardrail turns them on.
//
// TokenLimits, Budget, and Retention have moved off guardrails — token
// and budget caps now live on the Policy itself (Policy.Limits) and
// retention moves to account-level Settings — so they are not merged here.
func mergeGuardrail(g *types.Guardrail, merged *MergedGuardrails, allowlist map[string]struct{}, allowlistEnabled *bool) {
if g.Checks.ModelAllowlist.Enabled {
*allowlistEnabled = true
for _, m := range g.Checks.ModelAllowlist.Models {
if m != "" {
allowlist[m] = struct{}{}
}
}
}
if g.Checks.PromptCapture.Enabled {
merged.PromptCapture.Enabled = true
if g.Checks.PromptCapture.RedactPii {

View File

@@ -81,8 +81,8 @@ func TestSynthesizeServices_RealStore_PromptCaptureAccountIsSoleControl(t *testi
require.Len(t, services, 1)
cfg := decodeServiceGuardrailConfig(t, services[0])
assert.Equal(t, map[string][]string{"prov-1": {"gpt-5.4"}}, cfg.ProviderAllowlists,
"model allowlist is a pure policy guardrail and must reach the per-provider config")
assert.Equal(t, []string{"gpt-5.4"}, cfg.ModelAllowlist,
"model allowlist is a pure policy guardrail and must always reach the config")
assert.False(t, cfg.PromptCapture.Enabled,
"prompt capture must be off when the account toggle is off, even with a capture-enabled guardrail")
}
@@ -172,7 +172,7 @@ func TestSynthesizeServices_RealStore_NoGuardrail_CaptureOff(t *testing.T) {
require.Len(t, services, 1, "exactly one synth service expected")
cfg := decodeServiceGuardrailConfig(t, services[0])
assert.Empty(t, cfg.ProviderAllowlists, "no guardrail → provider unrestricted (absent from map)")
assert.Empty(t, cfg.ModelAllowlist, "no guardrail → no allowlist")
assert.False(t, cfg.PromptCapture.Enabled, "no guardrail → prompt capture off by default")
assert.False(t, cfg.PromptCapture.RedactPii, "no guardrail → redact off by default")
}

View File

@@ -1,95 +0,0 @@
package agentnetwork
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
)
// policyForProviders builds an enabled policy authorising the given providers
// under the given guardrails (both optional). Groups are irrelevant to
// buildProviderAllowlists, which keys purely on destination provider.
func policyForProviders(id string, guardrailIDs []string, providerIDs ...string) *types.Policy {
return &types.Policy{
ID: id,
Enabled: true,
DestinationProviderIDs: providerIDs,
GuardrailIDs: guardrailIDs,
}
}
func TestBuildProviderAllowlists(t *testing.T) {
byID := map[string]*types.Guardrail{
"g-4o": allowlistGuardrail("g-4o", "acc-1", "gpt-4o"),
"g-opus": allowlistGuardrail("g-opus", "acc-1", "claude-opus-4"),
"g-disabled": {ID: "g-disabled", Checks: types.GuardrailChecks{ModelAllowlist: types.GuardrailModelAllowlist{Enabled: false, Models: []string{"gpt-4o"}}}},
}
t.Run("all authorising policies restrict yields per-provider union", func(t *testing.T) {
policies := []*types.Policy{
policyForProviders("p1", []string{"g-4o"}, "prov-x"),
policyForProviders("p2", []string{"g-opus"}, "prov-x"),
}
got := buildProviderAllowlists(policies, byID)
assert.Equal(t, map[string][]string{"prov-x": {"claude-opus-4", "gpt-4o"}}, got,
"a provider every policy restricts carries the sorted union of their models")
})
t.Run("any un-guardrailed policy leaves the provider unrestricted (omitted)", func(t *testing.T) {
policies := []*types.Policy{
policyForProviders("p1", []string{"g-4o"}, "prov-x"),
policyForProviders("p2", nil, "prov-x"), // no guardrail
}
got := buildProviderAllowlists(policies, byID)
assert.NotContains(t, got, "prov-x",
"a provider reachable by an un-guardrailed policy must be omitted so the proxy treats it as unrestricted")
})
t.Run("a disabled allowlist counts as unrestricted", func(t *testing.T) {
policies := []*types.Policy{
policyForProviders("p1", []string{"g-disabled"}, "prov-x"),
}
got := buildProviderAllowlists(policies, byID)
assert.NotContains(t, got, "prov-x",
"a policy whose only guardrail has a disabled allowlist is unrestricted")
})
t.Run("providers are isolated from one another", func(t *testing.T) {
policies := []*types.Policy{
policyForProviders("p1", []string{"g-4o"}, "prov-x"),
policyForProviders("p2", []string{"g-opus"}, "prov-y"),
}
got := buildProviderAllowlists(policies, byID)
assert.Equal(t, []string{"gpt-4o"}, got["prov-x"], "prov-x keeps only its own model")
assert.Equal(t, []string{"claude-opus-4"}, got["prov-y"], "prov-y keeps only its own model")
})
t.Run("one policy authorising two providers restricts both", func(t *testing.T) {
policies := []*types.Policy{
policyForProviders("p1", []string{"g-4o"}, "prov-x", "prov-y"),
}
got := buildProviderAllowlists(policies, byID)
assert.Equal(t, []string{"gpt-4o"}, got["prov-x"])
assert.Equal(t, []string{"gpt-4o"}, got["prov-y"])
})
t.Run("union across a single policy's guardrails", func(t *testing.T) {
policies := []*types.Policy{
policyForProviders("p1", []string{"g-4o", "g-opus"}, "prov-x"),
}
got := buildProviderAllowlists(policies, byID)
assert.ElementsMatch(t, []string{"claude-opus-4", "gpt-4o"}, got["prov-x"],
"a policy's own multiple allowlist guardrails union together")
})
t.Run("an enabled allowlist with no models denies everything", func(t *testing.T) {
empty := map[string]*types.Guardrail{"g-empty": allowlistGuardrail("g-empty", "acc-1")}
got := buildProviderAllowlists([]*types.Policy{
policyForProviders("p1", []string{"g-empty"}, "prov-x"),
}, empty)
assert.Equal(t, map[string][]string{"prov-x": {}}, got,
"an enabled-but-empty allowlist is restricted with an empty set, not unrestricted")
})
}

View File

@@ -1031,12 +1031,8 @@ func TestSynthesizeServices_GuardrailMerge_AllowlistUnion_LimitsRestrictive(t *t
var cfg guardrailConfig
require.NoError(t, json.Unmarshal(guardrailJSON, &cfg), "guardrail config must unmarshal cleanly")
// Both policies restrict the same provider, so the per-provider backstop
// carries the union of their models — a coarse gate that management's
// per-policy/group check narrows; it only blocks models outside the union
// when management is down.
assert.ElementsMatch(t, []string{"gpt-5.4-mini", "gpt-5.4-pro"}, cfg.ProviderAllowlists["prov-1"],
"per-provider allowlist union must keep both models")
assert.ElementsMatch(t, []string{"gpt-5.4-mini", "gpt-5.4-pro"}, cfg.ModelAllowlist,
"model allowlist union must keep both models")
}
func TestSynthesizeServices_BackfillsMissingSessionKeys(t *testing.T) {
@@ -1223,3 +1219,102 @@ 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

@@ -0,0 +1,19 @@
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,6 +33,10 @@ 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
@@ -97,15 +101,16 @@ func NewProvider(accountID string) *Provider {
}
}
// 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.
// 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.
func (p *Provider) FromAPIRequest(req *api.AgentNetworkProviderRequest) {
p.ProviderID = req.ProviderId
p.Name = req.Name
p.UpstreamURL = req.UpstreamUrl
if req.ApiKey != nil && strings.TrimSpace(*req.ApiKey) != "" {
p.APIKey = *req.ApiKey
p.APIKeyProvided = req.ApiKey != nil
if req.ApiKey != nil {
p.APIKey = strings.TrimSpace(*req.ApiKey)
}
if req.ExtraValues != nil {
// Replace the whole map (rather than merge) so unsetting a
@@ -178,6 +183,7 @@ 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,3 +77,45 @@ 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

@@ -4,11 +4,16 @@ 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)
@@ -38,6 +43,7 @@ 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,6 +39,12 @@ 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()
@@ -50,10 +56,12 @@ func (c *GRPCController) RegisterProxyToCluster(ctx context.Context, clusterAddr
return nil
}
proxySet, _ := c.clusterProxies.LoadOrStore(clusterAddr, &sync.Map{})
proxySet.(*sync.Map).Store(proxyID, struct{}{})
_, alreadyRegistered := proxySet.(*sync.Map).LoadOrStore(proxyID, struct{}{})
log.WithContext(ctx).Debugf("Registered proxy %s to cluster %s", proxyID, clusterAddr)
c.metrics.IncrementProxyConnectionCount(clusterAddr)
if !alreadyRegistered {
c.metrics.IncrementProxyConnectionCount(clusterAddr)
}
return nil
}
@@ -64,10 +72,10 @@ func (c *GRPCController) UnregisterProxyFromCluster(ctx context.Context, cluster
return nil
}
if proxySet, ok := c.clusterProxies.Load(clusterAddr); ok {
proxySet.(*sync.Map).Delete(proxyID)
log.WithContext(ctx).Debugf("Unregistered proxy %s from cluster %s", proxyID, clusterAddr)
c.metrics.DecrementProxyConnectionCount(clusterAddr)
if _, registered := proxySet.(*sync.Map).LoadAndDelete(proxyID); registered {
log.WithContext(ctx).Debugf("Unregistered proxy %s from cluster %s", proxyID, clusterAddr)
c.metrics.DecrementProxyConnectionCount(clusterAddr)
}
}
return nil
}

View File

@@ -259,6 +259,21 @@ 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

@@ -0,0 +1,404 @@
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,6 +15,7 @@ import (
"net/http"
"net/url"
"os"
"sort"
"strconv"
"strings"
"sync"
@@ -29,6 +30,7 @@ import (
"github.com/netbirdio/netbird/shared/management/domain"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
"github.com/netbirdio/netbird/management/internals/modules/peers"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
@@ -36,7 +38,6 @@ import (
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/sessionkey"
"github.com/netbirdio/netbird/management/server/idp"
"github.com/netbirdio/netbird/management/server/peer"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
"github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/management/server/users"
proxyauth "github.com/netbirdio/netbird/proxy/auth"
@@ -84,6 +85,10 @@ 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
@@ -143,6 +148,8 @@ 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 {
@@ -171,10 +178,26 @@ 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
ctx context.Context
cancel context.CancelFunc
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
}
func enforceAccountScope(ctx context.Context, requestAccountID string) error {
@@ -192,17 +215,18 @@ 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(),
cancel: cancel,
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,
}
go s.cleanupStaleProxies(ctx)
return s
@@ -285,7 +309,6 @@ func (s *ProxyServiceServer) CheckLLMPolicyLimits(ctx context.Context, req *prot
UserID: req.GetUserId(),
GroupIDs: req.GetGroupIds(),
ProviderID: req.GetProviderId(),
Model: req.GetModel(),
})
if err != nil {
log.WithContext(ctx).Errorf("select policy for request: %v", err)
@@ -393,6 +416,7 @@ 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)
@@ -426,9 +450,10 @@ 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(stream, errChan)
go s.drainRecv(conn, stream, errChan)
return s.serveProxyConnection(conn, proxyRecord, errChan, true)
}
@@ -497,6 +522,8 @@ 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
@@ -544,10 +571,13 @@ 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 {
@@ -555,15 +585,19 @@ func (s *ProxyServiceServer) cleanupFailedSnapshot(ctx context.Context, conn *pr
}
}
// 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) {
// 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) {
for {
if _, err := stream.Recv(); err != nil {
msg, err := stream.Recv()
if err != nil {
errChan <- err
return
}
if result := msg.GetModelDiscoveryResult(); result != nil {
s.completeModelDiscovery(conn, result)
}
}
}
@@ -600,7 +634,9 @@ 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
@@ -617,6 +653,26 @@ 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 {
@@ -853,6 +909,20 @@ 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
}
@@ -870,6 +940,15 @@ 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()
@@ -1018,6 +1097,181 @@ 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
@@ -1050,6 +1304,12 @@ 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

@@ -1,138 +0,0 @@
package grpc
import (
"context"
"errors"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc/codes"
grpcstatus "google.golang.org/grpc/status"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
"github.com/netbirdio/netbird/shared/management/proto"
)
// fakeAgentNetworkLimits records the PolicySelectionInput it was invoked with
// and returns a pre-programmed result, so tests can assert what the handler
// forwards to the selector.
type fakeAgentNetworkLimits struct {
gotInput agentnetwork.PolicySelectionInput
result *agentnetwork.PolicySelectionResult
err error
}
func (f *fakeAgentNetworkLimits) SelectPolicyForRequest(_ context.Context, in agentnetwork.PolicySelectionInput) (*agentnetwork.PolicySelectionResult, error) {
f.gotInput = in
if f.err != nil {
return nil, f.err
}
return f.result, nil
}
func (f *fakeAgentNetworkLimits) RecordUsage(_ context.Context, _ agentnetwork.RecordUsageInput) error {
return nil
}
// TestCheckLLMPolicyLimits_ForwardsModelToSelector proves the wiring added here:
// the model the proxy extracted must reach the selector's Model unchanged,
// alongside the account/user/group/provider fields.
func TestCheckLLMPolicyLimits_ForwardsModelToSelector(t *testing.T) {
fake := &fakeAgentNetworkLimits{result: &agentnetwork.PolicySelectionResult{Allow: true, SelectedPolicyID: "pol-1"}}
s := &ProxyServiceServer{}
s.SetAgentNetworkLimitsService(fake)
req := &proto.CheckLLMPolicyLimitsRequest{
AccountId: "acc-1",
UserId: "user-1",
GroupIds: []string{"grp-a", "grp-b"},
ProviderId: "prov-1",
Model: "claude-opus-4",
}
resp, err := s.CheckLLMPolicyLimits(context.Background(), req)
require.NoError(t, err)
require.NotNil(t, resp)
assert.Equal(t, "acc-1", fake.gotInput.AccountID)
assert.Equal(t, "user-1", fake.gotInput.UserID)
assert.Equal(t, []string{"grp-a", "grp-b"}, fake.gotInput.GroupIDs)
assert.Equal(t, "prov-1", fake.gotInput.ProviderID)
assert.Equal(t, "claude-opus-4", fake.gotInput.Model,
"the request's model must be forwarded to the selector")
}
// TestCheckLLMPolicyLimits_EmptyModelForwardedAsEmpty proves an undetermined
// model (empty string) is forwarded as-is; the selector decides how to treat it.
func TestCheckLLMPolicyLimits_EmptyModelForwardedAsEmpty(t *testing.T) {
fake := &fakeAgentNetworkLimits{result: &agentnetwork.PolicySelectionResult{Allow: true}}
s := &ProxyServiceServer{}
s.SetAgentNetworkLimitsService(fake)
req := &proto.CheckLLMPolicyLimitsRequest{
AccountId: "acc-1",
ProviderId: "prov-1",
}
_, err := s.CheckLLMPolicyLimits(context.Background(), req)
require.NoError(t, err)
assert.Equal(t, "", fake.gotInput.Model, "an absent model must be forwarded as empty, not fabricated")
}
// TestCheckLLMPolicyLimits_DenyResponseCarriesModelBlockedCode proves the deny
// envelope surfaces the model-allowlist deny code + reason through the response.
func TestCheckLLMPolicyLimits_DenyResponseCarriesModelBlockedCode(t *testing.T) {
fake := &fakeAgentNetworkLimits{result: &agentnetwork.PolicySelectionResult{
Allow: false,
DenyCode: "llm_policy.model_blocked",
DenyReason: `model "claude-opus-4" is not permitted by any applicable policy allowlist`,
}}
s := &ProxyServiceServer{}
s.SetAgentNetworkLimitsService(fake)
resp, err := s.CheckLLMPolicyLimits(context.Background(), &proto.CheckLLMPolicyLimitsRequest{
AccountId: "acc-1",
ProviderId: "prov-1",
Model: "claude-opus-4",
})
require.NoError(t, err)
require.NotNil(t, resp)
assert.Equal(t, "deny", resp.Decision)
assert.Equal(t, "llm_policy.model_blocked", resp.DenyCode)
assert.NotEmpty(t, resp.DenyReason)
assert.Empty(t, resp.SelectedPolicyId, "a denied request must carry no selected policy")
}
// TestCheckLLMPolicyLimits_SelectorErrorSurfacesAsInternal proves a selector
// failure surfaces as an Internal gRPC error rather than a silent allow.
func TestCheckLLMPolicyLimits_SelectorErrorSurfacesAsInternal(t *testing.T) {
fake := &fakeAgentNetworkLimits{err: errors.New("boom")}
s := &ProxyServiceServer{}
s.SetAgentNetworkLimitsService(fake)
_, err := s.CheckLLMPolicyLimits(context.Background(), &proto.CheckLLMPolicyLimitsRequest{
AccountId: "acc-1",
ProviderId: "prov-1",
Model: "gpt-4o",
})
require.Error(t, err)
st, ok := grpcstatus.FromError(err)
require.True(t, ok)
assert.Equal(t, codes.Internal, st.Code(), "selector errors must never fail open on the hot path")
}
// TestCheckLLMPolicyLimits_UnconfiguredServiceReturnsUnimplemented locks the
// fallback: with no limits service wired the RPC returns Unimplemented.
func TestCheckLLMPolicyLimits_UnconfiguredServiceReturnsUnimplemented(t *testing.T) {
s := &ProxyServiceServer{}
_, err := s.CheckLLMPolicyLimits(context.Background(), &proto.CheckLLMPolicyLimitsRequest{
AccountId: "acc-1",
ProviderId: "prov-1",
})
require.Error(t, err)
st, ok := grpcstatus.FromError(err)
require.True(t, ok)
assert.Equal(t, codes.Unimplemented, st.Code())
}

View File

@@ -42,6 +42,10 @@ 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{}
}
@@ -218,6 +222,65 @@ 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

@@ -3,30 +3,20 @@ package auth
import (
"context"
"net/netip"
"os"
"strings"
"sync"
"time"
log "github.com/sirupsen/logrus"
"golang.org/x/sync/singleflight"
"github.com/netbirdio/netbird/proxy/internal/types"
"github.com/netbirdio/netbird/shared/management/proto"
)
// tunnelCacheTTL is the default cap on how long a positive ValidateTunnelPeer
// result is reused before re-fetching from management. 5 minutes balances
// freshness against management load on busy mesh networks. Override it with
// envTunnelCacheTTL when an account needs authorization changes to take effect
// sooner (at the cost of more ValidateTunnelPeer RPCs).
// tunnelCacheTTL caps how long a positive ValidateTunnelPeer result is
// reused before re-fetching from management. 5 minutes balances freshness
// against management load on busy mesh networks.
const tunnelCacheTTL = 300 * time.Second
// envTunnelCacheTTL overrides tunnelCacheTTL. The value is a Go duration string
// (e.g. "30s", "2m"); an unset, unparseable, or non-positive value keeps the
// default.
const envTunnelCacheTTL = "NB_PROXY_TUNNEL_CACHE_TTL"
// tunnelCachePerAccount caps the number of cached identities per account.
// Bounded eviction avoids memory growth in pathological cases (huge peer
// roster, brief request bursts) while staying generous for normal use.
@@ -70,35 +60,16 @@ type accountBucket struct {
order []tunnelCacheKey
}
// newTunnelValidationCache constructs a cache with the configured TTL
// (envTunnelCacheTTL override or default) and default bounds.
// newTunnelValidationCache constructs a cache with default TTL and bounds.
func newTunnelValidationCache() *tunnelValidationCache {
return &tunnelValidationCache{
entries: make(map[types.AccountID]*accountBucket),
ttl: tunnelCacheTTLFromEnv(),
ttl: tunnelCacheTTL,
maxSize: tunnelCachePerAccount,
now: time.Now,
}
}
// tunnelCacheTTLFromEnv returns the tunnel-cache TTL, honoring the
// envTunnelCacheTTL override. The override must be a positive Go duration
// string (e.g. "30s", "2m"); anything unset, unparseable, or non-positive
// falls back to tunnelCacheTTL.
func tunnelCacheTTLFromEnv() time.Duration {
raw := strings.TrimSpace(os.Getenv(envTunnelCacheTTL))
if raw == "" {
return tunnelCacheTTL
}
d, err := time.ParseDuration(raw)
if err != nil || d <= 0 {
log.Warnf("ignoring invalid %s=%q (want a positive Go duration like 30s or 2m); using default %s",
envTunnelCacheTTL, raw, tunnelCacheTTL)
return tunnelCacheTTL
}
return d
}
// get returns a cached response for the key, or nil when missing or
// expired. Expired entries are evicted lazily on read.
func (c *tunnelValidationCache) get(key tunnelCacheKey) *proto.ValidateTunnelPeerResponse {

View File

@@ -169,32 +169,3 @@ func TestTunnelCache_BoundedSizeEvictsOldest(t *testing.T) {
assert.NotNil(t, cache.get(keys[1]), "second-newest must remain cached")
assert.NotNil(t, cache.get(keys[2]), "newest must remain cached")
}
func TestTunnelCacheTTLFromEnv(t *testing.T) {
t.Run("unset uses default", func(t *testing.T) {
t.Setenv(envTunnelCacheTTL, "")
assert.Equal(t, tunnelCacheTTL, tunnelCacheTTLFromEnv())
})
t.Run("valid duration overrides", func(t *testing.T) {
t.Setenv(envTunnelCacheTTL, "45s")
assert.Equal(t, 45*time.Second, tunnelCacheTTLFromEnv())
})
t.Run("whitespace trimmed", func(t *testing.T) {
t.Setenv(envTunnelCacheTTL, " 2m ")
assert.Equal(t, 2*time.Minute, tunnelCacheTTLFromEnv())
})
t.Run("unparseable uses default", func(t *testing.T) {
t.Setenv(envTunnelCacheTTL, "nonsense")
assert.Equal(t, tunnelCacheTTL, tunnelCacheTTLFromEnv())
})
t.Run("non-positive uses default", func(t *testing.T) {
t.Setenv(envTunnelCacheTTL, "0s")
assert.Equal(t, tunnelCacheTTL, tunnelCacheTTLFromEnv())
t.Setenv(envTunnelCacheTTL, "-30s")
assert.Equal(t, tunnelCacheTTL, tunnelCacheTTLFromEnv())
})
t.Run("constructor honors override", func(t *testing.T) {
t.Setenv(envTunnelCacheTTL, "90s")
assert.Equal(t, 90*time.Second, newTunnelValidationCache().ttl)
})
}

View File

@@ -28,12 +28,12 @@ func TestDefaultTable_FirstPartyModelCoverage(t *testing.T) {
"ministral-8b-latest", "mistral-embed",
},
"anthropic": {
"claude-fable-5", "claude-opus-5", "claude-opus-4-8", "claude-opus-4-7", "claude-opus-4-6",
"claude-fable-5", "claude-opus-4-8", "claude-opus-4-7", "claude-opus-4-6",
"claude-opus-4-1", "claude-sonnet-4-6", "claude-sonnet-4-5", "claude-haiku-4-5",
},
// bedrock keys are the normalized ids the request parser emits.
"bedrock": {
"anthropic.claude-opus-5", "anthropic.claude-opus-4-8", "anthropic.claude-opus-4-7", "anthropic.claude-opus-4-6",
"anthropic.claude-opus-4-8", "anthropic.claude-opus-4-7", "anthropic.claude-opus-4-6",
"anthropic.claude-opus-4-1", "anthropic.claude-sonnet-4-6", "anthropic.claude-sonnet-4-5",
"anthropic.claude-haiku-4-5", "meta.llama3-3-70b-instruct",
"amazon.nova-pro", "amazon.nova-lite", "amazon.nova-micro", "amazon.nova-2-lite",

View File

@@ -180,11 +180,6 @@ anthropic:
output_per_1k: 0.050
cache_read_per_1k: 0.001
cache_creation_per_1k: 0.0125
claude-opus-5:
input_per_1k: 0.005
output_per_1k: 0.025
cache_read_per_1k: 0.0005
cache_creation_per_1k: 0.00625
claude-opus-4-8:
input_per_1k: 0.005
output_per_1k: 0.025
@@ -241,11 +236,6 @@ bedrock:
# eu.anthropic.claude-sonnet-4-5-20250929-v1:0 -> anthropic.claude-sonnet-4-5.
# Anthropic-on-Bedrock keeps the additive cache buckets (read ≈0.1x input,
# write ≈1.25x input); Nova / Llama report no cache, so cost is input+output.
anthropic.claude-opus-5:
input_per_1k: 0.005
output_per_1k: 0.025
cache_read_per_1k: 0.0005
cache_creation_per_1k: 0.00625
anthropic.claude-opus-4-8:
input_per_1k: 0.005
output_per_1k: 0.025

View File

@@ -10,15 +10,11 @@ import (
)
// Config is the JSON-decoded shape accepted by the factory. The
// runtime path consumes the normalised allowlists; raw config is not
// runtime path consumes the normalised allowlist; raw config is not
// retained beyond construction.
type Config struct {
// ProviderAllowlists maps a resolved provider id (KeyLLMResolvedProviderID) to
// its model allowlist. A provider present is restricted to those models; one
// absent is unrestricted. Kept per-provider so one provider's list can't leak
// onto another.
ProviderAllowlists map[string][]string `json:"provider_allowlists,omitempty"`
PromptCapture PromptCapture `json:"prompt_capture"`
ModelAllowlist []string `json:"model_allowlist"`
PromptCapture PromptCapture `json:"prompt_capture"`
}
// PromptCapture toggles the optional prompt capture + redaction step
@@ -58,28 +54,21 @@ func isEmptyJSON(raw []byte) bool {
return false
}
// normaliseConfig lowercases and trims allowlist entries for case-insensitive
// matching; empty entries drop. A provider whose entries all drop keeps an empty
// (non-nil) list — "deny every model" — distinct from an absent provider
// (unrestricted).
// normaliseConfig lowercases and trims allowlist entries so the runtime
// match is case-insensitive. Empty entries are dropped.
func normaliseConfig(cfg Config) Config {
if len(cfg.ProviderAllowlists) == 0 {
cfg.ProviderAllowlists = nil
if len(cfg.ModelAllowlist) == 0 {
return cfg
}
cleaned := make(map[string][]string, len(cfg.ProviderAllowlists))
for provider, models := range cfg.ProviderAllowlists {
list := make([]string, 0, len(models))
for _, entry := range models {
n := normaliseModel(entry)
if n == "" {
continue
}
list = append(list, n)
cleaned := make([]string, 0, len(cfg.ModelAllowlist))
for _, entry := range cfg.ModelAllowlist {
n := normaliseModel(entry)
if n == "" {
continue
}
cleaned[provider] = list
cleaned = append(cleaned, n)
}
cfg.ProviderAllowlists = cleaned
cfg.ModelAllowlist = cleaned
return cfg
}

View File

@@ -83,9 +83,8 @@ func (m *Middleware) MutationsSupported() bool { return false }
// prompt capture only affects the metadata emitted alongside an allow.
func (m *Middleware) Invoke(_ context.Context, in *middleware.Input) (*middleware.Output, error) {
model, modelPresent := lookupMetadata(in.Metadata, middleware.KeyLLMModel)
providerID, _ := lookupMetadata(in.Metadata, middleware.KeyLLMResolvedProviderID)
if denial := m.evaluateAllowlist(providerID, model, modelPresent); denial != nil {
if denial := m.evaluateAllowlist(model, modelPresent); denial != nil {
return denial, nil
}
@@ -111,32 +110,20 @@ func (m *Middleware) Invoke(_ context.Context, in *middleware.Input) (*middlewar
// is a no-op.
func (m *Middleware) Close() error { return nil }
// evaluateAllowlist denies when the resolved provider's allowlist rejects the
// model; nil means proceed. Scoped to the provider llm_router resolved, so an
// unrestricted provider (absent from config) is never caught by another's list.
func (m *Middleware) evaluateAllowlist(providerID, model string, modelPresent bool) *middleware.Output {
if len(m.cfg.ProviderAllowlists) == 0 {
// evaluateAllowlist returns a deny Output when the configured allowlist
// rejects the model. A nil return means the request should proceed.
func (m *Middleware) evaluateAllowlist(model string, modelPresent bool) *middleware.Output {
if len(m.cfg.ModelAllowlist) == 0 {
return nil
}
// Restrictions exist but the resolved provider is unknown, so we can't tell
// if this request targets a restricted provider — fail closed. llm_router
// normally stamps the provider first, so this is a defensive guard.
if providerID == "" {
return denyModel("", denyCodeModelUnknown, denyMessageModelUnknown, denyReasonModelUnknown)
}
allowlist, restricted := m.cfg.ProviderAllowlists[providerID]
if !restricted {
// This provider has no allowlist (some authorising policy left it
// unrestricted); management owns any per-policy/group decision.
return nil
}
// Fail closed: with an allowlist in effect for this provider, a request whose
// model the parser couldn't extract (absent/empty) is denied. This enforces
// the allowlist for path-routed providers (Bedrock, Vertex) with no body model.
// Fail closed: with an allowlist configured, a request whose model the
// upstream parser could not extract (absent or empty) must be denied rather
// than allowed. This is what enforces the allowlist for URL/path-routed
// providers (Bedrock, Vertex, ...) whose model lives outside the JSON body.
if !modelPresent || normaliseModel(model) == "" {
return denyModel("", denyCodeModelUnknown, denyMessageModelUnknown, denyReasonModelUnknown)
}
if modelInAllowlist(allowlist, model) {
if m.modelInAllowlist(model) {
return nil
}
return denyModel(model, denyCodeModel, denyMessageModel, denyReasonModel)
@@ -164,15 +151,14 @@ func denyModel(model, code, message, reason string) *middleware.Output {
}
}
// modelInAllowlist reports whether the model matches any entry in the supplied
// (already-normalised) allowlist under the case-insensitive, trim-tolerant
// comparison rule.
func modelInAllowlist(allowlist []string, model string) bool {
// modelInAllowlist reports whether the model matches any allowlist
// entry under the case-insensitive, trim-tolerant comparison rule.
func (m *Middleware) modelInAllowlist(model string) bool {
normalised := normaliseModel(model)
if normalised == "" {
return false
}
for _, allowed := range allowlist {
for _, allowed := range m.cfg.ModelAllowlist {
if allowed == normalised {
return true
}

View File

@@ -26,25 +26,6 @@ func newInput(meta ...middleware.KV) *middleware.Input {
return &middleware.Input{Slot: middleware.SlotOnRequest, Metadata: meta}
}
const (
testProvider = "prov-1"
otherProvider = "prov-2"
)
// providerCfg builds a Config restricting testProvider to the given models.
func providerCfg(models ...string) Config {
return Config{ProviderAllowlists: map[string][]string{testProvider: models}}
}
// newInputProvider builds an input that carries a resolved provider id (as
// llm_router would stamp) plus any extra metadata.
func newInputProvider(provider string, meta ...middleware.KV) *middleware.Input {
all := make([]middleware.KV, 0, len(meta)+1)
all = append(all, middleware.KV{Key: middleware.KeyLLMResolvedProviderID, Value: provider})
all = append(all, meta...)
return &middleware.Input{Slot: middleware.SlotOnRequest, Metadata: all}
}
func TestMiddlewareIdentity(t *testing.T) {
mw := New(Config{})
assert.Equal(t, ID, mw.ID(), "middleware ID must be llm_guardrail")
@@ -66,12 +47,12 @@ func TestMiddlewareIdentity(t *testing.T) {
func TestAllowlistEmptyAllowsAnyModel(t *testing.T) {
mw := New(Config{})
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
out, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: "gpt-4o"},
))
require.NoError(t, err)
require.NotNil(t, out)
assert.Equal(t, middleware.DecisionAllow, out.Decision, "no provider allowlists must allow any model")
assert.Equal(t, middleware.DecisionAllow, out.Decision, "empty allowlist must allow any model")
v, ok := metaValue(t, out.Metadata, middleware.KeyLLMPolicyDecision)
require.True(t, ok, "decision metadata must be emitted")
assert.Equal(t, "allow", v, "decision must be allow")
@@ -81,8 +62,8 @@ func TestAllowlistEmptyAllowsAnyModel(t *testing.T) {
}
func TestAllowlistMatchAllows(t *testing.T) {
mw := New(providerCfg("gpt-4o", "claude-opus-4"))
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
mw := New(Config{ModelAllowlist: []string{"gpt-4o", "claude-opus-4"}})
out, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: "gpt-4o"},
))
require.NoError(t, err)
@@ -90,8 +71,8 @@ func TestAllowlistMatchAllows(t *testing.T) {
}
func TestAllowlistMissDenies(t *testing.T) {
mw := New(providerCfg("gpt-4o"))
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
mw := New(Config{ModelAllowlist: []string{"gpt-4o"}})
out, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: "claude-opus-4"},
))
require.NoError(t, err)
@@ -110,10 +91,10 @@ func TestAllowlistMissDenies(t *testing.T) {
}
func TestAllowlistCaseInsensitive(t *testing.T) {
mw := New(providerCfg(" GPT-4o ", "Claude-OPUS-4"))
mw := New(Config{ModelAllowlist: []string{" GPT-4o ", "Claude-OPUS-4"}})
cases := []string{"gpt-4o", "GPT-4O", " claude-opus-4 "}
for _, model := range cases {
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
out, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: model},
))
require.NoError(t, err)
@@ -122,15 +103,14 @@ func TestAllowlistCaseInsensitive(t *testing.T) {
}
func TestAllowlistMissingModelKeyDenies(t *testing.T) {
// Fail closed: with an allowlist in effect for the resolved provider, a
// request whose model the parser could not extract (URL/path-routed
// providers such as Bedrock or Vertex whose shape wasn't recognised) must be
// denied, not allowed.
mw := New(providerCfg("gpt-4o"))
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider))
// Fail closed: with an allowlist configured, a request whose model the
// parser could not extract (URL/path-routed providers such as Bedrock or
// Vertex whose shape wasn't recognised) must be denied, not allowed.
mw := New(Config{ModelAllowlist: []string{"gpt-4o"}})
out, err := mw.Invoke(context.Background(), newInput())
require.NoError(t, err)
require.NotNil(t, out)
assert.Equal(t, middleware.DecisionDeny, out.Decision, "absent model must be denied when the provider is restricted")
assert.Equal(t, middleware.DecisionDeny, out.Decision, "absent model must be denied when an allowlist is set")
assert.Equal(t, 403, out.DenyStatus, "deny status must be 403")
require.NotNil(t, out.DenyReason, "deny reason must be populated")
assert.Equal(t, "llm_policy.model_unknown", out.DenyReason.Code, "deny code must be model_unknown")
@@ -142,101 +122,26 @@ func TestAllowlistMissingModelKeyDenies(t *testing.T) {
func TestAllowlistEmptyModelValueDenies(t *testing.T) {
// A present-but-empty model is as undeterminable as an absent one.
mw := New(providerCfg("gpt-4o"))
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
mw := New(Config{ModelAllowlist: []string{"gpt-4o"}})
out, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: " "},
))
require.NoError(t, err)
require.NotNil(t, out)
assert.Equal(t, middleware.DecisionDeny, out.Decision, "empty model must be denied when the provider is restricted")
assert.Equal(t, middleware.DecisionDeny, out.Decision, "empty model must be denied when an allowlist is set")
require.NotNil(t, out.DenyReason, "deny reason must be populated")
assert.Equal(t, "llm_policy.model_unknown", out.DenyReason.Code, "deny code must be model_unknown")
}
func TestAllowlistEmptyListAllowsMissingModel(t *testing.T) {
// Without any provider allowlists there is nothing to enforce, so a missing
// model is still allowed — the fail-closed rule only applies when a
// restriction is in effect.
// Without an allowlist there is nothing to enforce, so a missing model is
// still allowed — the fail-closed rule only applies when a list is set.
mw := New(Config{})
out, err := mw.Invoke(context.Background(), newInput())
require.NoError(t, err)
assert.Equal(t, middleware.DecisionAllow, out.Decision, "no allowlist must allow even without a model")
}
func TestUnrestrictedProviderAllowsAnyModel(t *testing.T) {
// The request resolved to otherProvider, which has no allowlist, so its
// traffic must not be caught by testProvider's restriction — the
// cross-provider-leak / false-deny guard.
mw := New(providerCfg("gpt-4o"))
out, err := mw.Invoke(context.Background(), newInputProvider(otherProvider,
middleware.KV{Key: middleware.KeyLLMModel, Value: "claude-opus-4"},
))
require.NoError(t, err)
assert.Equal(t, middleware.DecisionAllow, out.Decision, "an unrestricted provider must not inherit another provider's allowlist")
}
func TestPerProviderAllowlistsAreIsolated(t *testing.T) {
// gpt-4o is allowed only on testProvider; claude-opus-4 only on
// otherProvider. A model allowlisted for one provider must not be usable on
// the other — the fail-closed layer never unions allowlists across providers.
mw := New(Config{ProviderAllowlists: map[string][]string{
testProvider: {"gpt-4o"},
otherProvider: {"claude-opus-4"},
}})
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
middleware.KV{Key: middleware.KeyLLMModel, Value: "claude-opus-4"},
))
require.NoError(t, err)
assert.Equal(t, middleware.DecisionDeny, out.Decision, "claude-opus-4 is allowed only on otherProvider, not testProvider")
require.NotNil(t, out.DenyReason)
assert.Equal(t, "llm_policy.model_blocked", out.DenyReason.Code, "cross-provider model must be blocked, not model_unknown")
}
func TestRestrictionsButNoResolvedProviderFailsClosed(t *testing.T) {
// Restrictions exist for the account but the resolved provider id is absent,
// so the request cannot be scoped to a provider. Fail closed rather than
// wave it through.
mw := New(providerCfg("gpt-4o"))
out, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: "gpt-4o"},
))
require.NoError(t, err)
require.NotNil(t, out)
assert.Equal(t, middleware.DecisionDeny, out.Decision, "missing resolved provider must fail closed when restrictions exist")
require.NotNil(t, out.DenyReason)
assert.Equal(t, "llm_policy.model_unknown", out.DenyReason.Code, "deny code must be model_unknown")
}
func TestEnabledButEmptyAllowlistDeniesEveryModel(t *testing.T) {
// An allowlist-enabled provider with zero models is distinct from an
// unrestricted (absent) provider: it must deny every model.
mw := New(providerCfg())
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
middleware.KV{Key: middleware.KeyLLMModel, Value: "gpt-4o"},
))
require.NoError(t, err)
require.NotNil(t, out)
assert.Equal(t, middleware.DecisionDeny, out.Decision, "an enabled-but-empty allowlist must deny every model")
require.NotNil(t, out.DenyReason)
assert.Equal(t, "llm_policy.model_blocked", out.DenyReason.Code, "deny code must be model_blocked, not model_unknown")
}
func TestFactoryAllEmptyEntriesDenyEveryModel(t *testing.T) {
// All the provider's entries are blank; they collapse to a non-nil empty
// list (deny everything for that provider), not "no restriction".
raw := []byte(`{"provider_allowlists":{"prov-1":[""," "]}}`)
mw, err := Factory{}.New(raw)
require.NoError(t, err)
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
middleware.KV{Key: middleware.KeyLLMModel, Value: "gpt-4o"},
))
require.NoError(t, err)
require.NotNil(t, out)
assert.Equal(t, middleware.DecisionDeny, out.Decision, "all-blank allowlist entries must still restrict the provider")
require.NotNil(t, out.DenyReason)
assert.Equal(t, "llm_policy.model_blocked", out.DenyReason.Code, "deny code must be model_blocked")
}
func TestPromptCaptureDisabledEmitsNoPrompt(t *testing.T) {
mw := New(Config{})
out, err := mw.Invoke(context.Background(), newInput(
@@ -312,8 +217,8 @@ func TestFactoryAcceptsZeroConfigs(t *testing.T) {
func TestFactoryDecodesValidConfig(t *testing.T) {
cfg := Config{
ProviderAllowlists: map[string][]string{testProvider: {"gpt-4o"}},
PromptCapture: PromptCapture{Enabled: true, RedactPii: true},
ModelAllowlist: []string{"gpt-4o"},
PromptCapture: PromptCapture{Enabled: true, RedactPii: true},
}
raw, err := json.Marshal(cfg)
require.NoError(t, err, "marshalling test config must succeed")
@@ -329,15 +234,15 @@ func TestFactoryRejectsMalformedJSON(t *testing.T) {
}
func TestFactoryNormalisesAllowlist(t *testing.T) {
raw := []byte(`{"provider_allowlists":{"prov-1":[" GPT-4o ","",""," Claude-3 "]}}`)
raw := []byte(`{"model_allowlist":[" GPT-4o ","",""," Claude-3 "]}`)
mw, err := Factory{}.New(raw)
require.NoError(t, err)
out, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
out, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: "gpt-4o"},
))
require.NoError(t, err)
assert.Equal(t, middleware.DecisionAllow, out.Decision, "factory must lowercase + trim allowlist entries")
out2, err := mw.Invoke(context.Background(), newInputProvider(testProvider,
out2, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: "claude-3"},
))
require.NoError(t, err)

View File

@@ -175,7 +175,7 @@ func denyFromManagement(resp *proto.CheckLLMPolicyLimitsResponse) *middleware.Ou
DenyStatus: 403,
DenyReason: &middleware.DenyReason{
Code: code,
Message: denyMessageForCode(code),
Message: "LLM policy limit exceeded",
},
Metadata: []middleware.KV{
{Key: middleware.KeyLLMPolicyDecision, Value: "deny"},
@@ -184,21 +184,6 @@ func denyFromManagement(resp *proto.CheckLLMPolicyLimitsResponse) *middleware.Ou
}
}
// denyMessageForCode maps a management deny code to a public message.
// Model-allowlist rejections get a model-specific message matching the
// local guardrail; everything else keeps the generic quota wording. The
// message stays generic so it never leaks internal quota detail.
func denyMessageForCode(code string) string {
switch code {
case "llm_policy.model_blocked":
return "model is not in the policy allowlist"
case "llm_policy.model_unknown":
return "request model could not be determined for the policy allowlist"
default:
return "LLM policy limit exceeded"
}
}
// lookupKV returns the value associated with key, or the empty
// string when absent.
func lookupKV(kvs []middleware.KV, key string) string {

View File

@@ -115,46 +115,6 @@ func TestInvoke_DenyConvertsToProxyDeny(t *testing.T) {
assert.NotContains(t, out.DenyReason.Message, "1000", "internal cap numbers must not reach the caller")
}
// TestInvoke_ModelDenyMessages proves a model-allowlist rejection gets a
// model-specific public message rather than the generic quota wording, so a
// blocked or undetermined model reads consistently with the local guardrail.
func TestInvoke_ModelDenyMessages(t *testing.T) {
cases := []struct {
name string
code string
message string
}{
{"blocked", "llm_policy.model_blocked", "model is not in the policy allowlist"},
{"unknown", "llm_policy.model_unknown", "request model could not be determined for the policy allowlist"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
mgmt := &fakeMgmt{
checkResp: &proto.CheckLLMPolicyLimitsResponse{
Decision: "deny",
DenyCode: tc.code,
},
}
m := New(mgmt, nil)
out := runInvoke(t, m, &middleware.Input{
AccountID: "acc-1",
UserGroups: []string{"grp-engineers"},
Metadata: []middleware.KV{
{Key: middleware.KeyLLMResolvedProviderID, Value: "prov-1"},
{Key: middleware.KeyLLMModel, Value: "some-model"},
},
})
assert.Equal(t, middleware.DecisionDeny, out.Decision)
require.NotNil(t, out.DenyReason, "deny envelope must carry a reason payload")
assert.Equal(t, tc.code, out.DenyReason.Code, "canonical deny code surfaces to the caller")
assert.Equal(t, tc.message, out.DenyReason.Message,
"model denials must use a model-specific message, matching the local guardrail")
})
}
}
// TestInvoke_NoMgmtClientPassesThrough proves the partial-wiring
// safety: a middleware constructed without a management client
// allows every request without attribution. This makes a half-set-up

View File

@@ -25,17 +25,10 @@ func runParserGuardrail(t *testing.T, url string, body []byte, allowlist []strin
})
require.NoError(t, err, "parser must not error")
const providerID = "prov-under-test"
guard := llm_guardrail.New(llm_guardrail.Config{
ProviderAllowlists: map[string][]string{providerID: allowlist},
})
// The real chain has llm_router stamp the resolved provider id before the
// guardrail runs; the parser doesn't, so add it here so the guardrail can
// scope the allowlist to this provider.
meta := append([]middleware.KV{{Key: middleware.KeyLLMResolvedProviderID, Value: providerID}}, parsed.Metadata...)
guard := llm_guardrail.New(llm_guardrail.Config{ModelAllowlist: allowlist})
out, err := guard.Invoke(context.Background(), &middleware.Input{
Slot: middleware.SlotOnRequest,
Metadata: meta,
Metadata: parsed.Metadata,
})
require.NoError(t, err, "guardrail must not error")
require.NotNil(t, out, "guardrail must return an output")

View File

@@ -0,0 +1,337 @@
// 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

@@ -0,0 +1,315 @@
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

@@ -271,6 +271,10 @@ 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{}
}

View File

@@ -0,0 +1,349 @@
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

@@ -57,6 +57,7 @@ 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"
@@ -78,6 +79,10 @@ 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
@@ -100,16 +105,20 @@ type Server struct {
// middlewareRegistry is the source of registered middleware factories.
// Concrete middlewares register themselves through init().
middlewareRegistry *middleware.Registry
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
// 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
// hijackTracker tracks hijacked connections (e.g. WebSocket upgrades)
// so they can be closed during graceful shutdown, since http.Server.Shutdown
@@ -1277,12 +1286,16 @@ func (s *Server) proxyCapabilities() *proto.ProxyCapabilities {
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,
Private: &privateCapability,
SupportsPrivateService: &supportsPrivateService,
SupportsModelDiscovery: &supportsModelDiscovery,
}
}
@@ -1349,7 +1362,8 @@ 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.
// back-pressure. Model discovery commands are out-of-band: they run with
// bounded concurrency and return a correlated result instead of an ack.
func (s *Server) handleSyncMappingsStream(ctx context.Context, stream proto.ProxyService_SyncMappingsClient, initialSyncDone *bool, connectTime time.Time) error {
select {
case <-s.routerReady:
@@ -1358,6 +1372,30 @@ 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 {
@@ -1372,15 +1410,62 @@ 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")
tracker.recordBatch(ctx, s, msg.GetMapping(), msg.GetInitialSyncComplete(), batchStart)
syncComplete := msg.GetInitialSyncComplete()
tracker.recordBatch(ctx, s, msg.GetMapping(), syncComplete, batchStart)
if syncComplete {
initialSnapshotComplete = true
}
if err := stream.Send(&proto.SyncMappingsRequest{
if err := send(&proto.SyncMappingsRequest{
Msg: &proto.SyncMappingsRequest_Ack{Ack: &proto.SyncMappingsAck{}},
}); err != nil {
return fmt.Errorf("send ack: %w", err)
@@ -1389,6 +1474,31 @@ 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

View File

@@ -5138,6 +5138,9 @@ 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: |
@@ -5186,6 +5189,7 @@ components:
- name
- upstream_url
- models
- has_api_key
- enabled
- skip_tls_verification
- metadata_disabled
@@ -5212,7 +5216,8 @@ 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. Required on create; optional on update (omit to keep the existing key).
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.
example: "sk-..."
models:
type: array
@@ -5275,6 +5280,51 @@ 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:
@@ -5325,6 +5375,11 @@ 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).
@@ -5346,6 +5401,10 @@ 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: |
@@ -5364,10 +5423,12 @@ components:
- name
- description
- default_host
- auth_mode
- auth_header_template
- default_content_type
- brand_color
- kind
- supports_model_discovery
- models
AgentNetworkCatalogIdentityInjection:
type: object
@@ -14028,6 +14089,48 @@ 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

@@ -38,6 +38,27 @@ 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"
@@ -77,6 +98,24 @@ 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"
@@ -2038,6 +2077,9 @@ 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"`
@@ -2070,8 +2112,14 @@ 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.
@@ -2108,6 +2156,15 @@ 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.
@@ -2155,6 +2212,26 @@ 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.
@@ -2260,6 +2337,9 @@ 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"`
@@ -2305,7 +2385,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. Required on create; optional on update (omit to keep the existing key).
// 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 *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.

File diff suppressed because it is too large Load Diff

View File

@@ -73,6 +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;
}
// GetMappingUpdateRequest is sent to initialise a mapping stream.
@@ -420,6 +424,8 @@ message SyncMappingsRequest {
oneof msg {
SyncMappingsInit init = 1;
SyncMappingsAck ack = 2;
// Correlated response to a ModelDiscoveryRequest received from management.
ModelDiscoveryResult model_discovery_result = 3;
}
}
@@ -443,6 +449,9 @@ 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
@@ -501,3 +510,33 @@ 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;
}