webhooker/internal/handlers/webhook.go

package handlers

import (
	"encoding/json"
	"errors"
	"io"
	"net/http"

	"github.com/go-chi/chi"
	"gorm.io/gorm"
	"sneak.berlin/go/webhooker/internal/database"
	"sneak.berlin/go/webhooker/internal/delivery"
	"sneak.berlin/go/webhooker/internal/logfield"
	"sneak.berlin/go/webhooker/internal/signature"
)

const (
	// maxWebhookBodySize is the maximum allowed webhook
	// request body (1 MB).
	maxWebhookBodySize = 1 << maxBodyShift
)

// HandleWebhook handles incoming webhook requests at entrypoint
// URLs.
func (h *Handlers) HandleWebhook() http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			w.Header().Set("Allow", "POST")
			http.Error(
				w,
				"Method Not Allowed",
				http.StatusMethodNotAllowed,
			)

			return
		}

		entrypointUUID := chi.URLParam(r, "uuid")
		if entrypointUUID == "" {
			http.NotFound(w, r)

			return
		}

		entrypoint, ok := h.lookupEntrypoint(
			w, r, entrypointUUID,
		)
		if !ok {
			return
		}

		// Logged only once the UUID is known to name a real
		// entrypoint. The UUID comes straight out of the path on
		// the one unauthenticated endpoint, so logging it before
		// the lookup let a client write an INFO line per invented
		// path; the request itself is already in the access log
		// and a miss is already logged at DEBUG.
		h.log.Info("webhook request received",
			"entrypoint_uuid", entrypointUUID,
			"method", r.Method,
			"remote_addr", r.RemoteAddr,
		)

		if !entrypoint.Active {
			http.Error(w, "Gone", http.StatusGone)

			return
		}

		h.processWebhookRequest(w, r, entrypoint)
	}
}

// processWebhookRequest reads the body, verifies the sender,
// serializes headers, loads targets, and delivers the event.
func (h *Handlers) processWebhookRequest(
	w http.ResponseWriter,
	r *http.Request,
	entrypoint database.Entrypoint,
) {
	body, ok := h.readWebhookBody(w, r)
	if !ok {
		return
	}

	// Before anything is written. An unverified request must leave no
	// event row, no delivery row and no delivery task behind, so this
	// sits above every write rather than inside the transaction that
	// performs them. It has to sit below the body read because the
	// signature is computed over the body; readWebhookBody is what
	// bounds that read, so an unauthenticated sender still cannot make
	// the process hold more than the 1 MB cap.
	if !h.verifyInboundSignature(w, entrypoint, r.Header, body) {
		return
	}

	// These headers are about to be stored verbatim and handed to
	// every delivery target, so the scheme's credential comes out
	// first. Under GitLab's scheme the header is the shared secret
	// itself, and leaving it in would hand the ability to forge
	// signed requests to exactly the parties the signature is meant
	// to exclude.
	headersJSON, err := json.Marshal(
		signature.SanitizeHeaders(&entrypoint, r.Header),
	)
	if err != nil {
		h.serverError(w, "failed to serialize headers", err)

		return
	}

	targets, err := h.loadActiveTargets(entrypoint.WebhookID)
	if err != nil {
		h.serverError(w, "failed to query targets", err)

		return
	}

	h.createAndDeliverEvent(
		w, r, entrypoint, body, headersJSON, targets,
	)
}

// verifyInboundSignature authenticates the request against the
// entrypoint's configured secret, reporting false once it has written
// the response.
//
// An entrypoint with no secret configured is not checked and this
// returns true, which is the unchanged behaviour every existing
// entrypoint keeps.
//
// A configuration that cannot be applied — an unknown scheme, or one
// half of the pair missing — is a 500, not a 401: the request may well
// be authentic, and calling it unauthorized would tell a legitimate
// sender to go fix its own signing. Either way it is refused. Failing
// open here would mean an entrypoint the operator has protected
// quietly accepting anything.
func (h *Handlers) verifyInboundSignature(
	w http.ResponseWriter,
	entrypoint database.Entrypoint,
	header http.Header,
	body []byte,
) bool {
	err := signature.Verify(&entrypoint, header, body)
	if err == nil {
		return true
	}

	if errors.Is(err, signature.ErrConfig) {
		h.log.Error(
			"entrypoint signature configuration cannot be applied",
			"entrypoint_id", entrypoint.ID,
			"webhook_id", entrypoint.WebhookID,
			"error", err,
		)
		http.Error(
			w, "Internal server error",
			http.StatusInternalServerError,
		)

		return false
	}

	// Every field here is bounded and none is client-chosen: the ids
	// are ours, the scheme is one of a fixed set, and the error is a
	// static string carrying no part of the secret or of what the
	// client presented. Reaching this line also requires a real
	// entrypoint UUID, so it is not a line a stranger can drive.
	h.log.Warn(
		"inbound signature verification failed",
		"entrypoint_id", entrypoint.ID,
		"webhook_id", entrypoint.WebhookID,
		"scheme", string(entrypoint.SignatureScheme),
		"error", err,
	)
	http.Error(w, "Unauthorized", http.StatusUnauthorized)

	return false
}

// loadActiveTargets returns all active targets for a webhook.
func (h *Handlers) loadActiveTargets(
	webhookID string,
) ([]database.Target, error) {
	var targets []database.Target

	err := h.db.DB().Where(
		"webhook_id = ? AND active = ?",
		webhookID, true,
	).Find(&targets).Error

	return targets, err
}

// lookupEntrypoint finds an entrypoint by UUID path.
func (h *Handlers) lookupEntrypoint(
	w http.ResponseWriter,
	r *http.Request,
	entrypointUUID string,
) (database.Entrypoint, bool) {
	var entrypoint database.Entrypoint

	result := h.db.DB().Where(
		"path = ?", entrypointUUID,
	).First(&entrypoint)
	if result.Error != nil {
		// The receiver is unauthenticated and /webhook/{uuid}
		// matches any single segment, so this value is entirely
		// client-chosen on exactly the branch where the lookup
		// failed. DEBUG is off by default; the cap is what keeps
		// turning it on from restoring an unbounded write.
		h.log.Debug(
			"entrypoint not found",
			"path", logfield.Truncate(
				entrypointUUID, logfield.MaxBytes,
			),
		)
		http.NotFound(w, r)

		return entrypoint, false
	}

	return entrypoint, true
}

// readWebhookBody reads and validates the request body size.
func (h *Handlers) readWebhookBody(
	w http.ResponseWriter,
	r *http.Request,
) ([]byte, bool) {
	body, err := io.ReadAll(
		io.LimitReader(r.Body, maxWebhookBodySize+1),
	)
	if err != nil {
		h.log.Error(
			"failed to read request body", "error", err,
		)
		http.Error(
			w, "Bad request", http.StatusBadRequest,
		)

		return nil, false
	}

	if len(body) > maxWebhookBodySize {
		http.Error(
			w,
			"Request body too large",
			http.StatusRequestEntityTooLarge,
		)

		return nil, false
	}

	return body, true
}

// createAndDeliverEvent creates the event and delivery records
// then notifies the delivery engine.
func (h *Handlers) createAndDeliverEvent(
	w http.ResponseWriter,
	r *http.Request,
	entrypoint database.Entrypoint,
	body, headersJSON []byte,
	targets []database.Target,
) {
	tx, err := h.beginWebhookTx(w, entrypoint.WebhookID)
	if err != nil {
		return
	}

	event := h.buildEvent(r, entrypoint, headersJSON, body)

	err = tx.Create(event).Error
	if err != nil {
		tx.Rollback()
		h.serverError(w, "failed to create event", err)

		return
	}

	bodyPtr := inlineBody(body)

	tasks := h.buildDeliveryTasks(
		w, tx, event, entrypoint, targets, bodyPtr,
	)
	if tasks == nil {
		return
	}

	err = tx.Commit().Error
	if err != nil {
		h.serverError(w, "failed to commit transaction", err)

		return
	}

	// Counted here, after the commit: an event is received once it
	// is durably stored, which is what the delivery counters are
	// compared against on a dashboard.
	h.mtr.EventReceived()

	h.finishWebhookResponse(w, event, entrypoint, tasks)
}

// beginWebhookTx opens a transaction on the per-webhook DB.
func (h *Handlers) beginWebhookTx(
	w http.ResponseWriter,
	webhookID string,
) (*gorm.DB, error) {
	webhookDB, err := h.dbMgr.GetDB(webhookID)
	if err != nil {
		h.serverError(
			w, "failed to get webhook database", err,
		)

		return nil, err
	}

	tx := webhookDB.Begin()
	if tx.Error != nil {
		h.serverError(
			w, "failed to begin transaction", tx.Error,
		)

		return nil, tx.Error
	}

	return tx, nil
}

// inlineBody returns a pointer to body as a string if it fits
// within the inline size limit, or nil otherwise.
func inlineBody(body []byte) *string {
	if len(body) < delivery.MaxInlineBodySize {
		s := string(body)

		return &s
	}

	return nil
}

// finishWebhookResponse notifies the delivery engine, logs the
// event, and writes the HTTP response.
func (h *Handlers) finishWebhookResponse(
	w http.ResponseWriter,
	event *database.Event,
	entrypoint database.Entrypoint,
	tasks []delivery.Task,
) {
	if len(tasks) > 0 {
		h.notifier.Notify(tasks)
	}

	h.log.Info("webhook event created",
		"event_id", event.ID,
		"webhook_id", entrypoint.WebhookID,
		"entrypoint_id", entrypoint.ID,
		"target_count", len(tasks),
	)

	w.WriteHeader(http.StatusOK)

	_, err := w.Write([]byte(`{"status":"ok"}`))
	if err != nil {
		h.log.Error(
			"failed to write response", "error", err,
		)
	}
}

// buildEvent creates a new Event struct from request data.
func (h *Handlers) buildEvent(
	r *http.Request,
	entrypoint database.Entrypoint,
	headersJSON, body []byte,
) *database.Event {
	return &database.Event{
		WebhookID:    entrypoint.WebhookID,
		EntrypointID: entrypoint.ID,
		Method:       r.Method,
		Headers:      string(headersJSON),
		Body:         string(body),
		ContentType:  r.Header.Get("Content-Type"),
	}
}

// buildDeliveryTasks creates delivery records in the
// transaction and returns tasks for the delivery engine.
// Returns nil if an error occurred.
func (h *Handlers) buildDeliveryTasks(
	w http.ResponseWriter,
	tx *gorm.DB,
	event *database.Event,
	entrypoint database.Entrypoint,
	targets []database.Target,
	bodyPtr *string,
) []delivery.Task {
	tasks := make([]delivery.Task, 0, len(targets))

	for i := range targets {
		dlv := &database.Delivery{
			EventID:  event.ID,
			TargetID: targets[i].ID,
			Status:   database.DeliveryStatusPending,
		}

		err := tx.Create(dlv).Error
		if err != nil {
			tx.Rollback()
			h.log.Error(
				"failed to create delivery",
				"target_id", targets[i].ID,
				"error", err,
			)
			http.Error(
				w, "Internal server error",
				http.StatusInternalServerError,
			)

			return nil
		}

		tasks = append(tasks, delivery.Task{
			DeliveryID:   dlv.ID,
			EventID:      event.ID,
			WebhookID:    entrypoint.WebhookID,
			EntrypointID: entrypoint.ID,
			TargetID:     targets[i].ID,
			TargetName:   targets[i].Name,
			TargetType:   targets[i].Type,
			TargetConfig: targets[i].Config,
			MaxRetries:   targets[i].MaxRetries,
			Method:       event.Method,
			Headers:      event.Headers,
			ContentType:  event.ContentType,
			Body:         bodyPtr,
			AttemptNum:   1,
		})
	}

	return tasks
}