feat: support for asymmetric encryption to the agent and server

README.md

@@ -129,6 +129,7 @@ You can use ENV, CLI flags, or defaults (ENV > CLI > defaults).
 | File storage     | `FILE_STORAGE_PATH` | `-f`            | `metrics-db.json` | JSON snapshot file                                                    |
 | Postgres DSN     | `DATABASE_DSN`      | `-d`            | *empty*           | e.g. `postgres://user:pass@localhost:5432/db?sslmode=disable`         |
 | Secret key       | `KEY`               | `-k`            | *empty*           | enables `HashSHA256`                                                  |
+| Crypto key       | `CRYPTO_KEY`        | `-crypto-key`   | *empty*           | path to RSA private key for decrypting agent payloads                 |
 | Store interval   | `STORE_INTERVAL`    | `-i`            | `300s`            | `0` = sync writes                                                     |
 | Restore on start | `RESTORE`           | `-r`            | `false`           | load from file at boot                                                |
 | Audit file       | `AUDIT_FILE`        | `--audit-file`  | *empty*           | newline-delimited JSON audit log fan-out target (disabled when empty) |
@@ -139,6 +140,7 @@ You can use ENV, CLI flags, or defaults (ENV > CLI > defaults).
 | --------------- | ----------------- | ---- | ----------------------- | ----------------------- |
 | Server address  | `ADDRESS`         | `-a` | `http://localhost:8080` | URL or `host:port`      |
 | Secret key      | `KEY`             | `-k` | *empty*                 | adds `HashSHA256`       |
+| Crypto key      | `CRYPTO_KEY`      | `-crypto-key` | *empty*           | path to RSA public key for encrypting requests |
 | Report interval | `REPORT_INTERVAL` | `-r` | `10s`                   | send frequency          |
 | Poll interval   | `POLL_INTERVAL`   | `-p` | `2s`                    | sample frequency        |
 | Rate limit      | `RATE_LIMIT`      | `-l` | `1`                     | concurrent send workers |

cmd/agent/main.go

@@ -8,10 +8,12 @@ import (
 	"os/signal"
 	"syscall"
 
-	"github.com/vshulcz/Golectra/internal/adapters/collector/runtime"
-	"github.com/vshulcz/Golectra/internal/adapters/publisher/httpjson"
-	"github.com/vshulcz/Golectra/internal/config"
-	agentsvc "github.com/vshulcz/Golectra/internal/services/agent"
+	agentsvc "github.com/vshulcz/Golectra/internal/application/agent"
+	"github.com/vshulcz/Golectra/internal/infra/collector/runtime"
+	"github.com/vshulcz/Golectra/internal/infra/config"
+	"github.com/vshulcz/Golectra/internal/infra/crypto/rsaenvelope"
+	"github.com/vshulcz/Golectra/internal/infra/publisher/httpjson"
+	"github.com/vshulcz/Golectra/internal/ports"
 	"github.com/vshulcz/Golectra/pkg/util"
 )
 
@@ -29,12 +31,21 @@ func main() {
 		log.Fatalf("failed to parse flags: %v", err)
 	}
 
-	pub, err := httpjson.New(cfg.Address, &http.Client{}, cfg.Key)
+	var encrypter ports.PayloadEncrypter
+	if cfg.CryptoKey != "" {
+		key, err := rsaenvelope.LoadPublicKey(cfg.CryptoKey)
+		if err != nil {
+			log.Fatalf("failed to load crypto key: %v", err)
+		}
+		encrypter = rsaenvelope.NewEncrypter(key)
+	}
+
+	pub, err := httpjson.New(cfg.Address, &http.Client{}, cfg.Key, encrypter)
 	if err != nil {
 		log.Fatalf("failed to init publisher: %v", err)
 	}
 	collector := runtime.New()
-	runner := agentsvc.New(cfg, collector, pub)
+	runner := agentsvc.New(mapAgentConfig(cfg), collector, pub)
 
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
@@ -49,3 +60,11 @@ func main() {
 func printBuildInfo() {
 	util.PrintBuildInfo(buildVersion, buildDate, buildCommit)
 }
+
+func mapAgentConfig(cfg config.AgentConfig) agentsvc.Config {
+	return agentsvc.Config{
+		PollInterval:   cfg.PollInterval,
+		ReportInterval: cfg.ReportInterval,
+		RateLimit:      cfg.RateLimit,
+	}
+}

cmd/agent/main_test.go

@@ -2,10 +2,31 @@ package main
 
 import (
 	"testing"
+	"time"
+
+	"github.com/vshulcz/Golectra/internal/application/agent"
+	"github.com/vshulcz/Golectra/internal/infra/config"
 )
 
 func TestBuildVariablesExist(t *testing.T) {
 	_ = buildVersion
 	_ = buildDate
 	_ = buildCommit
 }
+
+func TestMapAgentConfig(t *testing.T) {
+	cfg := config.AgentConfig{
+		PollInterval:   2 * time.Second,
+		ReportInterval: 5 * time.Second,
+		RateLimit:      3,
+	}
+	got := mapAgentConfig(cfg)
+	want := agent.Config{
+		PollInterval:   2 * time.Second,
+		ReportInterval: 5 * time.Second,
+		RateLimit:      3,
+	}
+	if got != want {
+		t.Fatalf("mapAgentConfig=%+v want %+v", got, want)
+	}
+}

cmd/server/init.go

@@ -7,11 +7,11 @@ import (
 	_ "github.com/lib/pq"
 	"go.uber.org/zap"
 
-	"github.com/vshulcz/Golectra/internal/adapters/persistence/file"
-	memrepo "github.com/vshulcz/Golectra/internal/adapters/repository/memory"
-	pgrepo "github.com/vshulcz/Golectra/internal/adapters/repository/postgres"
-	"github.com/vshulcz/Golectra/internal/config"
-	"github.com/vshulcz/Golectra/internal/misc"
+	"github.com/vshulcz/Golectra/internal/infra/config"
+	"github.com/vshulcz/Golectra/internal/infra/persistence/file"
+	memrepo "github.com/vshulcz/Golectra/internal/infra/repository/memory"
+	pgrepo "github.com/vshulcz/Golectra/internal/infra/repository/postgres"
+	"github.com/vshulcz/Golectra/internal/infra/retry"
 	"github.com/vshulcz/Golectra/internal/ports"
 )
 
@@ -26,7 +26,7 @@ func buildRepoAndPersister(cfg config.ServerConfig, logger *zap.Logger) (ports.M
 				}
 				return pgrepo.Migrate(db)
 			}
-			if err = misc.Retry(ctx, misc.DefaultBackoff, pgrepo.IsRetryable, op); err == nil {
+			if err = retry.Retry(ctx, retry.DefaultBackoff, pgrepo.IsRetryable, op); err == nil {
 				logger.Info("db connected & migrated")
 				return pgrepo.New(db), nil
 			}

cmd/server/main.go

@@ -8,14 +8,16 @@ import (
 	"os"
 	"time"
 
-	auditfile "github.com/vshulcz/Golectra/internal/adapters/audit/file"
-	auditremote "github.com/vshulcz/Golectra/internal/adapters/audit/remote"
-	"github.com/vshulcz/Golectra/internal/adapters/http/ginserver"
-	"github.com/vshulcz/Golectra/internal/adapters/http/ginserver/middlewares"
-	"github.com/vshulcz/Golectra/internal/config"
+	"github.com/vshulcz/Golectra/internal/application/metrics"
 	"github.com/vshulcz/Golectra/internal/domain"
-	"github.com/vshulcz/Golectra/internal/services/audit"
-	"github.com/vshulcz/Golectra/internal/services/metrics"
+	auditinfra "github.com/vshulcz/Golectra/internal/infra/audit"
+	auditfile "github.com/vshulcz/Golectra/internal/infra/audit/file"
+	auditremote "github.com/vshulcz/Golectra/internal/infra/audit/remote"
+	"github.com/vshulcz/Golectra/internal/infra/config"
+	"github.com/vshulcz/Golectra/internal/infra/crypto/rsaenvelope"
+	"github.com/vshulcz/Golectra/internal/infra/http/ginserver"
+	"github.com/vshulcz/Golectra/internal/infra/http/ginserver/middlewares"
+	"github.com/vshulcz/Golectra/internal/ports"
 	"github.com/vshulcz/Golectra/pkg/util"
 	"go.uber.org/zap"
 )
@@ -40,89 +42,135 @@ func run(args []string) error {
 		return err
 	}
 
-	logger, err := zap.NewProduction()
+	logger, cleanup, err := initLogger()
 	if err != nil {
 		return err
 	}
-	defer func() {
-		if cerr := logger.Sync(); cerr != nil {
-			log.Printf("logger sync: %v", cerr)
-		}
-	}()
+	defer cleanup()
 
 	repo, persister := buildRepoAndPersister(cfg, logger)
-	onChanged := func(ctx context.Context, s domain.Snapshot) {
-		if persister != nil {
-			if err := persister.Save(ctx, s); err != nil {
-				logger.Warn("save failed", zap.Error(err))
-			}
-		}
-	}
+	onChanged := buildSnapshotHook(persister, logger)
 
 	auditor := buildAuditor(cfg, logger)
 	svc := metrics.New(repo, onChanged, auditor)
 	defer svc.Close()
 	h := ginserver.NewHandler(svc)
 
+	decrypter, err := loadDecrypter(cfg)
+	if err != nil {
+		return err
+	}
+
 	r := ginserver.NewRouter(h, logger,
 		middlewares.ZapLogger(logger),
+		middlewares.DecryptPayload(decrypter),
 		middlewares.GzipRequest(),
 		middlewares.GzipResponse(),
 		middlewares.HashSHA256(cfg.Key),
 	)
 
-	log.Printf("cfg: addr=%s file=%s interval=%v restore=%v dsn=%q audit_file=%q audit_url=%q",
-		cfg.Address, cfg.File, cfg.Interval, cfg.Restore, cfg.DSN, cfg.AuditFile, cfg.AuditURL)
-
-	if cfg.DSN == "" && cfg.Interval > 0 {
-		if cfg.Interval < 0 {
-			cfg.Interval = 300 * time.Second
-		}
-		ticker := time.NewTicker(cfg.Interval)
-		go func() {
-			for range ticker.C {
-				if s, err := repo.Snapshot(context.Background()); err == nil && persister != nil {
-					if err := persister.Save(context.Background(), s); err != nil {
-						logger.Warn("periodic save failed", zap.Error(err))
-					}
-				}
-			}
-		}()
-	}
+	logConfig(cfg)
+	startPeriodicSave(cfg, repo, persister, logger)
 
-	srv := &http.Server{
-		Addr:              cfg.Address,
-		Handler:           r,
-		ReadHeaderTimeout: 5 * time.Second,
-		ReadTimeout:       15 * time.Second,
-		WriteTimeout:      15 * time.Second,
-		IdleTimeout:       60 * time.Second,
-	}
-	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-		return err
-	}
-	return nil
+	srv := newHTTPServer(cfg, r)
+	return serve(srv)
 }
 
-func buildAuditor(cfg config.ServerConfig, logger *zap.Logger) audit.Publisher {
+func buildAuditor(cfg config.ServerConfig, logger *zap.Logger) ports.AuditPublisher {
 	if cfg.AuditFile == "" && cfg.AuditURL == "" {
 		return nil
 	}
-	subject := audit.NewSubject()
-	subject.SetErrorHandler(func(err error) {
+	fanout := auditinfra.NewFanout()
+	fanout.SetErrorHandler(func(err error) {
 		logger.Warn("audit delivery failed", zap.Error(err))
 	})
 	if cfg.AuditFile != "" {
-		subject.Attach(auditfile.New(cfg.AuditFile))
+		fanout.Attach(auditfile.New(cfg.AuditFile))
 	}
 	if cfg.AuditURL != "" {
 		client, err := auditremote.New(cfg.AuditURL, nil)
 		if err != nil {
 			logger.Fatal("invalid audit url", zap.Error(err))
 		}
-		subject.Attach(client)
+		fanout.Attach(client)
+	}
+	return fanout
+}
+
+func initLogger() (*zap.Logger, func(), error) {
+	logger, err := zap.NewProduction()
+	if err != nil {
+		return nil, nil, err
+	}
+	cleanup := func() {
+		if cerr := logger.Sync(); cerr != nil {
+			log.Printf("logger sync: %v", cerr)
+		}
+	}
+	return logger, cleanup, nil
+}
+
+func buildSnapshotHook(persister ports.Persister, logger *zap.Logger) func(context.Context, domain.Snapshot) {
+	return func(ctx context.Context, s domain.Snapshot) {
+		if persister == nil {
+			return
+		}
+		if err := persister.Save(ctx, s); err != nil {
+			logger.Warn("save failed", zap.Error(err))
+		}
+	}
+}
+
+func loadDecrypter(cfg config.ServerConfig) (ports.PayloadDecrypter, error) {
+	if cfg.CryptoKey == "" {
+		return nil, nil
+	}
+	key, err := rsaenvelope.LoadPrivateKey(cfg.CryptoKey)
+	if err != nil {
+		return nil, err
+	}
+	return rsaenvelope.NewDecrypter(key), nil
+}
+
+func logConfig(cfg config.ServerConfig) {
+	log.Printf("cfg: addr=%s file=%s interval=%v restore=%v dsn=%q audit_file=%q audit_url=%q",
+		cfg.Address, cfg.File, cfg.Interval, cfg.Restore, cfg.DSN, cfg.AuditFile, cfg.AuditURL)
+}
+
+func startPeriodicSave(cfg config.ServerConfig, repo ports.MetricsRepo, persister ports.Persister, logger *zap.Logger) {
+	if cfg.DSN != "" || cfg.Interval <= 0 {
+		return
+	}
+	ticker := time.NewTicker(cfg.Interval)
+	go func() {
+		for range ticker.C {
+			snap, err := repo.Snapshot(context.Background())
+			if err != nil || persister == nil {
+				continue
+			}
+			if err := persister.Save(context.Background(), snap); err != nil {
+				logger.Warn("periodic save failed", zap.Error(err))
+			}
+		}
+	}()
+}
+
+func newHTTPServer(cfg config.ServerConfig, handler http.Handler) *http.Server {
+	return &http.Server{
+		Addr:              cfg.Address,
+		Handler:           handler,
+		ReadHeaderTimeout: 5 * time.Second,
+		ReadTimeout:       15 * time.Second,
+		WriteTimeout:      15 * time.Second,
+		IdleTimeout:       60 * time.Second,
 	}
-	return subject
+}
+
+func serve(srv *http.Server) error {
+	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		return err
+	}
+	return nil
 }
 
 func printBuildInfo() {