yovision/Sense/server/app/sense/integration/bell_connector/client.go

package bell_connector

import (
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"net/http"
	"net/url"
	"regexp"
	"strconv"
	"strings"

	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
)

type HTTPDoer interface {
	Do(*http.Request) (*http.Response, error)
}

var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$`)

type Client struct {
	Endpoint        string
	RelayID         string
	Signer          machine_identity.Signer
	HTTP            HTTPDoer
	Enabled         bool
	MaxRequestBytes int64
}

func NewClient(endpoint, relayID string, signer machine_identity.Signer, policy machine_identity.TransportPolicy) (*Client, error) {
	parsed, err := url.Parse(endpoint)
	if err != nil || parsed.Scheme != "https" || parsed.Host == "" || parsed.User != nil || parsed.Path != "" || parsed.RawQuery != "" || parsed.Fragment != "" {
		return nil, errors.New("Bell connector endpoint must be an HTTPS origin")
	}
	httpClient, err := policy.HTTPClient()
	if err != nil {
		return nil, err
	}
	return &Client{Endpoint: strings.TrimRight(endpoint, "/"), RelayID: relayID, Signer: signer, HTTP: httpClient, Enabled: true, MaxRequestBytes: policy.MaxRequestBytes}, nil
}

func (c Client) Send(ctx context.Context, body []byte) (IngestResult, error) {
	if !c.Enabled {
		return IngestResult{}, &DeliveryError{Code: "connector_disabled", Detail: "Bell connector is disabled", Terminal: true}
	}
	if c.HTTP == nil || !json.Valid(body) || (strings.TrimSpace(c.RelayID) != "" && !safeIdentifier(c.RelayID)) {
		return IngestResult{}, errors.New("Bell connector is not configured")
	}
	maximum := c.MaxRequestBytes
	if maximum == 0 {
		maximum = MaxInboundBytes
	}
	if maximum < 1 || int64(len(body)) > maximum {
		return IngestResult{}, &DeliveryError{Code: "event_request_too_large", Detail: "event exceeds the configured request limit", Terminal: true}
	}
	requestID, err := newRequestID()
	if err != nil {
		return IngestResult{}, fmt.Errorf("generate request correlation id: %w", err)
	}
	token, err := c.Signer.Mint("yovision-bell", []string{"events:ingest"}, http.MethodPost, "/v1/events", body)
	if err != nil {
		return IngestResult{}, fmt.Errorf("mint Bell machine token: %w", err)
	}
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(c.Endpoint, "/")+"/v1/events", bytes.NewReader(body))
	if err != nil {
		return IngestResult{}, err
	}
	request.Header.Set("Authorization", "Bearer "+token)
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("X-Request-ID", requestID)
	if strings.TrimSpace(c.RelayID) != "" {
		request.Header.Set("X-YoVision-Relay-ID", c.RelayID)
	}
	response, err := c.HTTP.Do(request)
	if err != nil {
		return IngestResult{}, fmt.Errorf("deliver event to Bell: %w", err)
	}
	defer response.Body.Close()
	responseBody, err := io.ReadAll(io.LimitReader(response.Body, 64*1024+1))
	if err != nil || len(responseBody) > 64*1024 {
		return IngestResult{}, errors.New("Bell response is invalid")
	}
	if response.StatusCode == http.StatusCreated || response.StatusCode == http.StatusOK {
		if response.Header.Get("X-Request-ID") != requestID {
			return IngestResult{}, errors.New("Bell response request id is invalid")
		}
		var result IngestResult
		if json.Unmarshal(responseBody, &result) != nil || result.EventID == "" || result.PayloadSHA256 == "" || (result.Disposition != "created" && result.Disposition != "duplicate") {
			return IngestResult{}, errors.New("Bell response is invalid")
		}
		identity, parseErr := parseEventIdentity(body)
		if parseErr != nil || result.ProducerID != identity.ProducerID || result.SourceEventID != identity.SourceEventID {
			return IngestResult{}, errors.New("Bell response changed event identity")
		}
		canonical, canonicalErr := canonicalPayload(body)
		if canonicalErr != nil {
			return IngestResult{}, errors.New("delivered event cannot be canonicalized")
		}
		digest := sha256.Sum256(canonical)
		if result.PayloadSHA256 != hex.EncodeToString(digest[:]) {
			return IngestResult{}, errors.New("Bell response payload digest does not match the delivered event")
		}
		return result, nil
	}
	var problem Problem
	_ = json.Unmarshal(responseBody, &problem)
	terminal := response.StatusCode >= 400 && response.StatusCode < 500 && response.StatusCode != http.StatusTooManyRequests
	if problem.Code == "" {
		problem.Code = "bell_unavailable"
	}
	return IngestResult{}, &DeliveryError{Code: problem.Code, Detail: problem.Message, Terminal: terminal}
}

func newRequestID() (string, error) {
	raw := make([]byte, 16)
	if _, err := rand.Read(raw); err != nil {
		return "", err
	}
	value := base64.RawURLEncoding.EncodeToString(raw)
	if !requestIDPattern.MatchString(value) {
		return "", errors.New("generated request id is invalid")
	}
	return value, nil
}

func canonicalPayload(raw []byte) ([]byte, error) {
	decoder := json.NewDecoder(bytes.NewReader(raw))
	decoder.UseNumber()
	var value any
	if err := decoder.Decode(&value); err != nil {
		return nil, err
	}
	value, err := normalizeJCSNumbers(value)
	if err != nil {
		return nil, err
	}
	var buffer bytes.Buffer
	encoder := json.NewEncoder(&buffer)
	encoder.SetEscapeHTML(false)
	if err := encoder.Encode(value); err != nil {
		return nil, err
	}
	canonical := bytes.TrimSuffix(buffer.Bytes(), []byte("\n"))
	canonical = bytes.ReplaceAll(canonical, []byte(`\u2028`), []byte("\u2028"))
	canonical = bytes.ReplaceAll(canonical, []byte(`\u2029`), []byte("\u2029"))
	return canonical, nil
}

func normalizeJCSNumbers(value any) (any, error) {
	switch typed := value.(type) {
	case json.Number:
		number, err := strconv.ParseFloat(string(typed), 64)
		if err != nil || math.IsNaN(number) || math.IsInf(number, 0) {
			return nil, errors.New("JSON number is outside the RFC 8785 domain")
		}
		if number == 0 {
			return float64(0), nil
		}
		return number, nil
	case []any:
		for index, item := range typed {
			normalized, err := normalizeJCSNumbers(item)
			if err != nil {
				return nil, err
			}
			typed[index] = normalized
		}
	case map[string]any:
		for key, item := range typed {
			normalized, err := normalizeJCSNumbers(item)
			if err != nil {
				return nil, err
			}
			typed[key] = normalized
		}
	}
	return value, nil
}