siem-backend/internal/ingress/server.go

package ingress

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"strings"
	"time"

	"example.com/siem-greenfield/internal/config"
	"example.com/siem-greenfield/internal/contracts"
	"example.com/siem-greenfield/internal/metrics"
	"example.com/siem-greenfield/internal/postgres"
	"example.com/siem-greenfield/internal/queue"
)

func Run(ctx context.Context, cfg config.Config) error {
	pg, err := postgres.Open(ctx, cfg.PostgresURL)
	if err != nil {
		return fmt.Errorf("postgres: %w", err)
	}
	defer pg.Close()
	prod := queue.NewProducer(cfg.KafkaBrokers, cfg.KafkaTopic)
	defer prod.Close()
	m := metrics.New()
	accepted := m.Counter("siem_ingress_events_accepted_total")
	rejected := m.Counter("siem_ingress_requests_rejected_total")
	batches := m.Counter("siem_ingress_batches_total")
	mux := http.NewServeMux()
	mux.Handle("/metrics", m.Handler())
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, 200, map[string]string{"status": "ok"}) })
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
		cctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
		defer cancel()
		if e := pg.Pool.Ping(cctx); e != nil {
			writeJSON(w, 503, map[string]string{"status": "not_ready"})
			return
		}
		writeJSON(w, 200, map[string]string{"status": "ready"})
	})
	mux.HandleFunc("/ingest", func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			rejected.Add(1)
			writeJSON(w, 405, map[string]string{"error": "method not allowed"})
			return
		}
		apiKey := strings.TrimSpace(r.Header.Get("X-API-Key"))
		if apiKey == "" {
			rejected.Add(1)
			writeJSON(w, 401, map[string]string{"error": "missing api key"})
			return
		}
		r.Body = http.MaxBytesReader(w, r.Body, cfg.MaxBodyBytes)
		defer r.Body.Close()
		dec := json.NewDecoder(r.Body)
		dec.DisallowUnknownFields()
		var batch []contracts.LogPayload
		if e := dec.Decode(&batch); e != nil || (func() bool { var x any; return dec.Decode(&x) != io.EOF })() {
			rejected.Add(1)
			writeJSON(w, 400, map[string]string{"error": "invalid json"})
			return
		}
		if len(batch) == 0 || len(batch) > cfg.MaxBatchEvents {
			rejected.Add(1)
			writeJSON(w, 400, map[string]string{"error": "invalid batch size"})
			return
		}
		host := strings.TrimSpace(batch[0].Hostname)
		for i := range batch {
			if e := validate(&batch[i]); e != nil {
				rejected.Add(1)
				writeJSON(w, 400, map[string]string{"error": fmt.Sprintf("invalid payload at index %d: %v", i, e)})
				return
			}
			if batch[i].Hostname != host {
				rejected.Add(1)
				writeJSON(w, 400, map[string]string{"error": "all events in a batch must use the same hostname"})
				return
			}
		}
		cctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
		defer cancel()
		agentID, e := pg.AuthenticateOrEnroll(cctx, cfg.TenantID, host, apiKey, strings.TrimSpace(r.Header.Get("X-Enrollment-Key")), cfg.EnrollmentKey, postgres.ClientIP(r.RemoteAddr))
		if e != nil {
			rejected.Add(1)
			if errors.Is(e, postgres.ErrUnauthorized) {
				writeJSON(w, 401, map[string]string{"error": "invalid api key or hostname"})
			} else {
				log.Printf("auth: %v", e)
				writeJSON(w, 503, map[string]string{"error": "control plane unavailable"})
			}
			return
		}
		env := contracts.IngestEnvelope{Version: 1, TenantID: cfg.TenantID, AgentID: agentID, BatchUID: batchUID(agentID, batch), RemoteIP: postgres.ClientIP(r.RemoteAddr), ReceivedAt: time.Now().UTC(), Events: batch}
		payload, e := json.Marshal(env)
		if e != nil {
			writeJSON(w, 500, map[string]string{"error": "internal error"})
			return
		}
		qctx, qcancel := context.WithTimeout(r.Context(), 8*time.Second)
		defer qcancel()
		if e = prod.Write(qctx, agentID, payload); e != nil {
			log.Printf("queue write: %v", e)
			writeJSON(w, 503, map[string]string{"error": "ingest queue unavailable"})
			return
		}
		accepted.Add(uint64(len(batch)))
		batches.Add(1)
		writeJSON(w, 202, map[string]int{"accepted": len(batch)})
	})
	srv := &http.Server{Addr: cfg.ServiceAddr, Handler: withLimits(mux), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 15 * time.Second, IdleTimeout: 60 * time.Second}
	go func() {
		<-ctx.Done()
		cctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_ = srv.Shutdown(cctx)
	}()
	log.Printf("ingress listening on %s", cfg.ServiceAddr)
	e := srv.ListenAndServe()
	if e == http.ErrServerClosed {
		return nil
	}
	return e
}
func batchUID(agentID string, batch []contracts.LogPayload) string {
	b, _ := json.Marshal(batch)
	h := sha256.New()
	_, _ = h.Write([]byte(agentID))
	_, _ = h.Write([]byte{0})
	_, _ = h.Write(b)
	return hex.EncodeToString(h.Sum(nil))
}

func validate(p *contracts.LogPayload) error {
	p.Hostname = strings.TrimSpace(p.Hostname)
	p.Channel = strings.TrimSpace(p.Channel)
	p.Source = strings.TrimSpace(p.Source)
	if p.Hostname == "" || len(p.Hostname) > 255 {
		return fmt.Errorf("invalid host")
	}
	if p.Channel == "" || len(p.Channel) > 128 {
		return fmt.Errorf("invalid channel")
	}
	if p.Source == "" || len(p.Source) > 255 {
		return fmt.Errorf("invalid source")
	}
	if p.EventID == 0 {
		return fmt.Errorf("event id required")
	}
	if p.Time.IsZero() {
		return fmt.Errorf("ts required")
	}
	if strings.TrimSpace(p.Message) == "" && p.Metadata == nil {
		return fmt.Errorf("either msg or meta required")
	}
	if len(p.Message) > 2*1024*1024 {
		return fmt.Errorf("msg too large")
	}
	return nil
}
func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(v)
}
func withLimits(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("X-Content-Type-Options", "nosniff")
		w.Header().Set("Cache-Control", "no-store")
		next.ServeHTTP(w, r)
	})
}