From 50ba80e981193ca0791353253cad6850e9d616c8 Mon Sep 17 00:00:00 2001 From: Pascal Fischer Date: Wed, 16 Apr 2025 09:27:54 +0200 Subject: [PATCH] extend with additional packages --- go.mod | 128 ++- .../controllers/ephemeral_peers/config.go | 1 + .../controllers/network_map/controller.go | 4 + .../controllers/network_map/updatechannel.go | 178 ++++ .../network_map/updatechannel_test.go | 79 ++ .../controllers/peer_expiration/config.go | 1 + internal/modules/accounts/api.go | 112 +++ internal/modules/accounts/manager.go | 10 + internal/modules/accounts/repository.go | 19 + internal/modules/accounts/settings/api.go | 82 ++ internal/modules/accounts/settings/manager.go | 80 ++ .../modules/accounts/settings/repository.go | 29 + .../accounts/settings/types/settings.go | 89 ++ internal/modules/users/api.go | 8 +- internal/modules/users/config.go | 1 - internal/server/server.go | 8 +- internal/shared/api/grpc/grpcserver.go | 902 ++++++++++++++++++ .../{ => rest}/middleware/auth_middleware.go | 0 .../middleware/auth_middleware_test.go | 0 .../{ => rest}/middleware/bypass/bypass.go | 0 .../middleware/bypass/bypass_test.go | 0 .../middleware/metrics_middleware.go | 0 internal/shared/api/{ => rest}/router.go | 2 +- internal/shared/auth/jwt/extractor.go | 144 +++ internal/shared/auth/jwt/validator.go | 302 ++++++ internal/shared/auth/manager.go | 176 ++++ internal/shared/auth/manager_mock.go | 54 ++ internal/shared/auth/manager_test.go | 407 ++++++++ internal/shared/auth/test_data/jwks.json | 11 + internal/shared/auth/test_data/sample_key | 27 + internal/shared/auth/test_data/sample_key.pub | 9 + internal/shared/db/config.go | 5 +- internal/shared/db/database_connection.go | 24 +- 33 files changed, 2865 insertions(+), 27 deletions(-) create mode 100644 internal/controllers/ephemeral_peers/config.go create mode 100644 internal/controllers/network_map/controller.go create mode 100644 internal/controllers/network_map/updatechannel.go create mode 100644 internal/controllers/network_map/updatechannel_test.go create mode 100644 internal/controllers/peer_expiration/config.go create mode 100644 internal/modules/accounts/api.go create mode 100644 internal/modules/accounts/manager.go create mode 100644 internal/modules/accounts/repository.go create mode 100644 internal/modules/accounts/settings/api.go create mode 100644 internal/modules/accounts/settings/manager.go create mode 100644 internal/modules/accounts/settings/repository.go create mode 100644 internal/modules/accounts/settings/types/settings.go delete mode 100644 internal/modules/users/config.go create mode 100644 internal/shared/api/grpc/grpcserver.go rename internal/shared/api/{ => rest}/middleware/auth_middleware.go (100%) rename internal/shared/api/{ => rest}/middleware/auth_middleware_test.go (100%) rename internal/shared/api/{ => rest}/middleware/bypass/bypass.go (100%) rename internal/shared/api/{ => rest}/middleware/bypass/bypass_test.go (100%) rename internal/shared/api/{ => rest}/middleware/metrics_middleware.go (100%) rename internal/shared/api/{ => rest}/router.go (98%) create mode 100644 internal/shared/auth/jwt/extractor.go create mode 100644 internal/shared/auth/jwt/validator.go create mode 100644 internal/shared/auth/manager.go create mode 100644 internal/shared/auth/manager_mock.go create mode 100644 internal/shared/auth/manager_test.go create mode 100644 internal/shared/auth/test_data/jwks.json create mode 100644 internal/shared/auth/test_data/sample_key create mode 100644 internal/shared/auth/test_data/sample_key.pub diff --git a/go.mod b/go.mod index ed393da..8aad61e 100644 --- a/go.mod +++ b/go.mod @@ -6,22 +6,88 @@ toolchain go1.23.8 require ( github.com/caarlos0/env/v11 v11.3.1 + github.com/golang-jwt/jwt v3.2.2+incompatible + github.com/golang/mock v1.6.0 + github.com/golang/protobuf v1.5.4 + github.com/google/uuid v1.6.0 github.com/gorilla/mux v1.8.1 + github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.2-0.20240212192251-757544f21357 + github.com/hashicorp/go-secure-stdlib/base62 v0.1.2 + github.com/mattn/go-sqlite3 v1.14.22 + github.com/netbirdio/management-integrations/integrations v0.0.0-20250330143713-7901e0a82203 github.com/netbirdio/netbird v0.41.0 github.com/petermattis/goid v0.0.0-20250319124200-ccd6737f222a + github.com/rs/xid v1.3.0 github.com/sirupsen/logrus v1.9.3 github.com/spf13/cobra v1.9.1 github.com/spf13/viper v1.20.1 github.com/stretchr/testify v1.10.0 + golang.zx2c4.com/wireguard/wgctrl v0.0.0-20230429144221-925a1e7659e6 + google.golang.org/grpc v1.67.3 gorm.io/driver/postgres v1.5.11 gorm.io/driver/sqlite v1.5.7 gorm.io/gorm v1.25.12 ) require ( + cloud.google.com/go/auth v0.13.0 // indirect + cloud.google.com/go/auth/oauth2adapt v0.2.6 // indirect + cloud.google.com/go/compute/metadata v0.6.0 // indirect + dario.cat/mergo v1.0.0 // indirect + filippo.io/edwards25519 v1.1.0 // indirect + github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect + github.com/BurntSushi/toml v1.4.0 // indirect + github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/Microsoft/hcsshim v0.12.3 // indirect + github.com/TheJumpCloud/jcapi-go v3.0.0+incompatible // indirect + github.com/aws/aws-sdk-go-v2 v1.30.3 // indirect + github.com/aws/aws-sdk-go-v2/config v1.27.27 // indirect + github.com/aws/aws-sdk-go-v2/credentials v1.17.27 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.11 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.15 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.15 // indirect + github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.3 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.17 // indirect + github.com/aws/aws-sdk-go-v2/service/route53 v1.42.3 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.22.4 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.4 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.30.3 // indirect + github.com/aws/smithy-go v1.20.3 // indirect + github.com/beorn7/perks v1.0.1 // indirect + github.com/c-robinson/iplib v1.0.3 // indirect + github.com/caddyserver/certmagic v0.21.3 // indirect + github.com/caddyserver/zerossl v0.1.3 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/containerd/containerd v1.7.26 // indirect + github.com/containerd/log v0.1.0 // indirect + github.com/containerd/platforms v0.2.1 // indirect + github.com/cpuguy83/dockercfg v0.3.2 // indirect github.com/davecgh/go-spew v1.1.1 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/distribution/reference v0.6.0 // indirect + github.com/docker/docker v26.1.5+incompatible // indirect + github.com/docker/go-connections v0.5.0 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/eko/gocache/lib/v4 v4.2.0 // indirect + github.com/eko/gocache/store/go_cache/v4 v4.2.2 // indirect + github.com/eko/gocache/store/redis/v4 v4.2.2 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.8.0 // indirect + github.com/go-logr/logr v1.4.2 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-ole/go-ole v1.3.0 // indirect + github.com/go-sql-driver/mysql v1.8.1 // indirect github.com/go-viper/mapstructure/v2 v2.2.1 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/google/s2a-go v0.1.8 // indirect + github.com/googleapis/enterprise-certificate-proxy v0.3.4 // indirect + github.com/googleapis/gax-go/v2 v2.14.1 // indirect + github.com/hashicorp/errwrap v1.1.0 // indirect + github.com/hashicorp/go-multierror v1.1.1 // indirect + github.com/hashicorp/go-uuid v1.0.3 // indirect + github.com/hashicorp/go-version v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect @@ -29,19 +95,79 @@ require ( github.com/jackc/puddle/v2 v2.2.1 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect - github.com/mattn/go-sqlite3 v1.14.22 // indirect + github.com/jmespath/go-jmespath v0.4.0 // indirect + github.com/kelseyhightower/envconfig v1.4.0 // indirect + github.com/klauspost/compress v1.17.8 // indirect + github.com/klauspost/cpuid/v2 v2.2.7 // indirect + github.com/libdns/libdns v0.2.2 // indirect + github.com/libdns/route53 v1.5.0 // indirect + github.com/lufia/plan9stats v0.0.0-20240513124658-fba389f38bae // indirect + github.com/magiconair/properties v1.8.7 // indirect + github.com/mholt/acmez/v2 v2.0.1 // indirect + github.com/miekg/dns v1.1.59 // indirect + github.com/moby/docker-image-spec v1.3.1 // indirect + github.com/moby/patternmatcher v0.6.0 // indirect + github.com/moby/sys/sequential v0.5.0 // indirect + github.com/moby/sys/user v0.3.0 // indirect + github.com/moby/sys/userns v0.1.0 // indirect + github.com/moby/term v0.5.0 // indirect + github.com/morikuni/aec v1.0.0 // indirect + github.com/okta/okta-sdk-golang/v2 v2.18.0 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.0 // indirect + github.com/patrickmn/go-cache v2.1.0+incompatible // indirect github.com/pelletier/go-toml/v2 v2.2.3 // indirect + github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect + github.com/prometheus/client_golang v1.19.1 // indirect + github.com/prometheus/client_model v0.6.1 // indirect + github.com/prometheus/common v0.53.0 // indirect + github.com/prometheus/procfs v0.15.0 // indirect + github.com/redis/go-redis/v9 v9.7.1 // indirect github.com/sagikazarmark/locafero v0.7.0 // indirect + github.com/shirou/gopsutil/v3 v3.24.4 // indirect + github.com/shoenig/go-m1cpu v0.1.6 // indirect github.com/sourcegraph/conc v0.3.0 // indirect github.com/spf13/afero v1.12.0 // indirect github.com/spf13/cast v1.7.1 // indirect github.com/spf13/pflag v1.0.6 // indirect github.com/subosito/gotenv v1.6.0 // indirect + github.com/testcontainers/testcontainers-go v0.31.0 // indirect + github.com/testcontainers/testcontainers-go/modules/mysql v0.31.0 // indirect + github.com/testcontainers/testcontainers-go/modules/postgres v0.31.0 // indirect + github.com/tklauser/go-sysconf v0.3.14 // indirect + github.com/tklauser/numcpus v0.8.0 // indirect + github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect + github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect + github.com/yusufpapurcu/wmi v1.2.4 // indirect + github.com/zeebo/blake3 v0.2.3 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.54.0 // indirect + go.opentelemetry.io/otel v1.29.0 // indirect + go.opentelemetry.io/otel/exporters/prometheus v0.48.0 // indirect + go.opentelemetry.io/otel/metric v1.29.0 // indirect + go.opentelemetry.io/otel/sdk v1.29.0 // indirect + go.opentelemetry.io/otel/sdk/metric v1.29.0 // indirect + go.opentelemetry.io/otel/trace v1.29.0 // indirect + go.uber.org/mock v0.4.0 // indirect go.uber.org/multierr v1.11.0 // indirect + go.uber.org/zap v1.27.0 // indirect + goauthentik.io/api/v3 v3.2023051.3 // indirect golang.org/x/crypto v0.36.0 // indirect + golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect + golang.org/x/mod v0.17.0 // indirect + golang.org/x/net v0.36.0 // indirect + golang.org/x/oauth2 v0.25.0 // indirect golang.org/x/sync v0.12.0 // indirect golang.org/x/sys v0.31.0 // indirect golang.org/x/text v0.23.0 // indirect + golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect + golang.zx2c4.com/wireguard/windows v0.5.3 // indirect + google.golang.org/api v0.215.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20241223144023-3abc09e42ca8 // indirect + google.golang.org/protobuf v1.36.1 // indirect + gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect + gopkg.in/square/go-jose.v2 v2.6.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect + gorm.io/driver/mysql v1.5.7 // indirect ) diff --git a/internal/controllers/ephemeral_peers/config.go b/internal/controllers/ephemeral_peers/config.go new file mode 100644 index 0000000..abb4e43 --- /dev/null +++ b/internal/controllers/ephemeral_peers/config.go @@ -0,0 +1 @@ +package server diff --git a/internal/controllers/network_map/controller.go b/internal/controllers/network_map/controller.go new file mode 100644 index 0000000..327053e --- /dev/null +++ b/internal/controllers/network_map/controller.go @@ -0,0 +1,4 @@ +package network_map + +type Controller struct { +} diff --git a/internal/controllers/network_map/updatechannel.go b/internal/controllers/network_map/updatechannel.go new file mode 100644 index 0000000..1b40b3e --- /dev/null +++ b/internal/controllers/network_map/updatechannel.go @@ -0,0 +1,178 @@ +package server + +import ( + "context" + "sync" + "time" + + log "github.com/sirupsen/logrus" + + "github.com/netbirdio/netbird/management/proto" + "github.com/netbirdio/netbird/management/server/telemetry" + "github.com/netbirdio/netbird/management/server/types" +) + +const channelBufferSize = 100 + +type UpdateMessage struct { + Update *proto.SyncResponse + NetworkMap *types.NetworkMap +} + +type UpdateChannel struct { + // peerChannels is an update channel indexed by Peer.ID + peerChannels map[string]chan *UpdateMessage + // channelsMux keeps the mutex to access peerChannels + channelsMux *sync.RWMutex + // metrics provides method to collect application metrics + metrics telemetry.AppMetrics +} + +// NewUpdateChannel returns a new instance of UpdateChannel +func NewUpdateChannel(metrics telemetry.AppMetrics) *UpdateChannel { + return &UpdateChannel{ + peerChannels: make(map[string]chan *UpdateMessage), + channelsMux: &sync.RWMutex{}, + metrics: metrics, + } +} + +// SendUpdate sends update message to the peer's channel +func (p *UpdateChannel) SendUpdate(ctx context.Context, peerID string, update *UpdateMessage) { + start := time.Now() + var found, dropped bool + + p.channelsMux.RLock() + + defer func() { + p.channelsMux.RUnlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountSendUpdateDuration(time.Since(start), found, dropped) + } + }() + + if channel, ok := p.peerChannels[peerID]; ok { + found = true + select { + case channel <- update: + log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID) + default: + dropped = true + log.WithContext(ctx).Warnf("channel for peer %s is %d full or closed", peerID, len(channel)) + } + } else { + log.WithContext(ctx).Debugf("peer %s has no channel", peerID) + } +} + +// CreateChannel creates a go channel for a given peer used to deliver updates relevant to the peer. +func (p *UpdateChannel) CreateChannel(ctx context.Context, peerID string) chan *UpdateMessage { + start := time.Now() + + closed := false + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountCreateChannelDuration(time.Since(start), closed) + } + }() + + if channel, ok := p.peerChannels[peerID]; ok { + closed = true + delete(p.peerChannels, peerID) + close(channel) + } + // mbragin: todo shouldn't it be more? or configurable? + channel := make(chan *UpdateMessage, channelBufferSize) + p.peerChannels[peerID] = channel + + log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID) + + return channel +} + +func (p *UpdateChannel) closeChannel(ctx context.Context, peerID string) { + if channel, ok := p.peerChannels[peerID]; ok { + delete(p.peerChannels, peerID) + close(channel) + + log.WithContext(ctx).Debugf("closed updates channel of a peer %s", peerID) + return + } + + log.WithContext(ctx).Debugf("closing updates channel: peer %s has no channel", peerID) +} + +// CloseChannels closes updates channel for each given peer +func (p *UpdateChannel) CloseChannels(ctx context.Context, peerIDs []string) { + start := time.Now() + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountCloseChannelsDuration(time.Since(start), len(peerIDs)) + } + }() + + for _, id := range peerIDs { + p.closeChannel(ctx, id) + } +} + +// CloseChannel closes updates channel of a given peer +func (p *UpdateChannel) CloseChannel(ctx context.Context, peerID string) { + start := time.Now() + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountCloseChannelDuration(time.Since(start)) + } + }() + + p.closeChannel(ctx, peerID) +} + +// GetAllConnectedPeers returns a copy of the connected peers map +func (p *UpdateChannel) GetAllConnectedPeers() map[string]struct{} { + start := time.Now() + + p.channelsMux.RLock() + + m := make(map[string]struct{}) + + defer func() { + p.channelsMux.RUnlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountGetAllConnectedPeersDuration(time.Since(start), len(m)) + } + }() + + for ID := range p.peerChannels { + m[ID] = struct{}{} + } + + return m +} + +// HasChannel returns true if peers has channel in update manager, otherwise false +func (p *UpdateChannel) HasChannel(peerID string) bool { + start := time.Now() + + p.channelsMux.RLock() + + defer func() { + p.channelsMux.RUnlock() + if p.metrics != nil { + p.metrics.UpdateChannelMetrics().CountHasChannelDuration(time.Since(start)) + } + }() + + _, ok := p.peerChannels[peerID] + + return ok +} diff --git a/internal/controllers/network_map/updatechannel_test.go b/internal/controllers/network_map/updatechannel_test.go new file mode 100644 index 0000000..d067ab9 --- /dev/null +++ b/internal/controllers/network_map/updatechannel_test.go @@ -0,0 +1,79 @@ +package server + +import ( + "context" + "testing" + "time" + + "github.com/netbirdio/netbird/management/proto" +) + +// var peersUpdater *UpdateChannel + +func TestCreateChannel(t *testing.T) { + peer := "test-create" + peersUpdater := NewUpdateChannel(nil) + defer peersUpdater.CloseChannel(context.Background(), peer) + + _ = peersUpdater.CreateChannel(context.Background(), peer) + if _, ok := peersUpdater.peerChannels[peer]; !ok { + t.Error("Error creating the channel") + } +} + +func TestSendUpdate(t *testing.T) { + peer := "test-sendupdate" + peersUpdater := NewUpdateChannel(nil) + update1 := &UpdateMessage{Update: &proto.SyncResponse{ + NetworkMap: &proto.NetworkMap{ + Serial: 0, + }, + }} + _ = peersUpdater.CreateChannel(context.Background(), peer) + if _, ok := peersUpdater.peerChannels[peer]; !ok { + t.Error("Error creating the channel") + } + peersUpdater.SendUpdate(context.Background(), peer, update1) + select { + case <-peersUpdater.peerChannels[peer]: + default: + t.Error("Update wasn't send") + } + + for range [channelBufferSize]int{} { + peersUpdater.SendUpdate(context.Background(), peer, update1) + } + + update2 := &UpdateMessage{Update: &proto.SyncResponse{ + NetworkMap: &proto.NetworkMap{ + Serial: 10, + }, + }} + + peersUpdater.SendUpdate(context.Background(), peer, update2) + timeout := time.After(5 * time.Second) + for range [channelBufferSize]int{} { + select { + case <-timeout: + t.Error("timed out reading previously sent updates") + case updateReader := <-peersUpdater.peerChannels[peer]: + if updateReader.Update.NetworkMap.Serial == update2.Update.NetworkMap.Serial { + t.Error("got the update that shouldn't have been sent") + } + } + } + +} + +func TestCloseChannel(t *testing.T) { + peer := "test-close" + peersUpdater := NewUpdateChannel(nil) + _ = peersUpdater.CreateChannel(context.Background(), peer) + if _, ok := peersUpdater.peerChannels[peer]; !ok { + t.Error("Error creating the channel") + } + peersUpdater.CloseChannel(context.Background(), peer) + if _, ok := peersUpdater.peerChannels[peer]; ok { + t.Error("Error closing the channel") + } +} diff --git a/internal/controllers/peer_expiration/config.go b/internal/controllers/peer_expiration/config.go new file mode 100644 index 0000000..ca50a75 --- /dev/null +++ b/internal/controllers/peer_expiration/config.go @@ -0,0 +1 @@ +package peer_expiration diff --git a/internal/modules/accounts/api.go b/internal/modules/accounts/api.go new file mode 100644 index 0000000..710f75d --- /dev/null +++ b/internal/modules/accounts/api.go @@ -0,0 +1,112 @@ +package accounts + +import ( + "encoding/json" + "net/http" + + "github.com/gorilla/mux" + nbcontext "github.com/netbirdio/netbird/management/server/context" + "github.com/netbirdio/netbird/management/server/http/util" + + "management/internal/shared/db" + "management/internal/shared/errors" + "management/internal/shared/permissions" + "management/internal/shared/permissions/modules" + "management/internal/shared/permissions/operations" +) + +type handler struct { + manager *Manager + permissionsManager permissions.Manager +} + +func newHandler(manager *Manager, permissionsManager permissions.Manager) *handler { + return &handler{ + manager: manager, + permissionsManager: permissionsManager, + } +} + +func (h *handler) RegisterAPI(router *mux.Router) { + router.HandleFunc("/accounts/{accountId}", h.updateAccount).Methods("PUT", "OPTIONS") + router.HandleFunc("/accounts/{accountId}", h.deleteAccount).Methods("DELETE", "OPTIONS") + router.HandleFunc("/accounts", h.getAllAccounts).Methods("GET", "OPTIONS") +} + +func (h *handler) updateAccount(w http.ResponseWriter, r *http.Request) { + userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Accounts, operations.Write) + if err != nil { + util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) + return + } + if !allowed { + util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) + } + + vars := mux.Vars(r) + accountId := vars["accountId"] + + users, err := h.manager.UpdateAccount(r.Context(), nil, accountId) + if err != nil { + http.Error(w, "Internal Server Error", http.StatusInternalServerError) + return + } + _ = json.NewEncoder(w).Encode(users) +} + +func (h *handler) deleteAccount(w http.ResponseWriter, r *http.Request) { + userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Accounts, operations.Write) + if err != nil { + util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) + return + } + if !allowed { + util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) + } + + vars := mux.Vars(r) + accountId := vars["accountId"] + + user, err := h.manager.DeleteAccount(r.Context(), nil, db.LockingStrengthShare, accountId) + if err != nil { + http.Error(w, "Not Found", http.StatusNotFound) + return + } + _ = json.NewEncoder(w).Encode(user) +} + +func (h *handler) getAllAccounts(w http.ResponseWriter, r *http.Request) { + userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Accounts, operations.Read) + if err != nil { + util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) + return + } + if !allowed { + util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) + } + + accounts, err := h.manager.GetAllAccounts(r.Context(), nil, db.LockingStrengthShare) + if err != nil { + http.Error(w, "Internal Server Error", http.StatusInternalServerError) + return + } + _ = json.NewEncoder(w).Encode(accounts) +} diff --git a/internal/modules/accounts/manager.go b/internal/modules/accounts/manager.go new file mode 100644 index 0000000..12ef7cc --- /dev/null +++ b/internal/modules/accounts/manager.go @@ -0,0 +1,10 @@ +package accounts + +import "management/pkg/logging" + +var log = logging.LoggerForThisPackage() + +type Manager struct { + repo Repository + handler *handler +} diff --git a/internal/modules/accounts/repository.go b/internal/modules/accounts/repository.go new file mode 100644 index 0000000..5057b8e --- /dev/null +++ b/internal/modules/accounts/repository.go @@ -0,0 +1,19 @@ +package accounts + +import "management/internal/shared/db" + +type Repository interface { + RunInTx(fn func(tx db.Transaction) error) error +} + +type repository struct { + store *db.Store +} + +func newRepository(s *db.Store) Repository { + return &repository{store: s} +} + +func (r *repository) RunInTx(fn func(tx db.Transaction) error) error { + return r.store.RunInTx(fn) +} diff --git a/internal/modules/accounts/settings/api.go b/internal/modules/accounts/settings/api.go new file mode 100644 index 0000000..53077b5 --- /dev/null +++ b/internal/modules/accounts/settings/api.go @@ -0,0 +1,82 @@ +package settings + +import ( + "encoding/json" + "net/http" + + "github.com/gorilla/mux" + nbcontext "github.com/netbirdio/netbird/management/server/context" + "github.com/netbirdio/netbird/management/server/http/util" + + "management/internal/shared/db" + "management/internal/shared/errors" + "management/internal/shared/permissions" +) + +type handler struct { + manager *Manager + permissionsManager permissions.Manager +} + +func newHandler(manager *Manager, permissionsManager permissions.Manager) *handler { + return &handler{ + manager: manager, + permissionsManager: permissionsManager, + } +} + +func (h *handler) RegisterAPI(router *mux.Router) { + router.HandleFunc("/account/{accountID}/settings", h.GetAllUsers).Methods("GET", "OPTIONS") + router.HandleFunc("/account/{accountID}/settings", h.GetUser).Methods("PUT", "OPTIONS") +} + +func (h *handler) GetAllUsers(w http.ResponseWriter, r *http.Request) { + userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Users, operations.Read) + if err != nil { + util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) + return + } + if !allowed { + util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) + } + + users, err := h.manager.GetAllUsers(r.Context(), nil, db.LockingStrengthShare, userAuth.AccountId) + if err != nil { + http.Error(w, "Internal Server Error", http.StatusInternalServerError) + return + } + _ = json.NewEncoder(w).Encode(users) +} + +func (h *handler) GetUser(w http.ResponseWriter, r *http.Request) { + userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) + if err != nil { + util.WriteError(r.Context(), err, w) + return + } + + allowed, err := h.permissionsManager.ValidateUserPermissions(r.Context(), userAuth.AccountId, userAuth.UserId, modules.Users, operations.Read) + if err != nil { + util.WriteError(r.Context(), errors.NewPermissionValidationError(err), w) + return + } + if !allowed { + util.WriteError(r.Context(), errors.NewPermissionDeniedError(), w) + } + + vars := mux.Vars(r) + userId := vars["userId"] + + user, err := h.manager.GetUserByID(r.Context(), nil, db.LockingStrengthShare, userId) + if err != nil { + http.Error(w, "Not Found", http.StatusNotFound) + return + } + _ = json.NewEncoder(w).Encode(user) +} diff --git a/internal/modules/accounts/settings/manager.go b/internal/modules/accounts/settings/manager.go new file mode 100644 index 0000000..1e35d6f --- /dev/null +++ b/internal/modules/accounts/settings/manager.go @@ -0,0 +1,80 @@ +package settings + +import ( + "context" + "fmt" + + "github.com/netbirdio/netbird/management/server/integrations/extra_settings" + + "management/internal/modules/accounts/settings/types" + "management/internal/modules/users" + "management/internal/shared/db" + "management/pkg/logging" +) + +var log = logging.LoggerForThisPackage() + +type Manager struct { + repository Repository + extraSettingsManager extra_settings.Manager + userManager *users.Manager +} + +func NewManager(store *db.Store, userManager *users.Manager, extraSettingsManager extra_settings.Manager) *Manager { + return &Manager{ + repository: newRepository(store), + extraSettingsManager: extraSettingsManager, + userManager: userManager, + } +} + +func (m *Manager) GetExtraSettingsManager() extra_settings.Manager { + return m.extraSettingsManager +} + +func (m *Manager) GetSettings(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID string) (*types.Settings, error) { + extraSettings, err := m.extraSettingsManager.GetExtraSettings(ctx, accountID) + if err != nil { + return nil, fmt.Errorf("get extra settings: %w", err) + } + + settings, err := m.repository.GetAccountSettings(tx, strength, accountID) + if err != nil { + return nil, fmt.Errorf("get account settings: %w", err) + } + + // Once we migrate the peer approval to settings manager this merging is obsolete + if settings.Extra != nil { + settings.Extra.FlowEnabled = extraSettings.FlowEnabled + settings.Extra.FlowPacketCounterEnabled = extraSettings.FlowPacketCounterEnabled + settings.Extra.FlowENCollectionEnabled = extraSettings.FlowENCollectionEnabled + settings.Extra.FlowDnsCollectionEnabled = extraSettings.FlowDnsCollectionEnabled + } + + return settings, nil +} + +func (m *Manager) GetExtraSettings(ctx context.Context, tx db.Transaction, accountID string) (*types.ExtraSettings, error) { + extraSettings, err := m.extraSettingsManager.GetExtraSettings(ctx, accountID) + if err != nil { + return nil, fmt.Errorf("get extra settings: %w", err) + } + + settings, err := m.repository.GetAccountSettings(tx, db.LockingStrengthShare, accountID) + if err != nil { + return nil, fmt.Errorf("get account settings: %w", err) + } + + // Once we migrate the peer approval to settings manager this merging is obsolete + if settings.Extra == nil { + settings.Extra = &types.ExtraSettings{} + } + + settings.Extra.FlowEnabled = extraSettings.FlowEnabled + + return settings.Extra, nil +} + +func (m *Manager) UpdateExtraSettings(ctx context.Context, accountID, userID string, extraSettings *types.ExtraSettings) (bool, error) { + return m.extraSettingsManager.UpdateExtraSettings(ctx, accountID, userID, extraSettings) +} diff --git a/internal/modules/accounts/settings/repository.go b/internal/modules/accounts/settings/repository.go new file mode 100644 index 0000000..4323fa8 --- /dev/null +++ b/internal/modules/accounts/settings/repository.go @@ -0,0 +1,29 @@ +package settings + +import ( + "management/internal/modules/accounts/settings/types" + "management/internal/shared/db" +) + +type Repository interface { + RunInTx(fn func(tx db.Transaction) error) error + GetAccountSettings(tx db.Transaction, strength db.LockingStrength, accountID string) (*types.Settings, error) +} + +type repository struct { + store *db.Store +} + +func newRepository(s *db.Store) Repository { + return &repository{store: s} +} + +func (r *repository) RunInTx(fn func(tx db.Transaction) error) error { + return r.store.RunInTx(fn) +} + +func (r *repository) GetAccountSettings(tx db.Transaction, strength db.LockingStrength, accountID string) (*types.Settings, error) { + var settings types.Settings + err := r.store.GetOne(tx, strength, &settings, "account_id = ?", accountID) + return &settings, err +} diff --git a/internal/modules/accounts/settings/types/settings.go b/internal/modules/accounts/settings/types/settings.go new file mode 100644 index 0000000..7054ede --- /dev/null +++ b/internal/modules/accounts/settings/types/settings.go @@ -0,0 +1,89 @@ +package types + +import ( + "time" +) + +// Settings represents Account settings structure that can be modified via API and Dashboard +type Settings struct { + // PeerLoginExpirationEnabled globally enables or disables peer login expiration + PeerLoginExpirationEnabled bool + + // PeerLoginExpiration is a setting that indicates when peer login expires. + // Applies to all peers that have Peer.LoginExpirationEnabled set to true. + PeerLoginExpiration time.Duration + + // PeerInactivityExpirationEnabled globally enables or disables peer inactivity expiration + PeerInactivityExpirationEnabled bool + + // PeerInactivityExpiration is a setting that indicates when peer inactivity expires. + // Applies to all peers that have Peer.PeerInactivityExpirationEnabled set to true. + PeerInactivityExpiration time.Duration + + // RegularUsersViewBlocked allows to block regular users from viewing even their own peers and some UI elements + RegularUsersViewBlocked bool + + // GroupsPropagationEnabled allows to propagate auto groups from the user to the peer + GroupsPropagationEnabled bool + + // JWTGroupsEnabled allows extract groups from JWT claim, which name defined in the JWTGroupsClaimName + // and add it to account groups. + JWTGroupsEnabled bool + + // JWTGroupsClaimName from which we extract groups name to add it to account groups + JWTGroupsClaimName string + + // JWTAllowGroups list of groups to which users are allowed access + JWTAllowGroups []string `gorm:"serializer:json"` + + // RoutingPeerDNSResolutionEnabled enabled the DNS resolution on the routing peers + RoutingPeerDNSResolutionEnabled bool + + // Extra is a dictionary of Account settings + Extra *ExtraSettings `gorm:"embedded;embeddedPrefix:extra_"` +} + +// Copy copies the Settings struct +func (s *Settings) Copy() *Settings { + settings := &Settings{ + PeerLoginExpirationEnabled: s.PeerLoginExpirationEnabled, + PeerLoginExpiration: s.PeerLoginExpiration, + JWTGroupsEnabled: s.JWTGroupsEnabled, + JWTGroupsClaimName: s.JWTGroupsClaimName, + GroupsPropagationEnabled: s.GroupsPropagationEnabled, + JWTAllowGroups: s.JWTAllowGroups, + RegularUsersViewBlocked: s.RegularUsersViewBlocked, + + PeerInactivityExpirationEnabled: s.PeerInactivityExpirationEnabled, + PeerInactivityExpiration: s.PeerInactivityExpiration, + + RoutingPeerDNSResolutionEnabled: s.RoutingPeerDNSResolutionEnabled, + } + if s.Extra != nil { + settings.Extra = s.Extra.Copy() + } + return settings +} + +type ExtraSettings struct { + // PeerApprovalEnabled enables or disables the need for peers bo be approved by an administrator + PeerApprovalEnabled bool + + // IntegratedValidatorGroups list of group IDs to be used with integrated approval configurations + IntegratedValidatorGroups []string `gorm:"serializer:json"` + + FlowEnabled bool `gorm:"-"` + FlowPacketCounterEnabled bool `gorm:"-"` + FlowENCollectionEnabled bool `gorm:"-"` + FlowDnsCollectionEnabled bool `gorm:"-"` +} + +// Copy copies the ExtraSettings struct +func (e *ExtraSettings) Copy() *ExtraSettings { + var cpGroup []string + + return &ExtraSettings{ + PeerApprovalEnabled: e.PeerApprovalEnabled, + IntegratedValidatorGroups: append(cpGroup, e.IntegratedValidatorGroups...), + } +} diff --git a/internal/modules/users/api.go b/internal/modules/users/api.go index c834e1f..6e11f71 100644 --- a/internal/modules/users/api.go +++ b/internal/modules/users/api.go @@ -28,11 +28,11 @@ func newHandler(manager *Manager, permissionsManager permissions.Manager) *handl } func (h *handler) RegisterAPI(router *mux.Router) { - router.HandleFunc("/users", h.GetAllUsers).Methods("GET", "OPTIONS") - router.HandleFunc("/users/{userId}", h.GetUser).Methods("GET", "OPTIONS") + router.HandleFunc("/users", h.getAllUsers).Methods("GET", "OPTIONS") + router.HandleFunc("/users/{userId}", h.getUser).Methods("GET", "OPTIONS") } -func (h *handler) GetAllUsers(w http.ResponseWriter, r *http.Request) { +func (h *handler) getAllUsers(w http.ResponseWriter, r *http.Request) { userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) if err != nil { util.WriteError(r.Context(), err, w) @@ -56,7 +56,7 @@ func (h *handler) GetAllUsers(w http.ResponseWriter, r *http.Request) { _ = json.NewEncoder(w).Encode(users) } -func (h *handler) GetUser(w http.ResponseWriter, r *http.Request) { +func (h *handler) getUser(w http.ResponseWriter, r *http.Request) { userAuth, err := nbcontext.GetUserAuthFromContext(r.Context()) if err != nil { util.WriteError(r.Context(), err, w) diff --git a/internal/modules/users/config.go b/internal/modules/users/config.go deleted file mode 100644 index 82abcb9..0000000 --- a/internal/modules/users/config.go +++ /dev/null @@ -1 +0,0 @@ -package users diff --git a/internal/server/server.go b/internal/server/server.go index 7f083aa..9d5986d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -5,8 +5,12 @@ import ( "net/http" "time" + "github.com/netbirdio/management-integrations/integrations" + + "management/internal/modules/settings" "management/internal/modules/users" "management/internal/shared/api" + "management/internal/shared/api/rest" "management/internal/shared/db" "management/internal/shared/permissions" "management/pkg/logging" @@ -31,10 +35,12 @@ func NewServer() *Server { store := db.NewStore(ctx, dbConn) - router := api.NewRouter() + router := rest.NewRouter() + extraSettingsManager := integrations.NewManager() permissionsManager := permissions.NewManager(store) userManager := users.NewManager(store, permissions.NewManager(store)) + settingsManager := settings.NewManager(store, userManager, extraSettingsManager) return &Server{ httpServer: &http.Server{ diff --git a/internal/shared/api/grpc/grpcserver.go b/internal/shared/api/grpc/grpcserver.go new file mode 100644 index 0000000..4ef7a22 --- /dev/null +++ b/internal/shared/api/grpc/grpcserver.go @@ -0,0 +1,902 @@ +package server + +import ( + "context" + "fmt" + "net" + "net/netip" + "strings" + "sync" + "time" + + pb "github.com/golang/protobuf/proto" // nolint + "github.com/golang/protobuf/ptypes/timestamp" + "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip" + log "github.com/sirupsen/logrus" + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/peer" + "google.golang.org/grpc/status" + + integrationsConfig "github.com/netbirdio/management-integrations/integrations/config" + + "github.com/netbirdio/netbird/encryption" + "github.com/netbirdio/netbird/management/proto" + "github.com/netbirdio/netbird/management/server/account" + "github.com/netbirdio/netbird/management/server/activity" + "github.com/netbirdio/netbird/management/server/auth" + nbContext "github.com/netbirdio/netbird/management/server/context" + nbpeer "github.com/netbirdio/netbird/management/server/peer" + "github.com/netbirdio/netbird/management/server/posture" + "github.com/netbirdio/netbird/management/server/settings" + internalStatus "github.com/netbirdio/netbird/management/server/status" + "github.com/netbirdio/netbird/management/server/telemetry" + "github.com/netbirdio/netbird/management/server/types" +) + +// GRPCServer an instance of a Management gRPC API server +type GRPCServer struct { + // accountManager account.Manager + settingsManager settings.Manager + wgKey wgtypes.Key + proto.UnimplementedManagementServiceServer + peersUpdateManager *PeersUpdateManager + config *types.Config + secretsManager SecretsManager + appMetrics telemetry.AppMetrics + ephemeralManager *EphemeralManager + peerLocks sync.Map + authManager auth.Manager +} + +// NewServer creates a new Management server +func NewServer( + ctx context.Context, + config *types.Config, + accountManager account.Manager, + settingsManager settings.Manager, + peersUpdateManager *PeersUpdateManager, + secretsManager SecretsManager, + appMetrics telemetry.AppMetrics, + ephemeralManager *EphemeralManager, + authManager auth.Manager, +) (*GRPCServer, error) { + key, err := wgtypes.GeneratePrivateKey() + if err != nil { + return nil, err + } + + if appMetrics != nil { + // update gauge based on number of connected peers which is equal to open gRPC streams + err = appMetrics.GRPCMetrics().RegisterConnectedStreams(func() int64 { + return int64(len(peersUpdateManager.peerChannels)) + }) + if err != nil { + return nil, err + } + } + + return &GRPCServer{ + wgKey: key, + // peerKey -> event channel + peersUpdateManager: peersUpdateManager, + accountManager: accountManager, + settingsManager: settingsManager, + config: config, + secretsManager: secretsManager, + authManager: authManager, + appMetrics: appMetrics, + ephemeralManager: ephemeralManager, + }, nil +} + +func (s *GRPCServer) GetServerKey(ctx context.Context, req *proto.Empty) (*proto.ServerKeyResponse, error) { + ip := "" + p, ok := peer.FromContext(ctx) + if ok { + ip = p.Addr.String() + } + + log.WithContext(ctx).Tracef("GetServerKey request from %s", ip) + start := time.Now() + defer func() { + log.WithContext(ctx).Tracef("GetServerKey from %s took %v", ip, time.Since(start)) + }() + + // todo introduce something more meaningful with the key expiration/rotation + if s.appMetrics != nil { + s.appMetrics.GRPCMetrics().CountGetKeyRequest() + } + now := time.Now().Add(24 * time.Hour) + secs := int64(now.Second()) + nanos := int32(now.Nanosecond()) + expiresAt := ×tamp.Timestamp{Seconds: secs, Nanos: nanos} + + return &proto.ServerKeyResponse{ + Key: s.wgKey.PublicKey().String(), + ExpiresAt: expiresAt, + }, nil +} + +func getRealIP(ctx context.Context) net.IP { + if addr, ok := realip.FromContext(ctx); ok { + return net.IP(addr.AsSlice()) + } + return nil +} + +// Sync validates the existence of a connecting peer, sends an initial state (all available for the connecting peers) and +// notifies the connected peer of any updates (e.g. new peers under the same account) +func (s *GRPCServer) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_SyncServer) error { + reqStart := time.Now() + if s.appMetrics != nil { + s.appMetrics.GRPCMetrics().CountSyncRequest() + } + + ctx := srv.Context() + + syncReq := &proto.SyncRequest{} + peerKey, err := s.parseRequest(ctx, req, syncReq) + if err != nil { + return err + } + + // nolint:staticcheck + ctx = context.WithValue(ctx, nbContext.PeerIDKey, peerKey.String()) + + unlock := s.acquirePeerLockByUID(ctx, peerKey.String()) + defer func() { + if unlock != nil { + unlock() + } + }() + + accountID, err := s.accountManager.GetAccountIDForPeerKey(ctx, peerKey.String()) + if err != nil { + // nolint:staticcheck + ctx = context.WithValue(ctx, nbContext.AccountIDKey, "UNKNOWN") + log.WithContext(ctx).Tracef("peer %s is not registered", peerKey.String()) + if errStatus, ok := internalStatus.FromError(err); ok && errStatus.Type() == internalStatus.NotFound { + return status.Errorf(codes.PermissionDenied, "peer is not registered") + } + return err + } + + // nolint:staticcheck + ctx = context.WithValue(ctx, nbContext.AccountIDKey, accountID) + + realIP := getRealIP(ctx) + log.WithContext(ctx).Debugf("Sync request from peer [%s] [%s]", req.WgPubKey, realIP.String()) + + if syncReq.GetMeta() == nil { + log.WithContext(ctx).Tracef("peer system meta has to be provided on sync. Peer %s, remote addr %s", peerKey.String(), realIP) + } + + peer, netMap, postureChecks, err := s.accountManager.SyncAndMarkPeer(ctx, accountID, peerKey.String(), extractPeerMeta(ctx, syncReq.GetMeta()), realIP) + if err != nil { + log.WithContext(ctx).Debugf("error while syncing peer %s: %v", peerKey.String(), err) + return mapError(ctx, err) + } + + err = s.sendInitialSync(ctx, peerKey, peer, netMap, postureChecks, srv) + if err != nil { + log.WithContext(ctx).Debugf("error while sending initial sync for %s: %v", peerKey.String(), err) + return err + } + + updates := s.peersUpdateManager.CreateChannel(ctx, peer.ID) + + s.ephemeralManager.OnPeerConnected(ctx, peer) + + s.secretsManager.SetupRefresh(ctx, accountID, peer.ID) + + if s.appMetrics != nil { + s.appMetrics.GRPCMetrics().CountSyncRequestDuration(time.Since(reqStart)) + } + + unlock() + unlock = nil + + log.WithContext(ctx).Debugf("Sync: took %v", time.Since(reqStart)) + + return s.handleUpdates(ctx, accountID, peerKey, peer, updates, srv) +} + +// handleUpdates sends updates to the connected peer until the updates channel is closed. +func (s *GRPCServer) handleUpdates(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, updates chan *UpdateMessage, srv proto.ManagementService_SyncServer) error { + log.WithContext(ctx).Tracef("starting to handle updates for peer %s", peerKey.String()) + for { + select { + // condition when there are some updates + case update, open := <-updates: + if s.appMetrics != nil { + s.appMetrics.GRPCMetrics().UpdateChannelQueueLength(len(updates) + 1) + } + + if !open { + log.WithContext(ctx).Debugf("updates channel for peer %s was closed", peerKey.String()) + s.cancelPeerRoutines(ctx, accountID, peer) + return nil + } + log.WithContext(ctx).Debugf("received an update for peer %s", peerKey.String()) + + if err := s.sendUpdate(ctx, accountID, peerKey, peer, update, srv); err != nil { + return err + } + + // condition when client <-> server connection has been terminated + case <-srv.Context().Done(): + // happens when connection drops, e.g. client disconnects + log.WithContext(ctx).Debugf("stream of peer %s has been closed", peerKey.String()) + s.cancelPeerRoutines(ctx, accountID, peer) + return srv.Context().Err() + } + } +} + +// sendUpdate encrypts the update message using the peer key and the server's wireguard key, +// then sends the encrypted message to the connected peer via the sync server. +func (s *GRPCServer) sendUpdate(ctx context.Context, accountID string, peerKey wgtypes.Key, peer *nbpeer.Peer, update *UpdateMessage, srv proto.ManagementService_SyncServer) error { + encryptedResp, err := encryption.EncryptMessage(peerKey, s.wgKey, update.Update) + if err != nil { + s.cancelPeerRoutines(ctx, accountID, peer) + return status.Errorf(codes.Internal, "failed processing update message") + } + err = srv.SendMsg(&proto.EncryptedMessage{ + WgPubKey: s.wgKey.PublicKey().String(), + Body: encryptedResp, + }) + if err != nil { + s.cancelPeerRoutines(ctx, accountID, peer) + return status.Errorf(codes.Internal, "failed sending update message") + } + log.WithContext(ctx).Debugf("sent an update to peer %s", peerKey.String()) + return nil +} + +func (s *GRPCServer) cancelPeerRoutines(ctx context.Context, accountID string, peer *nbpeer.Peer) { + unlock := s.acquirePeerLockByUID(ctx, peer.Key) + defer unlock() + + err := s.accountManager.OnPeerDisconnected(ctx, accountID, peer.Key) + if err != nil { + log.WithContext(ctx).Errorf("failed to disconnect peer %s properly: %v", peer.Key, err) + } + s.peersUpdateManager.CloseChannel(ctx, peer.ID) + s.secretsManager.CancelRefresh(peer.ID) + s.ephemeralManager.OnPeerDisconnected(ctx, peer) + + log.WithContext(ctx).Tracef("peer %s has been disconnected", peer.Key) +} + +func (s *GRPCServer) validateToken(ctx context.Context, jwtToken string) (string, error) { + if s.authManager == nil { + return "", status.Errorf(codes.Internal, "missing auth manager") + } + + userAuth, token, err := s.authManager.ValidateAndParseToken(ctx, jwtToken) + if err != nil { + return "", status.Errorf(codes.InvalidArgument, "invalid jwt token, err: %v", err) + } + + // we need to call this method because if user is new, we will automatically add it to existing or create a new account + accountId, _, err := s.accountManager.GetAccountIDFromUserAuth(ctx, userAuth) + if err != nil { + return "", status.Errorf(codes.Internal, "unable to fetch account with claims, err: %v", err) + } + + if userAuth.AccountId != accountId { + log.WithContext(ctx).Debugf("gRPC server sets accountId from ensure, before %s, now %s", userAuth.AccountId, accountId) + userAuth.AccountId = accountId + } + + userAuth, err = s.authManager.EnsureUserAccessByJWTGroups(ctx, userAuth, token) + if err != nil { + return "", status.Error(codes.PermissionDenied, err.Error()) + } + + err = s.accountManager.SyncUserJWTGroups(ctx, userAuth) + if err != nil { + log.WithContext(ctx).Errorf("gRPC server failed to sync user JWT groups: %s", err) + } + + return userAuth.UserId, nil +} + +func (s *GRPCServer) acquirePeerLockByUID(ctx context.Context, uniqueID string) (unlock func()) { + log.WithContext(ctx).Tracef("acquiring peer lock for ID %s", uniqueID) + + start := time.Now() + value, _ := s.peerLocks.LoadOrStore(uniqueID, &sync.RWMutex{}) + mtx := value.(*sync.RWMutex) + mtx.Lock() + log.WithContext(ctx).Tracef("acquired peer lock for ID %s in %v", uniqueID, time.Since(start)) + start = time.Now() + + unlock = func() { + mtx.Unlock() + log.WithContext(ctx).Tracef("released peer lock for ID %s in %v", uniqueID, time.Since(start)) + } + + return unlock +} + +// maps internal internalStatus.Error to gRPC status.Error +func mapError(ctx context.Context, err error) error { + if e, ok := internalStatus.FromError(err); ok { + switch e.Type() { + case internalStatus.PermissionDenied: + return status.Error(codes.PermissionDenied, e.Message) + case internalStatus.Unauthorized: + return status.Error(codes.PermissionDenied, e.Message) + case internalStatus.Unauthenticated: + return status.Error(codes.PermissionDenied, e.Message) + case internalStatus.PreconditionFailed: + return status.Error(codes.FailedPrecondition, e.Message) + case internalStatus.NotFound: + return status.Error(codes.NotFound, e.Message) + default: + } + } + log.WithContext(ctx).Errorf("got an unhandled error: %s", err) + return status.Errorf(codes.Internal, "failed handling request") +} + +func extractPeerMeta(ctx context.Context, meta *proto.PeerSystemMeta) nbpeer.PeerSystemMeta { + if meta == nil { + return nbpeer.PeerSystemMeta{} + } + + osVersion := meta.GetOSVersion() + if osVersion == "" { + osVersion = meta.GetCore() + } + + networkAddresses := make([]nbpeer.NetworkAddress, 0, len(meta.GetNetworkAddresses())) + for _, addr := range meta.GetNetworkAddresses() { + netAddr, err := netip.ParsePrefix(addr.GetNetIP()) + if err != nil { + log.WithContext(ctx).Warnf("failed to parse netip address, %s: %v", addr.GetNetIP(), err) + continue + } + networkAddresses = append(networkAddresses, nbpeer.NetworkAddress{ + NetIP: netAddr, + Mac: addr.GetMac(), + }) + } + + files := make([]nbpeer.File, 0, len(meta.GetFiles())) + for _, file := range meta.GetFiles() { + files = append(files, nbpeer.File{ + Path: file.GetPath(), + Exist: file.GetExist(), + ProcessIsRunning: file.GetProcessIsRunning(), + }) + } + + return nbpeer.PeerSystemMeta{ + Hostname: meta.GetHostname(), + GoOS: meta.GetGoOS(), + Kernel: meta.GetKernel(), + Platform: meta.GetPlatform(), + OS: meta.GetOS(), + OSVersion: osVersion, + WtVersion: meta.GetNetbirdVersion(), + UIVersion: meta.GetUiVersion(), + KernelVersion: meta.GetKernelVersion(), + NetworkAddresses: networkAddresses, + SystemSerialNumber: meta.GetSysSerialNumber(), + SystemProductName: meta.GetSysProductName(), + SystemManufacturer: meta.GetSysManufacturer(), + Environment: nbpeer.Environment{ + Cloud: meta.GetEnvironment().GetCloud(), + Platform: meta.GetEnvironment().GetPlatform(), + }, + Files: files, + } +} + +func (s *GRPCServer) parseRequest(ctx context.Context, req *proto.EncryptedMessage, parsed pb.Message) (wgtypes.Key, error) { + peerKey, err := wgtypes.ParseKey(req.GetWgPubKey()) + if err != nil { + log.WithContext(ctx).Warnf("error while parsing peer's WireGuard public key %s.", req.WgPubKey) + return wgtypes.Key{}, status.Errorf(codes.InvalidArgument, "provided wgPubKey %s is invalid", req.WgPubKey) + } + + err = encryption.DecryptMessage(peerKey, s.wgKey, req.Body, parsed) + if err != nil { + return wgtypes.Key{}, status.Errorf(codes.InvalidArgument, "invalid request message") + } + + return peerKey, nil +} + +// Login endpoint first checks whether peer is registered under any account +// In case it is, the login is successful +// In case it isn't, the endpoint checks whether setup key is provided within the request and tries to register a peer. +// In case of the successful registration login is also successful +func (s *GRPCServer) Login(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) { + reqStart := time.Now() + defer func() { + if s.appMetrics != nil { + s.appMetrics.GRPCMetrics().CountLoginRequestDuration(time.Since(reqStart)) + } + }() + if s.appMetrics != nil { + s.appMetrics.GRPCMetrics().CountLoginRequest() + } + realIP := getRealIP(ctx) + log.WithContext(ctx).Debugf("Login request from peer [%s] [%s]", req.WgPubKey, realIP.String()) + + loginReq := &proto.LoginRequest{} + peerKey, err := s.parseRequest(ctx, req, loginReq) + if err != nil { + return nil, err + } + + //nolint + ctx = context.WithValue(ctx, nbContext.PeerIDKey, peerKey.String()) + accountID, err := s.accountManager.GetAccountIDForPeerKey(ctx, peerKey.String()) + if err != nil { + // this case should not happen and already indicates an issue but we don't want the system to fail due to being unable to log in detail + accountID = "UNKNOWN" + } + //nolint + ctx = context.WithValue(ctx, nbContext.AccountIDKey, accountID) + + if loginReq.GetMeta() == nil { + msg := status.Errorf(codes.FailedPrecondition, + "peer system meta has to be provided to log in. Peer %s, remote addr %s", peerKey.String(), realIP) + log.WithContext(ctx).Warn(msg) + return nil, msg + } + + userID, err := s.processJwtToken(ctx, loginReq, peerKey) + if err != nil { + return nil, err + } + + var sshKey []byte + if loginReq.GetPeerKeys() != nil { + sshKey = loginReq.GetPeerKeys().GetSshPubKey() + } + + peer, netMap, postureChecks, err := s.accountManager.LoginPeer(ctx, types.PeerLogin{ + WireGuardPubKey: peerKey.String(), + SSHKey: string(sshKey), + Meta: extractPeerMeta(ctx, loginReq.GetMeta()), + UserID: userID, + SetupKey: loginReq.GetSetupKey(), + ConnectionIP: realIP, + ExtraDNSLabels: loginReq.GetDnsLabels(), + }) + if err != nil { + log.WithContext(ctx).Warnf("failed logging in peer %s: %s", peerKey, err) + return nil, mapError(ctx, err) + } + + // if the login request contains setup key then it is a registration request + if loginReq.GetSetupKey() != "" { + s.ephemeralManager.OnPeerDisconnected(ctx, peer) + } + + var relayToken *Token + if s.config.Relay != nil && len(s.config.Relay.Addresses) > 0 { + relayToken, err = s.secretsManager.GenerateRelayToken() + if err != nil { + log.Errorf("failed generating Relay token: %v", err) + } + } + + // if peer has reached this point then it has logged in + loginResp := &proto.LoginResponse{ + NetbirdConfig: toNetbirdConfig(s.config, nil, relayToken, nil), + PeerConfig: toPeerConfig(peer, netMap.Network, s.accountManager.GetDNSDomain(), false), + Checks: toProtocolChecks(ctx, postureChecks), + } + encryptedResp, err := encryption.EncryptMessage(peerKey, s.wgKey, loginResp) + if err != nil { + log.WithContext(ctx).Warnf("failed encrypting peer %s message", peer.ID) + return nil, status.Errorf(codes.Internal, "failed logging in peer") + } + + return &proto.EncryptedMessage{ + WgPubKey: s.wgKey.PublicKey().String(), + Body: encryptedResp, + }, nil +} + +// processJwtToken validates the existence of a JWT token in the login request, and returns the corresponding user ID if +// the token is valid. +// +// The user ID can be empty if the token is not provided, which is acceptable if the peer is already +// registered or if it uses a setup key to register. +func (s *GRPCServer) processJwtToken(ctx context.Context, loginReq *proto.LoginRequest, peerKey wgtypes.Key) (string, error) { + userID := "" + if loginReq.GetJwtToken() != "" { + var err error + for i := 0; i < 3; i++ { + userID, err = s.validateToken(ctx, loginReq.GetJwtToken()) + if err == nil { + break + } + log.WithContext(ctx).Warnf("failed validating JWT token sent from peer %s with error %v. "+ + "Trying again as it may be due to the IdP cache issue", peerKey.String(), err) + time.Sleep(200 * time.Millisecond) + } + if err != nil { + return "", err + } + } + return userID, nil +} + +func ToResponseProto(configProto types.Protocol) proto.HostConfig_Protocol { + switch configProto { + case types.UDP: + return proto.HostConfig_UDP + case types.DTLS: + return proto.HostConfig_DTLS + case types.HTTP: + return proto.HostConfig_HTTP + case types.HTTPS: + return proto.HostConfig_HTTPS + case types.TCP: + return proto.HostConfig_TCP + default: + panic(fmt.Errorf("unexpected config protocol type %v", configProto)) + } +} + +func toNetbirdConfig(config *types.Config, turnCredentials *Token, relayToken *Token, extraSettings *types.ExtraSettings) *proto.NetbirdConfig { + if config == nil { + return nil + } + + var stuns []*proto.HostConfig + for _, stun := range config.Stuns { + stuns = append(stuns, &proto.HostConfig{ + Uri: stun.URI, + Protocol: ToResponseProto(stun.Proto), + }) + } + + var turns []*proto.ProtectedHostConfig + if config.TURNConfig != nil { + for _, turn := range config.TURNConfig.Turns { + var username string + var password string + if turnCredentials != nil { + username = turnCredentials.Payload + password = turnCredentials.Signature + } else { + username = turn.Username + password = turn.Password + } + turns = append(turns, &proto.ProtectedHostConfig{ + HostConfig: &proto.HostConfig{ + Uri: turn.URI, + Protocol: ToResponseProto(turn.Proto), + }, + User: username, + Password: password, + }) + } + } + + var relayCfg *proto.RelayConfig + if config.Relay != nil && len(config.Relay.Addresses) > 0 { + relayCfg = &proto.RelayConfig{ + Urls: config.Relay.Addresses, + } + + if relayToken != nil { + relayCfg.TokenPayload = relayToken.Payload + relayCfg.TokenSignature = relayToken.Signature + } + } + + var signalCfg *proto.HostConfig + if config.Signal != nil { + signalCfg = &proto.HostConfig{ + Uri: config.Signal.URI, + Protocol: ToResponseProto(config.Signal.Proto), + } + } + + nbConfig := &proto.NetbirdConfig{ + Stuns: stuns, + Turns: turns, + Signal: signalCfg, + Relay: relayCfg, + } + + return nbConfig +} + +func toPeerConfig(peer *nbpeer.Peer, network *types.Network, dnsName string, dnsResolutionOnRoutingPeerEnabled bool) *proto.PeerConfig { + netmask, _ := network.Net.Mask.Size() + fqdn := peer.FQDN(dnsName) + return &proto.PeerConfig{ + Address: fmt.Sprintf("%s/%d", peer.IP.String(), netmask), // take it from the network + SshConfig: &proto.SSHConfig{SshEnabled: peer.SSHEnabled}, + Fqdn: fqdn, + RoutingPeerDnsResolutionEnabled: dnsResolutionOnRoutingPeerEnabled, + } +} + +func toSyncResponse(ctx context.Context, config *types.Config, peer *nbpeer.Peer, turnCredentials *Token, relayCredentials *Token, networkMap *types.NetworkMap, dnsName string, checks []*posture.Checks, dnsCache *DNSConfigCache, dnsResolutionOnRoutingPeerEnabled bool, extraSettings *types.ExtraSettings) *proto.SyncResponse { + response := &proto.SyncResponse{ + PeerConfig: toPeerConfig(peer, networkMap.Network, dnsName, dnsResolutionOnRoutingPeerEnabled), + NetworkMap: &proto.NetworkMap{ + Serial: networkMap.Network.CurrentSerial(), + Routes: toProtocolRoutes(networkMap.Routes), + DNSConfig: toProtocolDNSConfig(networkMap.DNSConfig, dnsCache), + }, + Checks: toProtocolChecks(ctx, checks), + } + + nbConfig := toNetbirdConfig(config, turnCredentials, relayCredentials, extraSettings) + extendedConfig := integrationsConfig.ExtendNetBirdConfig(peer.ID, nbConfig, extraSettings) + response.NetbirdConfig = extendedConfig + + response.NetworkMap.PeerConfig = response.PeerConfig + + allPeers := make([]*proto.RemotePeerConfig, 0, len(networkMap.Peers)+len(networkMap.OfflinePeers)) + allPeers = appendRemotePeerConfig(allPeers, networkMap.Peers, dnsName) + response.RemotePeers = allPeers + response.NetworkMap.RemotePeers = allPeers + response.RemotePeersIsEmpty = len(allPeers) == 0 + response.NetworkMap.RemotePeersIsEmpty = response.RemotePeersIsEmpty + + response.NetworkMap.OfflinePeers = appendRemotePeerConfig(nil, networkMap.OfflinePeers, dnsName) + + firewallRules := toProtocolFirewallRules(networkMap.FirewallRules) + response.NetworkMap.FirewallRules = firewallRules + response.NetworkMap.FirewallRulesIsEmpty = len(firewallRules) == 0 + + routesFirewallRules := toProtocolRoutesFirewallRules(networkMap.RoutesFirewallRules) + response.NetworkMap.RoutesFirewallRules = routesFirewallRules + response.NetworkMap.RoutesFirewallRulesIsEmpty = len(routesFirewallRules) == 0 + + if networkMap.ForwardingRules != nil { + forwardingRules := make([]*proto.ForwardingRule, 0, len(networkMap.ForwardingRules)) + for _, rule := range networkMap.ForwardingRules { + forwardingRules = append(forwardingRules, rule.ToProto()) + } + response.NetworkMap.ForwardingRules = forwardingRules + } + + return response +} + +func appendRemotePeerConfig(dst []*proto.RemotePeerConfig, peers []*nbpeer.Peer, dnsName string) []*proto.RemotePeerConfig { + for _, rPeer := range peers { + dst = append(dst, &proto.RemotePeerConfig{ + WgPubKey: rPeer.Key, + AllowedIps: []string{rPeer.IP.String() + "/32"}, + SshConfig: &proto.SSHConfig{SshPubKey: []byte(rPeer.SSHKey)}, + Fqdn: rPeer.FQDN(dnsName), + }) + } + return dst +} + +// IsHealthy indicates whether the service is healthy +func (s *GRPCServer) IsHealthy(ctx context.Context, req *proto.Empty) (*proto.Empty, error) { + return &proto.Empty{}, nil +} + +// sendInitialSync sends initial proto.SyncResponse to the peer requesting synchronization +func (s *GRPCServer) sendInitialSync(ctx context.Context, peerKey wgtypes.Key, peer *nbpeer.Peer, networkMap *types.NetworkMap, postureChecks []*posture.Checks, srv proto.ManagementService_SyncServer) error { + var err error + + var turnToken *Token + if s.config.TURNConfig != nil && s.config.TURNConfig.TimeBasedCredentials { + turnToken, err = s.secretsManager.GenerateTurnToken() + if err != nil { + log.Errorf("failed generating TURN token: %v", err) + } + } + + var relayToken *Token + if s.config.Relay != nil && len(s.config.Relay.Addresses) > 0 { + relayToken, err = s.secretsManager.GenerateRelayToken() + if err != nil { + log.Errorf("failed generating Relay token: %v", err) + } + } + + settings, err := s.settingsManager.GetSettings(ctx, peer.AccountID, activity.SystemInitiator) + if err != nil { + return status.Errorf(codes.Internal, "error handling request") + } + + plainResp := toSyncResponse(ctx, s.config, peer, turnToken, relayToken, networkMap, s.accountManager.GetDNSDomain(), postureChecks, nil, settings.RoutingPeerDNSResolutionEnabled, settings.Extra) + + encryptedResp, err := encryption.EncryptMessage(peerKey, s.wgKey, plainResp) + if err != nil { + return status.Errorf(codes.Internal, "error handling request") + } + + err = srv.Send(&proto.EncryptedMessage{ + WgPubKey: s.wgKey.PublicKey().String(), + Body: encryptedResp, + }) + + if err != nil { + log.WithContext(ctx).Errorf("failed sending SyncResponse %v", err) + return status.Errorf(codes.Internal, "error handling request") + } + + return nil +} + +// GetDeviceAuthorizationFlow returns a device authorization flow information +// This is used for initiating an Oauth 2 device authorization grant flow +// which will be used by our clients to Login +func (s *GRPCServer) GetDeviceAuthorizationFlow(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) { + log.WithContext(ctx).Tracef("GetDeviceAuthorizationFlow request for pubKey: %s", req.WgPubKey) + start := time.Now() + defer func() { + log.WithContext(ctx).Tracef("GetDeviceAuthorizationFlow for pubKey: %s took %v", req.WgPubKey, time.Since(start)) + }() + + peerKey, err := wgtypes.ParseKey(req.GetWgPubKey()) + if err != nil { + errMSG := fmt.Sprintf("error while parsing peer's Wireguard public key %s on GetDeviceAuthorizationFlow request.", req.WgPubKey) + log.WithContext(ctx).Warn(errMSG) + return nil, status.Error(codes.InvalidArgument, errMSG) + } + + err = encryption.DecryptMessage(peerKey, s.wgKey, req.Body, &proto.DeviceAuthorizationFlowRequest{}) + if err != nil { + errMSG := fmt.Sprintf("error while decrypting peer's message with Wireguard public key %s.", req.WgPubKey) + log.WithContext(ctx).Warn(errMSG) + return nil, status.Error(codes.InvalidArgument, errMSG) + } + + if s.config.DeviceAuthorizationFlow == nil || s.config.DeviceAuthorizationFlow.Provider == string(types.NONE) { + return nil, status.Error(codes.NotFound, "no device authorization flow information available") + } + + provider, ok := proto.DeviceAuthorizationFlowProvider_value[strings.ToUpper(s.config.DeviceAuthorizationFlow.Provider)] + if !ok { + return nil, status.Errorf(codes.InvalidArgument, "no provider found in the protocol for %s", s.config.DeviceAuthorizationFlow.Provider) + } + + flowInfoResp := &proto.DeviceAuthorizationFlow{ + Provider: proto.DeviceAuthorizationFlowProvider(provider), + ProviderConfig: &proto.ProviderConfig{ + ClientID: s.config.DeviceAuthorizationFlow.ProviderConfig.ClientID, + ClientSecret: s.config.DeviceAuthorizationFlow.ProviderConfig.ClientSecret, + Domain: s.config.DeviceAuthorizationFlow.ProviderConfig.Domain, + Audience: s.config.DeviceAuthorizationFlow.ProviderConfig.Audience, + DeviceAuthEndpoint: s.config.DeviceAuthorizationFlow.ProviderConfig.DeviceAuthEndpoint, + TokenEndpoint: s.config.DeviceAuthorizationFlow.ProviderConfig.TokenEndpoint, + Scope: s.config.DeviceAuthorizationFlow.ProviderConfig.Scope, + UseIDToken: s.config.DeviceAuthorizationFlow.ProviderConfig.UseIDToken, + }, + } + + encryptedResp, err := encryption.EncryptMessage(peerKey, s.wgKey, flowInfoResp) + if err != nil { + return nil, status.Error(codes.Internal, "failed to encrypt no device authorization flow information") + } + + return &proto.EncryptedMessage{ + WgPubKey: s.wgKey.PublicKey().String(), + Body: encryptedResp, + }, nil +} + +// GetPKCEAuthorizationFlow returns a pkce authorization flow information +// This is used for initiating an Oauth 2 pkce authorization grant flow +// which will be used by our clients to Login +func (s *GRPCServer) GetPKCEAuthorizationFlow(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) { + log.WithContext(ctx).Tracef("GetPKCEAuthorizationFlow request for pubKey: %s", req.WgPubKey) + start := time.Now() + defer func() { + log.WithContext(ctx).Tracef("GetPKCEAuthorizationFlow for pubKey %s took %v", req.WgPubKey, time.Since(start)) + }() + + peerKey, err := wgtypes.ParseKey(req.GetWgPubKey()) + if err != nil { + errMSG := fmt.Sprintf("error while parsing peer's Wireguard public key %s on GetPKCEAuthorizationFlow request.", req.WgPubKey) + log.WithContext(ctx).Warn(errMSG) + return nil, status.Error(codes.InvalidArgument, errMSG) + } + + err = encryption.DecryptMessage(peerKey, s.wgKey, req.Body, &proto.PKCEAuthorizationFlowRequest{}) + if err != nil { + errMSG := fmt.Sprintf("error while decrypting peer's message with Wireguard public key %s.", req.WgPubKey) + log.WithContext(ctx).Warn(errMSG) + return nil, status.Error(codes.InvalidArgument, errMSG) + } + + if s.config.PKCEAuthorizationFlow == nil { + return nil, status.Error(codes.NotFound, "no pkce authorization flow information available") + } + + flowInfoResp := &proto.PKCEAuthorizationFlow{ + ProviderConfig: &proto.ProviderConfig{ + Audience: s.config.PKCEAuthorizationFlow.ProviderConfig.Audience, + ClientID: s.config.PKCEAuthorizationFlow.ProviderConfig.ClientID, + ClientSecret: s.config.PKCEAuthorizationFlow.ProviderConfig.ClientSecret, + TokenEndpoint: s.config.PKCEAuthorizationFlow.ProviderConfig.TokenEndpoint, + AuthorizationEndpoint: s.config.PKCEAuthorizationFlow.ProviderConfig.AuthorizationEndpoint, + Scope: s.config.PKCEAuthorizationFlow.ProviderConfig.Scope, + RedirectURLs: s.config.PKCEAuthorizationFlow.ProviderConfig.RedirectURLs, + UseIDToken: s.config.PKCEAuthorizationFlow.ProviderConfig.UseIDToken, + DisablePromptLogin: s.config.PKCEAuthorizationFlow.ProviderConfig.DisablePromptLogin, + }, + } + + encryptedResp, err := encryption.EncryptMessage(peerKey, s.wgKey, flowInfoResp) + if err != nil { + return nil, status.Error(codes.Internal, "failed to encrypt no pkce authorization flow information") + } + + return &proto.EncryptedMessage{ + WgPubKey: s.wgKey.PublicKey().String(), + Body: encryptedResp, + }, nil +} + +// SyncMeta endpoint is used to synchronize peer's system metadata and notifies the connected, +// peer's under the same account of any updates. +func (s *GRPCServer) SyncMeta(ctx context.Context, req *proto.EncryptedMessage) (*proto.Empty, error) { + realIP := getRealIP(ctx) + log.WithContext(ctx).Debugf("Sync meta request from peer [%s] [%s]", req.WgPubKey, realIP.String()) + + syncMetaReq := &proto.SyncMetaRequest{} + peerKey, err := s.parseRequest(ctx, req, syncMetaReq) + if err != nil { + return nil, err + } + + if syncMetaReq.GetMeta() == nil { + msg := status.Errorf(codes.FailedPrecondition, + "peer system meta has to be provided on sync. Peer %s, remote addr %s", peerKey.String(), realIP) + log.WithContext(ctx).Warn(msg) + return nil, msg + } + + err = s.accountManager.SyncPeerMeta(ctx, peerKey.String(), extractPeerMeta(ctx, syncMetaReq.GetMeta())) + if err != nil { + return nil, mapError(ctx, err) + } + + return &proto.Empty{}, nil +} + +// toProtocolChecks converts posture checks to protocol checks. +func toProtocolChecks(ctx context.Context, postureChecks []*posture.Checks) []*proto.Checks { + protoChecks := make([]*proto.Checks, 0, len(postureChecks)) + for _, postureCheck := range postureChecks { + protoChecks = append(protoChecks, toProtocolCheck(postureCheck)) + } + + return protoChecks +} + +// toProtocolCheck converts a posture.Checks to a proto.Checks. +func toProtocolCheck(postureCheck *posture.Checks) *proto.Checks { + protoCheck := &proto.Checks{} + + if check := postureCheck.Checks.ProcessCheck; check != nil { + for _, process := range check.Processes { + if process.LinuxPath != "" { + protoCheck.Files = append(protoCheck.Files, process.LinuxPath) + } + if process.MacPath != "" { + protoCheck.Files = append(protoCheck.Files, process.MacPath) + } + if process.WindowsPath != "" { + protoCheck.Files = append(protoCheck.Files, process.WindowsPath) + } + } + } + + return protoCheck +} diff --git a/internal/shared/api/middleware/auth_middleware.go b/internal/shared/api/rest/middleware/auth_middleware.go similarity index 100% rename from internal/shared/api/middleware/auth_middleware.go rename to internal/shared/api/rest/middleware/auth_middleware.go diff --git a/internal/shared/api/middleware/auth_middleware_test.go b/internal/shared/api/rest/middleware/auth_middleware_test.go similarity index 100% rename from internal/shared/api/middleware/auth_middleware_test.go rename to internal/shared/api/rest/middleware/auth_middleware_test.go diff --git a/internal/shared/api/middleware/bypass/bypass.go b/internal/shared/api/rest/middleware/bypass/bypass.go similarity index 100% rename from internal/shared/api/middleware/bypass/bypass.go rename to internal/shared/api/rest/middleware/bypass/bypass.go diff --git a/internal/shared/api/middleware/bypass/bypass_test.go b/internal/shared/api/rest/middleware/bypass/bypass_test.go similarity index 100% rename from internal/shared/api/middleware/bypass/bypass_test.go rename to internal/shared/api/rest/middleware/bypass/bypass_test.go diff --git a/internal/shared/api/middleware/metrics_middleware.go b/internal/shared/api/rest/middleware/metrics_middleware.go similarity index 100% rename from internal/shared/api/middleware/metrics_middleware.go rename to internal/shared/api/rest/middleware/metrics_middleware.go diff --git a/internal/shared/api/router.go b/internal/shared/api/rest/router.go similarity index 98% rename from internal/shared/api/router.go rename to internal/shared/api/rest/router.go index 9c56169..7c9d882 100644 --- a/internal/shared/api/router.go +++ b/internal/shared/api/rest/router.go @@ -1,4 +1,4 @@ -package api +package rest import ( "net/http" diff --git a/internal/shared/auth/jwt/extractor.go b/internal/shared/auth/jwt/extractor.go new file mode 100644 index 0000000..fab4291 --- /dev/null +++ b/internal/shared/auth/jwt/extractor.go @@ -0,0 +1,144 @@ +package jwt + +import ( + "errors" + "net/url" + "time" + + "github.com/golang-jwt/jwt" + log "github.com/sirupsen/logrus" + + nbcontext "github.com/netbirdio/netbird/management/server/context" +) + +const ( + // AccountIDSuffix suffix for the account id claim + AccountIDSuffix = "wt_account_id" + // DomainIDSuffix suffix for the domain id claim + DomainIDSuffix = "wt_account_domain" + // DomainCategorySuffix suffix for the domain category claim + DomainCategorySuffix = "wt_account_domain_category" + // UserIDClaim claim for the user id + UserIDClaim = "sub" + // LastLoginSuffix claim for the last login + LastLoginSuffix = "nb_last_login" + // Invited claim indicates that an incoming JWT is from a user that just accepted an invitation + Invited = "nb_invited" +) + +var ( + errUserIDClaimEmpty = errors.New("user ID claim token value is empty") +) + +// ClaimsExtractor struct that holds the extract function +type ClaimsExtractor struct { + authAudience string + userIDClaim string +} + +// ClaimsExtractorOption is a function that configures the ClaimsExtractor +type ClaimsExtractorOption func(*ClaimsExtractor) + +// WithAudience sets the audience for the extractor +func WithAudience(audience string) ClaimsExtractorOption { + return func(c *ClaimsExtractor) { + c.authAudience = audience + } +} + +// WithUserIDClaim sets the user id claim for the extractor +func WithUserIDClaim(userIDClaim string) ClaimsExtractorOption { + return func(c *ClaimsExtractor) { + c.userIDClaim = userIDClaim + } +} + +// NewClaimsExtractor returns an extractor, and if provided with a function with ExtractClaims signature, +// then it will use that logic. Uses ExtractClaimsFromRequestContext by default +func NewClaimsExtractor(options ...ClaimsExtractorOption) *ClaimsExtractor { + ce := &ClaimsExtractor{} + for _, option := range options { + option(ce) + } + + if ce.userIDClaim == "" { + ce.userIDClaim = UserIDClaim + } + return ce +} + +func parseTime(timeString string) time.Time { + if timeString == "" { + return time.Time{} + } + parsedTime, err := time.Parse(time.RFC3339, timeString) + if err != nil { + return time.Time{} + } + return parsedTime +} + +func (c ClaimsExtractor) audienceClaim(claimName string) string { + url, err := url.JoinPath(c.authAudience, claimName) + if err != nil { + return c.authAudience + claimName // as it was previously + } + + return url +} + +func (c *ClaimsExtractor) ToUserAuth(token *jwt.Token) (nbcontext.UserAuth, error) { + claims := token.Claims.(jwt.MapClaims) + userAuth := nbcontext.UserAuth{} + + userID, ok := claims[c.userIDClaim].(string) + if !ok { + return userAuth, errUserIDClaimEmpty + } + userAuth.UserId = userID + + if accountIDClaim, ok := claims[c.audienceClaim(AccountIDSuffix)]; ok { + userAuth.AccountId = accountIDClaim.(string) + } + + if domainClaim, ok := claims[c.audienceClaim(DomainIDSuffix)]; ok { + userAuth.Domain = domainClaim.(string) + } + + if domainCategoryClaim, ok := claims[c.audienceClaim(DomainCategorySuffix)]; ok { + userAuth.DomainCategory = domainCategoryClaim.(string) + } + + if lastLoginClaimString, ok := claims[c.audienceClaim(LastLoginSuffix)]; ok { + userAuth.LastLogin = parseTime(lastLoginClaimString.(string)) + } + + if invitedBool, ok := claims[c.audienceClaim(Invited)]; ok { + if value, ok := invitedBool.(bool); ok { + userAuth.Invited = value + } + } + + return userAuth, nil +} + +func (c *ClaimsExtractor) ToGroups(token *jwt.Token, claimName string) []string { + claims := token.Claims.(jwt.MapClaims) + userJWTGroups := make([]string, 0) + + if claim, ok := claims[claimName]; ok { + if claimGroups, ok := claim.([]interface{}); ok { + for _, g := range claimGroups { + if group, ok := g.(string); ok { + userJWTGroups = append(userJWTGroups, group) + } else { + log.Debugf("JWT claim %q contains a non-string group (type: %T): %v", claimName, g, g) + } + } + } + } else { + log.Debugf("JWT claim %q is not a string array", claimName) + } + + return userJWTGroups +} diff --git a/internal/shared/auth/jwt/validator.go b/internal/shared/auth/jwt/validator.go new file mode 100644 index 0000000..5b38ca7 --- /dev/null +++ b/internal/shared/auth/jwt/validator.go @@ -0,0 +1,302 @@ +package jwt + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rsa" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "math/big" + "net/http" + "net/url" + "strconv" + "strings" + "sync" + "time" + + "github.com/golang-jwt/jwt" + + log "github.com/sirupsen/logrus" +) + +// Jwks is a collection of JSONWebKey obtained from Config.HttpServerConfig.AuthKeysLocation +type Jwks struct { + Keys []JSONWebKey `json:"keys"` + expiresInTime time.Time +} + +// The supported elliptic curves types +const ( + // p256 represents a cryptographic elliptical curve type. + p256 = "P-256" + + // p384 represents a cryptographic elliptical curve type. + p384 = "P-384" + + // p521 represents a cryptographic elliptical curve type. + p521 = "P-521" +) + +// JSONWebKey is a representation of a Jason Web Key +type JSONWebKey struct { + Kty string `json:"kty"` + Kid string `json:"kid"` + Use string `json:"use"` + N string `json:"n"` + E string `json:"e"` + Crv string `json:"crv"` + X string `json:"x"` + Y string `json:"y"` + X5c []string `json:"x5c"` +} + +type Validator struct { + lock sync.Mutex + issuer string + audienceList []string + keysLocation string + idpSignkeyRefreshEnabled bool + keys *Jwks +} + +var ( + errKeyNotFound = errors.New("unable to find appropriate key") + errInvalidAudience = errors.New("invalid audience") + errInvalidIssuer = errors.New("invalid issuer") + errTokenEmpty = errors.New("required authorization token not found") + errTokenInvalid = errors.New("token is invalid") + errTokenParsing = errors.New("token could not be parsed") +) + +func NewValidator(issuer string, audienceList []string, keysLocation string, idpSignkeyRefreshEnabled bool) *Validator { + keys, err := getPemKeys(keysLocation) + if err != nil { + log.WithField("keysLocation", keysLocation).Errorf("could not get keys from location: %s", err) + } + + return &Validator{ + keys: keys, + issuer: issuer, + audienceList: audienceList, + keysLocation: keysLocation, + idpSignkeyRefreshEnabled: idpSignkeyRefreshEnabled, + } +} + +func (v *Validator) getKeyFunc(ctx context.Context) jwt.Keyfunc { + return func(token *jwt.Token) (interface{}, error) { + // Verify 'aud' claim + var checkAud bool + for _, audience := range v.audienceList { + checkAud = token.Claims.(jwt.MapClaims).VerifyAudience(audience, false) + if checkAud { + break + } + } + if !checkAud { + return token, errInvalidAudience + } + + // Verify 'issuer' claim + checkIss := token.Claims.(jwt.MapClaims).VerifyIssuer(v.issuer, false) + if !checkIss { + return token, errInvalidIssuer + } + + // If keys are rotated, verify the keys prior to token validation + if v.idpSignkeyRefreshEnabled { + // If the keys are invalid, retrieve new ones + // @todo propose a separate go routine to regularly check these to prevent blocking when actually + // validating the token + if !v.keys.stillValid() { + v.lock.Lock() + defer v.lock.Unlock() + + refreshedKeys, err := getPemKeys(v.keysLocation) + if err != nil { + log.WithContext(ctx).Debugf("cannot get JSONWebKey: %v, falling back to old keys", err) + refreshedKeys = v.keys + } + + log.WithContext(ctx).Debugf("keys refreshed, new UTC expiration time: %s", refreshedKeys.expiresInTime.UTC()) + + v.keys = refreshedKeys + } + } + + publicKey, err := getPublicKey(token, v.keys) + if err == nil { + return publicKey, nil + } + + msg := fmt.Sprintf("getPublicKey error: %s", err) + if errors.Is(err, errKeyNotFound) && !v.idpSignkeyRefreshEnabled { + msg = fmt.Sprintf("getPublicKey error: %s. You can enable key refresh by setting HttpServerConfig.IdpSignKeyRefreshEnabled to true in your management.json file and restart the service", err) + } + + log.WithContext(ctx).Error(msg) + + return nil, err + } +} + +// ValidateAndParse validates the token and returns the parsed token +func (m *Validator) ValidateAndParse(ctx context.Context, token string) (*jwt.Token, error) { + // If the token is empty... + if token == "" { + // If we get here, the required token is missing + log.WithContext(ctx).Debugf(" Error: No credentials found (CredentialsOptional=false)") + return nil, errTokenEmpty + } + + // Now parse the token + parsedToken, err := jwt.Parse(token, m.getKeyFunc(ctx)) + + // Check if there was an error in parsing... + if err != nil { + err = fmt.Errorf("%w: %s", errTokenParsing, err) + log.WithContext(ctx).Error(err.Error()) + return nil, err + } + + // Check if the parsed token is valid... + if !parsedToken.Valid { + log.WithContext(ctx).Debug(errTokenInvalid.Error()) + return nil, errTokenInvalid + } + + return parsedToken, nil +} + +// stillValid returns true if the JSONWebKey still valid and have enough time to be used +func (jwks *Jwks) stillValid() bool { + return !jwks.expiresInTime.IsZero() && time.Now().Add(5*time.Second).Before(jwks.expiresInTime) +} + +func getPemKeys(keysLocation string) (*Jwks, error) { + jwks := &Jwks{} + + url, err := url.ParseRequestURI(keysLocation) + if err != nil { + return jwks, err + } + + resp, err := http.Get(url.String()) + if err != nil { + return jwks, err + } + defer resp.Body.Close() + + err = json.NewDecoder(resp.Body).Decode(jwks) + if err != nil { + return jwks, err + } + + cacheControlHeader := resp.Header.Get("Cache-Control") + expiresIn := getMaxAgeFromCacheHeader(cacheControlHeader) + jwks.expiresInTime = time.Now().Add(time.Duration(expiresIn) * time.Second) + + return jwks, nil +} + +func getPublicKey(token *jwt.Token, jwks *Jwks) (interface{}, error) { + // todo as we load the jkws when the server is starting, we should build a JKS map with the pem cert at the boot time + for k := range jwks.Keys { + if token.Header["kid"] != jwks.Keys[k].Kid { + continue + } + + if len(jwks.Keys[k].X5c) != 0 { + cert := "-----BEGIN CERTIFICATE-----\n" + jwks.Keys[k].X5c[0] + "\n-----END CERTIFICATE-----" + return jwt.ParseRSAPublicKeyFromPEM([]byte(cert)) + } + + if jwks.Keys[k].Kty == "RSA" { + return getPublicKeyFromRSA(jwks.Keys[k]) + } + if jwks.Keys[k].Kty == "EC" { + return getPublicKeyFromECDSA(jwks.Keys[k]) + } + } + + return nil, errKeyNotFound +} + +func getPublicKeyFromECDSA(jwk JSONWebKey) (publicKey *ecdsa.PublicKey, err error) { + if jwk.X == "" || jwk.Y == "" || jwk.Crv == "" { + return nil, fmt.Errorf("ecdsa key incomplete") + } + + var xCoordinate []byte + if xCoordinate, err = base64.RawURLEncoding.DecodeString(jwk.X); err != nil { + return nil, err + } + + var yCoordinate []byte + if yCoordinate, err = base64.RawURLEncoding.DecodeString(jwk.Y); err != nil { + return nil, err + } + + publicKey = &ecdsa.PublicKey{} + + var curve elliptic.Curve + switch jwk.Crv { + case p256: + curve = elliptic.P256() + case p384: + curve = elliptic.P384() + case p521: + curve = elliptic.P521() + } + + publicKey.Curve = curve + publicKey.X = big.NewInt(0).SetBytes(xCoordinate) + publicKey.Y = big.NewInt(0).SetBytes(yCoordinate) + + return publicKey, nil +} + +func getPublicKeyFromRSA(jwk JSONWebKey) (*rsa.PublicKey, error) { + decodedE, err := base64.RawURLEncoding.DecodeString(jwk.E) + if err != nil { + return nil, err + } + decodedN, err := base64.RawURLEncoding.DecodeString(jwk.N) + if err != nil { + return nil, err + } + + var n, e big.Int + e.SetBytes(decodedE) + n.SetBytes(decodedN) + + return &rsa.PublicKey{ + E: int(e.Int64()), + N: &n, + }, nil +} + +// getMaxAgeFromCacheHeader extracts max-age directive from the Cache-Control header +func getMaxAgeFromCacheHeader(cacheControl string) int { + // Split into individual directives + directives := strings.Split(cacheControl, ",") + + for _, directive := range directives { + directive = strings.TrimSpace(directive) + if strings.HasPrefix(directive, "max-age=") { + // Extract the max-age value + maxAgeStr := strings.TrimPrefix(directive, "max-age=") + maxAge, err := strconv.Atoi(maxAgeStr) + if err != nil { + return 0 + } + + return maxAge + } + } + + return 0 +} diff --git a/internal/shared/auth/manager.go b/internal/shared/auth/manager.go new file mode 100644 index 0000000..daa8f9d --- /dev/null +++ b/internal/shared/auth/manager.go @@ -0,0 +1,176 @@ +package auth + +import ( + "context" + "crypto/sha256" + "encoding/base64" + "fmt" + "hash/crc32" + + "github.com/golang-jwt/jwt" + "github.com/netbirdio/netbird/base62" + nbjwt "github.com/netbirdio/netbird/management/server/auth/jwt" + nbcontext "github.com/netbirdio/netbird/management/server/context" + "github.com/netbirdio/netbird/management/server/store" + + "management/internal/modules/accounts/settings" + "management/internal/modules/users" + pattypes "management/internal/modules/users/pats/types" + "management/internal/modules/users/types" + "management/internal/shared/db" +) + +var _ Manager = (*manager)(nil) + +type Manager interface { + ValidateAndParseToken(ctx context.Context, value string) (nbcontext.UserAuth, *jwt.Token, error) + EnsureUserAccessByJWTGroups(ctx context.Context, userAuth nbcontext.UserAuth, token *jwt.Token) (nbcontext.UserAuth, error) + MarkPATUsed(ctx context.Context, tokenID string) error + GetPATInfo(ctx context.Context, token string) (user *types.User, pat *pattypes.PersonalAccessToken, domain string, category string, err error) +} + +type manager struct { + userManager *users.Manager + settingsManager *settings.Manager + + validator *nbjwt.Validator + extractor *nbjwt.ClaimsExtractor +} + +func NewManager(userManager *users.Manager, settingsManager *settings.Manager, issuer, audience, keysLocation, userIdClaim string, allAudiences []string, idpRefreshKeys bool) Manager { + // @note if invalid/missing parameters are sent the validator will instantiate + // but it will fail when validating and parsing the token + jwtValidator := nbjwt.NewValidator( + issuer, + allAudiences, + keysLocation, + idpRefreshKeys, + ) + + claimsExtractor := nbjwt.NewClaimsExtractor( + nbjwt.WithAudience(audience), + nbjwt.WithUserIDClaim(userIdClaim), + ) + + return &manager{ + userManager: userManager, + settingsManager: settingsManager, + + validator: jwtValidator, + extractor: claimsExtractor, + } +} + +func (m *manager) ValidateAndParseToken(ctx context.Context, value string) (nbcontext.UserAuth, *jwt.Token, error) { + token, err := m.validator.ValidateAndParse(ctx, value) + if err != nil { + return nbcontext.UserAuth{}, nil, err + } + + userAuth, err := m.extractor.ToUserAuth(token) + if err != nil { + return nbcontext.UserAuth{}, nil, err + } + return userAuth, token, err +} + +func (m *manager) EnsureUserAccessByJWTGroups(ctx context.Context, userAuth nbcontext.UserAuth, token *jwt.Token) (nbcontext.UserAuth, error) { + if userAuth.IsChild || userAuth.IsPAT { + return userAuth, nil + } + + settings, err := m.settingsManager.GetSettings(ctx, nil, db.LockingStrengthShare, userAuth.AccountId, userAuth.UserId) + if err != nil { + return userAuth, err + } + + // Ensures JWT group synchronization to the management is enabled before, + // filtering access based on the allowed groups. + if settings != nil && settings.JWTGroupsEnabled { + userAuth.Groups = m.extractor.ToGroups(token, settings.JWTGroupsClaimName) + if allowedGroups := settings.JWTAllowGroups; len(allowedGroups) > 0 { + if !userHasAllowedGroup(allowedGroups, userAuth.Groups) { + return userAuth, fmt.Errorf("user does not belong to any of the allowed JWT groups") + } + } + } + + return userAuth, nil +} + +// MarkPATUsed marks a personal access token as used +func (am *manager) MarkPATUsed(ctx context.Context, tokenID string) error { + return am.store.MarkPATUsed(ctx, store.LockingStrengthUpdate, tokenID) +} + +// GetPATInfo retrieves user, personal access token, domain, and category details from a personal access token. +func (am *manager) GetPATInfo(ctx context.Context, token string) (user *types.User, pat *pattypes.PersonalAccessToken, domain string, category string, err error) { + user, pat, err = am.extractPATFromToken(ctx, token) + if err != nil { + return nil, nil, "", "", err + } + + domain, category, err = am.store.GetAccountDomainAndCategory(ctx, store.LockingStrengthShare, user.AccountID) + if err != nil { + return nil, nil, "", "", err + } + + return user, pat, domain, category, nil +} + +// extractPATFromToken validates the token structure and retrieves associated User and PAT. +func (am *manager) extractPATFromToken(ctx context.Context, token string) (*types.User, *pattypes.PersonalAccessToken, error) { + if len(token) != pattypes.PATLength { + return nil, nil, fmt.Errorf("PAT has incorrect length") + } + + prefix := token[:len(pattypes.PATPrefix)] + if prefix != pattypes.PATPrefix { + return nil, nil, fmt.Errorf("PAT has wrong prefix") + } + secret := token[len(pattypes.PATPrefix) : len(pattypes.PATPrefix)+pattypes.PATSecretLength] + encodedChecksum := token[len(pattypes.PATPrefix)+pattypes.PATSecretLength : len(pattypes.PATPrefix)+pattypes.PATSecretLength+pattypes.PATChecksumLength] + + verificationChecksum, err := base62.Decode(encodedChecksum) + if err != nil { + return nil, nil, fmt.Errorf("PAT checksum decoding failed: %w", err) + } + + secretChecksum := crc32.ChecksumIEEE([]byte(secret)) + if secretChecksum != verificationChecksum { + return nil, nil, fmt.Errorf("PAT checksum does not match") + } + + hashedToken := sha256.Sum256([]byte(token)) + encodedHashedToken := base64.StdEncoding.EncodeToString(hashedToken[:]) + + var user *types.User + var pat *pattypes.PersonalAccessToken + + err = am.store.ExecuteInTransaction(ctx, func(transaction store.Store) error { + pat, err = transaction.GetPATByHashedToken(ctx, store.LockingStrengthShare, encodedHashedToken) + if err != nil { + return err + } + + user, err = transaction.GetUserByPATID(ctx, store.LockingStrengthShare, pat.ID) + return err + }) + if err != nil { + return nil, nil, err + } + + return user, pat, nil +} + +// userHasAllowedGroup checks if a user belongs to any of the allowed groups. +func userHasAllowedGroup(allowedGroups []string, userGroups []string) bool { + for _, userGroup := range userGroups { + for _, allowedGroup := range allowedGroups { + if userGroup == allowedGroup { + return true + } + } + } + return false +} diff --git a/internal/shared/auth/manager_mock.go b/internal/shared/auth/manager_mock.go new file mode 100644 index 0000000..bc70665 --- /dev/null +++ b/internal/shared/auth/manager_mock.go @@ -0,0 +1,54 @@ +package auth + +import ( + "context" + + "github.com/golang-jwt/jwt" + + nbcontext "github.com/netbirdio/netbird/management/server/context" + "github.com/netbirdio/netbird/management/server/types" +) + +var ( + _ Manager = (*MockManager)(nil) +) + +// @note really dislike this mocking approach but rather than have to do additional test refactoring. +type MockManager struct { + ValidateAndParseTokenFunc func(ctx context.Context, value string) (nbcontext.UserAuth, *jwt.Token, error) + EnsureUserAccessByJWTGroupsFunc func(ctx context.Context, userAuth nbcontext.UserAuth, token *jwt.Token) (nbcontext.UserAuth, error) + MarkPATUsedFunc func(ctx context.Context, tokenID string) error + GetPATInfoFunc func(ctx context.Context, token string) (user *types.User, pat *types.PersonalAccessToken, domain string, category string, err error) +} + +// EnsureUserAccessByJWTGroups implements Manager. +func (m *MockManager) EnsureUserAccessByJWTGroups(ctx context.Context, userAuth nbcontext.UserAuth, token *jwt.Token) (nbcontext.UserAuth, error) { + if m.EnsureUserAccessByJWTGroupsFunc != nil { + return m.EnsureUserAccessByJWTGroupsFunc(ctx, userAuth, token) + } + return nbcontext.UserAuth{}, nil +} + +// GetPATInfo implements Manager. +func (m *MockManager) GetPATInfo(ctx context.Context, token string) (user *types.User, pat *types.PersonalAccessToken, domain string, category string, err error) { + if m.GetPATInfoFunc != nil { + return m.GetPATInfoFunc(ctx, token) + } + return &types.User{}, &types.PersonalAccessToken{}, "", "", nil +} + +// MarkPATUsed implements Manager. +func (m *MockManager) MarkPATUsed(ctx context.Context, tokenID string) error { + if m.MarkPATUsedFunc != nil { + return m.MarkPATUsedFunc(ctx, tokenID) + } + return nil +} + +// ValidateAndParseToken implements Manager. +func (m *MockManager) ValidateAndParseToken(ctx context.Context, value string) (nbcontext.UserAuth, *jwt.Token, error) { + if m.ValidateAndParseTokenFunc != nil { + return m.ValidateAndParseTokenFunc(ctx, value) + } + return nbcontext.UserAuth{}, &jwt.Token{}, nil +} diff --git a/internal/shared/auth/manager_test.go b/internal/shared/auth/manager_test.go new file mode 100644 index 0000000..55fb1e3 --- /dev/null +++ b/internal/shared/auth/manager_test.go @@ -0,0 +1,407 @@ +package auth_test + +import ( + "context" + "crypto/sha256" + "encoding/base64" + "fmt" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" + "time" + + "github.com/golang-jwt/jwt" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/netbirdio/netbird/management/server/auth" + nbjwt "github.com/netbirdio/netbird/management/server/auth/jwt" + nbcontext "github.com/netbirdio/netbird/management/server/context" + "github.com/netbirdio/netbird/management/server/store" + "github.com/netbirdio/netbird/management/server/types" +) + +func TestAuthManager_GetAccountInfoFromPAT(t *testing.T) { + store, cleanup, err := store.NewTestStoreFromSQL(context.Background(), "", t.TempDir()) + if err != nil { + t.Fatalf("Error when creating store: %s", err) + } + t.Cleanup(cleanup) + + token := "nbp_9999EUDNdkeusjentDLSJEn1902u84390W6W" + hashedToken := sha256.Sum256([]byte(token)) + encodedHashedToken := base64.StdEncoding.EncodeToString(hashedToken[:]) + account := &types.Account{ + Id: "account_id", + Users: map[string]*types.User{"someUser": { + Id: "someUser", + PATs: map[string]*types.PersonalAccessToken{ + "tokenId": { + ID: "tokenId", + UserID: "someUser", + HashedToken: encodedHashedToken, + }, + }, + }}, + } + + err = store.SaveAccount(context.Background(), account) + if err != nil { + t.Fatalf("Error when saving account: %s", err) + } + + manager := auth.NewManager(store, "", "", "", "", []string{}, false) + + user, pat, _, _, err := manager.GetPATInfo(context.Background(), token) + if err != nil { + t.Fatalf("Error when getting Account from PAT: %s", err) + } + + assert.Equal(t, "account_id", user.AccountID) + assert.Equal(t, "someUser", user.Id) + assert.Equal(t, account.Users["someUser"].PATs["tokenId"].ID, pat.ID) +} + +func TestAuthManager_MarkPATUsed(t *testing.T) { + store, cleanup, err := store.NewTestStoreFromSQL(context.Background(), "", t.TempDir()) + if err != nil { + t.Fatalf("Error when creating store: %s", err) + } + t.Cleanup(cleanup) + + token := "nbp_9999EUDNdkeusjentDLSJEn1902u84390W6W" + hashedToken := sha256.Sum256([]byte(token)) + encodedHashedToken := base64.StdEncoding.EncodeToString(hashedToken[:]) + account := &types.Account{ + Id: "account_id", + Users: map[string]*types.User{"someUser": { + Id: "someUser", + PATs: map[string]*types.PersonalAccessToken{ + "tokenId": { + ID: "tokenId", + HashedToken: encodedHashedToken, + }, + }, + }}, + } + + err = store.SaveAccount(context.Background(), account) + if err != nil { + t.Fatalf("Error when saving account: %s", err) + } + + manager := auth.NewManager(store, "", "", "", "", []string{}, false) + + err = manager.MarkPATUsed(context.Background(), "tokenId") + if err != nil { + t.Fatalf("Error when marking PAT used: %s", err) + } + + account, err = store.GetAccount(context.Background(), "account_id") + if err != nil { + t.Fatalf("Error when getting account: %s", err) + } + assert.True(t, !account.Users["someUser"].PATs["tokenId"].GetLastUsed().IsZero()) +} + +func TestAuthManager_EnsureUserAccessByJWTGroups(t *testing.T) { + store, cleanup, err := store.NewTestStoreFromSQL(context.Background(), "", t.TempDir()) + if err != nil { + t.Fatalf("Error when creating store: %s", err) + } + t.Cleanup(cleanup) + + userId := "user-id" + domain := "test.domain" + + account := &types.Account{ + Id: "account_id", + Domain: domain, + Users: map[string]*types.User{"someUser": { + Id: "someUser", + }}, + Settings: &types.Settings{}, + } + + err = store.SaveAccount(context.Background(), account) + if err != nil { + t.Fatalf("Error when saving account: %s", err) + } + + // this has been validated and parsed by ValidateAndParseToken + userAuth := nbcontext.UserAuth{ + AccountId: account.Id, + Domain: domain, + UserId: userId, + DomainCategory: "test-category", + // Groups: []string{"group1", "group2"}, + } + + // these tests only assert groups are parsed from token as per account settings + token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{"idp-groups": []interface{}{"group1", "group2"}}) + + manager := auth.NewManager(store, "", "", "", "", []string{}, false) + + t.Run("JWT groups disabled", func(t *testing.T) { + userAuth, err := manager.EnsureUserAccessByJWTGroups(context.Background(), userAuth, token) + require.NoError(t, err, "ensure user access by JWT groups failed") + require.Len(t, userAuth.Groups, 0, "account not enabled to ensure access by groups") + }) + + t.Run("User impersonated", func(t *testing.T) { + userAuth, err := manager.EnsureUserAccessByJWTGroups(context.Background(), userAuth, token) + require.NoError(t, err, "ensure user access by JWT groups failed") + require.Len(t, userAuth.Groups, 0, "account not enabled to ensure access by groups") + }) + + t.Run("User PAT", func(t *testing.T) { + userAuth, err := manager.EnsureUserAccessByJWTGroups(context.Background(), userAuth, token) + require.NoError(t, err, "ensure user access by JWT groups failed") + require.Len(t, userAuth.Groups, 0, "account not enabled to ensure access by groups") + }) + + t.Run("JWT groups enabled without claim name", func(t *testing.T) { + account.Settings.JWTGroupsEnabled = true + err := store.SaveAccount(context.Background(), account) + require.NoError(t, err, "save account failed") + + userAuth, err := manager.EnsureUserAccessByJWTGroups(context.Background(), userAuth, token) + require.NoError(t, err, "ensure user access by JWT groups failed") + require.Len(t, userAuth.Groups, 0, "account missing groups claim name") + }) + + t.Run("JWT groups enabled without allowed groups", func(t *testing.T) { + account.Settings.JWTGroupsEnabled = true + account.Settings.JWTGroupsClaimName = "idp-groups" + err := store.SaveAccount(context.Background(), account) + require.NoError(t, err, "save account failed") + + userAuth, err := manager.EnsureUserAccessByJWTGroups(context.Background(), userAuth, token) + require.NoError(t, err, "ensure user access by JWT groups failed") + require.Equal(t, []string{"group1", "group2"}, userAuth.Groups, "group parsed do not match") + }) + + t.Run("User in allowed JWT groups", func(t *testing.T) { + account.Settings.JWTGroupsEnabled = true + account.Settings.JWTGroupsClaimName = "idp-groups" + account.Settings.JWTAllowGroups = []string{"group1"} + err := store.SaveAccount(context.Background(), account) + require.NoError(t, err, "save account failed") + + userAuth, err := manager.EnsureUserAccessByJWTGroups(context.Background(), userAuth, token) + require.NoError(t, err, "ensure user access by JWT groups failed") + + require.Equal(t, []string{"group1", "group2"}, userAuth.Groups, "group parsed do not match") + }) + + t.Run("User not in allowed JWT groups", func(t *testing.T) { + account.Settings.JWTGroupsEnabled = true + account.Settings.JWTGroupsClaimName = "idp-groups" + account.Settings.JWTAllowGroups = []string{"not-a-group"} + err := store.SaveAccount(context.Background(), account) + require.NoError(t, err, "save account failed") + + _, err = manager.EnsureUserAccessByJWTGroups(context.Background(), userAuth, token) + require.Error(t, err, "ensure user access is not in allowed groups") + }) +} + +func TestAuthManager_ValidateAndParseToken(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Add("Cache-Control", "max-age=30") // set a 30s expiry to these keys + http.ServeFile(w, r, "test_data/jwks.json") + })) + defer server.Close() + + issuer := "http://issuer.local" + audience := "http://audience.local" + userIdClaim := "" // defaults to "sub" + + // we're only testing with RSA256 + keyData, _ := os.ReadFile("test_data/sample_key") + key, _ := jwt.ParseRSAPrivateKeyFromPEM(keyData) + keyId := "test-key" + + // note, we can use a nil store because ValidateAndParseToken does not use it in it's flow + manager := auth.NewManager(nil, issuer, audience, server.URL, userIdClaim, []string{audience}, false) + + customClaim := func(name string) string { + return fmt.Sprintf("%s/%s", audience, name) + } + + lastLogin := time.Date(2025, 2, 12, 14, 25, 26, 0, time.UTC) //"2025-02-12T14:25:26.186Z" + + tests := []struct { + name string + tokenFunc func() string + expected *nbcontext.UserAuth // nil indicates expected error + }{ + { + name: "Valid with custom claims", + tokenFunc: func() string { + token := jwt.New(jwt.SigningMethodRS256) + token.Header["kid"] = keyId + token.Claims = jwt.MapClaims{ + "iss": issuer, + "aud": []string{audience}, + "iat": time.Now().Unix(), + "exp": time.Now().Add(time.Hour * 1).Unix(), + "sub": "user-id|123", + customClaim(nbjwt.AccountIDSuffix): "account-id|567", + customClaim(nbjwt.DomainIDSuffix): "http://localhost", + customClaim(nbjwt.DomainCategorySuffix): "private", + customClaim(nbjwt.LastLoginSuffix): lastLogin.Format(time.RFC3339), + customClaim(nbjwt.Invited): false, + } + tokenString, _ := token.SignedString(key) + return tokenString + }, + expected: &nbcontext.UserAuth{ + UserId: "user-id|123", + AccountId: "account-id|567", + Domain: "http://localhost", + DomainCategory: "private", + LastLogin: lastLogin, + Invited: false, + }, + }, + { + name: "Valid without custom claims", + tokenFunc: func() string { + token := jwt.New(jwt.SigningMethodRS256) + token.Header["kid"] = keyId + token.Claims = jwt.MapClaims{ + "iss": issuer, + "aud": []string{audience}, + "iat": time.Now().Unix(), + "exp": time.Now().Add(time.Hour).Unix(), + "sub": "user-id|123", + } + tokenString, _ := token.SignedString(key) + return tokenString + }, + expected: &nbcontext.UserAuth{ + UserId: "user-id|123", + }, + }, + { + name: "Expired token", + tokenFunc: func() string { + token := jwt.New(jwt.SigningMethodRS256) + token.Header["kid"] = keyId + token.Claims = jwt.MapClaims{ + "iss": issuer, + "aud": []string{audience}, + "iat": time.Now().Add(time.Hour * -2).Unix(), + "exp": time.Now().Add(time.Hour * -1).Unix(), + "sub": "user-id|123", + } + tokenString, _ := token.SignedString(key) + return tokenString + }, + }, + { + name: "Not yet valid", + tokenFunc: func() string { + token := jwt.New(jwt.SigningMethodRS256) + token.Header["kid"] = keyId + token.Claims = jwt.MapClaims{ + "iss": issuer, + "aud": []string{audience}, + "iat": time.Now().Add(time.Hour).Unix(), + "exp": time.Now().Add(time.Hour * 2).Unix(), + "sub": "user-id|123", + } + tokenString, _ := token.SignedString(key) + return tokenString + }, + }, + { + name: "Invalid signature", + tokenFunc: func() string { + token := jwt.New(jwt.SigningMethodRS256) + token.Header["kid"] = keyId + token.Claims = jwt.MapClaims{ + "iss": issuer, + "aud": []string{audience}, + "iat": time.Now().Unix(), + "exp": time.Now().Add(time.Hour).Unix(), + "sub": "user-id|123", + } + tokenString, _ := token.SignedString(key) + parts := strings.Split(tokenString, ".") + parts[2] = "invalid-signature" + return strings.Join(parts, ".") + }, + }, + { + name: "Invalid issuer", + tokenFunc: func() string { + token := jwt.New(jwt.SigningMethodRS256) + token.Header["kid"] = keyId + token.Claims = jwt.MapClaims{ + "iss": "not-the-issuer", + "aud": []string{audience}, + "iat": time.Now().Unix(), + "exp": time.Now().Add(time.Hour).Unix(), + "sub": "user-id|123", + } + tokenString, _ := token.SignedString(key) + return tokenString + }, + }, + { + name: "Invalid audience", + tokenFunc: func() string { + token := jwt.New(jwt.SigningMethodRS256) + token.Header["kid"] = keyId + token.Claims = jwt.MapClaims{ + "iss": issuer, + "aud": []string{"not-the-audience"}, + "iat": time.Now().Unix(), + "exp": time.Now().Add(time.Hour).Unix(), + "sub": "user-id|123", + } + tokenString, _ := token.SignedString(key) + return tokenString + }, + }, + { + name: "Invalid user claim", + tokenFunc: func() string { + token := jwt.New(jwt.SigningMethodRS256) + token.Header["kid"] = keyId + token.Claims = jwt.MapClaims{ + "iss": issuer, + "aud": []string{audience}, + "iat": time.Now().Unix(), + "exp": time.Now().Add(time.Hour).Unix(), + "not-sub": "user-id|123", + } + tokenString, _ := token.SignedString(key) + return tokenString + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tokenString := tt.tokenFunc() + + userAuth, token, err := manager.ValidateAndParseToken(context.Background(), tokenString) + + if tt.expected != nil { + assert.NoError(t, err) + assert.True(t, token.Valid) + assert.Equal(t, *tt.expected, userAuth) + } else { + assert.Error(t, err) + assert.Nil(t, token) + assert.Empty(t, userAuth) + } + }) + } + +} diff --git a/internal/shared/auth/test_data/jwks.json b/internal/shared/auth/test_data/jwks.json new file mode 100644 index 0000000..8080f55 --- /dev/null +++ b/internal/shared/auth/test_data/jwks.json @@ -0,0 +1,11 @@ +{ + "keys": [ + { + "kty": "RSA", + "kid": "test-key", + "use": "sig", + "n": "4f5wg5l2hKsTeNem_V41fGnJm6gOdrj8ym3rFkEU_wT8RDtnSgFEZOQpHEgQ7JL38xUfU0Y3g6aYw9QT0hJ7mCpz9Er5qLaMXJwZxzHzAahlfA0icqabvJOMvQtzD6uQv6wPEyZtDTWiQi9AXwBpHssPnpYGIn20ZZuNlX2BrClciHhCPUIIZOQn_MmqTD31jSyjoQoV7MhhMTATKJx2XrHhR-1DcKJzQBSTAGnpYVaqpsARap-nwRipr3nUTuxyGohBTSmjJ2usSeQXHI3bODIRe1AuTyHceAbewn8b462yEWKARdpd9AjQW5SIVPfdsz5B6GlYQ5LdYKtznTuy7w", + "e": "AQAB" + } + ] +} \ No newline at end of file diff --git a/internal/shared/auth/test_data/sample_key b/internal/shared/auth/test_data/sample_key new file mode 100644 index 0000000..e69284a --- /dev/null +++ b/internal/shared/auth/test_data/sample_key @@ -0,0 +1,27 @@ +-----BEGIN RSA PRIVATE KEY----- +MIIEowIBAAKCAQEA4f5wg5l2hKsTeNem/V41fGnJm6gOdrj8ym3rFkEU/wT8RDtn +SgFEZOQpHEgQ7JL38xUfU0Y3g6aYw9QT0hJ7mCpz9Er5qLaMXJwZxzHzAahlfA0i +cqabvJOMvQtzD6uQv6wPEyZtDTWiQi9AXwBpHssPnpYGIn20ZZuNlX2BrClciHhC +PUIIZOQn/MmqTD31jSyjoQoV7MhhMTATKJx2XrHhR+1DcKJzQBSTAGnpYVaqpsAR +ap+nwRipr3nUTuxyGohBTSmjJ2usSeQXHI3bODIRe1AuTyHceAbewn8b462yEWKA +Rdpd9AjQW5SIVPfdsz5B6GlYQ5LdYKtznTuy7wIDAQABAoIBAQCwia1k7+2oZ2d3 +n6agCAbqIE1QXfCmh41ZqJHbOY3oRQG3X1wpcGH4Gk+O+zDVTV2JszdcOt7E5dAy +MaomETAhRxB7hlIOnEN7WKm+dGNrKRvV0wDU5ReFMRHg31/Lnu8c+5BvGjZX+ky9 +POIhFFYJqwCRlopGSUIxmVj5rSgtzk3iWOQXr+ah1bjEXvlxDOWkHN6YfpV5ThdE +KdBIPGEVqa63r9n2h+qazKrtiRqJqGnOrHzOECYbRFYhexsNFz7YT02xdfSHn7gM +IvabDDP/Qp0PjE1jdouiMaFHYnLBbgvlnZW9yuVf/rpXTUq/njxIXMmvmEyyvSDn +FcFikB8pAoGBAPF77hK4m3/rdGT7X8a/gwvZ2R121aBcdPwEaUhvj/36dx596zvY +mEOjrWfZhF083/nYWE2kVquj2wjs+otCLfifEEgXcVPTnEOPO9Zg3uNSL0nNQghj +FuD3iGLTUBCtM66oTe0jLSslHe8gLGEQqyMzHOzYxNqibxcOZIe8Qt0NAoGBAO+U +I5+XWjWEgDmvyC3TrOSf/KCGjtu0TSv30ipv27bDLMrpvPmD/5lpptTFwcxvVhCs +2b+chCjlghFSWFbBULBrfci2FtliClOVMYrlNBdUSJhf3aYSG2Doe6Bgt1n2CpNn +/iu37Y3NfemZBJA7hNl4dYe+f+uzM87cdQ214+jrAoGAXA0XxX8ll2+ToOLJsaNT +OvNB9h9Uc5qK5X5w+7G7O998BN2PC/MWp8H+2fVqpXgNENpNXttkRm1hk1dych86 +EunfdPuqsX+as44oCyJGFHVBnWpm33eWQw9YqANRI+pCJzP08I5WK3osnPiwshd+ +hR54yjgfYhBFNI7B95PmEQkCgYBzFSz7h1+s34Ycr8SvxsOBWxymG5zaCsUbPsL0 +4aCgLScCHb9J+E86aVbbVFdglYa5Id7DPTL61ixhl7WZjujspeXZGSbmq0Kcnckb +mDgqkLECiOJW2NHP/j0McAkDLL4tysF8TLDO8gvuvzNC+WQ6drO2ThrypLVZQ+ry +eBIPmwKBgEZxhqa0gVvHQG/7Od69KWj4eJP28kq13RhKay8JOoN0vPmspXJo1HY3 +CKuHRG+AP579dncdUnOMvfXOtkdM4vk0+hWASBQzM9xzVcztCa+koAugjVaLS9A+ +9uQoqEeVNTckxx0S2bYevRy7hGQmUJTyQm3j1zEUR5jpdbL83Fbq +-----END RSA PRIVATE KEY----- \ No newline at end of file diff --git a/internal/shared/auth/test_data/sample_key.pub b/internal/shared/auth/test_data/sample_key.pub new file mode 100644 index 0000000..d5b7f71 --- /dev/null +++ b/internal/shared/auth/test_data/sample_key.pub @@ -0,0 +1,9 @@ +-----BEGIN PUBLIC KEY----- +MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA4f5wg5l2hKsTeNem/V41 +fGnJm6gOdrj8ym3rFkEU/wT8RDtnSgFEZOQpHEgQ7JL38xUfU0Y3g6aYw9QT0hJ7 +mCpz9Er5qLaMXJwZxzHzAahlfA0icqabvJOMvQtzD6uQv6wPEyZtDTWiQi9AXwBp +HssPnpYGIn20ZZuNlX2BrClciHhCPUIIZOQn/MmqTD31jSyjoQoV7MhhMTATKJx2 +XrHhR+1DcKJzQBSTAGnpYVaqpsARap+nwRipr3nUTuxyGohBTSmjJ2usSeQXHI3b +ODIRe1AuTyHceAbewn8b462yEWKARdpd9AjQW5SIVPfdsz5B6GlYQ5LdYKtznTuy +7wIDAQAB +-----END PUBLIC KEY----- \ No newline at end of file diff --git a/internal/shared/db/config.go b/internal/shared/db/config.go index d4b2462..99a06b0 100644 --- a/internal/shared/db/config.go +++ b/internal/shared/db/config.go @@ -1,6 +1,7 @@ package db type config struct { - Engine string `env:"NB_STORE_ENGINE" envDefault:"sqlite"` - PostgresDsnEnv string `env:"NB_STORE_ENGINE_POSTGRES_DSN" envDefault:""` + Engine string `env:"NB_STORE_ENGINE" envDefault:"sqlite"` + PostgresDsn string `env:"NB_STORE_ENGINE_POSTGRES_DSN" envDefault:""` + DataDir string `env:"NB_STORE_DATA_DIR" envDefault:"/var/lib/netbird"` } diff --git a/internal/shared/db/database_connection.go b/internal/shared/db/database_connection.go index d3fff24..68d0a63 100644 --- a/internal/shared/db/database_connection.go +++ b/internal/shared/db/database_connection.go @@ -3,7 +3,6 @@ package db import ( "context" "fmt" - "os" "path/filepath" "runtime" @@ -18,8 +17,6 @@ import ( const ( storeSqliteFileName = "licenses.db" - storeDataDirEnv = "NB_STORE_DATA_DIR" - storeDefaultDataDir = "/var/lib/netbird" ) // DatabaseConn is a wrapper around the gorm database connection @@ -39,7 +36,7 @@ func NewDatabaseConn(ctx context.Context) (*DatabaseConn, error) { var db *gorm.DB switch Engine(cfg.Engine) { case SqliteStoreEngine: - db, err = openSQLiteDB() + db, err = openSQLiteDB(cfg) case PostgresStoreEngine: db, err = openPostgresDB(cfg) case MemoryStoreEngine: @@ -58,6 +55,9 @@ func NewDatabaseConn(ctx context.Context) (*DatabaseConn, error) { } conns := runtime.NumCPU() + if Engine(cfg.Engine) == SqliteStoreEngine { + conns = 1 + } sql.SetMaxOpenConns(conns) return &DatabaseConn{ DB: db, @@ -75,19 +75,14 @@ func openMemoryDB() (*gorm.DB, error) { } // openSQLiteDB opens a new connection to a SQLite database -func openSQLiteDB() (*gorm.DB, error) { +func openSQLiteDB(cfg *config) (*gorm.DB, error) { storeStr := fmt.Sprintf("%s?cache=shared", storeSqliteFileName) if runtime.GOOS == "windows" { // To avoid `The process cannot access the file because it is being used by another process` on Windows storeStr = storeSqliteFileName } - dataDir, ok := os.LookupEnv(storeDataDirEnv) - if !ok { - dataDir = storeDefaultDataDir - } - - file := filepath.Join(dataDir, storeStr) + file := filepath.Join(cfg.DataDir, storeStr) db, err := gorm.Open(sqlite.Open(file), getGormConfig()) if err != nil { return nil, err @@ -98,12 +93,7 @@ func openSQLiteDB() (*gorm.DB, error) { // openPostgresDB opens a new connection to a Postgres database func openPostgresDB(cfg *config) (*gorm.DB, error) { - dsn, ok := os.LookupEnv(cfg.PostgresDsnEnv) - if !ok { - return nil, fmt.Errorf("%s is not set", cfg.PostgresDsnEnv) - } - - db, err := gorm.Open(postgres.Open(dsn), getGormConfig()) + db, err := gorm.Open(postgres.Open(cfg.PostgresDsn), getGormConfig()) if err != nil { return nil, err }