feat: webhook outbound

Author: kastov

infra/conf/webhook.go

@@ -0,0 +1,18 @@
+package conf
+
+import (
+	"github.com/xtls/xray-core/proxy/webhook"
+	"google.golang.org/protobuf/proto"
+)
+
+type WebhookConfig struct {
+	URL           string `json:"url"`
+	Deduplication uint32 `json:"deduplication"`
+}
+
+func (c *WebhookConfig) Build() (proto.Message, error) {
+	return &webhook.Config{
+		Url:           c.URL,
+		Deduplication: c.Deduplication,
+	}, nil
+}

infra/conf/xray.go

@@ -52,6 +52,7 @@ var (
 		"hysteria":    func() interface{} { return new(HysteriaClientConfig) },
 		"dns":         func() interface{} { return new(DNSOutboundConfig) },
 		"wireguard":   func() interface{} { return &WireGuardConfig{IsClient: true} },
+		"webhook":     func() interface{} { return new(WebhookConfig) },
 	}, "protocol", "settings")
 )
 

main/distro/all/all.go

@@ -47,6 +47,7 @@ import (
 	_ "github.com/xtls/xray-core/proxy/vless/outbound"
 	_ "github.com/xtls/xray-core/proxy/vmess/inbound"
 	_ "github.com/xtls/xray-core/proxy/vmess/outbound"
+	_ "github.com/xtls/xray-core/proxy/webhook"
 	_ "github.com/xtls/xray-core/proxy/wireguard"
 
 	// Transports

proxy/webhook/config.proto

@@ -0,0 +1,12 @@
+syntax = "proto3";
+
+package xray.proxy.webhook;
+option csharp_namespace = "Xray.Proxy.Webhook";
+option go_package = "github.com/xtls/xray-core/proxy/webhook";
+option java_package = "com.xray.proxy.webhook";
+option java_multiple_files = true;
+
+message Config {
+  string url = 1;
+  uint32 deduplication = 2;
+}

proxy/webhook/webhook.go

@@ -0,0 +1,235 @@
+// Package webhook is an outbound handler that drops traffic and fires an HTTP
+// POST with connection metadata to a configured URL.
+//
+// URL formats:
+//
+//	https://api.example.com/alert          — regular HTTP(S)
+//	unix:///var/run/webhook.sock           — Unix socket, POST to /
+//	unix:///var/run/webhook.sock:/alert    — Unix socket, POST to /alert
+package webhook
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/xtls/xray-core/common"
+	"github.com/xtls/xray-core/common/errors"
+	"github.com/xtls/xray-core/common/session"
+	"github.com/xtls/xray-core/transport"
+	"github.com/xtls/xray-core/transport/internet"
+)
+
+const unixScheme = "unix://"
+
+// parseURL splits a webhook URL into an HTTP URL and an optional Unix socket
+// path. For regular http/https URLs the input is returned unchanged with an
+// empty socketPath. For Unix sockets the format is:
+//
+//	unix:///path/to/socket.sock:/http/path
+//
+// The :/ separator after the socket path delimits the HTTP request path.
+// If omitted, "/" is used.
+func parseURL(raw string) (httpURL, socketPath string) {
+	if !strings.HasPrefix(raw, unixScheme) {
+		return raw, ""
+	}
+	rest := raw[len(unixScheme):]
+	if idx := strings.Index(rest, ":/"); idx >= 0 {
+		return "http://localhost" + rest[idx+1:], rest[:idx]
+	}
+	return "http://localhost/", rest
+}
+
+type event struct {
+	Email          *string `json:"email"`
+	Level          *uint32 `json:"level"`
+	Protocol       *string `json:"protocol"`
+	Network        *string `json:"network"`
+	Source         *string `json:"source"`
+	Destination    *string `json:"destination"`
+	OriginalTarget *string `json:"originalTarget"`
+	RouteTarget    *string `json:"routeTarget"`
+	InboundTag     *string `json:"inboundTag"`
+	InboundName    *string `json:"inboundName"`
+	InboundLocal   *string `json:"inboundLocal"`
+	OutboundTag    *string `json:"outboundTag"`
+	Timestamp      int64   `json:"ts"`
+}
+
+func ptr[T any](v T) *T { return &v }
+
+type Handler struct {
+	url           string
+	deduplication uint32
+	client        *http.Client
+	seen          sync.Map
+	done          chan struct{}
+	wg            sync.WaitGroup
+}
+
+func New(ctx context.Context, config *Config) (*Handler, error) {
+	if config.Url == "" {
+		return nil, errors.New("webhook: url must not be empty")
+	}
+	httpURL, socketPath := parseURL(config.Url)
+	h := &Handler{
+		url:           httpURL,
+		deduplication: config.Deduplication,
+		client: &http.Client{
+			Timeout: 5 * time.Second,
+		},
+		done: make(chan struct{}),
+	}
+	if socketPath != "" {
+		h.client.Transport = &http.Transport{
+			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+				var d net.Dialer
+				return d.DialContext(ctx, "unix", socketPath)
+			},
+		}
+	}
+	if h.deduplication > 0 {
+		go h.cleanupLoop()
+	}
+	return h, nil
+}
+
+func (h *Handler) Close() error {
+	select {
+	case <-h.done:
+	default:
+		close(h.done)
+	}
+	h.wg.Wait()
+	return nil
+}
+
+func (h *Handler) cleanupLoop() {
+	ttl := time.Duration(h.deduplication) * time.Second
+	ticker := time.NewTicker(ttl)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-h.done:
+			return
+		case <-ticker.C:
+			now := time.Now()
+			h.seen.Range(func(key, value any) bool {
+				if now.Sub(value.(time.Time)) >= ttl {
+					h.seen.Delete(key)
+				}
+				return true
+			})
+		}
+	}
+}
+
+func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer internet.Dialer) error {
+	outbounds := session.OutboundsFromContext(ctx)
+	ob := outbounds[len(outbounds)-1]
+	ob.Name = "webhook"
+
+	ev := event{
+		Timestamp:   time.Now().Unix(),
+		Network:     ptr(ob.Target.Network.SystemString()),
+		Destination: ptr(ob.Target.String()),
+		OutboundTag: ptr(ob.Tag),
+	}
+	if ob.OriginalTarget.Address != nil {
+		ev.OriginalTarget = ptr(ob.OriginalTarget.String())
+	}
+	if ob.RouteTarget.Address != nil {
+		ev.RouteTarget = ptr(ob.RouteTarget.String())
+	}
+
+	if ib := session.InboundFromContext(ctx); ib != nil {
+		ev.Source = ptr(ib.Source.String())
+		ev.InboundTag = ptr(ib.Tag)
+		ev.InboundName = ptr(ib.Name)
+		if ib.Local.Address != nil {
+			ev.InboundLocal = ptr(ib.Local.String())
+		}
+		if ib.User != nil {
+			ev.Email = ptr(ib.User.Email)
+			ev.Level = ptr(ib.User.Level)
+		}
+	}
+
+	if ct := session.ContentFromContext(ctx); ct != nil {
+		ev.Protocol = ptr(ct.Protocol)
+	}
+
+	email := ""
+	if ev.Email != nil {
+		email = *ev.Email
+	}
+	if !h.isDuplicate(email) {
+		select {
+		case <-h.done:
+		default:
+			h.wg.Add(1)
+			go func() {
+				defer h.wg.Done()
+				h.post(ctx, &ev)
+			}()
+		}
+	}
+
+	common.Interrupt(link.Writer)
+	return nil
+}
+
+func (h *Handler) isDuplicate(email string) bool {
+	if h.deduplication == 0 || email == "" {
+		return false
+	}
+	ttl := time.Duration(h.deduplication) * time.Second
+	now := time.Now()
+	if v, loaded := h.seen.LoadOrStore(email, now); loaded {
+		if now.Sub(v.(time.Time)) < ttl {
+			return true
+		}
+		h.seen.Store(email, now)
+	}
+	return false
+}
+
+func (h *Handler) post(ctx context.Context, ev *event) {
+	body, err := json.Marshal(ev)
+	if err != nil {
+		errors.LogWarning(ctx, "webhook: marshal failed: ", err)
+		return
+	}
+
+	postCtx, cancel := context.WithTimeout(context.Background(), h.client.Timeout)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(postCtx, http.MethodPost, h.url, bytes.NewReader(body))
+	if err != nil {
+		errors.LogWarning(ctx, "webhook: request build failed: ", err)
+		return
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := h.client.Do(req)
+	if err != nil {
+		errors.LogInfo(ctx, "webhook: POST failed: ", err)
+		return
+	}
+	resp.Body.Close()
+	if resp.StatusCode >= 400 {
+		errors.LogWarning(ctx, "webhook: POST returned status ", resp.StatusCode)
+	}
+}
+
+func init() {
+	common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
+		return New(ctx, config.(*Config))
+	}))
+}