Verify signatures

import express from 'express';
import { createHmac, timingSafeEqual } from 'node:crypto';
import { gunzipSync } from 'node:zlib';

const TOLERANCE_SECONDS = 300;
const CANONICAL_T = /^(0|[1-9][0-9]*)$/;
const LOWER_HEX_64 = /^[0-9a-f]{64}$/;
// Refuse above this before inflating; cap what inflating can produce.
const MAX_COMPRESSED_BYTES = 2 * 1024 * 1024;
const MAX_UNCOMPRESSED_BYTES = 8 * 1024 * 1024;
const SECRETS = [
  process.env.LUA_DRAIN_SECRET,
  process.env.LUA_DRAIN_SECRET_PREVIOUS,
].filter(Boolean);

export function verifySignature(header, rawBody, secrets, nowSeconds = Math.floor(Date.now() / 1000)) {
  if (!header || secrets.length === 0) return false;

  let t = null;
  const candidates = [];
  for (const piece of header.split(',')) {
    const eq = piece.indexOf('=');
    if (eq < 0) return false;
    const key = piece.slice(0, eq).trim();
    const value = piece.slice(eq + 1).trim();
    // Reject, never coerce: Number() would happily take '01789217999',
    // '1789217999.5' and '0x6aa54ccd', and none of those is a `t` Lua sends.
    if (key === 'v1') {
      if (!LOWER_HEX_64.test(value)) return false;
      candidates.push(value);
    } else if (key === 't') {
      if (!CANONICAL_T.test(value)) return false;
      t = Number(value);
    }
  }
  if (t === null || !Number.isSafeInteger(t) || candidates.length === 0) return false;
  if (Math.abs(nowSeconds - t) > TOLERANCE_SECONDS) return false; // inclusive at 300

  const base = Buffer.concat([Buffer.from(`${t}.`, 'utf8'), rawBody]);

  let ok = false;
  for (const secret of secrets) {
    const expected = createHmac('sha256', secret).update(base).digest();
    for (const candidate of candidates) {
      const given = Buffer.from(candidate, 'hex');
      if (given.length === expected.length && timingSafeEqual(given, expected)) ok = true;
    }
  }
  return ok;
}

const app = express();

// `limit` is the COMPRESSED cap: express checks Content-Length and the bytes as
// they arrive, and its default error handler answers 413 — all before we see the
// body. `inflate:false` keeps the compressed bytes so we decompress them ourselves.
app.post('/lua', express.raw({ type: '*/*', limit: MAX_COMPRESSED_BYTES, inflate: false }), (req, res) => {
  let rawBody = req.body;
  if (req.get('content-encoding') === 'gzip') {
    try {
      // maxOutputLength makes zlib refuse to ALLOCATE past the cap: it throws
      // rather than returning the buffer, so a bomb is never held in memory.
      rawBody = gunzipSync(req.body, { maxOutputLength: MAX_UNCOMPRESSED_BYTES });
    } catch (err) {
      const tooLarge = err?.code === 'ERR_BUFFER_TOO_LARGE';
      return res.status(tooLarge ? 413 : 400).json({ error: tooLarge ? 'too large' : 'bad gzip' });
    }
  }

  if (!verifySignature(req.get('X-Lua-Signature'), rawBody, SECRETS)) {
    return res.status(401).json({ error: 'bad signature' });
  }

  const token = req.get('X-Lua-Verify');
  if (token) res.set('X-Lua-Verify', token);

  const batch = JSON.parse(rawBody.toString('utf8'));
  for (const record of batch.records) store(batch.resource, record);

  res.status(204).end();
});

function store(resource, record) {
  console.log(resource['lua.agent.id'], record.id, record.eventName, record.severityText);
}

app.listen(8080);

import hashlib
import hmac
import json
import os
import re
import time
import zlib

from flask import Flask, request, Response

TOLERANCE_SECONDS = 300
CANONICAL_T = re.compile(r"\A(0|[1-9][0-9]*)\Z")
LOWER_HEX_64 = re.compile(r"\A[0-9a-f]{64}\Z")
# Refuse above this before inflating; cap what inflating can produce.
MAX_COMPRESSED_BYTES = 2 * 1024 * 1024
MAX_UNCOMPRESSED_BYTES = 8 * 1024 * 1024
SECRETS = [
    s for s in (os.environ.get("LUA_DRAIN_SECRET"), os.environ.get("LUA_DRAIN_SECRET_PREVIOUS")) if s
]


def verify_signature(header, raw_body, secrets, now=None):
    if not header or not secrets:
        return False

    timestamp = None
    candidates = []
    for piece in header.split(","):
        key, sep, value = piece.partition("=")
        if not sep:
            return False
        key, value = key.strip(), value.strip()
        # Reject, never coerce: int() accepts "01789217999", "+1789217999" and
        # surrounding whitespace, and none of those is a `t` Lua sends.
        if key == "v1":
            if not LOWER_HEX_64.match(value):
                return False
            candidates.append(bytes.fromhex(value))
        elif key == "t":
            if not CANONICAL_T.match(value):
                return False
            timestamp = value
    if timestamp is None or not candidates:
        return False

    t = int(timestamp)
    if abs(int(now if now is not None else time.time()) - t) > TOLERANCE_SECONDS:  # inclusive at 300
        return False

    base = f"{t}.".encode("utf-8") + raw_body

    ok = False
    for secret in secrets:
        expected = hmac.new(secret.encode("utf-8"), base, hashlib.sha256).digest()
        for candidate in candidates:
            # compare_digest is constant time, over bytes; no early return, so
            # timing says nothing about which secret or which v1 matched.
            if hmac.compare_digest(expected, candidate):
                ok = True
    return ok


app = Flask(__name__)
# The COMPRESSED cap, enforced by Werkzeug before your view runs.
app.config["MAX_CONTENT_LENGTH"] = MAX_COMPRESSED_BYTES


def too_large():
    return Response('{"error":"too large"}', status=413, mimetype="application/json")


@app.post("/lua")
def receive():
    # Content-Length is a claim, so check it and then the bytes that arrived.
    if request.content_length is not None and request.content_length > MAX_COMPRESSED_BYTES:
        return too_large()

    raw_body = request.get_data()  # Werkzeug does not inflate request bodies for us
    if len(raw_body) > MAX_COMPRESSED_BYTES:
        return too_large()

    if request.headers.get("Content-Encoding") == "gzip":
        # decompress(data, max_length) stops the inflater at max_length and parks
        # the rest of the INPUT in unconsumed_tail, so the output buffer can never
        # grow past the cap. wbits=31 selects gzip framing.
        inflater = zlib.decompressobj(wbits=31)
        try:
            raw_body = inflater.decompress(raw_body, MAX_UNCOMPRESSED_BYTES + 1)
        except zlib.error:
            return Response('{"error":"bad gzip"}', status=400, mimetype="application/json")
        if len(raw_body) > MAX_UNCOMPRESSED_BYTES or inflater.unconsumed_tail:
            return too_large()
        # Lua sends exactly one gzip member, and this object decodes only the
        # first, so a truncated stream or trailing bytes means something else.
        if not inflater.eof or inflater.unused_data:
            return Response('{"error":"bad gzip"}', status=400, mimetype="application/json")

    if not verify_signature(request.headers.get("X-Lua-Signature", ""), raw_body, SECRETS):
        return Response('{"error":"bad signature"}', status=401, mimetype="application/json")

    response = Response(status=204)
    token = request.headers.get("X-Lua-Verify")
    if token:
        response.headers["X-Lua-Verify"] = token

    batch = json.loads(raw_body)
    for record in batch["records"]:
        store(batch["resource"], record)

    return response


def store(resource, record):
    print(resource["lua.agent.id"], record["id"], record["eventName"], record["severityText"])

package main

import (
	"bytes"
	"compress/gzip"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"
)

const toleranceSeconds int64 = 300

// Refuse above this before inflating; cap what inflating can produce.
const (
	maxCompressedBytes   int64 = 2 << 20
	maxUncompressedBytes int64 = 8 << 20
)

var (
	canonicalT = regexp.MustCompile(`^(0|[1-9][0-9]*)$`)
	lowerHex64 = regexp.MustCompile(`^[0-9a-f]{64}$`)
	secrets    = loadSecrets()
)

func loadSecrets() []string {
	var out []string
	for _, name := range []string{"LUA_DRAIN_SECRET", "LUA_DRAIN_SECRET_PREVIOUS"} {
		if v := os.Getenv(name); v != "" {
			out = append(out, v)
		}
	}
	return out
}

// VerifySignature checks X-Lua-Signature against the UNCOMPRESSED body.
func VerifySignature(header string, rawBody []byte, keys []string, now time.Time) bool {
	if header == "" || len(keys) == 0 {
		return false
	}

	var ts int64
	haveTS := false
	var candidates []string
	for _, piece := range strings.Split(header, ",") {
		key, value, found := strings.Cut(piece, "=")
		if !found {
			return false
		}
		key, value = strings.TrimSpace(key), strings.TrimSpace(value)
		// Reject, never coerce: ParseInt accepts "01789217999" and "+1789217999",
		// and hex.DecodeString accepts uppercase — neither is what Lua sends.
		switch key {
		case "t":
			if !canonicalT.MatchString(value) {
				return false
			}
			n, err := strconv.ParseInt(value, 10, 64)
			if err != nil {
				return false
			}
			ts, haveTS = n, true
		case "v1":
			if !lowerHex64.MatchString(value) {
				return false
			}
			candidates = append(candidates, value)
		}
	}
	if !haveTS || len(candidates) == 0 {
		return false
	}
	if delta := now.Unix() - ts; delta > toleranceSeconds || delta < -toleranceSeconds { // inclusive at 300
		return false
	}

	base := append([]byte(strconv.FormatInt(ts, 10)+"."), rawBody...)

	ok := false
	for _, key := range keys {
		mac := hmac.New(sha256.New, []byte(key))
		mac.Write(base)
		expected := mac.Sum(nil)
		for _, candidate := range candidates {
			given, err := hex.DecodeString(candidate)
			if err != nil {
				continue
			}
			// hmac.Equal is constant time; no early return, so timing says nothing.
			if hmac.Equal(given, expected) {
				ok = true
			}
		}
	}
	return ok
}

type batch struct {
	BatchID  string            `json:"batchId"`
	Resource map[string]any    `json:"resource"`
	Records  []json.RawMessage `json:"records"`
}

func receive(w http.ResponseWriter, r *http.Request) {
	// MaxBytesReader answers on the COMPRESSED body, before anything is inflated.
	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxCompressedBytes))
	if err != nil {
		http.Error(w, "too large", http.StatusRequestEntityTooLarge)
		return
	}

	rawBody := body
	if r.Header.Get("Content-Encoding") == "gzip" {
		zr, gzErr := gzip.NewReader(bytes.NewReader(body))
		if gzErr != nil {
			http.Error(w, "bad gzip", http.StatusBadRequest)
			return
		}
		defer zr.Close()
		// LimitReader stops the copy one byte past the cap, so the destination
		// buffer can never grow beyond it: a bomb is refused, not held in memory.
		// The extra byte distinguishes "exactly at the cap" from "over it".
		if rawBody, err = io.ReadAll(io.LimitReader(zr, maxUncompressedBytes+1)); err != nil {
			http.Error(w, "bad gzip", http.StatusBadRequest)
			return
		}
		if int64(len(rawBody)) > maxUncompressedBytes {
			http.Error(w, "too large", http.StatusRequestEntityTooLarge)
			return
		}
	}

	if !VerifySignature(r.Header.Get("X-Lua-Signature"), rawBody, secrets, time.Now()) {
		http.Error(w, "bad signature", http.StatusUnauthorized)
		return
	}

	if token := r.Header.Get("X-Lua-Verify"); token != "" {
		w.Header().Set("X-Lua-Verify", token)
	}

	var b batch
	if err := json.Unmarshal(rawBody, &b); err != nil {
		http.Error(w, "bad json", http.StatusBadRequest)
		return
	}
	for _, record := range b.Records {
		store(b.Resource, record)
	}

	w.WriteHeader(http.StatusNoContent)
}

func store(resource map[string]any, record json.RawMessage) {
	var r struct {
		ID           string `json:"id"`
		EventName    string `json:"eventName"`
		SeverityText string `json:"severityText"`
	}
	if err := json.Unmarshal(record, &r); err != nil {
		return
	}
	fmt.Println(resource["lua.agent.id"], r.ID, r.EventName, r.SeverityText)
}

func main() {
	http.HandleFunc("/lua", receive)
	log.Fatal(http.ListenAndServe(":8080", nil))
}