// 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=,v1=`. // 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("payload_too_large").Inc() writeErr(w, http.StatusRequestEntityTooLarge, "payload_too_large", "") return } d.Metrics.AlertsReceived.WithLabelValues("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=,v1=` and checks // HMAC-SHA256(secret, ".") == 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 }