Netpulse_SasS/server/cmd/netpulse-server/main.go

// Команда netpulse-server — приймальна сторона AgentService.
//
// Зонди самі підключаються сюди; сервер до них не ходить.
package main

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"flag"
	"fmt"
	"log/slog"
	"net"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/netpulse/netpulse/server/internal/crypto"
	"github.com/netpulse/netpulse/server/internal/gitstore"
	"github.com/netpulse/netpulse/server/internal/grpcapi"
	"github.com/netpulse/netpulse/server/internal/store"
	npv1 "github.com/netpulse/netpulse/gen/go/netpulse/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"
)

var version = "dev"

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, "netpulse-server:", err)
		os.Exit(1)
	}
}

func run() error {
	var (
		listen   = flag.String("listen", envOr("NETPULSE_LISTEN", ":9443"), "адреса прослуховування gRPC")
		dsn      = flag.String("dsn", os.Getenv("NETPULSE_DSN"), "DSN PostgreSQL")
		certFile = flag.String("cert", os.Getenv("NETPULSE_CERT"), "сертифікат сервера")
		keyFile  = flag.String("key", os.Getenv("NETPULSE_KEY"), "приватний ключ")
		caFile   = flag.String("client-ca", os.Getenv("NETPULSE_CLIENT_CA"), "CA для перевірки сертифікатів зондів (mTLS)")
		insecure = flag.Bool("insecure", os.Getenv("NETPULSE_INSECURE") == "1", "без TLS — лише локальний стенд")
		keysFlag = flag.String("dek", os.Getenv("NETPULSE_DEK"), "ключі шифрування: key_id=<hex|base64>[,...]")
		logLevel = flag.String("log-level", envOr("NETPULSE_LOG_LEVEL", "info"), "debug|info|warn|error")
		gitRoot  = flag.String("git-root", envOr("NETPULSE_GIT_ROOT", "/var/lib/netpulse/git"),
			"корінь сховища версій конфігів; порожньо — без Git")
	)
	flag.Parse()

	if *dsn == "" {
		return errors.New("не вказано -dsn (або NETPULSE_DSN)")
	}

	log := newLogger(*logLevel)

	ring, err := buildKeyring(*keysFlag)
	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	st, err := store.New(ctx, *dsn)
	if err != nil {
		return fmt.Errorf("підключення до БД: %w", err)
	}
	defer st.Close()

	// Сховище версій конфігів. Спільний каталог для API й колектора:
	// колектор туди пише під час бекапу, API звідти читає для diff.
	// Різні машини потребують спільного тому — інакше половина історії
	// буде видна одному процесу, а половина другому.
	if *gitRoot != "" {
		st.UseGit(gitstore.New(*gitRoot))
	}

	svc := grpcapi.New(st, ring, log)

	opts := []grpc.ServerOption{
		grpc.ChainUnaryInterceptor(svc.UnaryInterceptor),
		grpc.ChainStreamInterceptor(svc.StreamInterceptor),
		// Зонди сидять за NAT: без keepalive проміжний маршрутизатор
		// тихо викидає сесію, і сервер вважає мертвого агента живим.
		grpc.KeepaliveParams(keepalive.ServerParameters{
			Time:    60 * time.Second,
			Timeout: 20 * time.Second,
		}),
		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
			MinTime:             20 * time.Second,
			PermitWithoutStream: true,
		}),
		// Конфіги великих маршрутизаторів бувають на кілька МБ,
		// але їх шлють чанками — ліміт лишається захистом.
		grpc.MaxRecvMsgSize(16 << 20),
	}

	if !*insecure {
		tc, err := serverTLS(*certFile, *keyFile, *caFile)
		if err != nil {
			return err
		}
		opts = append(opts, grpc.Creds(credentials.NewTLS(tc)))
	} else {
		log.Warn("запуск без TLS — припустимо лише для локального стенду")
	}

	srv := grpc.NewServer(opts...)
	npv1.RegisterAgentServiceServer(srv, svc)
	// Реєстрація йде БЕЗ токена зонда, тож окремим сервісом: інакше
	// довелося б робити виняток усередині інтерсептора автентифікації.
	npv1.RegisterEnrollmentServiceServer(srv, grpcapi.NewEnrollService(st, *listen))

	lis, err := net.Listen("tcp", *listen)
	if err != nil {
		return fmt.Errorf("listen %s: %w", *listen, err)
	}

	log.Info("запуск", "version", version, "listen", *listen, "tls", !*insecure)

	// Диспетчер збору конфігів: черга наповнюється REST-процесом, а
	// живі сесії зондів тримає саме цей.
	go svc.DispatchConfigJobs(ctx, 5*time.Second)
	// Розклад бекапів. Кілька екземплярів безпечні: тік бере
	// advisory-блокування, тож розкручує його рівно один.
	go svc.ScheduleBackups(ctx)
	// Звірка планів: чеки міняє REST-процес, а перезалити план може
	// лише той, хто тримає сесію зонда.
	go svc.SyncPlans(ctx)

	errCh := make(chan error, 1)
	go func() { errCh <- srv.Serve(lis) }()

	select {
	case <-ctx.Done():
		log.Info("зупинка", "сесій_онлайн", svc.SessionCount())
		// GracefulStop дає активним стрімам догратись: обірваний
		// посеред батчу агент однаково перешле його, але зайвих
		// ретраїв краще уникнути.
		done := make(chan struct{})
		go func() { srv.GracefulStop(); close(done) }()
		select {
		case <-done:
		case <-time.After(15 * time.Second):
			log.Warn("м'яка зупинка не встигла — примусова")
			srv.Stop()
		}
		return nil
	case err := <-errCh:
		return err
	}
}

func envOr(key, def string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return def
}

func newLogger(level string) *slog.Logger {
	lv := slog.LevelInfo
	switch level {
	case "debug":
		lv = slog.LevelDebug
	case "warn":
		lv = slog.LevelWarn
	case "error":
		lv = slog.LevelError
	}
	return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: lv}))
}

// buildKeyring розбирає ключі шифрування.
//
// Ключі приходять через оточення або секрет-менеджер і ніколи не
// лежать поруч із дампом БД: інакше шифрування конфігів і паролів
// від обладнання не давало б нічого. На відміну від REST-процесу,
// тут ключ обов'язковий: без нього AgentService не може ані видати
// креденшели, ані зберегти конфіг.
func buildKeyring(spec string) (*crypto.Keyring, error) {
	ring, err := crypto.ParseKeyring(spec)
	if err != nil {
		return nil, err
	}
	if ring == nil {
		return nil, errors.New("не вказано -dek: без ключа неможливо ані видати креденшели, ані зберегти конфіг")
	}
	return ring, nil
}

func serverTLS(certFile, keyFile, clientCA string) (*tls.Config, error) {
	if certFile == "" || keyFile == "" {
		return nil, errors.New("потрібні -cert і -key (або -insecure для локального стенду)")
	}
	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		return nil, fmt.Errorf("сертифікат сервера: %w", err)
	}

	tc := &tls.Config{
		Certificates: []tls.Certificate{cert},
		MinVersion:   tls.VersionTLS13,
	}

	if clientCA != "" {
		pem, err := os.ReadFile(clientCA)
		if err != nil {
			return nil, fmt.Errorf("CA зондів: %w", err)
		}
		pool := x509.NewCertPool()
		if !pool.AppendCertsFromPEM(pem) {
			return nil, errors.New("CA зондів: не вдалося розібрати PEM")
		}
		tc.ClientCAs = pool
		// Токен каже, ЯКИЙ це зонд; сертифікат — що він узагалі має
		// право говорити з сервером. Обидва обов'язкові.
		tc.ClientAuth = tls.RequireAndVerifyClientCert
	}

	return tc, nil
}