Use webhook dispatcher when creating and updating endpoints (#2280)

* Use webhook dispatcher when creating and updating endpoints

Author: jirevwe

api/handlers/endpoint.go

@@ -84,6 +84,8 @@ func (h *Handler) CreateEndpoint(w http.ResponseWriter, r *http.Request) {
 		Licenser:       h.A.Licenser,
 		E:              e,
 		ProjectID:      project.UID,
+		FeatureFlag:    h.A.FFlag,
+		Logger:         h.A.Logger,
 	}
 
 	endpoint, err := ce.Run(r.Context())
@@ -319,6 +321,8 @@ func (h *Handler) UpdateEndpoint(w http.ResponseWriter, r *http.Request) {
 		EndpointRepo: postgres.NewEndpointRepo(h.A.DB),
 		ProjectRepo:  postgres.NewProjectRepo(h.A.DB),
 		Licenser:     h.A.Licenser,
+		FeatureFlag:  h.A.FFlag,
+		Logger:       h.A.Logger,
 		E:            e,
 		Endpoint:     endpoint,
 		Project:      project,

internal/pkg/fflag/fflag.go

@@ -78,6 +78,12 @@ func NewFFlag(enableFeatureFlags []string) *FFlag {
 	return f
 }
 
+func NoopFflag() *FFlag {
+	return &FFlag{
+		Features: clone(DefaultFeaturesState),
+	}
+}
+
 func clone(src map[FeatureFlagKey]FeatureFlagState) map[FeatureFlagKey]FeatureFlagState {
 	dst := make(map[FeatureFlagKey]FeatureFlagState)
 	for k, v := range src {

internal/pkg/middleware/middleware.go

@@ -6,12 +6,13 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
-	"github.com/frain-dev/convoy/internal/pkg/license"
 	"net/http"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/frain-dev/convoy/internal/pkg/license"
+
 	"github.com/frain-dev/convoy/internal/pkg/limiter"
 	rlimiter "github.com/frain-dev/convoy/internal/pkg/limiter/redis"
 
@@ -43,7 +44,49 @@ const (
 	pageableCtx types.ContextKey = "pageable"
 )
 
-var ErrValidLicenseRequired = errors.New("access to this resource requires a valid license")
+var (
+	ErrValidLicenseRequired = errors.New("access to this resource requires a valid license")
+
+	// skipLoggingPaths defines paths that should not be logged by the request logger
+	skipLoggingPaths []string
+)
+
+// shouldSkipLogging checks if the given path should be excluded from logging
+func shouldSkipLogging(r map[string]interface{}, w map[string]interface{}) bool {
+	for _, skipPath := range skipLoggingPaths {
+		if strings.Contains(r["requestURL"].(string), skipPath) {
+			return true
+		}
+	}
+
+	headers := w["header"].(map[string]string)
+
+	if strings.Contains(headers["content-type"], "application/javascript") {
+		return true
+	}
+
+	if strings.Contains(headers["content-type"], "image") {
+		return true
+	}
+
+	if strings.Contains(headers["content-type"], "font") {
+		return true
+	}
+
+	if strings.Contains(headers["content-type"], "text/html") {
+		return true
+	}
+
+	if strings.Contains(headers["content-type"], "text/javascript") {
+		return true
+	}
+
+	if strings.Contains(headers["content-type"], "text/css") {
+		return true
+	}
+
+	return false
+}
 
 type AuthorizedLogin struct {
 	Username   string    `json:"username,omitempty"`
@@ -316,6 +359,10 @@ func LogHttpRequest(a *types.APIOptions) func(next http.Handler) http.Handler {
 					"httpResponse": responseFields,
 				}
 
+				if shouldSkipLogging(requestFields, responseFields) {
+					return
+				}
+
 				log.FromContext(r.Context()).WithFields(logFields).Log(lvl, requestFields["requestURL"])
 			}()
 

net/dispatcher.go

@@ -37,6 +37,7 @@ var (
 	ErrLoggerIsRequired    = errors.New("logger is required")
 	ErrInvalidIPPrefix     = errors.New("invalid IP prefix")
 	ErrTracerIsRequired    = errors.New("tracer cannot be nil")
+	ErrNon2xxResponse      = errors.New("endpoint returned a non-2xx response")
 )
 
 type DispatcherOption func(d *Dispatcher) error
@@ -96,7 +97,7 @@ type Dispatcher struct {
 	ff *fflag.FFlag
 	l  license.Licenser
 
-	logger        *log.Logger
+	logger        log.StdLogger
 	transport     *http.Transport
 	client        *http.Client
 	rules         *netjail.Rules
@@ -242,7 +243,7 @@ func TLSConfigOption(insecureSkipVerify bool, licenser license.Licenser, caCertT
 	}
 }
 
-func LoggerOption(logger *log.Logger) DispatcherOption {
+func LoggerOption(logger log.StdLogger) DispatcherOption {
 	return func(d *Dispatcher) error {
 		if logger == nil {
 			return ErrLoggerIsRequired
@@ -289,7 +290,7 @@ func (d *Dispatcher) validateProxy(proxyURL string) (*url.URL, bool, error) {
 	return nil, false, nil
 }
 
-func (d *Dispatcher) SendRequest(ctx context.Context, endpoint, method string, jsonData json.RawMessage, signatureHeader string, hmac string, maxResponseSize int64, headers httpheader.HTTPHeader, idempotencyKey string, timeout time.Duration) (*Response, error) {
+func (d *Dispatcher) SendWebhook(ctx context.Context, endpoint string, jsonData json.RawMessage, signatureHeader string, hmac string, maxResponseSize int64, headers httpheader.HTTPHeader, idempotencyKey string, timeout time.Duration) (*Response, error) {
 	d.logger.Debugf("rules: %+v", d.rules)
 
 	ctx, cancel := context.WithTimeout(ctx, timeout)
@@ -307,7 +308,7 @@ func (d *Dispatcher) SendRequest(ctx context.Context, endpoint, method string, j
 		ctx = netjail.ContextWithRules(ctx, d.rules)
 	}
 
-	req, err := http.NewRequestWithContext(ctx, method, endpoint, bytes.NewBuffer(jsonData))
+	req, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewBuffer(jsonData))
 	if err != nil {
 		d.logger.WithError(err).Error("error occurred while creating request")
 		return r, err
@@ -462,3 +463,48 @@ func (d *Dispatcher) do(ctx context.Context, req *http.Request, res *Response, m
 
 	return nil
 }
+
+// Ping sends a GET request to the specified endpoint and verifies it returns a 2xx response.
+// It returns an error if the endpoint is unreachable or returns a non-2xx status code.
+func (d *Dispatcher) Ping(ctx context.Context, endpoint string, timeout time.Duration) error {
+	d.logger.Debugf("rules: %+v", d.rules)
+
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	if d.ff.CanAccessFeature(fflag.IpRules) && d.l.IpRules() {
+		ctx = netjail.ContextWithRules(ctx, d.rules)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil)
+	if err != nil {
+		d.logger.WithError(err).Error("error creating ping request")
+		return err
+	}
+
+	trace := &httptrace.ClientTrace{
+		GotConn: func(connInfo httptrace.GotConnInfo) {
+			d.logger.Debugf("IP address resolved for %s to: %s", endpoint, connInfo.Conn.RemoteAddr())
+		},
+	}
+
+	ctx = httptrace.WithClientTrace(ctx, trace)
+	req = req.WithContext(ctx)
+
+	req.Header.Add("User-Agent", defaultUserAgent())
+
+	response, err := d.client.Do(req)
+	if err != nil {
+		d.logger.WithError(err).Error("error sending ping request")
+		return err
+	}
+	defer response.Body.Close()
+
+	if response.StatusCode < 200 || response.StatusCode > 299 {
+		err = fmt.Errorf("%w: got status code %d", ErrNon2xxResponse, response.StatusCode)
+		d.logger.WithError(err).Error("ping request failed")
+		return err
+	}
+
+	return nil
+}

net/dispatcher_test.go

@@ -295,7 +295,7 @@ func TestDispatcher_SendRequest(t *testing.T) {
 				defer deferFn()
 			}
 
-			got, err := d.SendRequest(context.Background(), tt.args.endpoint, tt.args.method, tt.args.jsonData, tt.args.project.Config.Signature.Header.String(), tt.args.hmac, config.MaxResponseSize, tt.args.headers, "", time.Minute)
+			got, err := d.SendWebhook(context.Background(), tt.args.endpoint, tt.args.jsonData, tt.args.project.Config.Signature.Header.String(), tt.args.hmac, config.MaxResponseSize, tt.args.headers, "", time.Minute)
 			if tt.wantErr {
 				require.NotNil(t, err)
 				require.Contains(t, err.Error(), tt.want.Error)
@@ -398,7 +398,7 @@ func TestNewDispatcher(t *testing.T) {
 	}
 }
 
-// TestDispatcherSendRequest tests the basic functionality of SendRequest
+// TestDispatcherSendRequest tests the basic functionality of SendWebhook
 func TestDispatcherSendRequest(t *testing.T) {
 	// Start a test server
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -437,10 +437,9 @@ func TestDispatcherSendRequest(t *testing.T) {
 	}
 
 	// Send request
-	resp, err := dispatcher.SendRequest(
+	resp, err := dispatcher.SendWebhook(
 		context.Background(),
 		server.URL,
-		"POST",
 		jsonData,
 		"X-Signature",
 		"test-hmac",
@@ -483,10 +482,9 @@ func TestDispatcherWithTimeout(t *testing.T) {
 	require.NoError(t, err)
 
 	// Send request with a short timeout
-	_, err = dispatcher.SendRequest(
+	_, err = dispatcher.SendWebhook(
 		context.Background(),
 		server.URL,
-		"GET",
 		nil,
 		"X-Signature",
 		"test-hmac",
@@ -529,10 +527,9 @@ func TestDispatcherWithBlockedIP(t *testing.T) {
 	require.NoError(t, err)
 
 	// Attempt to send a request
-	_, err = dispatcher.SendRequest(
+	_, err = dispatcher.SendWebhook(
 		context.Background(),
 		server.URL,
-		"GET",
 		nil,
 		"X-Signature",
 		"test-hmac",

services/create_endpoint.go

@@ -3,8 +3,13 @@ package services
 import (
 	"context"
 	"errors"
+	"fmt"
+	"github.com/frain-dev/convoy/config"
+	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	"github.com/frain-dev/convoy/internal/pkg/keys"
+	"github.com/frain-dev/convoy/net"
 	"net/http"
+	"net/url"
 	"time"
 
 	"github.com/frain-dev/convoy"
@@ -23,9 +28,10 @@ type CreateEndpointService struct {
 	EndpointRepo   datastore.EndpointRepository
 	ProjectRepo    datastore.ProjectRepository
 	Licenser       license.Licenser
-
-	E         models.CreateEndpoint
-	ProjectID string
+	FeatureFlag    *fflag.FFlag
+	Logger         log.StdLogger
+	E              models.CreateEndpoint
+	ProjectID      string
 }
 
 func (a *CreateEndpointService) Run(ctx context.Context) (*datastore.Endpoint, error) {
@@ -34,12 +40,12 @@ func (a *CreateEndpointService) Run(ctx context.Context) (*datastore.Endpoint, e
 		return nil, &ServiceError{ErrMsg: "failed to load endpoint project", Err: err}
 	}
 
-	url, err := util.ValidateEndpoint(a.E.URL, project.Config.SSL.EnforceSecureEndpoints, a.Licenser.CustomCertificateAuthority())
+	endpointUrl, err := a.ValidateEndpoint(ctx, project.Config.SSL.EnforceSecureEndpoints)
 	if err != nil {
 		return nil, &ServiceError{ErrMsg: err.Error()}
 	}
 
-	a.E.URL = url
+	a.E.URL = endpointUrl
 
 	truthValue := true
 	switch project.Type {
@@ -125,6 +131,56 @@ func (a *CreateEndpointService) Run(ctx context.Context) (*datastore.Endpoint, e
 	return endpoint, nil
 }
 
+func (a *CreateEndpointService) ValidateEndpoint(ctx context.Context, enforceSecure bool) (string, error) {
+	if util.IsStringEmpty(a.E.URL) {
+		return "", errors.New("please provide the endpoint url")
+	}
+
+	u, pingErr := url.Parse(a.E.URL)
+	if pingErr != nil {
+		return "", pingErr
+	}
+
+	switch u.Scheme {
+	case "http":
+		if enforceSecure {
+			return "", errors.New("only https endpoints allowed")
+		}
+	case "https":
+		cfg, innerErr := config.Get()
+		if innerErr != nil {
+			return "", innerErr
+		}
+
+		caCertTLSCfg, innerErr := config.GetCaCert()
+		if innerErr != nil {
+			return "", innerErr
+		}
+
+		dispatcher, innerErr := net.NewDispatcher(
+			a.Licenser,
+			a.FeatureFlag,
+			net.LoggerOption(a.Logger),
+			net.ProxyOption(cfg.Server.HTTP.HttpProxy),
+			net.AllowListOption(cfg.Dispatcher.AllowList),
+			net.BlockListOption(cfg.Dispatcher.BlockList),
+			net.TLSConfigOption(cfg.Dispatcher.InsecureSkipVerify, a.Licenser, caCertTLSCfg),
+		)
+		if innerErr != nil {
+			return "", innerErr
+		}
+
+		pingErr = dispatcher.Ping(ctx, a.E.URL, 10*time.Second)
+		if pingErr != nil {
+			return "", fmt.Errorf("failed to ping tls endpoint: %v", pingErr)
+		}
+	default:
+		return "", errors.New("invalid endpoint scheme")
+	}
+
+	return u.String(), nil
+}
+
 func ValidateEndpointAuthentication(auth *datastore.EndpointAuthentication) (*datastore.EndpointAuthentication, error) {
 	if auth != nil && !util.IsStringEmpty(string(auth.Type)) {
 		if err := util.Validate(auth); err != nil {