[agent-network] Management: store, manager, synthesizer, policy engine, provider catalog, HTTP/gRPC API

Adds the account-scoped agent-network module: provider/policy/budget CRUD and
store, the reverse-proxy service synthesizer, policy selection + limit
enforcement, the provider catalog (incl. Vertex AI and AWS Bedrock entries),
and the management HTTP + proxy gRPC surfaces.
This commit is contained in:
mlsmaycon
2026-06-27 00:43:07 +02:00
parent 350a96c640
commit 769e12840d
59 changed files with 12576 additions and 14 deletions
@@ -3,7 +3,9 @@ package controller
import (
"context"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
"github.com/netbirdio/netbird/management/internals/modules/zones"
"github.com/netbirdio/netbird/management/server/agentnetwork"
"github.com/netbirdio/netbird/management/server/peer"
"github.com/netbirdio/netbird/management/server/store"
"github.com/netbirdio/netbird/management/server/types"
@@ -16,6 +18,10 @@ type Repository interface {
GetPeersByIDs(ctx context.Context, accountID string, peerIDs []string) (map[string]*peer.Peer, error)
GetPeerByID(ctx context.Context, accountID string, peerID string) (*peer.Peer, error)
GetAccountZones(ctx context.Context, accountID string) ([]*zones.Zone, error)
// SynthesizeAgentNetworkServices returns the in-memory reverse-proxy
// services synthesised from the account's agent-network provider/policy
// state. Empty for accounts without agent-network providers.
SynthesizeAgentNetworkServices(ctx context.Context, accountID string) ([]*service.Service, error)
}
type repository struct {
@@ -50,6 +56,10 @@ func (r *repository) GetPeerByID(ctx context.Context, accountID string, peerID s
return r.store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
}
func (r *repository) SynthesizeAgentNetworkServices(ctx context.Context, accountID string) ([]*service.Service, error) {
return agentnetwork.SynthesizeServices(ctx, r.store, accountID)
}
func (r *repository) GetAccountZones(ctx context.Context, accountID string) ([]*zones.Zone, error) {
return r.store.GetAccountZones(ctx, store.LockingStrengthNone, accountID)
}
+53 -3
View File
@@ -220,12 +220,36 @@ func (m *managerImpl) GetPeerID(ctx context.Context, peerKey string) (string, er
func (m *managerImpl) CreateProxyPeer(ctx context.Context, accountID string, peerKey string, cluster string) error {
existingPeerID, err := m.store.GetPeerIDByKey(ctx, store.LockingStrengthNone, peerKey)
if err == nil && existingPeerID != "" {
// Peer already exists
// Same pubkey already registered — idempotent.
return nil
}
// Dedupe stale embedded peer records for the same (account, cluster).
// The proxy generates a fresh WireGuard keypair on every startup
// (proxy/internal/roundtrip/netbird.go), so without this sweep the
// prior embedded peer would linger forever — holding its CGNAT IP
// allocation, polluting other peers' rosters, and (most visibly)
// leaving the synth DNS pointing at the dead address. The
// (account, cluster) tuple identifies "the embedded peer for this
// proxy instance at this cluster"; any record matching that tuple
// with a different pubkey is by definition stale and must go.
staleIDs, err := m.findStaleEmbeddedProxyPeers(ctx, accountID, cluster, peerKey)
if err != nil {
return fmt.Errorf("scan for stale embedded proxy peers: %w", err)
}
if len(staleIDs) > 0 {
// userID="" + checkConnected=false: the deletion is initiated
// by management itself on behalf of the freshly-registering
// proxy, not by an end user; the stale peer may still be
// marked Connected from its prior session, but its session is
// dead by definition (its key no longer exists).
if err := m.DeletePeers(ctx, accountID, staleIDs, "", false); err != nil {
return fmt.Errorf("delete stale embedded proxy peers %v: %w", staleIDs, err)
}
}
name := fmt.Sprintf("proxy-%s", xid.New().String())
peer := &peer.Peer{
newPeer := &peer.Peer{
Ephemeral: true,
ProxyMeta: peer.ProxyMeta{
Cluster: cluster,
@@ -242,10 +266,36 @@ func (m *managerImpl) CreateProxyPeer(ctx context.Context, accountID string, pee
},
}
_, _, _, _, err = m.accountManager.AddPeer(ctx, accountID, "", "", peer, true)
_, _, _, _, err = m.accountManager.AddPeer(ctx, accountID, "", "", newPeer, true)
if err != nil {
return fmt.Errorf("failed to create proxy peer: %w", err)
}
return nil
}
// findStaleEmbeddedProxyPeers returns the peer IDs of embedded proxy peer
// records in accountID that target the same cluster but carry a different
// WireGuard pubkey than the freshly-registering one. Used by CreateProxyPeer
// to garbage-collect stale records left behind when the proxy restarts with a
// regenerated keypair.
func (m *managerImpl) findStaleEmbeddedProxyPeers(ctx context.Context, accountID, cluster, newKey string) ([]string, error) {
account, err := m.store.GetAccount(ctx, accountID)
if err != nil {
return nil, err
}
var stale []string
for _, p := range account.Peers {
if p == nil || !p.ProxyMeta.Embedded {
continue
}
if p.ProxyMeta.Cluster != cluster {
continue
}
if p.Key == newKey {
continue
}
stale = append(stale, p.ID)
}
return stale, nil
}
@@ -39,6 +39,10 @@ type AccessLogEntry struct {
BytesDownload int64 `gorm:"index"`
Protocol AccessLogProtocol `gorm:"index"`
Metadata map[string]string `gorm:"serializer:json"`
// AgentNetwork marks the entry as emitted by a synthesised agent-network
// service. Sourced from proto.AccessLog.AgentNetwork the proxy stamps
// before shipping. Indexed so the agent-network log surface filters cheaply.
AgentNetwork bool `gorm:"index"`
}
// FromProto creates an AccessLogEntry from a proto.AccessLog
@@ -58,6 +62,7 @@ func (a *AccessLogEntry) FromProto(serviceLog *proto.AccessLog) {
a.BytesDownload = serviceLog.GetBytesDownload()
a.Protocol = AccessLogProtocol(serviceLog.GetProtocol())
a.Metadata = maps.Clone(serviceLog.GetMetadata())
a.AgentNetwork = serviceLog.GetAgentNetwork()
if sourceIP := serviceLog.GetSourceIp(); sourceIP != "" {
if addr, err := netip.ParseAddr(sourceIP); err == nil {
@@ -2,12 +2,15 @@ package manager
import (
"context"
"math"
"strconv"
"strings"
"time"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs"
agentNetworkTypes "github.com/netbirdio/netbird/management/server/agentnetwork/types"
"github.com/netbirdio/netbird/management/server/geolocation"
"github.com/netbirdio/netbird/management/server/permissions"
"github.com/netbirdio/netbird/management/server/permissions/modules"
@@ -16,6 +19,28 @@ import (
"github.com/netbirdio/netbird/shared/management/status"
)
// Metadata keys the proxy stamps on agent-network access-log entries. These
// mirror the constants in proxy/internal/middleware/keys.go and form the wire
// contract between the proxy and management; management flattens them into
// queryable columns. Keep in sync with the proxy side.
const (
metaKeyProvider = "llm.provider"
metaKeyModel = "llm.model"
metaKeyResolvedProviderID = "llm.resolved_provider_id"
metaKeySelectedPolicyID = "llm.selected_policy_id"
metaKeyPolicyDecision = "llm_policy.decision"
metaKeyPolicyReason = "llm_policy.reason"
metaKeyInputTokens = "llm.input_tokens" //nolint:gosec // metadata key name, not a credential
metaKeyOutputTokens = "llm.output_tokens" //nolint:gosec // metadata key name, not a credential
metaKeyTotalTokens = "llm.total_tokens" //nolint:gosec // metadata key name, not a credential
metaKeyCostUSDTotal = "cost.usd_total"
metaKeyStream = "llm.stream"
metaKeySessionID = "llm.session_id"
metaKeyAuthorisingGroups = "llm.authorising_groups"
metaKeyRequestPrompt = "llm.request_prompt"
metaKeyResponseCompletion = "llm.response_completion"
)
type managerImpl struct {
store store.Store
permissionsManager permissions.Manager
@@ -31,8 +56,14 @@ func NewManager(store store.Store, permissionsManager permissions.Manager, geo g
}
}
// SaveAccessLog saves an access log entry to the database after enriching it
// SaveAccessLog saves an access log entry to the database after enriching it.
// Agent-network entries are flattened into their own dedicated table (queryable
// LLM columns + group child rows) instead of the shared reverse-proxy table.
func (m *managerImpl) SaveAccessLog(ctx context.Context, logEntry *accesslogs.AccessLogEntry) error {
if logEntry.AgentNetwork {
return m.saveAgentNetworkAccessLog(ctx, logEntry)
}
if m.geo != nil && logEntry.GeoLocation.ConnectionIP != nil {
location, err := m.geo.Lookup(logEntry.GeoLocation.ConnectionIP)
if err != nil {
@@ -61,6 +92,184 @@ func (m *managerImpl) SaveAccessLog(ctx context.Context, logEntry *accesslogs.Ac
return nil
}
// saveAgentNetworkAccessLog flattens the metadata-bearing access-log entry and
// persists it in two parts:
//
// - The stripped usage record is written unconditionally — usage/cost is
// collected on every request regardless of the account's log-collection
// toggle (the proxy ships a usage-only entry when logging is disabled).
// - The full access-log row (with request detail + prompt) is written only
// when the account's EnableLogCollection setting is on. This setting read
// is the authoritative gate; the proxy-side strip is defense in depth.
func (m *managerImpl) saveAgentNetworkAccessLog(ctx context.Context, logEntry *accesslogs.AccessLogEntry) error {
entry, groups := flattenAgentNetworkLog(logEntry)
usage, usageGroups := usageFromFlattenedLog(entry, groups)
if err := m.store.CreateAgentNetworkUsage(ctx, usage, usageGroups); err != nil {
log.WithContext(ctx).WithFields(log.Fields{
"account_id": entry.AccountID,
"model": entry.Model,
}).Errorf("failed to save agent-network usage: %v", err)
return err
}
settings, err := m.store.GetAgentNetworkSettings(ctx, store.LockingStrengthNone, entry.AccountID)
if err != nil {
// No settings row (or a transient read error) means we can't confirm
// log collection is enabled — usage is already saved, so skip the full
// row rather than fail the whole ingest.
log.WithContext(ctx).Debugf("skipping full agent-network access-log row for account %s: %v", entry.AccountID, err)
return nil
}
if !settings.EnableLogCollection {
return nil
}
if err := m.store.CreateAgentNetworkAccessLog(ctx, entry, groups); err != nil {
log.WithContext(ctx).WithFields(log.Fields{
"account_id": entry.AccountID,
"service_id": entry.ServiceID,
"model": entry.Model,
"status": entry.StatusCode,
}).Errorf("failed to save agent-network access log: %v", err)
return err
}
return nil
}
// flattenAgentNetworkLog converts a reverse-proxy AccessLogEntry (whose LLM
// dimensions live in the opaque Metadata map) into the flattened
// agent-network row + authorising-group child rows.
func flattenAgentNetworkLog(e *accesslogs.AccessLogEntry) (*agentNetworkTypes.AgentNetworkAccessLog, []agentNetworkTypes.AgentNetworkAccessLogGroup) {
meta := e.Metadata
var sourceIP string
if e.GeoLocation.ConnectionIP != nil {
sourceIP = e.GeoLocation.ConnectionIP.String()
}
entry := &agentNetworkTypes.AgentNetworkAccessLog{
ID: e.ID,
AccountID: e.AccountID,
ServiceID: e.ServiceID,
Timestamp: e.Timestamp,
UserID: e.UserId,
SourceIP: sourceIP,
Method: e.Method,
Host: e.Host,
Path: e.Path,
Duration: e.Duration,
StatusCode: e.StatusCode,
AuthMethod: e.AuthMethodUsed,
BytesUpload: e.BytesUpload,
BytesDownload: e.BytesDownload,
Provider: meta[metaKeyProvider],
Model: meta[metaKeyModel],
SessionID: meta[metaKeySessionID],
ResolvedProviderID: meta[metaKeyResolvedProviderID],
SelectedPolicyID: meta[metaKeySelectedPolicyID],
Decision: meta[metaKeyPolicyDecision],
DenyReason: meta[metaKeyPolicyReason],
InputTokens: parseMetaInt(meta, metaKeyInputTokens),
OutputTokens: parseMetaInt(meta, metaKeyOutputTokens),
TotalTokens: parseMetaInt(meta, metaKeyTotalTokens),
CostUSD: parseMetaFloat(meta, metaKeyCostUSDTotal),
Stream: parseMetaBool(meta, metaKeyStream),
RequestPrompt: meta[metaKeyRequestPrompt],
ResponseCompletion: meta[metaKeyResponseCompletion],
}
var groups []agentNetworkTypes.AgentNetworkAccessLogGroup
for _, gid := range parseGroupCSV(meta[metaKeyAuthorisingGroups]) {
groups = append(groups, agentNetworkTypes.AgentNetworkAccessLogGroup{
LogID: entry.ID,
GroupID: gid,
AccountID: entry.AccountID,
})
}
return entry, groups
}
// usageFromFlattenedLog derives the stripped usage record (and its group child
// rows) from an already-flattened access-log entry. The usage row shares the
// log's ID so the two correlate.
func usageFromFlattenedLog(e *agentNetworkTypes.AgentNetworkAccessLog, groups []agentNetworkTypes.AgentNetworkAccessLogGroup) (*agentNetworkTypes.AgentNetworkUsage, []agentNetworkTypes.AgentNetworkUsageGroup) {
usage := &agentNetworkTypes.AgentNetworkUsage{
ID: e.ID,
AccountID: e.AccountID,
Timestamp: e.Timestamp,
UserID: e.UserID,
ResolvedProviderID: e.ResolvedProviderID,
Provider: e.Provider,
Model: e.Model,
SessionID: e.SessionID,
InputTokens: e.InputTokens,
OutputTokens: e.OutputTokens,
TotalTokens: e.TotalTokens,
CostUSD: e.CostUSD,
}
usageGroups := make([]agentNetworkTypes.AgentNetworkUsageGroup, 0, len(groups))
for _, g := range groups {
usageGroups = append(usageGroups, agentNetworkTypes.AgentNetworkUsageGroup{
UsageID: usage.ID,
GroupID: g.GroupID,
AccountID: g.AccountID,
})
}
return usage, usageGroups
}
// parseMetaInt parses a non-negative token count. Negative or unparseable
// values are clamped to 0 so a malformed metric can't persist a negative
// counter.
func parseMetaInt(meta map[string]string, key string) int64 {
if v, err := strconv.ParseInt(strings.TrimSpace(meta[key]), 10, 64); err == nil && v >= 0 {
return v
}
return 0
}
// parseMetaFloat parses a non-negative, finite cost. Negative, NaN, Inf, or
// unparseable values are clamped to 0 so a malformed metric can't poison the
// stored cost.
func parseMetaFloat(meta map[string]string, key string) float64 {
if v, err := strconv.ParseFloat(strings.TrimSpace(meta[key]), 64); err == nil && v >= 0 && !math.IsInf(v, 0) {
return v
}
return 0
}
func parseMetaBool(meta map[string]string, key string) bool {
v, _ := strconv.ParseBool(strings.TrimSpace(meta[key]))
return v
}
// parseGroupCSV splits the comma-separated authorising-group id list the proxy
// emits, trimming blanks and de-duplicating. Dedup matters because the group
// rows are keyed by (log_id, group_id) / (usage_id, group_id): a repeated id
// in the CSV would otherwise produce a duplicate primary key and fail the
// insert transaction.
func parseGroupCSV(raw string) []string {
if raw == "" {
return nil
}
parts := strings.Split(raw, ",")
out := make([]string, 0, len(parts))
seen := make(map[string]struct{}, len(parts))
for _, p := range parts {
if p = strings.TrimSpace(p); p != "" {
if _, dup := seen[p]; dup {
continue
}
seen[p] = struct{}{}
out = append(out, p)
}
}
return out
}
// GetAllAccessLogs retrieves access logs for an account with pagination and filtering
func (m *managerImpl) GetAllAccessLogs(ctx context.Context, accountID, userID string, filter *accesslogs.AccessLogFilter) ([]*accesslogs.AccessLogEntry, int64, error) {
ok, ctx, err := m.permissionsManager.ValidateUserPermissions(ctx, accountID, userID, modules.Services, operations.Read)
@@ -66,6 +66,51 @@ type TargetOptions struct {
// reachable without WireGuard (public APIs, LAN services, localhost
// sidecars). Default false.
DirectUpstream bool `json:"direct_upstream,omitempty"`
// Middlewares carries per-target agent-network middleware configs. Empty
// for private and operator-defined services; populated only by the
// agent-network synthesizer.
Middlewares []MiddlewareConfig `gorm:"serializer:json" json:"middlewares,omitempty"`
CaptureMaxRequestBytes int64 `json:"capture_max_request_bytes,omitempty"`
CaptureMaxResponseBytes int64 `json:"capture_max_response_bytes,omitempty"`
CaptureContentTypes []string `gorm:"serializer:json" json:"capture_content_types,omitempty"`
// AgentNetwork marks targets synthesised from Agent Network state. The
// proxy uses it to gate agent-network-specific behaviour (access log
// tagging, observability, etc.).
AgentNetwork bool `json:"agent_network,omitempty"`
// DisableAccessLog suppresses the per-request access-log emission for this
// target. Defaults false to preserve access-log behaviour for every
// non-agent-network target. The agent-network synthesizer sets this true
// only when the account's EnableLogCollection toggle is off.
DisableAccessLog bool `json:"disable_access_log,omitempty"`
}
// MiddlewareSlot mirrors proto.MiddlewareSlot / middleware.Slot.
type MiddlewareSlot string
const (
MiddlewareSlotOnRequest MiddlewareSlot = "on_request"
MiddlewareSlotOnResponse MiddlewareSlot = "on_response"
MiddlewareSlotTerminal MiddlewareSlot = "terminal"
)
// MiddlewareFailMode mirrors proto.MiddlewareConfig_FailMode.
type MiddlewareFailMode string
const (
MiddlewareFailOpen MiddlewareFailMode = "fail_open"
MiddlewareFailClosed MiddlewareFailMode = "fail_closed"
)
// MiddlewareConfig is the per-target configuration for a single
// middleware instance. Mirrors proto.MiddlewareConfig.
type MiddlewareConfig struct {
ID string `json:"id"`
Enabled bool `json:"enabled"`
Slot MiddlewareSlot `json:"slot"`
ConfigJSON []byte `json:"config_json,omitempty"`
FailMode MiddlewareFailMode `json:"fail_mode,omitempty"`
TimeoutMs int32 `json:"timeout_ms,omitempty"`
CanMutate bool `json:"can_mutate"`
}
type Target struct {
@@ -504,21 +549,75 @@ func targetOptionsToAPI(opts TargetOptions) *api.ServiceTargetOptions {
func targetOptionsToProto(opts TargetOptions) *proto.PathTargetOptions {
if !opts.SkipTLSVerify && opts.PathRewrite == "" && opts.RequestTimeout == 0 &&
len(opts.CustomHeaders) == 0 && !opts.DirectUpstream {
len(opts.CustomHeaders) == 0 && !opts.DirectUpstream &&
len(opts.Middlewares) == 0 && opts.CaptureMaxRequestBytes == 0 &&
opts.CaptureMaxResponseBytes == 0 && len(opts.CaptureContentTypes) == 0 &&
!opts.AgentNetwork && !opts.DisableAccessLog {
return nil
}
popts := &proto.PathTargetOptions{
SkipTlsVerify: opts.SkipTLSVerify,
PathRewrite: pathRewriteToProto(opts.PathRewrite),
CustomHeaders: opts.CustomHeaders,
DirectUpstream: opts.DirectUpstream,
SkipTlsVerify: opts.SkipTLSVerify,
PathRewrite: pathRewriteToProto(opts.PathRewrite),
CustomHeaders: opts.CustomHeaders,
DirectUpstream: opts.DirectUpstream,
AgentNetwork: opts.AgentNetwork,
DisableAccessLog: opts.DisableAccessLog,
}
if opts.RequestTimeout != 0 {
popts.RequestTimeout = durationpb.New(opts.RequestTimeout)
}
if len(opts.Middlewares) > 0 {
popts.Middlewares = middlewaresToProto(opts.Middlewares)
}
popts.CaptureMaxRequestBytes = opts.CaptureMaxRequestBytes
popts.CaptureMaxResponseBytes = opts.CaptureMaxResponseBytes
if len(opts.CaptureContentTypes) > 0 {
popts.CaptureContentTypes = append([]string(nil), opts.CaptureContentTypes...)
}
return popts
}
// middlewaresToProto converts the internal middleware slice to the proto
// representation sent to the proxy via the mapping stream.
func middlewaresToProto(in []MiddlewareConfig) []*proto.MiddlewareConfig {
out := make([]*proto.MiddlewareConfig, 0, len(in))
for _, m := range in {
pm := &proto.MiddlewareConfig{
Id: m.ID,
Enabled: m.Enabled,
Slot: middlewareSlotToProto(m.Slot),
ConfigJson: append([]byte(nil), m.ConfigJSON...),
CanMutate: m.CanMutate,
FailMode: middlewareFailModeToProto(m.FailMode),
}
if m.TimeoutMs > 0 {
pm.Timeout = durationpb.New(time.Duration(m.TimeoutMs) * time.Millisecond)
}
out = append(out, pm)
}
return out
}
func middlewareSlotToProto(s MiddlewareSlot) proto.MiddlewareSlot {
switch s {
case MiddlewareSlotOnRequest:
return proto.MiddlewareSlot_MIDDLEWARE_SLOT_ON_REQUEST
case MiddlewareSlotOnResponse:
return proto.MiddlewareSlot_MIDDLEWARE_SLOT_ON_RESPONSE
case MiddlewareSlotTerminal:
return proto.MiddlewareSlot_MIDDLEWARE_SLOT_TERMINAL
default:
return proto.MiddlewareSlot_MIDDLEWARE_SLOT_UNSPECIFIED
}
}
func middlewareFailModeToProto(m MiddlewareFailMode) proto.MiddlewareConfig_FailMode {
if m == MiddlewareFailClosed {
return proto.MiddlewareConfig_FAIL_CLOSED
}
return proto.MiddlewareConfig_FAIL_OPEN
}
// l4TargetOptionsToProto converts L4-relevant target options to proto.
func l4TargetOptionsToProto(target *Target) *proto.PathTargetOptions {
if !target.ProxyProtocol && target.Options.RequestTimeout == 0 && target.Options.SessionIdleTimeout == 0 {
+27 -1
View File
@@ -26,9 +26,11 @@ import (
"github.com/netbirdio/netbird/formatter/hook"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs"
accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager"
rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
"github.com/netbirdio/netbird/management/server/activity"
activitystore "github.com/netbirdio/netbird/management/server/activity/store"
"github.com/netbirdio/netbird/management/server/agentnetwork"
nbcache "github.com/netbirdio/netbird/management/server/cache"
nbContext "github.com/netbirdio/netbird/management/server/context"
nbhttp "github.com/netbirdio/netbird/management/server/http"
@@ -120,7 +122,7 @@ func (s *BaseServer) EventStore() activity.Store {
func (s *BaseServer) APIHandler() http.Handler {
return Create(s, func() http.Handler {
httpAPIHandler, err := nbhttp.NewAPIHandler(context.Background(), s.Router(), s.AccountManager(), s.NetworksManager(), s.ResourcesManager(), s.RoutesManager(), s.GroupsManager(), s.GeoLocationManager(), s.AuthManager(), s.Metrics(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(), s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(), s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies, s.RateLimiter(), s.IsValidChildAccount)
httpAPIHandler, err := nbhttp.NewAPIHandler(context.Background(), s.Router(), s.AccountManager(), s.NetworksManager(), s.ResourcesManager(), s.RoutesManager(), s.GroupsManager(), s.GeoLocationManager(), s.AuthManager(), s.Metrics(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(), s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(), s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies, s.RateLimiter(), s.IsValidChildAccount, s.AgentNetworkManager())
if err != nil {
log.Fatalf("failed to create API handler: %v", err)
}
@@ -223,11 +225,35 @@ func (s *BaseServer) ReverseProxyGRPCServer() *nbgrpc.ProxyServiceServer {
s.AfterInit(func(s *BaseServer) {
proxyService.SetServiceManager(s.ServiceManager())
proxyService.SetProxyController(s.ServiceProxyController())
proxyService.SetAgentNetworkSynthesizer(newAgentNetworkSynthesizer(s.Store()))
proxyService.SetAgentNetworkLimitsService(s.AgentNetworkManager())
})
return proxyService
})
}
// agentNetworkSynthesizerAdapter implements nbgrpc.AgentNetworkSynthesizer by
// delegating to the agentnetwork package's store-backed synthesiser.
type agentNetworkSynthesizerAdapter struct {
store store.Store
}
func newAgentNetworkSynthesizer(s store.Store) *agentNetworkSynthesizerAdapter {
return &agentNetworkSynthesizerAdapter{store: s}
}
func (a *agentNetworkSynthesizerAdapter) SynthesizeServicesForCluster(ctx context.Context, clusterAddr string) ([]*rpservice.Service, error) {
return agentnetwork.SynthesizeServicesForCluster(ctx, a.store, clusterAddr)
}
func (a *agentNetworkSynthesizerAdapter) SynthesizeServicesForAccount(ctx context.Context, accountID string) ([]*rpservice.Service, error) {
return agentnetwork.SynthesizeServices(ctx, a.store, accountID)
}
func (a *agentNetworkSynthesizerAdapter) SynthesizeServiceForDomain(ctx context.Context, domain string) (*rpservice.Service, error) {
return agentnetwork.SynthesizeServiceForDomain(ctx, a.store, domain)
}
func (s *BaseServer) proxyOIDCConfig() nbgrpc.ProxyOIDCConfig {
return Create(s, func() nbgrpc.ProxyOIDCConfig {
return nbgrpc.ProxyOIDCConfig{
+19
View File
@@ -20,6 +20,7 @@ import (
recordsManager "github.com/netbirdio/netbird/management/internals/modules/zones/records/manager"
"github.com/netbirdio/netbird/management/server"
"github.com/netbirdio/netbird/management/server/account"
"github.com/netbirdio/netbird/management/server/agentnetwork"
"github.com/netbirdio/netbird/management/server/geolocation"
"github.com/netbirdio/netbird/management/server/groups"
"github.com/netbirdio/netbird/management/server/idp"
@@ -194,6 +195,24 @@ func (s *BaseServer) NetworksManager() networks.Manager {
})
}
func (s *BaseServer) AgentNetworkManager() agentnetwork.Manager {
return Create(s, func() agentnetwork.Manager {
mgr := agentnetwork.NewManager(
s.Store(),
s.PermissionsManager(),
s.AccountManager(),
s.ServiceProxyController(),
)
// Sweep expired agent-network access logs per account retention,
// reusing the reverse-proxy cleanup interval config.
mgr.StartAccessLogCleanup(
context.Background(),
s.Config.ReverseProxy.AccessLogCleanupIntervalHours,
)
return mgr
})
}
func (s *BaseServer) ZonesManager() zones.Manager {
return Create(s, func() zones.Manager {
return zonesManager.NewManager(s.Store(), s.AccountManager(), s.PermissionsManager(), s.DNSDomain())
+199 -1
View File
@@ -10,6 +10,7 @@ import (
"errors"
"fmt"
"io"
"math"
"net"
"net/http"
"net/url"
@@ -35,6 +36,7 @@ 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/server/agentnetwork"
"github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/management/server/users"
proxyauth "github.com/netbirdio/netbird/proxy/auth"
@@ -60,6 +62,23 @@ type ProxyTokenChecker interface {
}
// ProxyServiceServer implements the ProxyService gRPC server
// AgentNetworkSynthesizer produces in-memory reverse-proxy services from
// Agent Network provider/policy state for the proxy snapshot path; synthesised
// services never appear in the reverseproxy_services table.
type AgentNetworkSynthesizer interface {
SynthesizeServicesForCluster(ctx context.Context, clusterAddr string) ([]*rpservice.Service, error)
SynthesizeServicesForAccount(ctx context.Context, accountID string) ([]*rpservice.Service, error)
SynthesizeServiceForDomain(ctx context.Context, domain string) (*rpservice.Service, error)
}
// AgentNetworkLimitsService is the minimal slice of agentnetwork.Manager the
// gRPC layer needs for CheckLLMPolicyLimits + RecordLLMUsage — kept narrow so
// the grpc package doesn't take a hard import on the full manager.
type AgentNetworkLimitsService interface {
SelectPolicyForRequest(ctx context.Context, in agentnetwork.PolicySelectionInput) (*agentnetwork.PolicySelectionResult, error)
RecordUsage(ctx context.Context, in agentnetwork.RecordUsageInput) error
}
type ProxyServiceServer struct {
proto.UnimplementedProxyServiceServer
@@ -72,6 +91,14 @@ type ProxyServiceServer struct {
mu sync.RWMutex
// Manager for reverse proxy operations
serviceManager rpservice.Manager
// agentNetworkSynth produces synthesised reverse-proxy services from
// Agent Network state. Optional — when nil the snapshot path only ships
// persisted services.
agentNetworkSynth AgentNetworkSynthesizer
// agentNetworkLimits handles the pre-flight selection (CheckLLMPolicyLimits)
// and the post-flight consumption write (RecordLLMUsage). Optional — when
// nil both RPCs return Unimplemented.
agentNetworkLimits AgentNetworkLimitsService
// ProxyController for service updates and cluster management
proxyController proxy.Controller
@@ -209,6 +236,127 @@ func (s *ProxyServiceServer) SetServiceManager(manager rpservice.Manager) {
s.serviceManager = manager
}
// SetAgentNetworkSynthesizer wires the agent-network service synthesiser.
// Optional — when nil the snapshot path skips agent-network synthesis. The
// modules layer injects this after both the proxy server and the agent-network
// manager are constructed.
func (s *ProxyServiceServer) SetAgentNetworkSynthesizer(synth AgentNetworkSynthesizer) {
s.mu.Lock()
s.agentNetworkSynth = synth
s.mu.Unlock()
}
// SetAgentNetworkLimitsService wires the policy-selection + post-flight
// consumption sink. Pass nil to disable; both RPCs return Unimplemented while
// unset so partial wiring surfaces during integration.
func (s *ProxyServiceServer) SetAgentNetworkLimitsService(svc AgentNetworkLimitsService) {
s.mu.Lock()
s.agentNetworkLimits = svc
s.mu.Unlock()
}
// agentNetworkSynthesizer returns the synthesiser under read lock.
func (s *ProxyServiceServer) agentNetworkSynthesizer() AgentNetworkSynthesizer {
s.mu.RLock()
defer s.mu.RUnlock()
return s.agentNetworkSynth
}
// CheckLLMPolicyLimits is the pre-flight policy gate the proxy calls before
// forwarding an LLM request upstream. Delegates to the agent-network selector,
// which scores applicable policies by remaining headroom and returns the
// policy that pays for this request (or a deny when all are exhausted).
func (s *ProxyServiceServer) CheckLLMPolicyLimits(ctx context.Context, req *proto.CheckLLMPolicyLimitsRequest) (*proto.CheckLLMPolicyLimitsResponse, error) {
s.mu.RLock()
svc := s.agentNetworkLimits
s.mu.RUnlock()
if svc == nil {
return nil, status.Errorf(codes.Unimplemented, "agent-network limits service not configured on management")
}
if req.GetAccountId() == "" {
return nil, status.Errorf(codes.InvalidArgument, "account_id is required")
}
if err := enforceAccountScope(ctx, req.GetAccountId()); err != nil {
return nil, err
}
res, err := svc.SelectPolicyForRequest(ctx, agentnetwork.PolicySelectionInput{
AccountID: req.GetAccountId(),
UserID: req.GetUserId(),
GroupIDs: req.GetGroupIds(),
ProviderID: req.GetProviderId(),
})
if err != nil {
log.WithContext(ctx).Errorf("select policy for request: %v", err)
return nil, status.Error(codes.Internal, "select policy failed")
}
if !res.Allow {
return &proto.CheckLLMPolicyLimitsResponse{
Decision: "deny",
SelectedPolicyId: res.SelectedPolicyID,
AttributionGroupId: res.AttributionGroupID,
WindowSeconds: res.WindowSeconds,
DenyCode: res.DenyCode,
DenyReason: res.DenyReason,
}, nil
}
return &proto.CheckLLMPolicyLimitsResponse{
Decision: "allow",
SelectedPolicyId: res.SelectedPolicyID,
AttributionGroupId: res.AttributionGroupID,
WindowSeconds: res.WindowSeconds,
}, nil
}
// RecordLLMUsage increments the per-(dimension, window) consumption counter for
// the user and optional attribution group after a served request. Returns
// Unimplemented when the agent-network limits service hasn't been wired.
func (s *ProxyServiceServer) RecordLLMUsage(ctx context.Context, req *proto.RecordLLMUsageRequest) (*proto.RecordLLMUsageResponse, error) {
s.mu.RLock()
svc := s.agentNetworkLimits
s.mu.RUnlock()
if svc == nil {
return nil, status.Errorf(codes.Unimplemented, "agent-network limits service not configured on management")
}
accountID := req.GetAccountId()
if accountID == "" {
return nil, status.Errorf(codes.InvalidArgument, "account_id is required")
}
if err := enforceAccountScope(ctx, accountID); err != nil {
return nil, err
}
tokensIn := req.GetTokensInput()
tokensOut := req.GetTokensOutput()
costUSD := req.GetCostUsd()
// Reject impossible counters at the boundary instead of recording them:
// a negative window, negative tokens, or a negative / non-finite cost
// would otherwise decrement or poison the persisted consumption totals.
if req.GetWindowSeconds() < 0 || tokensIn < 0 || tokensOut < 0 || costUSD < 0 || math.IsNaN(costUSD) || math.IsInf(costUSD, 0) {
return nil, status.Errorf(codes.InvalidArgument, "usage counters must be non-negative and finite")
}
// Book the policy-window dimensions (when a policy cap bound this request)
// and every applicable account budget rule's window in a single batched
// transaction.
if err := svc.RecordUsage(ctx, agentnetwork.RecordUsageInput{
AccountID: accountID,
UserID: req.GetUserId(),
AttributionGroupID: req.GetGroupId(),
GroupIDs: req.GetGroupIds(),
WindowSeconds: req.GetWindowSeconds(),
TokensIn: tokensIn,
TokensOut: tokensOut,
CostUSD: costUSD,
}); err != nil {
log.WithContext(ctx).Errorf("record usage: %v", err)
return nil, status.Error(codes.Internal, "record usage failed")
}
return &proto.RecordLLMUsageResponse{}, nil
}
// SetProxyController sets the proxy controller. Must be called before serving.
func (s *ProxyServiceServer) SetProxyController(proxyController proxy.Controller) {
s.mu.Lock()
@@ -623,12 +771,40 @@ func (s *ProxyServiceServer) snapshotServiceMappings(ctx context.Context, conn *
return nil, fmt.Errorf("get services from store: %w", err)
}
if synth := s.agentNetworkSynthesizer(); synth != nil {
var synthesised []*rpservice.Service
var serr error
// Account-scoped connections synthesise only their own account, so the
// snapshot can never carry another tenant's mappings (which embed the
// upstream auth header derived from that tenant's provider API key).
// Global connections still see the whole cluster.
if conn.accountID != nil {
synthesised, serr = synth.SynthesizeServicesForAccount(ctx, *conn.accountID)
} else {
synthesised, serr = synth.SynthesizeServicesForCluster(ctx, conn.address)
}
if serr != nil {
// Surface a real synthesis failure instead of silently shipping an
// incomplete snapshot (which would drop the account's agent-network
// routes). Consistent with the persisted-services error above; the
// proxy retries the snapshot on connection error.
return nil, fmt.Errorf("synthesise agent-network services: %w", serr)
}
services = append(services, synthesised...)
}
oidcCfg := s.GetOIDCValidationConfig()
var mappings []*proto.ProxyMapping
for _, service := range services {
if !service.Enabled || service.ProxyCluster == "" || service.ProxyCluster != conn.address {
continue
}
// Defense in depth: an account-scoped proxy must never receive another
// account's mapping, matching the per-account filtering the incremental
// update path already applies.
if conn.accountID != nil && service.AccountID != *conn.accountID {
continue
}
m := service.ToProtoMapping(rpservice.Create, "", oidcCfg)
if !proxyAcceptsMapping(conn, m) {
@@ -1617,7 +1793,29 @@ func (s *ProxyServiceServer) ValidateSession(ctx context.Context, req *proto.Val
}
func (s *ProxyServiceServer) getServiceByDomain(ctx context.Context, domain string) (*rpservice.Service, error) {
return s.serviceManager.GetServiceByDomain(ctx, domain)
service, err := s.serviceManager.GetServiceByDomain(ctx, domain)
if err == nil {
return service, nil
}
// Fall back to the Agent Network synthesiser scoped directly to the domain's
// account. Synthesised services are never persisted, so they must resolve
// here for OIDC / session / tunnel-peer flows against agent-network
// endpoints. Resolving by domain synthesises only the owning account rather
// than every tenant on the cluster.
if synth := s.agentNetworkSynthesizer(); synth != nil {
svc, serr := synth.SynthesizeServiceForDomain(ctx, domain)
if serr != nil {
// A real synthesis failure must surface, not be masked by the
// original store miss — otherwise a transient DB error looks like
// "no such service".
return nil, fmt.Errorf("synthesize agent-network service for %s: %w", domain, serr)
}
if svc != nil {
return svc, nil
}
}
return nil, err
}
func (s *ProxyServiceServer) checkGroupAccess(service *rpservice.Service, user *types.User) error {