From 6f0fd1d1b33dddaf4a927cede4f6854765431ea6 Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Thu, 19 Sep 2024 13:49:09 +0200 Subject: [PATCH 1/3] - Increase queue size and drop the overflowed messages (#2617) - Explicit close the net.Conn in user space wgProxy when close the wgProxy - Add extra logs --- client/internal/peer/conn.go | 9 +++++--- client/internal/wgproxy/proxy_userspace.go | 18 +++++++++++---- relay/client/client.go | 27 ++++++++++++++++------ 3 files changed, 40 insertions(+), 14 deletions(-) diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 5327e31d2..911ddd228 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -518,6 +518,9 @@ func (conn *Conn) relayConnectionIsReady(rci RelayConnInfo) { defer conn.mu.Unlock() if conn.ctx.Err() != nil { + if err := rci.relayedConn.Close(); err != nil { + log.Warnf("failed to close unnecessary relayed connection: %v", err) + } return } @@ -530,6 +533,7 @@ func (conn *Conn) relayConnectionIsReady(rci RelayConnInfo) { conn.log.Errorf("failed to add relayed net.Conn to local proxy: %v", err) return } + conn.log.Infof("created new wgProxy for relay connection: %s", endpoint) endpointUdpAddr, _ := net.ResolveUDPAddr(endpoint.Network(), endpoint.String()) conn.endpointRelay = endpointUdpAddr @@ -775,9 +779,8 @@ func (conn *Conn) getEndpointForICEConnInfo(iceConnInfo ICEConnInfo) (net.Addr, ep, err := wgProxy.AddTurnConn(iceConnInfo.RemoteConn) if err != nil { conn.log.Errorf("failed to add turn net.Conn to local proxy: %v", err) - err = wgProxy.CloseConn() - if err != nil { - conn.log.Warnf("failed to close turn proxy connection: %v", err) + if errClose := wgProxy.CloseConn(); errClose != nil { + conn.log.Warnf("failed to close turn proxy connection: %v", errClose) } return nil, nil, err } diff --git a/client/internal/wgproxy/proxy_userspace.go b/client/internal/wgproxy/proxy_userspace.go index c2c8a9b51..701f615b9 100644 --- a/client/internal/wgproxy/proxy_userspace.go +++ b/client/internal/wgproxy/proxy_userspace.go @@ -32,8 +32,8 @@ func NewWGUserSpaceProxy(ctx context.Context, wgPort int) *WGUserSpaceProxy { } // AddTurnConn start the proxy with the given remote conn -func (p *WGUserSpaceProxy) AddTurnConn(turnConn net.Conn) (net.Addr, error) { - p.remoteConn = turnConn +func (p *WGUserSpaceProxy) AddTurnConn(remoteConn net.Conn) (net.Addr, error) { + p.remoteConn = remoteConn var err error p.localConn, err = nbnet.NewDialer().DialContext(p.ctx, "udp", fmt.Sprintf(":%d", p.localWGListenPort)) @@ -54,6 +54,14 @@ func (p *WGUserSpaceProxy) CloseConn() error { if p.localConn == nil { return nil } + + if p.remoteConn == nil { + return nil + } + + if err := p.remoteConn.Close(); err != nil { + log.Warnf("failed to close remote conn: %s", err) + } return p.localConn.Close() } @@ -65,6 +73,8 @@ func (p *WGUserSpaceProxy) Free() error { // proxyToRemote proxies everything from Wireguard to the RemoteKey peer // blocks func (p *WGUserSpaceProxy) proxyToRemote() { + defer log.Infof("exit from proxyToRemote: %s", p.localConn.LocalAddr()) + buf := make([]byte, 1500) for { select { @@ -93,7 +103,8 @@ func (p *WGUserSpaceProxy) proxyToRemote() { // proxyToLocal proxies everything from the RemoteKey peer to local Wireguard // blocks func (p *WGUserSpaceProxy) proxyToLocal() { - + defer p.cancel() + defer log.Infof("exit from proxyToLocal: %s", p.localConn.LocalAddr()) buf := make([]byte, 1500) for { select { @@ -103,7 +114,6 @@ func (p *WGUserSpaceProxy) proxyToLocal() { n, err := p.remoteConn.Read(buf) if err != nil { if err == io.EOF { - p.cancel() return } log.Errorf("failed to read from remote conn: %s", err) diff --git a/relay/client/client.go b/relay/client/client.go index e431c029d..90bc3ac41 100644 --- a/relay/client/client.go +++ b/relay/client/client.go @@ -58,7 +58,10 @@ func (m *Msg) Free() { m.bufPool.Put(m.bufPtr) } +// connContainer is a container for the connection to the peer. It is responsible for managing the messages from the +// server and forwarding them to the upper layer content reader. type connContainer struct { + log *log.Entry conn *Conn messages chan Msg msgChanLock sync.Mutex @@ -67,10 +70,10 @@ type connContainer struct { cancel context.CancelFunc } -func newConnContainer(conn *Conn, messages chan Msg) *connContainer { +func newConnContainer(log *log.Entry, conn *Conn, messages chan Msg) *connContainer { ctx, cancel := context.WithCancel(context.Background()) - return &connContainer{ + log: log, conn: conn, messages: messages, ctx: ctx, @@ -91,6 +94,10 @@ func (cc *connContainer) writeMsg(msg Msg) { case cc.messages <- msg: case <-cc.ctx.Done(): msg.Free() + default: + msg.Free() + cc.log.Infof("message queue is full") + // todo consider to close the connection } } @@ -141,8 +148,8 @@ type Client struct { // NewClient creates a new client for the relay server. The client is not connected to the server until the Connect func NewClient(ctx context.Context, serverURL string, authTokenStore *auth.TokenStore, peerID string) *Client { hashedID, hashedStringId := messages.HashID(peerID) - return &Client{ - log: log.WithFields(log.Fields{"client_id": hashedStringId, "relay": serverURL}), + c := &Client{ + log: log.WithFields(log.Fields{"relay": serverURL}), parentCtx: ctx, connectionURL: serverURL, authTokenStore: authTokenStore, @@ -155,6 +162,8 @@ func NewClient(ctx context.Context, serverURL string, authTokenStore *auth.Token }, conns: make(map[string]*connContainer), } + c.log.Infof("create new relay connection: local peerID: %s, local peer hashedID: %s", peerID, hashedStringId) + return c } // Connect establishes a connection to the relay server. It blocks until the connection is established or an error occurs. @@ -203,10 +212,10 @@ func (c *Client) OpenConn(dstPeerID string) (net.Conn, error) { } c.log.Infof("open connection to peer: %s", hashedStringID) - msgChannel := make(chan Msg, 2) + msgChannel := make(chan Msg, 100) conn := NewConn(c, hashedID, hashedStringID, msgChannel, c.instanceURL) - c.conns[hashedStringID] = newConnContainer(conn, msgChannel) + c.conns[hashedStringID] = newConnContainer(c.log, conn, msgChannel) return conn, nil } @@ -455,7 +464,10 @@ func (c *Client) listenForStopEvents(hc *healthcheck.Receiver, conn net.Conn, in } c.log.Errorf("health check timeout") internalStopFlag.set() - _ = conn.Close() // ignore the err because the readLoop will handle it + if err := conn.Close(); err != nil { + // ignore the err handling because the readLoop will handle it + c.log.Warnf("failed to close connection: %s", err) + } return case <-c.parentCtx.Done(): err := c.close(true) @@ -486,6 +498,7 @@ func (c *Client) closeConn(connReference *Conn, id string) error { if container.conn != connReference { return fmt.Errorf("conn reference mismatch") } + c.log.Infof("free up connection to peer: %s", id) delete(c.conns, id) container.close() From fc4b37f7bcdc2de36f279c458ce79da312d8d29e Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Thu, 19 Sep 2024 13:49:28 +0200 Subject: [PATCH 2/3] Exit from processConnResults after all tries (#2621) * Exit from processConnResults after all tries If all server is unavailable then the server picker never return because we never close the result channel. Count the number of the results and exit when we reached the expected size --- relay/client/picker.go | 10 +++++++--- relay/client/picker_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 3 deletions(-) create mode 100644 relay/client/picker_test.go diff --git a/relay/client/picker.go b/relay/client/picker.go index b0888a4a0..13b0547aa 100644 --- a/relay/client/picker.go +++ b/relay/client/picker.go @@ -35,12 +35,15 @@ func (sp *ServerPicker) PickServer(parentCtx context.Context, urls []string) (*C connResultChan := make(chan connResult, totalServers) successChan := make(chan connResult, 1) - concurrentLimiter := make(chan struct{}, maxConcurrentServers) + for _, url := range urls { + // todo check if we have a successful connection so we do not need to connect to other servers concurrentLimiter <- struct{}{} go func(url string) { - defer func() { <-concurrentLimiter }() + defer func() { + <-concurrentLimiter + }() sp.startConnection(parentCtx, connResultChan, url) }(url) } @@ -72,7 +75,8 @@ func (sp *ServerPicker) startConnection(ctx context.Context, resultChan chan con func (sp *ServerPicker) processConnResults(resultChan chan connResult, successChan chan connResult) { var hasSuccess bool - for cr := range resultChan { + for numOfResults := 0; numOfResults < cap(resultChan); numOfResults++ { + cr := <-resultChan if cr.Err != nil { log.Debugf("failed to connect to Relay server: %s: %v", cr.Url, cr.Err) continue diff --git a/relay/client/picker_test.go b/relay/client/picker_test.go new file mode 100644 index 000000000..f5649d700 --- /dev/null +++ b/relay/client/picker_test.go @@ -0,0 +1,31 @@ +package client + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestServerPicker_UnavailableServers(t *testing.T) { + sp := ServerPicker{ + TokenStore: nil, + PeerID: "test", + } + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + go func() { + _, err := sp.PickServer(ctx, []string{"rel://dummy1", "rel://dummy2"}) + if err == nil { + t.Error(err) + } + cancel() + }() + + <-ctx.Done() + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + t.Errorf("PickServer() took too long to complete") + } +} From 35c892aea365fa9e3e710ecc45dc7c0a353b8c71 Mon Sep 17 00:00:00 2001 From: Bethuel Mmbaga Date: Fri, 20 Sep 2024 12:36:58 +0300 Subject: [PATCH 3/3] [management] Restrict accessible peers to user-owned peers for non-admins (#2618) * Restrict accessible peers to user-owned peers for non-admin users Signed-off-by: bcmmbaga * add tests Signed-off-by: bcmmbaga * add service user test Signed-off-by: bcmmbaga * reuse account from token Signed-off-by: bcmmbaga * return error when peer not found Signed-off-by: bcmmbaga --------- Signed-off-by: bcmmbaga --- management/server/http/peers_handler.go | 20 +- management/server/http/peers_handler_test.go | 194 +++++++++++++++++-- 2 files changed, 198 insertions(+), 16 deletions(-) diff --git a/management/server/http/peers_handler.go b/management/server/http/peers_handler.go index 1487bbc39..5a2190d83 100644 --- a/management/server/http/peers_handler.go +++ b/management/server/http/peers_handler.go @@ -7,8 +7,6 @@ import ( "net/http" "github.com/gorilla/mux" - log "github.com/sirupsen/logrus" - "github.com/netbirdio/netbird/management/server" nbgroup "github.com/netbirdio/netbird/management/server/group" "github.com/netbirdio/netbird/management/server/http/api" @@ -16,6 +14,7 @@ import ( "github.com/netbirdio/netbird/management/server/jwtclaims" nbpeer "github.com/netbirdio/netbird/management/server/peer" "github.com/netbirdio/netbird/management/server/status" + log "github.com/sirupsen/logrus" ) // PeersHandler is a handler that returns peers of the account @@ -215,7 +214,7 @@ func (h *PeersHandler) setApprovalRequiredFlag(respBody []*api.PeerBatch, approv // GetAccessiblePeers returns a list of all peers that the specified peer can connect to within the network. func (h *PeersHandler) GetAccessiblePeers(w http.ResponseWriter, r *http.Request) { claims := h.claimsExtractor.FromRequestContext(r) - account, _, err := h.accountManager.GetAccountFromToken(r.Context(), claims) + account, user, err := h.accountManager.GetAccountFromToken(r.Context(), claims) if err != nil { util.WriteError(r.Context(), err, w) return @@ -228,6 +227,21 @@ func (h *PeersHandler) GetAccessiblePeers(w http.ResponseWriter, r *http.Request return } + // If the user is regular user and does not own the peer + // with the given peerID return an empty list + if !user.HasAdminPower() && !user.IsServiceUser { + peer, ok := account.Peers[peerID] + if !ok { + util.WriteError(r.Context(), status.Errorf(status.NotFound, "peer not found"), w) + return + } + + if peer.UserID != user.Id { + util.WriteJSONObject(r.Context(), w, []api.AccessiblePeer{}) + return + } + } + dnsDomain := h.accountManager.GetDNSDomain() validPeers, err := h.accountManager.GetValidatedPeers(account) diff --git a/management/server/http/peers_handler_test.go b/management/server/http/peers_handler_test.go index 153c8f03a..dae264fff 100644 --- a/management/server/http/peers_handler_test.go +++ b/management/server/http/peers_handler_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "io" "net" "net/http" @@ -12,20 +13,30 @@ import ( "time" "github.com/gorilla/mux" - + nbgroup "github.com/netbirdio/netbird/management/server/group" "github.com/netbirdio/netbird/management/server/http/api" nbpeer "github.com/netbirdio/netbird/management/server/peer" + "golang.org/x/exp/maps" "github.com/netbirdio/netbird/management/server/jwtclaims" - "github.com/magiconair/properties/assert" + "github.com/stretchr/testify/assert" "github.com/netbirdio/netbird/management/server" "github.com/netbirdio/netbird/management/server/mock_server" ) -const testPeerID = "test_peer" -const noUpdateChannelTestPeerID = "no-update-channel" +type ctxKey string + +const ( + testPeerID = "test_peer" + noUpdateChannelTestPeerID = "no-update-channel" + + adminUser = "admin_user" + regularUser = "regular_user" + serviceUser = "service_user" + userIDKey ctxKey = "user_id" +) func initTestMetaData(peers ...*nbpeer.Peer) *PeersHandler { return &PeersHandler{ @@ -60,21 +71,57 @@ func initTestMetaData(peers ...*nbpeer.Peer) *PeersHandler { return "netbird.selfhosted" }, GetAccountFromTokenFunc: func(_ context.Context, claims jwtclaims.AuthorizationClaims) (*server.Account, *server.User, error) { - user := server.NewAdminUser("test_user") - return &server.Account{ + peersMap := make(map[string]*nbpeer.Peer) + for _, peer := range peers { + peersMap[peer.ID] = peer.Copy() + } + + policy := &server.Policy{ + ID: "policy", + AccountID: claims.AccountId, + Name: "policy", + Enabled: true, + Rules: []*server.PolicyRule{ + { + ID: "rule", + Name: "rule", + Enabled: true, + Action: "accept", + Destinations: []string{"group1"}, + Sources: []string{"group1"}, + Bidirectional: true, + Protocol: "all", + Ports: []string{"80"}, + }, + }, + } + + srvUser := server.NewRegularUser(serviceUser) + srvUser.IsServiceUser = true + + account := &server.Account{ Id: claims.AccountId, Domain: "hotmail.com", - Peers: map[string]*nbpeer.Peer{ - peers[0].ID: peers[0], - peers[1].ID: peers[1], - }, + Peers: peersMap, Users: map[string]*server.User{ - "test_user": user, + adminUser: server.NewAdminUser(adminUser), + regularUser: server.NewRegularUser(regularUser), + serviceUser: srvUser, + }, + Groups: map[string]*nbgroup.Group{ + "group1": { + ID: "group1", + AccountID: claims.AccountId, + Name: "group1", + Issued: "api", + Peers: maps.Keys(peersMap), + }, }, Settings: &server.Settings{ PeerLoginExpirationEnabled: true, PeerLoginExpiration: time.Hour, }, + Policies: []*server.Policy{policy}, Network: &server.Network{ Identifier: "ciclqisab2ss43jdn8q0", Net: net.IPNet{ @@ -83,7 +130,9 @@ func initTestMetaData(peers ...*nbpeer.Peer) *PeersHandler { }, Serial: 51, }, - }, user, nil + } + + return account, account.Users[claims.UserId], nil }, HasConnectedChannelFunc: func(peerID string) bool { statuses := make(map[string]struct{}) @@ -99,8 +148,9 @@ func initTestMetaData(peers ...*nbpeer.Peer) *PeersHandler { }, claimsExtractor: jwtclaims.NewClaimsExtractor( jwtclaims.WithFromRequestContext(func(r *http.Request) jwtclaims.AuthorizationClaims { + userID := r.Context().Value(userIDKey).(string) return jwtclaims.AuthorizationClaims{ - UserId: "test_user", + UserId: userID, Domain: "hotmail.com", AccountId: "test_id", } @@ -197,6 +247,8 @@ func TestGetPeers(t *testing.T) { recorder := httptest.NewRecorder() req := httptest.NewRequest(tc.requestType, tc.requestPath, tc.requestBody) + ctx := context.WithValue(context.Background(), userIDKey, "admin_user") + req = req.WithContext(ctx) router := mux.NewRouter() router.HandleFunc("/api/peers/", p.GetAllPeers).Methods("GET") @@ -251,3 +303,119 @@ func TestGetPeers(t *testing.T) { }) } } + +func TestGetAccessiblePeers(t *testing.T) { + peer1 := &nbpeer.Peer{ + ID: "peer1", + Key: "key1", + IP: net.ParseIP("100.64.0.1"), + Status: &nbpeer.PeerStatus{Connected: true}, + Name: "peer1", + LoginExpirationEnabled: false, + UserID: regularUser, + } + + peer2 := &nbpeer.Peer{ + ID: "peer2", + Key: "key2", + IP: net.ParseIP("100.64.0.2"), + Status: &nbpeer.PeerStatus{Connected: true}, + Name: "peer2", + LoginExpirationEnabled: false, + UserID: adminUser, + } + + peer3 := &nbpeer.Peer{ + ID: "peer3", + Key: "key3", + IP: net.ParseIP("100.64.0.3"), + Status: &nbpeer.PeerStatus{Connected: true}, + Name: "peer3", + LoginExpirationEnabled: false, + UserID: regularUser, + } + + p := initTestMetaData(peer1, peer2, peer3) + + tt := []struct { + name string + peerID string + callerUserID string + expectedStatus int + expectedPeers []string + }{ + { + name: "non admin user can access owned peer", + peerID: "peer1", + callerUserID: regularUser, + expectedStatus: http.StatusOK, + expectedPeers: []string{"peer2", "peer3"}, + }, + { + name: "non admin user can't access unowned peer", + peerID: "peer2", + callerUserID: regularUser, + expectedStatus: http.StatusOK, + expectedPeers: []string{}, + }, + { + name: "admin user can access owned peer", + peerID: "peer2", + callerUserID: adminUser, + expectedStatus: http.StatusOK, + expectedPeers: []string{"peer1", "peer3"}, + }, + { + name: "admin user can access unowned peer", + peerID: "peer3", + callerUserID: adminUser, + expectedStatus: http.StatusOK, + expectedPeers: []string{"peer1", "peer2"}, + }, + { + name: "service user can access unowned peer", + peerID: "peer3", + callerUserID: serviceUser, + expectedStatus: http.StatusOK, + expectedPeers: []string{"peer1", "peer2"}, + }, + } + + for _, tc := range tt { + t.Run(tc.name, func(t *testing.T) { + + recorder := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/peers/%s/accessible-peers", tc.peerID), nil) + ctx := context.WithValue(context.Background(), userIDKey, tc.callerUserID) + req = req.WithContext(ctx) + + router := mux.NewRouter() + router.HandleFunc("/api/peers/{peerId}/accessible-peers", p.GetAccessiblePeers).Methods("GET") + router.ServeHTTP(recorder, req) + + res := recorder.Result() + if res.StatusCode != tc.expectedStatus { + t.Fatalf("handler returned wrong status code: got %v want %v", res.StatusCode, tc.expectedStatus) + } + + body, err := io.ReadAll(res.Body) + if err != nil { + t.Fatalf("failed to read response body: %v", err) + } + defer res.Body.Close() + + var accessiblePeers []api.AccessiblePeer + err = json.Unmarshal(body, &accessiblePeers) + if err != nil { + t.Fatalf("failed to unmarshal response: %v", err) + } + + peerIDs := make([]string, len(accessiblePeers)) + for i, peer := range accessiblePeers { + peerIDs[i] = peer.Id + } + + assert.ElementsMatch(t, peerIDs, tc.expectedPeers) + }) + } +}