diff --git a/PROJECTS/beginner/canary-token-generator/backend/cmd/canary/main.go b/PROJECTS/beginner/canary-token-generator/backend/cmd/canary/main.go index f91a06c2..38592634 100644 --- a/PROJECTS/beginner/canary-token-generator/backend/cmd/canary/main.go +++ b/PROJECTS/beginner/canary-token-generator/backend/cmd/canary/main.go @@ -5,10 +5,12 @@ package main import ( "context" + "encoding/json" "errors" "flag" "fmt" "log/slog" + "net/http" "os" "os/signal" "sync" @@ -22,6 +24,9 @@ import ( "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/event" "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/health" "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/middleware" + "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/notify" + "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/notify/telegram" + "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/notify/webhook" "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/server" "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/token" "github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/token/generators/mysql" @@ -86,6 +91,14 @@ func run(configPath string) error { } logger.Info("redis connected") + notifySvc, eventSvc := buildEventStack( + cfg, + logger, + eventRepo, + tokenRepo, + rdb, + ) + genRegistry := registry.Build(registry.Config{ BaseURL: cfg.Canary.BaseURL, MySQLPublicHost: cfg.MySQL.PublicHost, @@ -108,19 +121,19 @@ func run(configPath string) error { healthH := health.NewHandler(db, rdb) tokenH := token.NewHandler( tokenSvc, - &directEventRecorder{ + &eventRecorderAdapter{svc: eventSvc}, + &fingerprintRecorderAdapter{ repo: eventRepo, - tokens: tokenRepo, - logger: logger, + window: cfg.Notify.FingerprintWindow, }, - nil, logger, ) srv := mountRouter(cfg, logger, rdb, healthH, tokenH, verifier) var wg sync.WaitGroup - spawnMySQLListener(ctx, cfg, logger, &wg, tokenSvc, eventRepo, tokenRepo) + spawnMySQLListener(ctx, cfg, logger, &wg, tokenSvc, eventSvc) + spawnRetentionLoop(ctx, cfg, &wg, eventSvc) errChan := make(chan error, 1) go func() { errChan <- srv.Start() }() @@ -133,18 +146,60 @@ func run(configPath string) error { } shutdownErr := gracefulShutdown(cfg, logger, srv, telemetry, rdb, db) + logger.Info("waiting for in-flight notifications") + notifySvc.Wait() wg.Wait() return shutdownErr } +func buildEventStack( + cfg *config.Config, + logger *slog.Logger, + eventRepo *event.Repository, + tokenRepo *token.Repository, + rdb *core.Redis, +) (*notify.Service, *event.Service) { + tgSender := telegram.NewSender(telegram.Config{ + APIBase: cfg.Notify.TelegramAPIBase, + ManageURL: cfg.Canary.ManageURL, + MaxTries: cfg.Notify.MaxTries, + MaxElapsed: cfg.Notify.MaxElapsed, + InitialInterval: cfg.Notify.InitialInterval, + }) + whSender := webhook.NewSender(webhook.Config{ + ManageURL: cfg.Canary.ManageURL, + HMACSecret: cfg.Notify.WebhookHMACSecret, + MaxTries: cfg.Notify.MaxTries, + MaxElapsed: cfg.Notify.MaxElapsed, + InitialInterval: cfg.Notify.InitialInterval, + }) + + notifySvc := notify.NewService(eventRepo, + notify.WithLogger(logger), + notify.WithSendTimeout(cfg.Notify.SendTimeout), + ) + notifySvc.Register(tgSender, whSender) + + eventSvc := event.NewService( + eventRepo, + tokenRepo, + rdb.Client, + notifySvc, + event.ServiceConfig{ + DedupTTL: cfg.Notify.DedupTTL, + Logger: logger, + }, + ) + return notifySvc, eventSvc +} + func spawnMySQLListener( ctx context.Context, cfg *config.Config, logger *slog.Logger, wg *sync.WaitGroup, tokenSvc *token.Service, - eventRepo *event.Repository, - tokenRepo *token.Repository, + eventSvc *event.Service, ) { if !cfg.MySQL.Enabled { return @@ -154,11 +209,7 @@ func spawnMySQLListener( defer wg.Done() handler := mysql.NewHandler( &mysqlTokenLookup{svc: tokenSvc}, - &mysqlEventRecorder{ - repo: eventRepo, - tokens: tokenRepo, - logger: logger, - }, + &eventRecorderAdapter{svc: eventSvc}, ) if mErr := mysql.Run(ctx, cfg.MySQL.Addr, handler); mErr != nil { logger.Error("mysql server error", "error", mErr) @@ -166,6 +217,26 @@ func spawnMySQLListener( }() } +func spawnRetentionLoop( + ctx context.Context, + cfg *config.Config, + wg *sync.WaitGroup, + eventSvc *event.Service, +) { + if cfg.Notify.RetentionInterval <= 0 || cfg.Notify.RetentionLimit <= 0 { + return + } + wg.Add(1) + go func() { + defer wg.Done() + eventSvc.RunRetentionLoop( + ctx, + cfg.Notify.RetentionInterval, + cfg.Notify.RetentionLimit, + ) + }() +} + func mountRouter( cfg *config.Config, logger *slog.Logger, @@ -189,6 +260,29 @@ func mountRouter( healthH.RegisterRoutes(r) tokenH.RegisterTriggerRoutes(r) + createMin := middleware.NewRateLimiter( + rdb.Client, + middleware.RateLimitConfig{ + Limit: middleware.PerMinute( + cfg.RateLimit.CreateMinRate, + cfg.RateLimit.CreateMinBurst, + ), + KeyFunc: keyByCreateMin, + FailOpen: true, + }, + ).Handler + createHour := middleware.NewRateLimiter( + rdb.Client, + middleware.RateLimitConfig{ + Limit: middleware.PerHour( + cfg.RateLimit.CreateHourRate, + cfg.RateLimit.CreateHourBurst, + ), + KeyFunc: keyByCreateHour, + FailOpen: true, + }, + ).Handler + r.Route("/api", func(api chi.Router) { api.Use(middleware.CORS(cfg.CORS)) api.Use( @@ -202,13 +296,24 @@ func mountRouter( }).Handler, ) api.Get("/tokens/types", tokenH.GetTypes) - api.With(middleware.TurnstileVerify(verifier)). - Post("/tokens", tokenH.CreateToken) + api.With( + createMin, + createHour, + middleware.TurnstileVerify(verifier), + ).Post("/tokens", tokenH.CreateToken) }) return srv } +func keyByCreateMin(r *http.Request) string { + return "ratelimit:create:min:" + middleware.ExtractFingerprint(r) +} + +func keyByCreateHour(r *http.Request) string { + return "ratelimit:create:hour:" + middleware.ExtractFingerprint(r) +} + func gracefulShutdown( cfg *config.Config, logger *slog.Logger, @@ -291,25 +396,35 @@ func (a registryAdapter) Get(t token.Type) (token.Generator, bool) { return g, ok } -type directEventRecorder struct { - repo *event.Repository - tokens *token.Repository - logger *slog.Logger +type eventRecorderAdapter struct { + svc *event.Service } -func (d *directEventRecorder) Record( +func (a *eventRecorderAdapter) Record( ctx context.Context, t *token.Token, evt *event.Event, ) error { - if err := d.repo.Insert(ctx, evt); err != nil { - return fmt.Errorf("insert event: %w", err) - } - if err := d.tokens.IncrementTriggerCount(ctx, t.ID); err != nil { - d.logger.WarnContext(ctx, "increment trigger count", - "error", err, "token_id", t.ID) - } - return nil + return a.svc.Record(ctx, t.NotifyInfo(), evt) +} + +type fingerprintRecorderAdapter struct { + repo *event.Repository + window time.Duration +} + +func (f *fingerprintRecorderAdapter) AttachFingerprint( + ctx context.Context, + tokenID, sourceIP string, + fingerprint json.RawMessage, +) error { + return f.repo.AttachFingerprint( + ctx, + tokenID, + sourceIP, + fingerprint, + f.window, + ) } type mysqlTokenLookup struct{ svc *token.Service } @@ -320,24 +435,3 @@ func (m *mysqlTokenLookup) GetByID( ) (*token.Token, error) { return m.svc.GetByID(ctx, id) } - -type mysqlEventRecorder struct { - repo *event.Repository - tokens *token.Repository - logger *slog.Logger -} - -func (m *mysqlEventRecorder) Record( - ctx context.Context, - t *token.Token, - evt *event.Event, -) error { - if err := m.repo.Insert(ctx, evt); err != nil { - return fmt.Errorf("insert event: %w", err) - } - if err := m.tokens.IncrementTriggerCount(ctx, t.ID); err != nil { - m.logger.WarnContext(ctx, "mysql: increment trigger count", - "error", err, "token_id", t.ID) - } - return nil -} diff --git a/PROJECTS/beginner/canary-token-generator/backend/internal/config/config.go b/PROJECTS/beginner/canary-token-generator/backend/internal/config/config.go index 1beb55b2..2e561c35 100644 --- a/PROJECTS/beginner/canary-token-generator/backend/internal/config/config.go +++ b/PROJECTS/beginner/canary-token-generator/backend/internal/config/config.go @@ -26,6 +26,20 @@ type Config struct { Canary CanaryConfig `koanf:"canary"` Turnstile TurnstileConfig `koanf:"turnstile"` MySQL MySQLConfig `koanf:"mysql"` + Notify NotifyConfig `koanf:"notify"` +} + +type NotifyConfig struct { + DedupTTL time.Duration `koanf:"dedup_ttl"` + SendTimeout time.Duration `koanf:"send_timeout"` + MaxTries uint `koanf:"max_tries"` + MaxElapsed time.Duration `koanf:"max_elapsed"` + InitialInterval time.Duration `koanf:"initial_interval"` + RetentionInterval time.Duration `koanf:"retention_interval"` + RetentionLimit int `koanf:"retention_limit"` + WebhookHMACSecret string `koanf:"webhook_hmac_secret"` + TelegramAPIBase string `koanf:"telegram_api_base"` + FingerprintWindow time.Duration `koanf:"fingerprint_window"` } type CanaryConfig struct { @@ -75,9 +89,13 @@ type RedisConfig struct { } type RateLimitConfig struct { - Requests int `koanf:"requests"` - Window time.Duration `koanf:"window"` - Burst int `koanf:"burst"` + Requests int `koanf:"requests"` + Window time.Duration `koanf:"window"` + Burst int `koanf:"burst"` + CreateMinRate int `koanf:"create_min_rate"` + CreateMinBurst int `koanf:"create_min_burst"` + CreateHourRate int `koanf:"create_hour_rate"` + CreateHourBurst int `koanf:"create_hour_burst"` } type CORSConfig struct { @@ -182,9 +200,13 @@ func loadDefaults(k *koanf.Koanf) error { "redis.pool_size": 10, "redis.min_idle_conns": 5, - "rate_limit.requests": 100, - "rate_limit.window": "1m", - "rate_limit.burst": 20, + "rate_limit.requests": 100, + "rate_limit.window": "1m", + "rate_limit.burst": 20, + "rate_limit.create_min_rate": 5, + "rate_limit.create_min_burst": 5, + "rate_limit.create_hour_rate": 20, + "rate_limit.create_hour_burst": 5, "cors.allowed_origins": []string{"http://localhost:3000"}, "cors.allowed_methods": []string{ @@ -222,6 +244,17 @@ func loadDefaults(k *koanf.Koanf) error { "mysql.addr": "0.0.0.0:3306", "mysql.public_host": "localhost", "mysql.public_port": 3306, + + "notify.dedup_ttl": "15m", + "notify.send_timeout": "30s", + "notify.max_tries": 3, + "notify.max_elapsed": "30s", + "notify.initial_interval": "500ms", + "notify.retention_interval": "1h", + "notify.retention_limit": 100, + "notify.webhook_hmac_secret": "", + "notify.telegram_api_base": "https://api.telegram.org", + "notify.fingerprint_window": "5m", } for key, value := range defaults { @@ -234,30 +267,44 @@ func loadDefaults(k *koanf.Koanf) error { } var envKeyMap = map[string]string{ - "DATABASE_URL": "database.url", - "REDIS_URL": "redis.url", - "APP_ENVIRONMENT": "app.environment", - "HOST": "server.host", - "PORT": "server.port", - "LOG_LEVEL": "log.level", - "LOG_FORMAT": "log.format", - "RATE_LIMIT_REQUESTS": "rate_limit.requests", - "RATE_LIMIT_WINDOW": "rate_limit.window", - "RATE_LIMIT_BURST": "rate_limit.burst", - "OTEL_ENDPOINT": "otel.endpoint", - "OTEL_EXPORTER_OTLP_ENDPOINT": "otel.endpoint", - "OTEL_SERVICE_NAME": "otel.service_name", - "OTEL_ENABLED": "otel.enabled", - "OTEL_INSECURE": "otel.insecure", - "OTEL_SAMPLE_RATE": "otel.sample_rate", - "CANARY_BASE_URL": "canary.base_url", - "CANARY_MANAGE_URL": "canary.manage_url", - "TURNSTILE_SECRET_KEY": "turnstile.secret_key", - "TURNSTILE_SITE_KEY": "turnstile.site_key", - "MYSQL_ENABLED": "mysql.enabled", - "MYSQL_ADDR": "mysql.addr", - "MYSQL_PUBLIC_HOST": "mysql.public_host", - "MYSQL_PUBLIC_PORT": "mysql.public_port", + "DATABASE_URL": "database.url", + "REDIS_URL": "redis.url", + "APP_ENVIRONMENT": "app.environment", + "HOST": "server.host", + "PORT": "server.port", + "LOG_LEVEL": "log.level", + "LOG_FORMAT": "log.format", + "RATE_LIMIT_REQUESTS": "rate_limit.requests", + "RATE_LIMIT_WINDOW": "rate_limit.window", + "RATE_LIMIT_BURST": "rate_limit.burst", + "OTEL_ENDPOINT": "otel.endpoint", + "OTEL_EXPORTER_OTLP_ENDPOINT": "otel.endpoint", + "OTEL_SERVICE_NAME": "otel.service_name", + "OTEL_ENABLED": "otel.enabled", + "OTEL_INSECURE": "otel.insecure", + "OTEL_SAMPLE_RATE": "otel.sample_rate", + "CANARY_BASE_URL": "canary.base_url", + "CANARY_MANAGE_URL": "canary.manage_url", + "TURNSTILE_SECRET_KEY": "turnstile.secret_key", + "TURNSTILE_SITE_KEY": "turnstile.site_key", + "MYSQL_ENABLED": "mysql.enabled", + "MYSQL_ADDR": "mysql.addr", + "MYSQL_PUBLIC_HOST": "mysql.public_host", + "MYSQL_PUBLIC_PORT": "mysql.public_port", + "RATE_LIMIT_CREATE_MIN_RATE": "rate_limit.create_min_rate", + "RATE_LIMIT_CREATE_MIN_BURST": "rate_limit.create_min_burst", + "RATE_LIMIT_CREATE_HOUR_RATE": "rate_limit.create_hour_rate", + "RATE_LIMIT_CREATE_HOUR_BURST": "rate_limit.create_hour_burst", + "NOTIFY_DEDUP_TTL": "notify.dedup_ttl", + "NOTIFY_SEND_TIMEOUT": "notify.send_timeout", + "NOTIFY_MAX_TRIES": "notify.max_tries", + "NOTIFY_MAX_ELAPSED": "notify.max_elapsed", + "NOTIFY_INITIAL_INTERVAL": "notify.initial_interval", + "NOTIFY_RETENTION_INTERVAL": "notify.retention_interval", + "NOTIFY_RETENTION_LIMIT": "notify.retention_limit", + "WEBHOOK_HMAC_SECRET": "notify.webhook_hmac_secret", + "NOTIFY_TELEGRAM_API_BASE": "notify.telegram_api_base", + "NOTIFY_FINGERPRINT_WINDOW": "notify.fingerprint_window", } func envKeyReplacer(s string) string { diff --git a/PROJECTS/beginner/canary-token-generator/backend/internal/token/dto.go b/PROJECTS/beginner/canary-token-generator/backend/internal/token/dto.go index e53daa03..ed326d85 100644 --- a/PROJECTS/beginner/canary-token-generator/backend/internal/token/dto.go +++ b/PROJECTS/beginner/canary-token-generator/backend/internal/token/dto.go @@ -17,7 +17,7 @@ type CreateRequest struct { TelegramChat string `json:"telegram_chat" validate:"required_if=AlertChannel telegram"` WebhookURL string `json:"webhook_url" validate:"required_if=AlertChannel webhook,omitempty,url"` Metadata json.RawMessage `json:"metadata"` - TurnstileResp string `json:"cf_turnstile_response" validate:"required"` + TurnstileResp string `json:"cf_turnstile_response"` } type Response struct {