Implemented Primitive Domain Socket Support described in https://github.com/v2ray/Planning/issues/25

app/commander/errors.generated.go

@@ -2,6 +2,4 @@ package commander
 
 import "v2ray.com/core/common/errors"
 
-func newError(values ...interface{}) *errors.Error {
- return errors.New(values...).Path("App", "Commander")
-}
+func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("App", "Commander") }

app/dispatcher/errors.generated.go

@@ -2,6 +2,4 @@ package dispatcher
 
 import "v2ray.com/core/common/errors"
 
-func newError(values ...interface{}) *errors.Error {
- return errors.New(values...).Path("App", "Dispatcher")
-}
+func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("App", "Dispatcher") }

app/proxyman/command/errors.generated.go

@@ -2,6 +2,4 @@ package command
 
 import "v2ray.com/core/common/errors"
 
-func newError(values ...interface{}) *errors.Error {
- return errors.New(values...).Path("App", "Proxyman", "Command")
-}
+func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("App", "Proxyman", "Command") }

app/proxyman/config.proto

@@ -9,6 +9,8 @@ option java_multiple_files = true;
 import "v2ray.com/core/common/net/address.proto";
 import "v2ray.com/core/common/net/port.proto";
 import "v2ray.com/core/transport/internet/config.proto";
+import "v2ray.com/core/transport/internet/domainsocket/config.proto";
+import "v2ray.com/core/common/serial/typed_message.proto";
 
 message InboundConfig {
 }
@@ -60,8 +62,23 @@ message ReceiverConfig {
  repeated KnownProtocols domain_override = 7;
 }
 
+message UnixReceiverConfig {
+ reserved 1;
+ v2ray.core.internet.domainsocket.DomainSocketSettings domainSockSettings = 2;
+ reserved 3;
+ v2ray.core.transport.internet.StreamConfig stream_settings = 4;
+ reserved 5,6;
+ repeated KnownProtocols domain_override = 7;
+}
+
+message InboundHandlerConfig {
+ string tag = 1;
+ v2ray.core.common.serial.TypedMessage receiver_settings = 2;
+ v2ray.core.common.serial.TypedMessage proxy_settings = 3;
+}
+
 message OutboundConfig {
- 
+
 }
 
 message SenderConfig {
@@ -72,6 +89,14 @@ message SenderConfig {
  MultiplexingConfig multiplex_settings = 4;
 }
 
+
+message UnixSenderConfig {
+ v2ray.core.internet.domainsocket.DomainSocketSettings domainSockSettings = 1;
+ v2ray.core.transport.internet.StreamConfig stream_settings = 2;
+ v2ray.core.transport.internet.ProxyConfig proxy_settings = 3;
+ MultiplexingConfig multiplex_settings = 4;
+}
+
 message MultiplexingConfig {
  // Whether or not Mux is enabled.
  bool enabled = 1;

app/proxyman/errors.generated.go

@@ -2,6 +2,4 @@ package proxyman
 
 import "v2ray.com/core/common/errors"
 
-func newError(values ...interface{}) *errors.Error {
- return errors.New(values...).Path("App", "Proxyman")
-}
+func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("App", "Proxyman") }

app/proxyman/inbound/errors.generated.go

@@ -2,6 +2,4 @@ package inbound
 
 import "v2ray.com/core/common/errors"
 
-func newError(values ...interface{}) *errors.Error {
- return errors.New(values...).Path("App", "Proxyman", "Inbound")
-}
+func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("App", "Proxyman", "Inbound") }

app/proxyman/inbound/inbound.go

@@ -126,15 +126,21 @@ func NewHandler(ctx context.Context, config *core.InboundHandlerConfig) (core.In
  if err != nil {
  return nil, err
  }
- receiverSettings, ok := rawReceiverSettings.(*proxyman.ReceiverConfig)
- if !ok {
- return nil, newError("not a ReceiverConfig").AtError()
- }
  proxySettings, err := config.ProxySettings.GetInstance()
  if err != nil {
  return nil, err
  }
  tag := config.Tag
+
+ receiverSettings, ok := rawReceiverSettings.(*proxyman.ReceiverConfig)
+ if !ok {
+ receiverSettings, ok := rawReceiverSettings.(*proxyman.UnixReceiverConfig)
+ if ok {
+ return NewUnixInboundHandler(ctx, tag, receiverSettings, proxySettings)
+ }
+ return nil, newError("not a ReceiverConfig").AtError()
+ }
+
  allocStrategy := receiverSettings.AllocationStrategy
  if allocStrategy == nil || allocStrategy.Type == proxyman.AllocationStrategy_Always {
  return NewAlwaysOnInboundHandler(ctx, tag, receiverSettings, proxySettings)

app/proxyman/inbound/unix.go

@@ -0,0 +1,100 @@
+package inbound
+
+import (
+ "context"
+
+ "v2ray.com/core/app/proxyman"
+ "v2ray.com/core/app/proxyman/mux"
+ "v2ray.com/core/common"
+ "v2ray.com/core/common/net"
+ "v2ray.com/core/proxy"
+ "v2ray.com/core/transport/internet/domainsocket"
+)
+
+type UnixInboundHandler struct {
+ tag string
+ listenerHolder *domainsocket.Listener
+ ctx context.Context
+ path string
+ proxy proxy.Inbound
+ mux *mux.Server
+ additional *proxyman.UnixReceiverConfig
+}
+
+func (uih *UnixInboundHandler) Start() error {
+ var err error
+ uih.listenerHolder, err = domainsocket.ListenDS(uih.ctx, uih.path)
+ if err != nil {
+ return newError(err).AtError()
+ }
+ err = uih.listenerHolder.LowerUP()
+ if err != nil {
+ return newError(err).AtError()
+ }
+ nchan := make(chan net.Conn, 2)
+ err = uih.listenerHolder.UP(nchan, false)
+ if err != nil {
+ return newError(err).AtError()
+ }
+ go uih.progressTraffic(nchan)
+ return nil
+}
+func (uih *UnixInboundHandler) progressTraffic(rece <-chan net.Conn) {
+
+ for {
+ conn, notclosed := <-rece
+ if !notclosed {
+ return
+ }
+ go func(conn net.Conn) {
+ ctx, cancel := context.WithCancel(uih.ctx)
+ if len(uih.tag) > 0 {
+ ctx = proxy.ContextWithInboundTag(ctx, uih.tag)
+ }
+ if err := uih.proxy.Process(ctx, net.Network_TCP, conn, uih.mux); err != nil {
+ newError("connection ends").Base(err).WriteToLog()
+ }
+ cancel()
+ conn.Close()
+ }(conn)
+ }
+}
+func (uih *UnixInboundHandler) Close() error {
+ if uih.listenerHolder != nil {
+ uih.listenerHolder.Down()
+ return nil
+ }
+ return newError("Called UnixInboundHandler.Close while listenerHolder is nil")
+
+}
+func (uih *UnixInboundHandler) Tag() string {
+ return uih.tag
+}
+
+func (uih *UnixInboundHandler) GetRandomInboundProxy() (interface{}, net.Port, int) {
+ //It makes bo sense to support it
+ return nil, 0, 0
+}
+
+func NewUnixInboundHandler(ctx context.Context, tag string, receiverConfig *proxyman.UnixReceiverConfig, proxyConfig interface{}) (*UnixInboundHandler, error) {
+ rawProxy, err := common.CreateObject(ctx, proxyConfig)
+ if err != nil {
+ return nil, err
+ }
+ p, ok := rawProxy.(proxy.Inbound)
+ if !ok {
+ return nil, newError("not an inbound proxy.")
+ }
+
+ h := &UnixInboundHandler{
+ proxy: p,
+ mux: mux.NewServer(ctx),
+ tag: tag,
+ ctx: ctx,
+ path: receiverConfig.DomainSockSettings.GetPath(),
+ additional: receiverConfig,
+ }
+
+ return h, nil
+
+}

app/proxyman/mux/errors.generated.go

@@ -2,6 +2,4 @@ package mux
 
 import "v2ray.com/core/common/errors"
 
-func newError(values ...interface{}) *errors.Error {
- return errors.New(values...).Path("App", "Proxyman", "Mux")
-}
+func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("App", "Proxyman", "Mux") }

app/proxyman/outbound/errors.generated.go

@@ -2,6 +2,4 @@ package outbound
 
 import "v2ray.com/core/common/errors"
 
-func newError(values ...interface{}) *errors.Error {
- return errors.New(values...).Path("App", "Proxyman", "Outbound")
-}
+func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("App", "Proxyman", "Outbound") }

app/proxyman/outbound/handler.go

@@ -21,7 +21,7 @@ type Handler struct {
  mux *mux.ClientManager
 }
 
-func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (*Handler, error) {
+func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (core.OutboundHandler, error) {
  v := core.MustFromContext(ctx)
  h := &Handler{
  config: config,
@@ -36,6 +36,8 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (*Handl
  switch s := senderSettings.(type) {
  case *proxyman.SenderConfig:
  h.senderSettings = s
+ case *proxyman.UnixSenderConfig:
+ return NewUnixHandler(ctx, config)
  default:
  return nil, newError("settings is not SenderConfig")
  }

app/proxyman/outbound/unix.go

@@ -0,0 +1,132 @@
+package outbound
+
+import (
+ "context"
+ "io"
+
+ "v2ray.com/core"
+ "v2ray.com/core/app/proxyman"
+ "v2ray.com/core/app/proxyman/mux"
+ "v2ray.com/core/common"
+ "v2ray.com/core/common/errors"
+ "v2ray.com/core/common/net"
+ "v2ray.com/core/proxy"
+ "v2ray.com/core/transport/internet"
+ "v2ray.com/core/transport/internet/domainsocket"
+ "v2ray.com/core/transport/ray"
+)
+
+type UnixHandler struct {
+ config *core.OutboundHandlerConfig
+ senderSettings *proxyman.UnixSenderConfig
+ proxy proxy.Outbound
+ outboundManager core.OutboundHandlerManager
+ mux *mux.ClientManager
+}
+
+func NewUnixHandler(ctx context.Context, config *core.OutboundHandlerConfig) (core.OutboundHandler, error) {
+ v := core.FromContext(ctx)
+ if v == nil {
+ return nil, newError("V is not in context")
+ }
+ h := &UnixHandler{
+ config: config,
+ outboundManager: v.OutboundHandlerManager(),
+ }
+
+ if config.SenderSettings != nil {
+ senderSettings, err := config.SenderSettings.GetInstance()
+ if err != nil {
+ return nil, err
+ }
+ switch s := senderSettings.(type) {
+ case *proxyman.UnixSenderConfig:
+ h.senderSettings = s
+ default:
+ return nil, newError("settings is not UnixSenderConfig")
+ }
+ }
+
+ proxyConfig, err := config.ProxySettings.GetInstance()
+ if err != nil {
+ return nil, err
+ }
+
+ rawProxyHandler, err := common.CreateObject(ctx, proxyConfig)
+ if err != nil {
+ return nil, err
+ }
+
+ proxyHandler, ok := rawProxyHandler.(proxy.Outbound)
+ if !ok {
+ return nil, newError("not an outbound handler")
+ }
+
+ if h.senderSettings != nil && h.senderSettings.MultiplexSettings != nil && h.senderSettings.MultiplexSettings.Enabled {
+ config := h.senderSettings.MultiplexSettings
+ if config.Concurrency < 1 || config.Concurrency > 1024 {
+ return nil, newError("invalid mux concurrency: ", config.Concurrency).AtWarning()
+ }
+ h.mux = mux.NewClientManager(proxyHandler, h, config)
+ }
+
+ h.proxy = proxyHandler
+ return h, nil
+}
+
+// Tag implements core.OutboundHandler.
+func (h *UnixHandler) Tag() string {
+ return h.config.Tag
+}
+
+// Dispatch implements proxy.Outbound.Dispatch.
+func (h *UnixHandler) Dispatch(ctx context.Context, outboundRay ray.OutboundRay) {
+ if h.mux != nil {
+ err := h.mux.Dispatch(ctx, outboundRay)
+ if err != nil {
+ newError("failed to process outbound traffic").Base(err).WriteToLog()
+ outboundRay.OutboundOutput().CloseError()
+ }
+ } else {
+ err := h.proxy.Process(ctx, outboundRay, h)
+ // Ensure outbound ray is properly closed.
+ if err != nil && errors.Cause(err) != io.EOF {
+ newError("failed to process outbound traffic").Base(err).WriteToLog()
+ outboundRay.OutboundOutput().CloseError()
+ } else {
+ outboundRay.OutboundOutput().Close()
+ }
+ outboundRay.OutboundInput().CloseError()
+ }
+}
+
+// Dial implements proxy.Dialer.Dial().
+func (h *UnixHandler) Dial(ctx context.Context, dest net.Destination) (internet.Connection, error) {
+ if h.senderSettings != nil {
+ if h.senderSettings.ProxySettings.HasTag() {
+ newError("Unix domain socket does not support redirect").AtWarning().WriteToLog()
+ }
+
+ if h.senderSettings.StreamSettings != nil {
+ newError("Unix domain socket does not support stream setting").AtWarning().WriteToLog()
+ }
+ }
+
+ return domainsocket.DialDS(ctx, h.senderSettings.GetDomainSockSettings().GetPath())
+}
+
+// GetOutbound implements proxy.GetOutbound.
+func (h *UnixHandler) GetOutbound() proxy.Outbound {
+ return h.proxy
+}
+
+// Start implements common.Runnable.
+func (h *UnixHandler) Start() error {
+ return nil
+}
+
+// Close implements common.Runnable.
+func (h *UnixHandler) Close() error {
+ common.Close(h.mux)
+ return nil
+}