Merge branch 'main' into feature/add-serial-to-proxy

2026-10-03 20:19:07 +02:00 · 2026-02-24 18:09:14 +01:00
parent b450fa2cca f341d69314
commit dc136c17e9
63 changed files with 4862 additions and 805 deletions
@@ -0,0 +1,192 @@
+package grpc
+
+import (
+	"context"
+
+	pb "github.com/golang/protobuf/proto" // nolint
+	log "github.com/sirupsen/logrus"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/netbirdio/netbird/encryption"
+	rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
+	nbContext "github.com/netbirdio/netbird/management/server/context"
+	nbpeer "github.com/netbirdio/netbird/management/server/peer"
+	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/shared/management/proto"
+	internalStatus "github.com/netbirdio/netbird/shared/management/status"
+)
+
+// CreateExpose handles a peer request to create a new expose service.
+func (s *Server) CreateExpose(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
+	exposeReq := &proto.ExposeServiceRequest{}
+	peerKey, err := s.parseRequest(ctx, req, exposeReq)
+	if err != nil {
+		return nil, err
+	}
+
+	accountID, peer, err := s.authenticateExposePeer(ctx, peerKey)
+	if err != nil {
+		return nil, err
+	}
+
+	// nolint:staticcheck
+	ctx = context.WithValue(ctx, nbContext.AccountIDKey, accountID)
+
+	reverseProxyMgr := s.getReverseProxyManager()
+	if reverseProxyMgr == nil {
+		return nil, status.Errorf(codes.Internal, "reverse proxy manager not available")
+	}
+
+	created, err := reverseProxyMgr.CreateServiceFromPeer(ctx, accountID, peer.ID, &rpservice.ExposeServiceRequest{
+		NamePrefix: exposeReq.NamePrefix,
+		Port:       int(exposeReq.Port),
+		Protocol:   exposeProtocolToString(exposeReq.Protocol),
+		Domain:     exposeReq.Domain,
+		Pin:        exposeReq.Pin,
+		Password:   exposeReq.Password,
+		UserGroups: exposeReq.UserGroups,
+	})
+	if err != nil {
+		return nil, mapExposeError(ctx, err)
+	}
+
+	return s.encryptResponse(peerKey, &proto.ExposeServiceResponse{
+		ServiceName: created.ServiceName,
+		ServiceUrl:  created.ServiceURL,
+		Domain:      created.Domain,
+	})
+}
+
+// RenewExpose extends the TTL of an active expose session.
+func (s *Server) RenewExpose(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
+	renewReq := &proto.RenewExposeRequest{}
+	peerKey, err := s.parseRequest(ctx, req, renewReq)
+	if err != nil {
+		return nil, err
+	}
+
+	accountID, peer, err := s.authenticateExposePeer(ctx, peerKey)
+	if err != nil {
+		return nil, err
+	}
+
+	reverseProxyMgr := s.getReverseProxyManager()
+	if reverseProxyMgr == nil {
+		return nil, status.Errorf(codes.Internal, "reverse proxy manager not available")
+	}
+
+	if err := reverseProxyMgr.RenewServiceFromPeer(ctx, accountID, peer.ID, renewReq.Domain); err != nil {
+		return nil, mapExposeError(ctx, err)
+	}
+
+	return s.encryptResponse(peerKey, &proto.RenewExposeResponse{})
+}
+
+// StopExpose terminates an active expose session.
+func (s *Server) StopExpose(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
+	stopReq := &proto.StopExposeRequest{}
+	peerKey, err := s.parseRequest(ctx, req, stopReq)
+	if err != nil {
+		return nil, err
+	}
+
+	accountID, peer, err := s.authenticateExposePeer(ctx, peerKey)
+	if err != nil {
+		return nil, err
+	}
+
+	reverseProxyMgr := s.getReverseProxyManager()
+	if reverseProxyMgr == nil {
+		return nil, status.Errorf(codes.Internal, "reverse proxy manager not available")
+	}
+
+	if err := reverseProxyMgr.StopServiceFromPeer(ctx, accountID, peer.ID, stopReq.Domain); err != nil {
+		return nil, mapExposeError(ctx, err)
+	}
+
+	return s.encryptResponse(peerKey, &proto.StopExposeResponse{})
+}
+
+func mapExposeError(ctx context.Context, err error) error {
+	s, ok := internalStatus.FromError(err)
+	if !ok {
+		log.WithContext(ctx).Errorf("expose service error: %v", err)
+		return status.Errorf(codes.Internal, "internal error")
+	}
+
+	switch s.Type() {
+	case internalStatus.InvalidArgument:
+		return status.Errorf(codes.InvalidArgument, "%s", s.Message)
+	case internalStatus.PermissionDenied:
+		return status.Errorf(codes.PermissionDenied, "%s", s.Message)
+	case internalStatus.NotFound:
+		return status.Errorf(codes.NotFound, "%s", s.Message)
+	case internalStatus.AlreadyExists:
+		return status.Errorf(codes.AlreadyExists, "%s", s.Message)
+	case internalStatus.PreconditionFailed:
+		return status.Errorf(codes.ResourceExhausted, "%s", s.Message)
+	default:
+		log.WithContext(ctx).Errorf("expose service error: %v", err)
+		return status.Errorf(codes.Internal, "internal error")
+	}
+}
+
+func (s *Server) encryptResponse(peerKey wgtypes.Key, msg pb.Message) (*proto.EncryptedMessage, error) {
+	wgKey, err := s.secretsManager.GetWGKey()
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "internal error")
+	}
+
+	encryptedResp, err := encryption.EncryptMessage(peerKey, wgKey, msg)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "encrypt response")
+	}
+
+	return &proto.EncryptedMessage{
+		WgPubKey: wgKey.PublicKey().String(),
+		Body:     encryptedResp,
+	}, nil
+}
+
+func (s *Server) authenticateExposePeer(ctx context.Context, peerKey wgtypes.Key) (string, *nbpeer.Peer, error) {
+	accountID, err := s.accountManager.GetAccountIDForPeerKey(ctx, peerKey.String())
+	if err != nil {
+		if errStatus, ok := internalStatus.FromError(err); ok && errStatus.Type() == internalStatus.NotFound {
+			return "", nil, status.Errorf(codes.PermissionDenied, "peer is not registered")
+		}
+		return "", nil, status.Errorf(codes.Internal, "lookup account for peer")
+	}
+
+	peer, err := s.accountManager.GetStore().GetPeerByPeerPubKey(ctx, store.LockingStrengthNone, peerKey.String())
+	if err != nil {
+		return "", nil, status.Errorf(codes.PermissionDenied, "peer is not registered")
+	}
+
+	return accountID, peer, nil
+}
+
+func (s *Server) getReverseProxyManager() rpservice.Manager {
+	s.reverseProxyMu.RLock()
+	defer s.reverseProxyMu.RUnlock()
+	return s.reverseProxyManager
+}
+
+// SetReverseProxyManager sets the reverse proxy manager on the server.
+func (s *Server) SetReverseProxyManager(mgr rpservice.Manager) {
+	s.reverseProxyMu.Lock()
+	defer s.reverseProxyMu.Unlock()
+	s.reverseProxyManager = mgr
+}
+
+func exposeProtocolToString(p proto.ExposeProtocol) string {
+	switch p {
+	case proto.ExposeProtocol_EXPOSE_HTTP:
+		return "http"
+	case proto.ExposeProtocol_EXPOSE_HTTPS:
+		return "https"
+	default:
+		return "http"
+	}
+}
@@ -76,6 +76,20 @@ func (m *mockReverseProxyManager) GetServiceIDByTargetID(_ context.Context, _, _
 	return "", nil
 }

+func (m *mockReverseProxyManager) CreateServiceFromPeer(_ context.Context, _, _ string, _ *service.ExposeServiceRequest) (*service.ExposeServiceResponse, error) {
+	return &service.ExposeServiceResponse{}, nil
+}
+
+func (m *mockReverseProxyManager) RenewServiceFromPeer(_ context.Context, _, _, _ string) error {
+	return nil
+}
+
+func (m *mockReverseProxyManager) StopServiceFromPeer(_ context.Context, _, _, _ string) error {
+	return nil
+}
+
+func (m *mockReverseProxyManager) StartExposeReaper(_ context.Context) {}
+
 type mockUsersManager struct {
 	users map[string]*types.User
 	err   error
@@ -8,12 +8,66 @@ import (
 	"testing"
 	"time"

+	"sync"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"

+	rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
 	"github.com/netbirdio/netbird/shared/management/proto"
 )

+type testProxyController struct {
+	mu             sync.Mutex
+	clusterProxies map[string]map[string]struct{}
+}
+
+func newTestProxyController() *testProxyController {
+	return &testProxyController{
+		clusterProxies: make(map[string]map[string]struct{}),
+	}
+}
+
+func (c *testProxyController) SendServiceUpdateToCluster(_ context.Context, _ string, _ *proto.ProxyMapping, _ string) {
+}
+
+func (c *testProxyController) GetOIDCValidationConfig() rpservice.OIDCValidationConfig {
+	return rpservice.OIDCValidationConfig{}
+}
+
+func (c *testProxyController) RegisterProxyToCluster(_ context.Context, clusterAddr, proxyID string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if _, ok := c.clusterProxies[clusterAddr]; !ok {
+		c.clusterProxies[clusterAddr] = make(map[string]struct{})
+	}
+	c.clusterProxies[clusterAddr][proxyID] = struct{}{}
+	return nil
+}
+
+func (c *testProxyController) UnregisterProxyFromCluster(_ context.Context, clusterAddr, proxyID string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if proxies, ok := c.clusterProxies[clusterAddr]; ok {
+		delete(proxies, proxyID)
+	}
+	return nil
+}
+
+func (c *testProxyController) GetProxiesForCluster(clusterAddr string) []string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	proxies, ok := c.clusterProxies[clusterAddr]
+	if !ok {
+		return nil
+	}
+	result := make([]string, 0, len(proxies))
+	for id := range proxies {
+		result = append(result, id)
+	}
+	return result
+}
+
 // registerFakeProxy adds a fake proxy connection to the server's internal maps
 // and returns the channel where messages will be received.
 func registerFakeProxy(s *ProxyServiceServer, proxyID, clusterAddr string) chan *proto.GetMappingUpdateResponse {
@@ -46,6 +100,7 @@ func TestSendServiceUpdateToCluster_UniqueTokensPerProxy(t *testing.T) {
 	s := &ProxyServiceServer{
 		tokenStore: tokenStore,
 	}
+	s.SetProxyController(newTestProxyController())

 	const cluster = "proxy.example.com"
 	const numProxies = 3
@@ -102,6 +157,7 @@ func TestSendServiceUpdateToCluster_DeleteNoToken(t *testing.T) {
 	s := &ProxyServiceServer{
 		tokenStore: tokenStore,
 	}
+	s.SetProxyController(newTestProxyController())

 	const cluster = "proxy.example.com"
 	ch1 := registerFakeProxy(s, "proxy-a", cluster)
@@ -135,6 +191,7 @@ func TestSendServiceUpdate_UniqueTokensPerProxy(t *testing.T) {
 	s := &ProxyServiceServer{
 		tokenStore: tokenStore,
 	}
+	s.SetProxyController(newTestProxyController())

 	// Register proxies in different clusters (SendServiceUpdate broadcasts to all)
 	ch1 := registerFakeProxy(s, "proxy-a", "cluster-a")
@@ -26,6 +26,7 @@ import (
 	"github.com/netbirdio/netbird/shared/management/client/common"

 	"github.com/netbirdio/netbird/management/internals/controllers/network_map"
+	rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
 	nbconfig "github.com/netbirdio/netbird/management/internals/server/config"
 	"github.com/netbirdio/netbird/management/server/idp"
 	"github.com/netbirdio/netbird/management/server/job"
@@ -80,6 +81,9 @@ type Server struct {
 	syncSem        atomic.Int32
 	syncLimEnabled bool
 	syncLim        int32
+
+	reverseProxyManager rpservice.Manager
+	reverseProxyMu      sync.RWMutex
 }

 // NewServer creates a new Management server
@@ -303,6 +303,20 @@ func (m *testValidateSessionServiceManager) GetServiceIDByTargetID(_ context.Con
 	return "", nil
 }

+func (m *testValidateSessionServiceManager) CreateServiceFromPeer(_ context.Context, _, _ string, _ *service.ExposeServiceRequest) (*service.ExposeServiceResponse, error) {
+	return nil, nil
+}
+
+func (m *testValidateSessionServiceManager) RenewServiceFromPeer(_ context.Context, _, _, _ string) error {
+	return nil
+}
+
+func (m *testValidateSessionServiceManager) StopServiceFromPeer(_ context.Context, _, _, _ string) error {
+	return nil
+}
+
+func (m *testValidateSessionServiceManager) StartExposeReaper(_ context.Context) {}
+
 type testValidateSessionProxyManager struct{}

 func (m *testValidateSessionProxyManager) Connect(_ context.Context, _, _, _ string) error {