feat(plugins): add inbound plugin jsonschema-validator (#239)

Signed-off-by: Yusheng Li <leeys.top@gmail.com>
Co-authored-by: Yusheng Li <leeys.top@gmail.com>

Author: cchenggit

README.md

@@ -33,6 +33,7 @@ WebhookX is an open-source webhooks gateway for message receiving, processing, a
   - `webhookx-signature`: Sign outbound requests with HMAC(SHA-256) by adding `Webhookx-Signature` and `Webhookx-Timestamp` headers.
   - `wasm`: Transform outbound requests using high-level languages such as AssemblyScript, Rust or TinyGo. See [plugin/wasm](plugins/wasm).
   - `function`: Customize inbound behavior with JavaScript, e.g. signature verification or request body transformation.
+  - `jsonschema-validator`: Validate event payloads against JSONSchema definitions. Up to Draft v6 is supported.
 - **Observability:** OpenTelemetry metrics and tracing for monitoring and troubleshooting.
 
 

plugins/jsonschema_validator/jsonschema/jsonschema.go

@@ -0,0 +1,40 @@
+package jsonschema
+
+import (
+	"github.com/getkin/kin-openapi/openapi3"
+	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/webhookx-io/webhookx/pkg/openapi"
+	"github.com/webhookx-io/webhookx/utils"
+)
+
+type JSONSchema struct {
+	schemaDef string
+	hex       string
+}
+
+func New(schemaDef []byte) *JSONSchema {
+	return &JSONSchema{
+		schemaDef: string(schemaDef),
+		hex:       utils.Sha256(string(schemaDef)),
+	}
+}
+
+var cache, _ = lru.New[string, *openapi3.Schema](128)
+
+func (s *JSONSchema) Validate(ctx *ValidatorContext) error {
+	schema, ok := cache.Get(s.hex)
+	if !ok {
+		schema = &openapi3.Schema{}
+		err := schema.UnmarshalJSON([]byte(s.schemaDef))
+		if err != nil {
+			return err
+		}
+		cache.Add(s.hex, schema)
+	}
+
+	err := openapi.Validate(schema, ctx.HTTPRequest.Data)
+	if err != nil {
+		return err
+	}
+	return nil
+}

plugins/jsonschema_validator/jsonschema/jsonschema_test.go

@@ -0,0 +1,66 @@
+package jsonschema
+
+import (
+	"encoding/json"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"testing"
+)
+
+func TestJSONSchema(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Schema Validator Suite")
+}
+
+var _ = Describe("Schema Validator Plugin", func() {
+
+	Context("JSONSchema Validator", func() {
+		It("should validate valid JSON data against the schema", func() {
+			schemaDef := `{
+				"type": "object",
+				"properties": {
+					"name": { "type": "string" },
+					"age": { "type": "integer", "minimum": 0 }
+				},
+				"required": ["name", "age"]
+			}`
+
+			validator := New([]byte(schemaDef))
+
+			validData := map[string]interface{}{"name": "John Doe", "age": 30}
+			ctx := &ValidatorContext{
+				HTTPRequest: &HTTPRequest{
+					Data: validData,
+				},
+			}
+
+			err := validator.Validate(ctx)
+			Expect(err).To(BeNil())
+		})
+
+		It("should return an error for invalid JSON data against the schema", func() {
+			schemaDef := `{
+				"type": "object",
+				"properties": {
+					"name": { "type": "string" },
+					"age": { "type": "integer", "minimum": 0 }
+				},
+				"required": ["name", "age"]
+			}`
+
+			validator := New([]byte(schemaDef))
+
+			invalidData := map[string]interface{}{"name": "John Doe", "age": -5}
+			ctx := &ValidatorContext{
+				HTTPRequest: &HTTPRequest{
+					Data: invalidData,
+				},
+			}
+
+			err := validator.Validate(ctx)
+			Expect(err).ToNot(BeNil())
+			b, _ := json.Marshal(err)
+			Expect(string(b)).To(Equal(`{"message":"request validation","fields":{"age":"number must be at least 0"}}`))
+		})
+	})
+})

plugins/jsonschema_validator/jsonschema/validator.go

@@ -0,0 +1,18 @@
+package jsonschema
+
+import (
+	"net/http"
+)
+
+type Validator interface {
+	Validate(ctx *ValidatorContext) error
+}
+
+type ValidatorContext struct {
+	HTTPRequest *HTTPRequest
+}
+
+type HTTPRequest struct {
+	R    *http.Request
+	Data map[string]any
+}

plugins/jsonschema_validator/plugin.go

@@ -0,0 +1,149 @@
+package jsonschema_validator
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/getkin/kin-openapi/openapi3"
+	"github.com/webhookx-io/webhookx/pkg/errs"
+	"github.com/webhookx-io/webhookx/pkg/http/response"
+	"github.com/webhookx-io/webhookx/pkg/plugin"
+	"github.com/webhookx-io/webhookx/pkg/types"
+	"github.com/webhookx-io/webhookx/plugins/jsonschema_validator/jsonschema"
+	"github.com/webhookx-io/webhookx/utils"
+)
+
+type Config struct {
+	Draft         string             `json:"draft" validate:"required,oneof=6 default:6"`
+	DefaultSchema string             `json:"default_schema" validate:"omitempty,json,max=1048576"`
+	Schemas       map[string]*Schema `json:"schemas" validate:"dive"`
+}
+
+type Schema struct {
+	Schema string `json:"schema" validate:"omitempty,json,max=1048576"`
+}
+
+type SchemaValidatorPlugin struct {
+	plugin.BasePlugin[Config]
+}
+
+func New(config []byte) (plugin.Plugin, error) {
+	p := &SchemaValidatorPlugin{}
+	p.Name = "jsonschema-validator"
+
+	if config != nil {
+		if err := p.UnmarshalConfig(config); err != nil {
+			return nil, err
+		}
+	}
+	return p, nil
+}
+
+func unmarshalAndValidateSchema(schema string) (*openapi3.Schema, error) {
+	openapiSchema := &openapi3.Schema{}
+	err := openapiSchema.UnmarshalJSON([]byte(schema))
+	if err != nil {
+		return nil, fmt.Errorf("value must be a valid jsonschema")
+	}
+	err = openapiSchema.Validate(context.Background(), openapi3.EnableSchemaFormatValidation())
+	if err != nil {
+		return openapiSchema, err
+	}
+	return openapiSchema, nil
+}
+
+func (p *SchemaValidatorPlugin) ValidateConfig() error {
+	err := utils.Validate(p.Config)
+	if err != nil {
+		return err
+	}
+
+	e := errs.NewValidateError(errors.New("request validation"))
+
+	var defaultErr error
+	if p.Config.DefaultSchema != "" {
+		_, err := unmarshalAndValidateSchema(p.Config.DefaultSchema)
+		if err != nil {
+			defaultErr = err
+			e.Fields = map[string]interface{}{
+				"default_schema": err.Error(),
+			}
+		}
+	}
+
+	for event, schema := range p.Config.Schemas {
+		field := fmt.Sprintf("schemas[%s]", event)
+		if schema == nil || schema.Schema == "" {
+			if defaultErr != nil {
+				e.Fields[field] = map[string]string{
+					"schema": "invalid due to reusing the default_schema definition",
+				}
+			}
+		} else {
+			_, err = unmarshalAndValidateSchema(schema.Schema)
+			if err != nil {
+				e.Fields[field] = map[string]string{
+					"schema": err.Error(),
+				}
+			}
+		}
+	}
+	if len(e.Fields) > 0 {
+		return e
+	}
+	return nil
+}
+
+func (p *SchemaValidatorPlugin) ExecuteInbound(inbound *plugin.Inbound) (res plugin.InboundResult, err error) {
+	var event map[string]any
+	body := inbound.RawBody
+	if err = json.Unmarshal(body, &event); err != nil {
+		return
+	}
+
+	eventType, ok := event["event_type"].(string)
+	if !ok || eventType == "" {
+		res.Payload = body
+		return
+	}
+
+	data := event["data"]
+	if data == nil {
+		res.Payload = body
+		return
+	}
+
+	schema, ok := p.Config.Schemas[eventType]
+	if !ok {
+		res.Payload = body
+		return
+	}
+	if schema == nil || schema.Schema == "" {
+		if p.Config.DefaultSchema == "" {
+			res.Payload = body
+			return
+		}
+		schema = &Schema{
+			Schema: p.Config.DefaultSchema,
+		}
+	}
+
+	validator := jsonschema.New([]byte(schema.Schema))
+	e := validator.Validate(&jsonschema.ValidatorContext{
+		HTTPRequest: &jsonschema.HTTPRequest{
+			R:    inbound.Request,
+			Data: data.(map[string]any),
+		},
+	})
+	if e != nil {
+		response.JSON(inbound.Response, 400, types.ErrorResponse{
+			Message: "Request Validation",
+			Error:   e,
+		})
+		res.Terminated = true
+		return
+	}
+	res.Payload = body
+	return
+}

plugins/plugins.go

@@ -3,12 +3,14 @@ package plugins
 import (
 	"github.com/webhookx-io/webhookx/pkg/plugin"
 	"github.com/webhookx-io/webhookx/plugins/function"
+	"github.com/webhookx-io/webhookx/plugins/jsonschema_validator"
 	"github.com/webhookx-io/webhookx/plugins/wasm"
 	"github.com/webhookx-io/webhookx/plugins/webhookx_signature"
 )
 
 func LoadPlugins() {
 	plugin.RegisterPlugin(plugin.TypeInbound, "function", function.New)
+	plugin.RegisterPlugin(plugin.TypeInbound, "jsonschema-validator", jsonschema_validator.New)
 	plugin.RegisterPlugin(plugin.TypeOutbound, "wasm", wasm.New)
 	plugin.RegisterPlugin(plugin.TypeOutbound, "webhookx-signature", webhookx_signature.New)
 }

proxy/gateway.go

@@ -260,6 +260,7 @@ func (gw *Gateway) handle(w http.ResponseWriter, r *http.Request) bool {
 			response.JSON(w, 500, types.ErrorResponse{Message: "internal error"})
 			return false
 		}
+
 		if result.Terminated {
 			return false
 		}

test/cmd/admin_test.go

@@ -101,10 +101,13 @@ var _ = Describe("admin", Ordered, func() {
 
 				plugins, err = db.Plugins.ListSourcePlugin(context.TODO(), source.ID)
 				assert.NoError(GinkgoT(), err)
-				assert.Equal(GinkgoT(), 1, len(plugins))
+				assert.Equal(GinkgoT(), 2, len(plugins))
 				assert.Equal(GinkgoT(), "function", plugins[0].Name)
 				assert.Equal(GinkgoT(), true, plugins[0].Enabled)
 				assert.Equal(GinkgoT(), `{"function": "function handle() {}"}`, string(plugins[0].Config))
+				assert.Equal(GinkgoT(), `jsonschema-validator`, plugins[1].Name)
+				assert.Equal(GinkgoT(), true, plugins[1].Enabled)
+				assert.Equal(GinkgoT(), `{"draft": "6", "schemas": {"charge.succeeded": {"schema": "{\n  \"type\": \"object\",\n  \"properties\": {\n      \"id\": { \"type\": \"string\" },\n      \"amount\": { \"type\": \"integer\", \"minimum\": 1 },\n      \"currency\": { \"type\": \"string\", \"minLength\": 3, \"maxLength\": 6 }\n  },\n  \"required\": [\"id\", \"amount\", \"currency\"]\n}\n"}}, "default_schema": "{\n  \"type\": \"object\",\n  \"properties\": {\n      \"id\": { \"type\": \"string\" }\n  },\n  \"required\": [\"id\"]\n}\n"}`, string(plugins[1].Config))
 			})
 
 			It("entities not defined in the declarative configuration should be deleted", func() {