lrosales
/
broad-announce


			
				
					
						
						
							123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
							// HTTP POST handler for ingestd. Implements the seven protection
// layers from SPEC §22 in order:
//
//   1. payload-size cap
//   2. (per-IP) — M1: deferred to M5 with the WS path
//   3. per-source token bucket
//   4. per-company token bucket
//   5. schema validate
//   6. (broker CB) — M9
//   7. (quarantine) — M9
//
// Auth: Stripe-style HMAC-SHA256 in header `X-BA-Signature: t=<ts>,v1=<hex>`.
// The source's secret is keyed by (company_id, source_id) — for M0
// the secret is fetched from the source table; for M0 the simplest
// impl is a single env var per source, but we go straight to the
// DB lookup so M2 doesn't have to change this path.
package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"crypto/subtle"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

	"git3.techno-world.net/lrosales/broad-announce/internal/config"
)

// httpDeps is the HTTP-handler-scoped wrapper. It embeds
// processDeps (the shared pipeline used by both HTTP and MQTT)
// and adds the HTTP-only fields (MaxBytes, raw request/response
// types). The body of the handler is a single ProcessAlert call.
type httpDeps struct {
	processDeps
	MaxBytes int
}

// AcceptResponse is the JSON body returned on 202.
type AcceptResponse struct {
	AlertID     string `json:"alert_id"`
	DedupeCount uint32 `json:"dedupe_count"`
	ReceivedAt  string `json:"received_at"`
}

// RegisterRoutes wires the ingestd HTTP routes onto the given mux.
func RegisterRoutes(mux *http.ServeMux, d *httpDeps) {
	mux.HandleFunc("POST /v1/ingest", d.handleIngest)
}

// handleIngest is the M0 HTTP POST endpoint. The body of the
// pipeline is the shared processDeps.ProcessAlert; this handler
// only adds the HTTP-specific bits (MaxBytesReader, header
// signature, response shape).
func (d *httpDeps) handleIngest(w http.ResponseWriter, r *http.Request) {
	// 1. Payload-size cap. We use MaxBytesReader so a streaming
	//    client can't lie about Content-Length and try to OOM us.
	r.Body = http.MaxBytesReader(w, r.Body, int64(d.MaxBytes))
	body, err := io.ReadAll(r.Body)
	if err != nil {
		var maxErr *http.MaxBytesError
		if errors.As(err, &maxErr) {
			d.Metrics.AlertsReceived.WithLabelValues("http", "payload_too_large").Inc()
			writeErr(w, http.StatusRequestEntityTooLarge, "payload_too_large", "")
			return
		}
		d.Metrics.AlertsReceived.WithLabelValues("http", "invalid").Inc()
		writeErr(w, http.StatusBadRequest, "bad_request", err.Error())
		return
	}
	_ = r.Body.Close()

	res := d.ProcessAlert(r.Context(), body, r.Header.Get("X-BA-Signature"))
	if !res.Accepted {
		// Rate-limit rejections get a Retry-After header.
		if res.RejectReason == "rate_limited_source" || res.RejectReason == "rate_limited_company" {
			if n, err := strconv.Atoi(res.Detail); err == nil {
				w.Header().Set("Retry-After", strconv.Itoa(n))
			}
		}
		writeErr(w, res.HTTPStatus, res.RejectReason, res.Detail)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusAccepted)
	_ = json.NewEncoder(w).Encode(AcceptResponse{
		AlertID:     res.AlertID,
		DedupeCount: res.DedupeCount,
		ReceivedAt:  time.Now().UTC().Format(time.RFC3339Nano),
	})
}

// now is a thin alias for processDeps.now (kept here so the
// existing tests that call d.now() on httpDeps still work).
func (d *httpDeps) now() time.Time { return d.processDeps.Now() }

// verifyHMAC parses `X-BA-Signature: t=<unix>,v1=<hex>` and checks
// HMAC-SHA256(secret, "<unix>.<body>") == hex. Replay window: 5 min.
func verifyHMAC(header string, secret, body []byte, now time.Time) bool {
	if header == "" || len(secret) == 0 {
		return false
	}
	var tsStr, sigHex string
	for _, part := range strings.Split(header, ",") {
		kv := strings.SplitN(strings.TrimSpace(part), "=", 2)
		if len(kv) != 2 {
			continue
		}
		switch kv[0] {
		case "t":
			tsStr = kv[1]
		case "v1":
			sigHex = kv[1]
		}
	}
	if tsStr == "" || sigHex == "" {
		return false
	}
	tsInt, err := strconv.ParseInt(tsStr, 10, 64)
	if err != nil {
		return false
	}
	ts := time.Unix(tsInt, 0)
	if abs(now.Sub(ts)) > 5*time.Minute {
		return false
	}
	mac := hmac.New(sha256.New, secret)
	mac.Write([]byte(tsStr))
	mac.Write([]byte("."))
	mac.Write(body)
	expected := mac.Sum(nil)
	got, err := hex.DecodeString(sigHex)
	if err != nil {
		return false
	}
	return subtle.ConstantTimeCompare(expected, got) == 1
}

func abs(d time.Duration) time.Duration {
	if d < 0 {
		return -d
	}
	return d
}

func writeErr(w http.ResponseWriter, code int, reason, detail string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	body := map[string]string{"error": reason}
	if detail != "" {
		body["detail"] = detail
	}
	_ = json.NewEncoder(w).Encode(body)
}

// loadSourcesFromEnv parses BA_INGESTD_SOURCES as a comma-separated
// list of company_id:source_id:secret triples. M0 dev-mode only;
// M2 swaps this for a Postgres lookup.
func loadSourcesFromEnv(logger *slog.Logger) map[string]SourceConfig {
	out := map[string]SourceConfig{}
	raw := os.Getenv("BA_INGESTD_SOURCES")
	if raw == "" {
		logger.Warn("no BA_INGESTD_SOURCES configured; ingestd will reject all sources")
		return out
	}
	for _, triple := range strings.Split(raw, ",") {
		parts := strings.SplitN(triple, ":", 3)
		if len(parts) != 3 {
			logger.Warn("bad BA_INGESTD_SOURCES entry, expected company:source:secret", "entry", triple)
			continue
		}
		key := parts[0] + ":" + parts[1]
		out[key] = SourceConfig{
			CompanyID:       parts[0],
			SourceID:        parts[1],
			HMACSecret:      []byte(parts[2]),
			RateLimitPerSec: config.GetInt("BA_INGESTD_RATE_LIMIT_PER_SOURCE", 100),
		}
	}
	logger.Info("loaded sources from env", "count", len(out))
	return out
}