From 31c4d09d09294a39f160aafe420d8e3eedd1e044 Mon Sep 17 00:00:00 2001 From: Meo597 <197331664+Meo597@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:31:00 +0800 Subject: [PATCH] add responder --- infra/conf/responder.go | 46 ++++++++ infra/conf/responder_test.go | 65 +++++++++++ infra/conf/xray.go | 1 + main/distro/all/all.go | 1 + proxy/responder/config.pb.go | 123 ++++++++++++++++++++ proxy/responder/config.proto | 11 ++ proxy/responder/responder.go | 34 ++++++ proxy/responder/responder_test.go | 187 ++++++++++++++++++++++++++++++ 8 files changed, 468 insertions(+) create mode 100644 infra/conf/responder.go create mode 100644 infra/conf/responder_test.go create mode 100644 proxy/responder/config.pb.go create mode 100644 proxy/responder/config.proto create mode 100644 proxy/responder/responder.go create mode 100644 proxy/responder/responder_test.go diff --git a/infra/conf/responder.go b/infra/conf/responder.go new file mode 100644 index 000000000..1ba4cb6d2 --- /dev/null +++ b/infra/conf/responder.go @@ -0,0 +1,46 @@ +package conf + +import ( + "bytes" + "encoding/base64" + "net/http" + "strings" + + "github.com/xtls/xray-core/common" + "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/proxy/responder" + "google.golang.org/protobuf/proto" +) + +type ResponderConfig struct { + Type string `json:"type"` + Response string `json:"response"` +} + +func (c *ResponderConfig) Build() (proto.Message, error) { + config := new(responder.Config) + switch strings.ToLower(c.Type) { + case "http": + response := http.Response{ + StatusCode: 403, + ProtoMajor: 1, + ProtoMinor: 1, + Header: http.Header{ + "Connection": {"close"}, + "Cache-Control": {"max-age=3600, public"}, + }, + } + var data bytes.Buffer + common.Must(response.Write(&data)) + config.Response = data.Bytes() + case "custom": + data, err := base64.StdEncoding.DecodeString(c.Response) + if err != nil { + return nil, errors.New("failed to decode response").Base(err) + } + config.Response = data + default: + return nil, errors.New("unknown responder response type: ", c.Type) + } + return config, nil +} diff --git a/infra/conf/responder_test.go b/infra/conf/responder_test.go new file mode 100644 index 000000000..70a99fd4a --- /dev/null +++ b/infra/conf/responder_test.go @@ -0,0 +1,65 @@ +package conf_test + +import ( + "encoding/json" + "testing" + + "github.com/xtls/xray-core/app/proxyman" + "github.com/xtls/xray-core/common/serial" + "github.com/xtls/xray-core/core" + . "github.com/xtls/xray-core/infra/conf" + "github.com/xtls/xray-core/proxy/responder" + "google.golang.org/protobuf/proto" +) + +func TestResponderConfig(t *testing.T) { + parser := loadJSON(func() Buildable { return new(ResponderConfig) }) + httpResponse := []byte("HTTP/1.1 403 Forbidden\r\n" + + "Cache-Control: max-age=3600, public\r\n" + + "Connection: close\r\n" + + "Content-Length: 0\r\n\r\n") + + runMultiTestCase(t, []TestCase{ + {Input: `{"type":"http"}`, Parser: parser, Output: &responder.Config{Response: httpResponse}}, + {Input: `{"type":"HtTp"}`, Parser: parser, Output: &responder.Config{Response: httpResponse}}, + {Input: `{"type":"custom","response":"aGVsbG8="}`, Parser: parser, Output: &responder.Config{Response: []byte("hello")}}, + {Input: `{"type":"CuStOm","response":"aGVsbG8="}`, Parser: parser, Output: &responder.Config{Response: []byte("hello")}}, + {Input: `{"type":"custom","response":"AP+A"}`, Parser: parser, Output: &responder.Config{Response: []byte{0, 255, 128}}}, + {Input: `{"type":"custom","response":""}`, Parser: parser, Output: &responder.Config{}}, + {Input: `{"type":"custom"}`, Parser: parser, Output: &responder.Config{}}, + { + Input: `{ + "protocol": "ReSpOnDeR", + "tag": "reply", + "settings": {"type": "custom", "response": "aGVsbG8="} + }`, + Parser: func(s string) (proto.Message, error) { + var config OutboundDetourConfig + if err := json.Unmarshal([]byte(s), &config); err != nil { + return nil, err + } + return config.Build() + }, + Output: &core.OutboundHandlerConfig{ + Tag: "reply", + SenderSettings: serial.ToTypedMessage(&proxyman.SenderConfig{}), + ProxySettings: serial.ToTypedMessage(&responder.Config{Response: []byte("hello")}), + }, + }, + }) +} + +func TestResponderConfigErrors(t *testing.T) { + parser := loadJSON(func() Buildable { return new(ResponderConfig) }) + for _, input := range []string{ + `{}`, + `{"type":""}`, + `{"type":"unknown"}`, + `{"type":"none"}`, + `{"type":"custom","response":"!"}`, + } { + if _, err := parser(input); err == nil { + t.Fatalf("expected error for %s", input) + } + } +} diff --git a/infra/conf/xray.go b/infra/conf/xray.go index 6bf6ba444..0ef14a12b 100644 --- a/infra/conf/xray.go +++ b/infra/conf/xray.go @@ -45,6 +45,7 @@ var ( "block": func() interface{} { return new(BlackholeConfig) }, "blackhole": func() interface{} { return new(BlackholeConfig) }, "loopback": func() interface{} { return new(LoopbackConfig) }, + "responder": func() interface{} { return new(ResponderConfig) }, "direct": func() interface{} { return new(FreedomConfig) }, "freedom": func() interface{} { return new(FreedomConfig) }, "http": func() interface{} { return new(HTTPClientConfig) }, diff --git a/main/distro/all/all.go b/main/distro/all/all.go index 840d0e146..abe4166b4 100644 --- a/main/distro/all/all.go +++ b/main/distro/all/all.go @@ -42,6 +42,7 @@ import ( _ "github.com/xtls/xray-core/proxy/http" _ "github.com/xtls/xray-core/proxy/loopback" _ "github.com/xtls/xray-core/proxy/masque" + _ "github.com/xtls/xray-core/proxy/responder" _ "github.com/xtls/xray-core/proxy/shadowsocks" _ "github.com/xtls/xray-core/proxy/socks" _ "github.com/xtls/xray-core/proxy/trojan" diff --git a/proxy/responder/config.pb.go b/proxy/responder/config.pb.go new file mode 100644 index 000000000..ab57ff1a9 --- /dev/null +++ b/proxy/responder/config.pb.go @@ -0,0 +1,123 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v6.33.5 +// source: proxy/responder/config.proto + +package responder + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type Config struct { + state protoimpl.MessageState `protogen:"open.v1"` + Response []byte `protobuf:"bytes,1,opt,name=response,proto3" json:"response,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Config) Reset() { + *x = Config{} + mi := &file_proxy_responder_config_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Config) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Config) ProtoMessage() {} + +func (x *Config) ProtoReflect() protoreflect.Message { + mi := &file_proxy_responder_config_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Config.ProtoReflect.Descriptor instead. +func (*Config) Descriptor() ([]byte, []int) { + return file_proxy_responder_config_proto_rawDescGZIP(), []int{0} +} + +func (x *Config) GetResponse() []byte { + if x != nil { + return x.Response + } + return nil +} + +var File_proxy_responder_config_proto protoreflect.FileDescriptor + +const file_proxy_responder_config_proto_rawDesc = "" + + "\n" + + "\x1cproxy/responder/config.proto\x12\x14xray.proxy.responder\"$\n" + + "\x06Config\x12\x1a\n" + + "\bresponse\x18\x01 \x01(\fR\bresponseB^\n" + + "\x18com.xray.proxy.responderP\x01Z)github.com/xtls/xray-core/proxy/responder\xaa\x02\x14Xray.Proxy.Responderb\x06proto3" + +var ( + file_proxy_responder_config_proto_rawDescOnce sync.Once + file_proxy_responder_config_proto_rawDescData []byte +) + +func file_proxy_responder_config_proto_rawDescGZIP() []byte { + file_proxy_responder_config_proto_rawDescOnce.Do(func() { + file_proxy_responder_config_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_proxy_responder_config_proto_rawDesc), len(file_proxy_responder_config_proto_rawDesc))) + }) + return file_proxy_responder_config_proto_rawDescData +} + +var file_proxy_responder_config_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_proxy_responder_config_proto_goTypes = []any{ + (*Config)(nil), // 0: xray.proxy.responder.Config +} +var file_proxy_responder_config_proto_depIdxs = []int32{ + 0, // [0:0] is the sub-list for method output_type + 0, // [0:0] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_proxy_responder_config_proto_init() } +func file_proxy_responder_config_proto_init() { + if File_proxy_responder_config_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_proxy_responder_config_proto_rawDesc), len(file_proxy_responder_config_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_proxy_responder_config_proto_goTypes, + DependencyIndexes: file_proxy_responder_config_proto_depIdxs, + MessageInfos: file_proxy_responder_config_proto_msgTypes, + }.Build() + File_proxy_responder_config_proto = out.File + file_proxy_responder_config_proto_goTypes = nil + file_proxy_responder_config_proto_depIdxs = nil +} diff --git a/proxy/responder/config.proto b/proxy/responder/config.proto new file mode 100644 index 000000000..1fc09cd18 --- /dev/null +++ b/proxy/responder/config.proto @@ -0,0 +1,11 @@ +syntax = "proto3"; + +package xray.proxy.responder; +option csharp_namespace = "Xray.Proxy.Responder"; +option go_package = "github.com/xtls/xray-core/proxy/responder"; +option java_package = "com.xray.proxy.responder"; +option java_multiple_files = true; + +message Config { + bytes response = 1; +} diff --git a/proxy/responder/responder.go b/proxy/responder/responder.go new file mode 100644 index 000000000..eaf623929 --- /dev/null +++ b/proxy/responder/responder.go @@ -0,0 +1,34 @@ +package responder + +import ( + "context" + + "github.com/xtls/xray-core/common" + "github.com/xtls/xray-core/common/buf" + "github.com/xtls/xray-core/common/session" + "github.com/xtls/xray-core/transport" + "github.com/xtls/xray-core/transport/internet" +) + +type Handler struct { + response []byte +} + +func New(_ context.Context, config *Config) (*Handler, error) { + return &Handler{response: config.Response}, nil +} + +// Process implements proxy.Outbound. +func (h *Handler) Process(ctx context.Context, link *transport.Link, _ internet.Dialer) error { + outbounds := session.OutboundsFromContext(ctx) + ob := outbounds[len(outbounds)-1] + ob.Name = "responder" + + return link.Writer.WriteMultiBuffer(buf.MergeBytes(nil, h.response)) +} + +func init() { + common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) { + return New(ctx, config.(*Config)) + })) +} diff --git a/proxy/responder/responder_test.go b/proxy/responder/responder_test.go new file mode 100644 index 000000000..63eb38fd6 --- /dev/null +++ b/proxy/responder/responder_test.go @@ -0,0 +1,187 @@ +package responder_test + +import ( + "bytes" + "context" + "errors" + "io" + "strconv" + "testing" + "time" + + "github.com/xtls/xray-core/app/proxyman" + _ "github.com/xtls/xray-core/app/proxyman/outbound" + "github.com/xtls/xray-core/common/buf" + "github.com/xtls/xray-core/common/net" + "github.com/xtls/xray-core/common/serial" + "github.com/xtls/xray-core/common/session" + "github.com/xtls/xray-core/core" + "github.com/xtls/xray-core/features/outbound" + "github.com/xtls/xray-core/proxy/responder" + "github.com/xtls/xray-core/transport" + "github.com/xtls/xray-core/transport/pipe" +) + +type writerFunc func(buf.MultiBuffer) error + +func (f writerFunc) WriteMultiBuffer(mb buf.MultiBuffer) error { + return f(mb) +} + +func newHandler(t *testing.T, response []byte) *responder.Handler { + t.Helper() + handler, err := responder.New(context.Background(), &responder.Config{Response: response}) + if err != nil { + t.Fatal(err) + } + return handler +} + +func responseContext(network net.Network) context.Context { + return session.ContextWithOutbounds(context.Background(), []*session.Outbound{{ + Target: net.Destination{Network: network, Address: net.DomainAddress("responder.invalid"), Port: 80}, + }}) +} + +func TestResponderResponse(t *testing.T) { + tests := []struct { + name string + response []byte + }{ + {name: "empty"}, + {name: "binary", response: []byte{0, 255, 128, 1}}, + {name: "large", response: bytes.Repeat([]byte("response\x00\xff"), buf.Size)}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + handler := newHandler(t, test.response) + want := string(test.response) + for _, network := range []net.Network{net.Network_TCP, net.Network_UDP} { + t.Run(network.String(), func(t *testing.T) { + for range 2 { + var output buf.MultiBufferContainer + writes := 0 + writer := writerFunc(func(mb buf.MultiBuffer) error { + writes++ + return output.WriteMultiBuffer(mb) + }) + ctx := responseContext(network) + // No reader or dialer: the handler must respond without using either. + if err := handler.Process(ctx, &transport.Link{Writer: writer}, nil); err != nil { + t.Fatal(err) + } + if writes != 1 || output.MultiBuffer.String() != want { + t.Fatalf("writes = %d, response length = %d, want one write of %d bytes", writes, output.MultiBuffer.Len(), len(want)) + } + if got := session.OutboundsFromContext(ctx)[0].Name; got != "responder" { + t.Fatalf("outbound name = %q, want responder", got) + } + if len(output.MultiBuffer) > 0 { + output.MultiBuffer[0].Bytes()[0] ^= 0xff + } + output.Close() + if string(test.response) != want { + t.Fatal("sending or modifying the response changed the configured payload") + } + } + }) + } + }) + } +} + +func TestResponderConcurrent(t *testing.T) { + response := bytes.Repeat([]byte("concurrent response\x00\xff"), buf.Size) + want := string(response) + handler := newHandler(t, response) + for i := range 8 { + t.Run(strconv.Itoa(i), func(t *testing.T) { + t.Parallel() + var output buf.MultiBufferContainer + defer output.Close() + if err := handler.Process(responseContext(net.Network_TCP), &transport.Link{Writer: &output}, nil); err != nil { + t.Fatal(err) + } + if output.MultiBuffer.String() != want { + t.Fatal("concurrent response mismatch") + } + output.MultiBuffer[0].Bytes()[0] ^= 0xff + }) + } +} + +func TestResponderWriteError(t *testing.T) { + handler := newHandler(t, []byte("response")) + want := errors.New("write failed") + writer := writerFunc(func(mb buf.MultiBuffer) error { + buf.ReleaseMulti(mb) + return want + }) + if err := handler.Process(responseContext(net.Network_TCP), &transport.Link{Writer: writer}, nil); !errors.Is(err, want) { + t.Fatalf("Process returned %v, want %v", err, want) + } +} + +func TestResponderDispatch(t *testing.T) { + for _, response := range [][]byte{nil, bytes.Repeat([]byte("response\x00\xff"), buf.Size)} { + name := "large" + if len(response) == 0 { + name = "empty" + } + t.Run(name, func(t *testing.T) { + instance, err := core.New(&core.Config{ + App: []*serial.TypedMessage{serial.ToTypedMessage(&proxyman.OutboundConfig{})}, + Outbound: []*core.OutboundHandlerConfig{{ + Tag: "reply", + ProxySettings: serial.ToTypedMessage(&responder.Config{Response: response}), + }}, + }) + if err != nil { + t.Fatal(err) + } + defer instance.Close() + manager := instance.GetFeature(outbound.ManagerType()).(outbound.Manager) + handler := manager.GetHandler("reply") + if handler == nil { + t.Fatal("responder handler not registered") + } + for _, network := range []net.Network{net.Network_TCP, net.Network_UDP} { + t.Run(network.String(), func(t *testing.T) { + input, upload := pipe.New() + output, download := pipe.New(pipe.WithoutSizeLimit()) + defer input.Interrupt() + defer output.Interrupt() + done := make(chan struct{}) + go func() { + handler.Dispatch(responseContext(network), &transport.Link{Reader: input, Writer: download}) + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("responder did not finish without incoming data") + } + // Read after Dispatch returns: normal closure must preserve queued data. + var actual []byte + for { + mb, err := output.ReadMultiBufferTimeout(time.Second) + actual = append(actual, []byte(mb.String())...) + buf.ReleaseMulti(mb) + if err == io.EOF { + break + } + if err != nil { + t.Fatal(err) + } + } + if !bytes.Equal(actual, response) { + t.Fatalf("response mismatch: got %d bytes, want %d", len(actual), len(response)) + } + if err := upload.WriteMultiBuffer(buf.MergeBytes(nil, []byte("late request"))); err != io.ErrClosedPipe { + t.Fatalf("input was not interrupted: got %v, want %v", err, io.ErrClosedPipe) + } + }) + } + }) + } +}