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/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) + }) + } +} 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() 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") + } +}