1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-12-23 02:26:49 -05:00
v2fly/proxy/shadowsocks/server.go

251 lines
6.7 KiB
Go
Raw Normal View History

2016-01-27 06:46:40 -05:00
package shadowsocks
import (
"context"
"time"
"v2ray.com/core"
"v2ray.com/core/common"
2016-12-09 05:35:27 -05:00
"v2ray.com/core/common/buf"
2017-12-19 15:28:12 -05:00
"v2ray.com/core/common/log"
2017-01-13 17:42:39 -05:00
"v2ray.com/core/common/net"
2016-08-20 14:55:45 -04:00
"v2ray.com/core/common/protocol"
2018-07-31 13:33:59 -04:00
"v2ray.com/core/common/session"
2016-12-29 18:32:20 -05:00
"v2ray.com/core/common/signal"
2018-07-31 13:33:59 -04:00
"v2ray.com/core/common/task"
2016-08-20 14:55:45 -04:00
"v2ray.com/core/proxy"
"v2ray.com/core/transport/internet"
"v2ray.com/core/transport/internet/udp"
2018-04-16 18:31:10 -04:00
"v2ray.com/core/transport/pipe"
2016-01-27 06:46:40 -05:00
)
2016-05-12 13:42:08 -04:00
type Server struct {
2018-04-03 11:51:01 -04:00
config ServerConfig
user *protocol.User
account *MemoryAccount
v *core.Instance
2016-01-31 11:01:28 -05:00
}
2017-04-07 16:45:07 -04:00
// NewServer create a new Shadowsocks server.
func NewServer(ctx context.Context, config *ServerConfig) (*Server, error) {
2016-09-17 18:41:21 -04:00
if config.GetUser() == nil {
2017-04-08 19:43:25 -04:00
return nil, newError("user is not specified")
2016-01-31 11:01:28 -05:00
}
2016-10-31 10:24:28 -04:00
2016-11-02 11:18:29 -04:00
rawAccount, err := config.User.GetTypedAccount()
2016-11-02 11:17:57 -04:00
if err != nil {
2017-04-08 19:43:25 -04:00
return nil, newError("failed to get user account").Base(err)
2016-11-02 11:17:57 -04:00
}
2017-11-29 16:57:18 -05:00
account := rawAccount.(*MemoryAccount)
2016-11-02 11:17:57 -04:00
2016-09-17 18:41:21 -04:00
s := &Server{
2018-04-03 11:51:01 -04:00
config: *config,
2016-11-02 11:17:57 -04:00
user: config.GetUser(),
account: account,
2018-02-21 11:05:29 -05:00
v: core.MustFromContext(ctx),
}
2017-11-27 16:09:30 -05:00
2016-09-17 18:41:21 -04:00
return s, nil
2016-01-27 06:46:40 -05:00
}
func (s *Server) Network() net.NetworkList {
2017-01-14 18:57:06 -05:00
list := net.NetworkList{
2018-04-03 11:51:01 -04:00
Network: s.config.Network,
}
if len(list.Network) == 0 {
list.Network = append(list.Network, net.Network_TCP)
2017-01-14 18:57:06 -05:00
}
if s.config.UdpEnabled {
2017-01-14 18:57:06 -05:00
list.Network = append(list.Network, net.Network_UDP)
}
return list
}
func (s *Server) Process(ctx context.Context, network net.Network, conn internet.Connection, dispatcher core.Dispatcher) error {
switch network {
case net.Network_TCP:
return s.handleConnection(ctx, conn, dispatcher)
case net.Network_UDP:
return s.handlerUDPPayload(ctx, conn, dispatcher)
default:
2017-04-08 19:43:25 -04:00
return newError("unknown network: ", network)
}
}
func (s *Server) handlerUDPPayload(ctx context.Context, conn internet.Connection, dispatcher core.Dispatcher) error {
udpServer := udp.NewDispatcher(dispatcher, func(ctx context.Context, payload *buf.Buffer) {
request := protocol.RequestHeaderFromContext(ctx)
if request == nil {
return
}
data, err := EncodeUDPPacket(request, payload.Bytes())
payload.Release()
if err != nil {
newError("failed to encode UDP packet").Base(err).AtWarning().WriteToLog(session.ExportIDToError(ctx))
return
}
defer data.Release()
conn.Write(data.Bytes())
})
2016-01-28 15:30:05 -05:00
reader := buf.NewReader(conn)
for {
2017-11-09 16:33:15 -05:00
mpayload, err := reader.ReadMultiBuffer()
2016-01-28 15:30:05 -05:00
if err != nil {
break
2016-01-28 15:30:05 -05:00
}
2017-04-15 15:07:23 -04:00
for _, payload := range mpayload {
2017-09-19 17:27:49 -04:00
request, data, err := DecodeUDPPacket(s.user, payload)
2017-04-15 15:07:23 -04:00
if err != nil {
if source, ok := proxy.SourceFromContext(ctx); ok {
newError("dropping invalid UDP packet from: ", source).Base(err).WriteToLog(session.ExportIDToError(ctx))
2017-12-19 15:28:12 -05:00
log.Record(&log.AccessMessage{
From: source,
To: "",
Status: log.AccessRejected,
Reason: err,
})
2017-04-15 15:07:23 -04:00
}
payload.Release()
continue
2017-02-09 16:49:38 -05:00
}
2017-09-19 17:27:49 -04:00
if request.Option.Has(RequestOptionOneTimeAuth) && s.account.OneTimeAuth == Account_Disabled {
newError("client payload enables OTA but server doesn't allow it").WriteToLog(session.ExportIDToError(ctx))
2017-04-15 15:07:23 -04:00
payload.Release()
continue
}
2016-01-28 15:30:05 -05:00
2017-09-19 17:27:49 -04:00
if !request.Option.Has(RequestOptionOneTimeAuth) && s.account.OneTimeAuth == Account_Enabled {
newError("client payload disables OTA but server forces it").WriteToLog(session.ExportIDToError(ctx))
2017-04-15 15:07:23 -04:00
payload.Release()
continue
}
2016-11-02 11:17:57 -04:00
2017-04-15 15:07:23 -04:00
dest := request.Destination()
if source, ok := proxy.SourceFromContext(ctx); ok {
2017-12-19 15:28:12 -05:00
log.Record(&log.AccessMessage{
From: source,
To: dest,
Status: log.AccessAccepted,
Reason: "",
})
2017-04-15 15:07:23 -04:00
}
newError("tunnelling request to ", dest).WriteToLog(session.ExportIDToError(ctx))
2016-11-02 11:17:57 -04:00
2017-04-15 15:07:23 -04:00
ctx = protocol.ContextWithUser(ctx, request.User)
ctx = protocol.ContextWithRequestHeader(ctx, request)
udpServer.Dispatch(ctx, dest, data)
2017-04-15 15:07:23 -04:00
}
}
2016-01-28 15:30:05 -05:00
return nil
2016-01-28 15:30:05 -05:00
}
func (s *Server) handleConnection(ctx context.Context, conn internet.Connection, dispatcher core.Dispatcher) error {
sessionPolicy := s.v.PolicyManager().ForLevel(s.user.Level)
conn.SetReadDeadline(time.Now().Add(sessionPolicy.Timeouts.Handshake))
var bufferedReader buf.BufferedReader
{
var reader buf.Reader
if sessionPolicy.Buffer.PerConnection == 0 {
reader = &buf.SingleReader{Reader: conn}
} else {
reader = buf.NewReader(conn)
}
bufferedReader = buf.BufferedReader{Reader: reader}
}
2018-04-20 18:54:53 -04:00
request, bodyReader, err := ReadTCPSession(s.user, &bufferedReader)
2016-01-28 06:33:58 -05:00
if err != nil {
2017-12-19 15:28:12 -05:00
log.Record(&log.AccessMessage{
From: conn.RemoteAddr(),
To: "",
Status: log.AccessRejected,
Reason: err,
})
2017-04-08 19:43:25 -04:00
return newError("failed to create request from: ", conn.RemoteAddr()).Base(err)
2016-01-28 06:33:58 -05:00
}
2017-01-31 10:49:59 -05:00
conn.SetReadDeadline(time.Time{})
2016-10-31 10:24:28 -04:00
2018-04-20 18:54:53 -04:00
bufferedReader.Direct = true
2016-01-28 06:33:58 -05:00
2016-10-31 10:24:28 -04:00
dest := request.Destination()
2017-12-19 15:28:12 -05:00
log.Record(&log.AccessMessage{
From: conn.RemoteAddr(),
To: dest,
Status: log.AccessAccepted,
Reason: "",
})
newError("tunnelling request to ", dest).WriteToLog(session.ExportIDToError(ctx))
2016-01-30 16:57:20 -05:00
ctx = protocol.ContextWithUser(ctx, request.User)
2017-01-31 06:42:05 -05:00
2017-11-14 18:36:14 -05:00
ctx, cancel := context.WithCancel(ctx)
timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle)
2018-05-25 06:08:28 -04:00
ctx = core.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer)
2018-04-16 18:31:10 -04:00
link, err := dispatcher.Dispatch(ctx, dest)
if err != nil {
return err
}
2016-01-28 06:33:58 -05:00
2018-04-11 10:45:09 -04:00
responseDone := func() error {
2018-02-23 11:07:45 -05:00
defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly)
2017-11-09 16:33:15 -05:00
bufferedWriter := buf.NewBufferedWriter(buf.NewWriter(conn))
2016-10-31 10:24:28 -04:00
responseWriter, err := WriteTCPResponse(request, bufferedWriter)
if err != nil {
2017-04-08 19:43:25 -04:00
return newError("failed to write response").Base(err)
2016-10-31 10:24:28 -04:00
}
2016-02-23 12:16:13 -05:00
2018-03-28 16:23:24 -04:00
{
2018-04-16 18:31:10 -04:00
payload, err := link.Reader.ReadMultiBuffer()
2018-03-28 16:23:24 -04:00
if err != nil {
return err
}
if err := responseWriter.WriteMultiBuffer(payload); err != nil {
return err
}
2017-01-06 05:40:49 -05:00
}
if err := bufferedWriter.SetBuffered(false); err != nil {
return err
}
2018-04-16 18:31:10 -04:00
if err := buf.Copy(link.Reader, responseWriter, buf.UpdateActivity(timer)); err != nil {
2017-04-08 19:43:25 -04:00
return newError("failed to transport all TCP response").Base(err)
2016-01-28 07:40:00 -05:00
}
2016-01-28 06:33:58 -05:00
2016-12-29 18:32:20 -05:00
return nil
2018-04-11 10:45:09 -04:00
}
2016-12-29 18:32:20 -05:00
2018-04-11 10:45:09 -04:00
requestDone := func() error {
2018-02-23 11:07:45 -05:00
defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly)
2016-12-29 18:32:20 -05:00
2018-04-16 18:31:10 -04:00
if err := buf.Copy(bodyReader, link.Writer, buf.UpdateActivity(timer)); err != nil {
2017-04-08 19:43:25 -04:00
return newError("failed to transport all TCP request").Base(err)
2016-12-29 18:32:20 -05:00
}
2018-02-23 11:07:45 -05:00
2016-12-29 18:32:20 -05:00
return nil
2018-04-11 10:45:09 -04:00
}
2016-01-28 06:33:58 -05:00
2018-05-27 07:02:29 -04:00
var requestDoneAndCloseWriter = task.Single(requestDone, task.OnSuccess(task.Close(link.Writer)))
if err := task.Run(task.WithContext(ctx), task.Parallel(requestDoneAndCloseWriter, responseDone))(); err != nil {
2018-04-16 18:31:10 -04:00
pipe.CloseError(link.Reader)
pipe.CloseError(link.Writer)
2017-04-08 19:43:25 -04:00
return newError("connection ends").Base(err)
2016-12-29 18:32:20 -05:00
}
return nil
2016-01-28 06:33:58 -05:00
}
func init() {
common.Must(common.RegisterConfig((*ServerConfig)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
return NewServer(ctx, config.(*ServerConfig))
}))
2016-06-14 16:54:08 -04:00
}