// Copyright (c) 2026 Petr BalvĂ­n (https://petrbalvin.org) // SPDX-License-Identifier: MIT //go:build linux || freebsd package handler import ( "encoding/json" "errors" "fmt" "io" "log/slog" "mime" "net/http" "net/url" "path/filepath" "slices" "strings" "sync" "time" "sourcedock.dev/petrbalvin/nuntius/internal/config" "sourcedock.dev/petrbalvin/nuntius/internal/contactform" "sourcedock.dev/petrbalvin/nuntius/internal/email" "sourcedock.dev/petrbalvin/nuntius/internal/storage" "sourcedock.dev/petrbalvin/nuntius/internal/telegram" ) const ( // secondsPerHour is the token-bucket refill window. It is part of the // meaning of rate_limit_per_hour, not an independent policy: the // bucket refills at perHour/3600 tokens per second. secondsPerHour = 3600.0 ) // formSender is the interface for delivering a submission. type formSender interface { Send(req contactform.Request) error } // subscriberStorer persists newsletter subscribers. // storage.NewsletterStore satisfies this interface. type subscriberStorer interface { Append(sub storage.Subscriber) error } // duplicateChecker lets the pipeline skip repeat newsletter subscriptions // for an already recorded address. *storage.DedupeNewsletterStore satisfies // it; stores without it keep their previous behaviour. type duplicateChecker interface { Has(email string) bool } // confirmationSender mails the double opt-in link to the subscriber. // *email.FormSender satisfies it once its SMTP identity is configured. type confirmationSender interface { SendConfirmation(to, link string) error } // archiveStorer persists submissions for forms that asked for durability. // *storage.ArchiveStore satisfies it. type archiveStorer interface { Append(sub storage.Submission) error } // acknowledgementSender mails the submitter a receipt. *email.FormSender // satisfies it once its SMTP identity is configured. type acknowledgementSender interface { SendAcknowledgement(to string) error } // telegramNotifier delivers a submission summary to the owner's chat. // *telegram.Notifier satisfies it. type telegramNotifier interface { Notify(formName string, req contactform.Request) error } // ContactHandler serves one or more contact forms, dispatched by URL path. // Each form has its own sender, rate limiter, CORS allowlist, and honeypot. // Newsletter-type forms also get a per-form subscriber store. type ContactHandler struct { trustProxy bool dataDir string // maxBodyBytes caps the request body size to prevent memory // exhaustion; it comes from server.max_body_bytes. maxBodyBytes int // metricsToken guards GET /metrics; empty keeps the endpoint open. metricsToken string forms map[string]*config.Form senders map[string]formSender rateLimits map[string]*rateLimiter stores map[string]subscriberStorer archives map[string]archiveStorer pendings map[string]*storage.PendingStore notifiers map[string]telegramNotifier stats *formStatsRegistry closeOnce sync.Once } // New constructs a ContactHandler that serves all forms defined in cfg. // The server-wide mechanics (body cap, rate limiter memory bounds) and the // per-form policies (validation, pending lifetime) all come from cfg. func New(cfg *config.Config) *ContactHandler { h := &ContactHandler{ trustProxy: cfg.Server.TrustProxyHeaders, dataDir: cfg.DataDir, maxBodyBytes: cfg.Server.BodyLimit(), metricsToken: cfg.Server.MetricsToken, forms: make(map[string]*config.Form, len(cfg.Forms)), senders: make(map[string]formSender, len(cfg.Forms)), rateLimits: make(map[string]*rateLimiter, len(cfg.Forms)), stores: make(map[string]subscriberStorer, len(cfg.Forms)), archives: make(map[string]archiveStorer, len(cfg.Forms)), pendings: make(map[string]*storage.PendingStore, len(cfg.Forms)), notifiers: make(map[string]telegramNotifier, len(cfg.Forms)), } for i := range cfg.Forms { f := &cfg.Forms[i] h.forms[f.Path] = f h.senders[f.Path] = email.NewFormSender(f) h.rateLimits[f.Path] = newRateLimiter(limiterSettings{ perHour: f.RateLimit(), maxBuckets: cfg.Server.MaxRateLimitBuckets(), cleanupEvery: cfg.Server.RateLimitCleanup(), maxBucketAge: cfg.Server.RateLimitMaxBucketAge(), }) if f.Type == "newsletter" { storePath := filepath.Join(cfg.DataDir, "newsletter-"+f.Name+".jsonl") h.stores[f.Path] = storage.NewDedupeNewsletterStore(storage.NewNewsletterStore(storePath)) h.pendings[f.Path] = storage.NewPendingStore( filepath.Join(cfg.DataDir, "newsletter-"+f.Name+"-pending.json"), f.PendingTTL()) } else if f.Archive { h.archives[f.Path] = storage.NewArchiveStore( filepath.Join(cfg.DataDir, "archive-"+f.Name+".jsonl")) } if f.Telegram != nil { h.notifiers[f.Path] = telegram.New( f.Telegram.BotToken, f.Telegram.ChatID, f.Telegram.Timeout()) } } // The registry needs the fully populated form map, hence after the loop. h.stats = newFormStatsRegistry(h.forms) h.restoreState() return h } // Close stops the background cleanup goroutines for all rate limiters. // It is idempotent: calling it more than once, from the shutdown path or // a caller's cleanup, is safe. func (h *ContactHandler) Close() { h.closeOnce.Do(func() { for _, lim := range h.rateLimits { lim.stop() } }) } // Register mounts one POST + OPTIONS handler per form, plus a single // GET /health handler and a GET /metrics endpoint. Newsletter forms also // get a GET /confirm endpoint redeeming their opt-in tokens. func (h *ContactHandler) Register(mux *http.ServeMux) { for path := range h.forms { mux.HandleFunc("POST "+path, h.makeHandler(path)) mux.HandleFunc("OPTIONS "+path, h.makeHandler(path)) if h.pendings[path] != nil { mux.HandleFunc("GET "+path+"/confirm", h.makeConfirmHandler(path)) } } mux.HandleFunc("GET /health", h.Health) mux.HandleFunc("GET /metrics", h.Metrics) } // makeHandler returns the per-form HTTP handler. func (h *ContactHandler) makeHandler(path string) http.HandlerFunc { form := h.forms[path] sender := h.senders[path] limiter := h.rateLimits[path] store := h.stores[path] // nil for non-newsletter forms return func(w http.ResponseWriter, r *http.Request) { origin := r.Header.Get("Origin") // CORS preflight. if r.Method == http.MethodOptions { if formAllowed(form, origin) { writeCORS(w, origin, form.AllowedOrigins) } w.WriteHeader(http.StatusNoContent) return } // CORS on actual request. if formAllowed(form, origin) { writeCORS(w, origin, form.AllowedOrigins) } else if origin != "" { h.bump(path, metricOriginBlocked) respondError(w, http.StatusForbidden, "origin_not_allowed", "") return } h.bump(path, metricReceived) // Rate limit. An explicit rate_limit_per_hour = 0 disables it. if form.RateLimit() > 0 { ip := ClientIP(r, h.trustProxy) if !limiter.allow(ip) { h.bump(path, metricRateLimited) respondError(w, http.StatusTooManyRequests, "rate_limited", "Too many requests, please try again later.") return } } // Parse the body. A plain HTML form post speaks urlencoded or // multipart and carries the same fields as the JSON contract // under fixed names; every other content type speaks JSON. The // size cap applies to all shapes alike. The honeypot check is // bound here as well, because the two shapes carry it // differently: a field on the form, a key in the raw JSON. r.Body = http.MaxBytesReader(w, r.Body, int64(h.maxBodyBytes)) var req contactform.Request mediaType := mediaTypeOf(r.Header.Get("Content-Type")) honeypotHit := func() bool { return false } switch mediaType { case "application/x-www-form-urlencoded", "multipart/form-data": // 1 MiB of in-memory multipart is plenty: only the value // parts are read, file parts are ignored, and the body cap // bounds the whole request anyway. if mediaType == "multipart/form-data" { if err := r.ParseMultipartForm(1 << 20); err != nil { h.parseFailed(w, path, err) return } } else if err := r.ParseForm(); err != nil { h.parseFailed(w, path, err) return } req = requestFromForm(r.PostForm) if hp := form.Honeypot(); hp != "" { honeypotHit = func() bool { return r.PostForm.Get(hp) != "" } } default: body, err := io.ReadAll(r.Body) if err != nil { h.parseFailed(w, path, err) return } if err := json.Unmarshal(body, &req); err != nil { h.bump(path, metricInvalidBody) respondError(w, http.StatusBadRequest, "invalid_body", "Could not parse request body.") return } if hp := form.Honeypot(); hp != "" { honeypotHit = func() bool { var raw map[string]any if err := json.Unmarshal(body, &raw); err != nil { return false } v, ok := raw[hp] return ok && v != nil && v != "" } } } // Honeypot: silently accept but never send. An empty field name // disables the check for this form. if honeypotHit() { h.bump(path, metricHoneypotBlocked) slog.Info("honeypot triggered, dropping silently", "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) respondSuccess(w, r, form.RedirectURL) return } // Validate against the form's configured policy. if errs := contactform.Validate(&req, form.Policy()); len(errs) > 0 { h.bump(path, metricValidationFailed) w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(http.StatusBadRequest) if err := json.NewEncoder(w).Encode(contactform.ErrorResponse{ Error: "validation", Details: errs, }); err != nil { slog.Error("failed to encode json response", "err", err) } return } // Skip repeat newsletter subscriptions for an address that is // already recorded: no second mail, no duplicate log line. The // caller sees the same success response as first-timers. if dup, ok := store.(duplicateChecker); ok && dup.Has(req.Email) { h.bump(path, metricDuplicateSignup) slog.Info("duplicate newsletter signup suppressed", "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) respondSuccess(w, r, form.RedirectURL) return } // Newsletter forms run the double opt-in flow: record a pending // subscription and mail the subscriber a confirmation link. The // owner is notified only once the link is redeemed, so bots that // fill the form cannot flood the inbox. if form.Type == "newsletter" { rawToken, err := storage.RandomToken() if err != nil { h.bump(path, metricSendFailed) slog.Error("token generation failed", "err", err, "form", form.Name, "path", path) respondError(w, http.StatusInternalServerError, "send_failed", "Could not start the subscription.") return } pend := storage.PendingSubscription{ Email: req.Email, IP: ClientIP(r, h.trustProxy), } pending := h.pendings[path] if pending == nil { // unreachable via New(): production always builds one slog.Error("newsletter form without a pending store", "path", path) respondError(w, http.StatusInternalServerError, "storage_failed", "Could not start the subscription.") return } if err := pending.Issue(rawToken, pend); err != nil { h.bump(path, metricPersistFailed) slog.Error("pending subscription store failed", "err", err, "form", form.Name, "path", path) respondError(w, http.StatusInternalServerError, "storage_failed", "Could not start the subscription.") return } link := h.confirmLink(r, path, rawToken) var sendErr error if cs, ok := sender.(confirmationSender); ok { sendErr = cs.SendConfirmation(req.Email, link) } else { sendErr = sender.Send(req) } if sendErr != nil { h.bump(path, metricSendFailed) slog.Error("confirmation mail failed", "err", sendErr, "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) respondError(w, http.StatusInternalServerError, "send_failed", "Could not send the confirmation email.") return } h.bump(path, metricConfirmationSent) slog.Info("confirmation mail sent", "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) respondSuccess(w, r, form.RedirectURL) return } // Archive before sending: the point of the log is that a failed // SMTP round-trip loses nothing. A failed append fails the // request without sending, so a retry cannot split the mail // from its record. if archive := h.archives[path]; archive != nil { if err := archive.Append(storage.Submission{ Form: form.Name, Name: req.Name, Email: req.Email, Service: req.Service, Message: req.Message, IP: ClientIP(r, h.trustProxy), }); err != nil { h.bump(path, metricPersistFailed) slog.Error("submission archive append failed", "err", err, "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) respondError(w, http.StatusInternalServerError, "storage_failed", "Could not record submission.") return } } // Deliver. The owner mail is the record and the Telegram // notification the bell: the submission counts as delivered // when either channel gets through, and only when both fail // (or no bell is configured) does the caller see an error. sendErr := sender.Send(req) if sendErr != nil { h.bump(path, metricSendFailed) slog.Error("send failed", "err", sendErr, "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) } delivered := sendErr == nil if notifier := h.notifiers[path]; notifier != nil { if err := notifier.Notify(form.Name, req); err != nil { h.bump(path, metricTelegramFailed) slog.Error("telegram notification failed", "err", err, "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) } else { delivered = true if sendErr != nil { slog.Warn("telegram delivered after the mail failed", "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) } } } if !delivered { respondError(w, http.StatusInternalServerError, "send_failed", "Could not send email.") return } if sendErr == nil { h.bump(path, metricSent) slog.Info("message sent", "form", form.Name, "path", path, "service", req.Service, "ip", ClientIP(r, h.trustProxy), ) } // The optional receipt to the submitter is best-effort: the // submission is delivered, so a failed acknowledgement must // not turn an accepted submission into an error. if form.AutoReply { if as, ok := sender.(acknowledgementSender); ok { if err := as.SendAcknowledgement(req.Email); err != nil { h.bump(path, metricAutoReplyFailed) slog.Warn("acknowledgement mail failed", "err", err, "form", form.Name, "path", path, "ip", ClientIP(r, h.trustProxy)) } } } respondSuccess(w, r, form.RedirectURL) } } // confirmLink builds the absolute opt-in URL for a token. The scheme is // https whenever a trusted proxy reports X-Forwarded-Proto=https, matching // the ClientIP trust model. func (h *ContactHandler) confirmLink(r *http.Request, path, rawToken string) string { scheme := "http" if h.trustProxy && r.Header.Get("X-Forwarded-Proto") == "https" { scheme = "https" } return fmt.Sprintf("%s://%s%s/confirm?token=%s", scheme, r.Host, path, rawToken) } // makeConfirmHandler redeems a double opt-in token: the pending entry moves // into the confirmed subscriber log and the owner is notified best-effort. // The response is HTML because humans open these links in browsers. func (h *ContactHandler) makeConfirmHandler(path string) http.HandlerFunc { form := h.forms[path] sender := h.senders[path] store := h.stores[path] pending := h.pendings[path] return func(w http.ResponseWriter, r *http.Request) { writePage := func(code int, title, detail string) { w.Header().Set("Content-Type", "text/html; charset=utf-8") w.WriteHeader(code) fmt.Fprintf(w, ""+ "nuntius"+ "

%s

%s

", title, detail) } rawToken := r.URL.Query().Get("token") sub, ok := pending.Peek(rawToken) if !ok { h.bump(path, metricConfirmFailed) writePage(http.StatusGone, "Link expired", "This confirmation link is invalid or has expired. Please sign up again.") return } // The main record lands before the pending entry is dropped, so a // storage failure keeps the token redeemable and nothing is lost. alreadyRecorded := false if dup, ok := store.(duplicateChecker); ok && dup.Has(sub.Email) { alreadyRecorded = true } if !alreadyRecorded { if err := store.Append(storage.Subscriber{ Email: sub.Email, IP: sub.IP, Form: form.Name, }); err != nil { h.bump(path, metricPersistFailed) slog.Error("confirmed subscription append failed", "err", err, "form", form.Name, "path", path) writePage(http.StatusInternalServerError, "Almost there", "The confirmation could not be saved. Please try the link again shortly.") return } } pending.Consume(rawToken) h.bump(path, metricConfirmed) slog.Info("newsletter subscription confirmed", "form", form.Name, "path", path) // Owner notification is best-effort and must not affect the // subscriber's result. if err := sender.Send(contactform.Request{ Name: "(nuntius)", Email: sub.Email, Message: "The address above confirmed its newsletter subscription.", }); err != nil { h.bump(path, metricSendFailed) slog.Error("owner notification failed", "err", err, "form", form.Name, "path", path) } writePage(http.StatusOK, "Subscription confirmed", "The address "+sub.Email+" is now subscribed to \""+form.Name+"\".") } } // Health handles GET /health. func (h *ContactHandler) Health(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json; charset=utf-8") if err := json.NewEncoder(w).Encode(map[string]any{ "status": "ok", "forms": len(h.forms), }); err != nil { slog.Error("failed to encode health response", "err", err) } } // --- internals --- func formAllowed(form *config.Form, origin string) bool { if origin == "" { return true } return slices.Contains(form.AllowedOrigins, origin) } func writeCORS(w http.ResponseWriter, origin string, allowed []string) { // Only echo the origin back if it is in the allowlist. if slices.Contains(allowed, origin) { w.Header().Set("Access-Control-Allow-Origin", origin) w.Header().Set("Vary", "Origin") w.Header().Set("Access-Control-Allow-Methods", "POST, OPTIONS") w.Header().Set("Access-Control-Allow-Headers", "Content-Type") return } } // rateLimiter is a per-IP token bucket. Its memory mechanics (the cap on // distinct buckets, the cleanup tick, the age at which an idle bucket is // dropped) come from the server configuration, not from constants here. type rateLimiter struct { mu sync.Mutex perHour int maxBuckets int cleanupEvery time.Duration maxBucketAge time.Duration buckets map[string]*bucket stopCh chan struct{} } type bucket struct { tokens float64 last time.Time } // limiterSettings carries the configured mechanics of one rate limiter. type limiterSettings struct { perHour int maxBuckets int cleanupEvery time.Duration maxBucketAge time.Duration } func newRateLimiter(s limiterSettings) *rateLimiter { r := &rateLimiter{ perHour: s.perHour, maxBuckets: s.maxBuckets, cleanupEvery: s.cleanupEvery, maxBucketAge: s.maxBucketAge, buckets: make(map[string]*bucket), stopCh: make(chan struct{}), } // Clean up entries older than the configured bucket age on every tick. r.startCleanup() return r } // startCleanup launches a background goroutine that periodically removes // expired bucket entries to prevent unbounded memory growth. func (r *rateLimiter) startCleanup() { go func() { ticker := time.NewTicker(r.cleanupEvery) defer ticker.Stop() for { select { case <-ticker.C: r.cleanup(r.maxBucketAge) case <-r.stopCh: return } } }() } // stop terminates the cleanup goroutine. func (r *rateLimiter) stop() { close(r.stopCh) } // snapshot returns a copy of every live bucket so state can be written to // disk without holding the lock while encoding. func (r *rateLimiter) snapshot() map[string]bucket { r.mu.Lock() defer r.mu.Unlock() out := make(map[string]bucket, len(r.buckets)) for ip, b := range r.buckets { out[ip] = *b } return out } // restore merges persisted buckets, dropping entries older than the // configured bucket age and stopping once the map cap is reached. Entries // newer than the cutoff keep their remaining tokens. func (r *rateLimiter) restore(entries map[string]bucket, now time.Time) { r.mu.Lock() defer r.mu.Unlock() for ip, b := range entries { if len(r.buckets) >= r.maxBuckets { return } cutoff := now.Add(-r.maxBucketAge) if b.last.Before(cutoff) || b.last.After(now) { continue } entry := b entry.tokens = min(entry.tokens, float64(r.perHour)) r.buckets[ip] = &entry } } func (r *rateLimiter) cleanup(maxAge time.Duration) { r.mu.Lock() defer r.mu.Unlock() cutoff := time.Now().Add(-maxAge) for ip, b := range r.buckets { if b.last.Before(cutoff) { delete(r.buckets, ip) } } } func (r *rateLimiter) allow(ip string) bool { r.mu.Lock() defer r.mu.Unlock() now := time.Now() b, ok := r.buckets[ip] if !ok { if len(r.buckets) >= r.maxBuckets { return false } b = &bucket{tokens: float64(r.perHour), last: now} r.buckets[ip] = b } rate := float64(r.perHour) / secondsPerHour elapsed := now.Sub(b.last).Seconds() b.tokens = min(b.tokens+elapsed*rate, float64(r.perHour)) b.last = now if b.tokens < 1 { return false } b.tokens-- return true } func respondOK(w http.ResponseWriter) { w.Header().Set("Content-Type", "application/json; charset=utf-8") if err := json.NewEncoder(w).Encode(contactform.Response{OK: true}); err != nil { slog.Error("failed to encode ok response", "err", err) } } // respondSuccess answers an accepted submission. A form with a // redirect_url speaks browser: 303 See Other to the configured page, so a // plain HTML form works without JavaScript and a bot hit is // indistinguishable from a real one. The JSON contract is the default. func respondSuccess(w http.ResponseWriter, r *http.Request, redirectURL string) { if redirectURL != "" { http.Redirect(w, r, redirectURL, http.StatusSeeOther) return } respondOK(w) } // parseFailed answers an unreadable, oversized or unparsable request body. // An over-cap body is 413 regardless of the shape; everything else is a // 400 invalid_body. func (h *ContactHandler) parseFailed(w http.ResponseWriter, path string, err error) { if _, ok := errors.AsType[*http.MaxBytesError](err); ok { h.bump(path, metricBodyTooLarge) respondError(w, http.StatusRequestEntityTooLarge, "body_too_large", "Request body too large.") return } h.bump(path, metricInvalidBody) respondError(w, http.StatusBadRequest, "invalid_body", "Could not parse request body.") } // mediaTypeOf extracts the bare media type from a Content-Type header, // lower-cased and without parameters. func mediaTypeOf(header string) string { mt, _, err := mime.ParseMediaType(header) if err != nil { return strings.ToLower(strings.TrimSpace(header)) } return mt } // requestFromForm builds the request from posted form fields. The names // are fixed for the plain HTML shape: name, email, service, message, // plus the configured honeypot field, which the pipeline reads // separately. File parts have no counterpart in the contract and are // ignored. func requestFromForm(v url.Values) contactform.Request { return contactform.Request{ Name: v.Get("name"), Email: v.Get("email"), Service: v.Get("service"), Message: v.Get("message"), } } func respondError(w http.ResponseWriter, code int, err, msg string) { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(code) if encErr := json.NewEncoder(w).Encode(contactform.ErrorResponse{ Error: err, Message: msg, }); encErr != nil { slog.Error("failed to encode error response", "err", encErr) } } // ClientIP returns the address used for rate limiting and logging. // // With trustProxy false only the connection peer address is considered; // it cannot be forged by the caller. With trustProxy true, headers set by // a trusted reverse proxy take precedence: the first X-Forwarded-For // entry, then X-Real-IP. Enable it only when such a proxy sits directly // in front of nuntius and overwrites those headers rather than appending // to them. func ClientIP(r *http.Request, trustProxy bool) string { if trustProxy { if xff := r.Header.Get("X-Forwarded-For"); xff != "" { if before, _, ok := strings.Cut(xff, ","); ok { return strings.TrimSpace(before) } return strings.TrimSpace(xff) } if xr := r.Header.Get("X-Real-IP"); xr != "" { return xr } } host := r.RemoteAddr if i := strings.LastIndexByte(host, ':'); i >= 0 { host = host[:i] } return host }