1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2025-02-20 23:47:21 -05:00

242 lines
6.5 KiB
Go
Raw Normal View History

2016-01-27 12:46:40 +01:00
package shadowsocks
import (
"context"
"time"
"v2ray.com/core/common/session"
2018-05-27 13:02:29 +02:00
"v2ray.com/core/common/task"
"v2ray.com/core"
"v2ray.com/core/common"
2016-12-09 11:35:27 +01:00
"v2ray.com/core/common/buf"
2017-12-19 21:28:12 +01:00
"v2ray.com/core/common/log"
2017-01-13 23:42:39 +01:00
"v2ray.com/core/common/net"
2016-08-20 20:55:45 +02:00
"v2ray.com/core/common/protocol"
2016-12-30 00:32:20 +01:00
"v2ray.com/core/common/signal"
2016-08-20 20:55:45 +02:00
"v2ray.com/core/proxy"
"v2ray.com/core/transport/internet"
"v2ray.com/core/transport/internet/udp"
2018-04-17 00:31:10 +02:00
"v2ray.com/core/transport/pipe"
2016-01-27 12:46:40 +01:00
)
2016-05-12 10:42:08 -07:00
type Server struct {
2018-04-03 17:51:01 +02:00
config ServerConfig
user *protocol.User
account *MemoryAccount
v *core.Instance
2016-01-31 17:01:28 +01:00
}
2017-04-07 22:45:07 +02:00
// NewServer create a new Shadowsocks server.
func NewServer(ctx context.Context, config *ServerConfig) (*Server, error) {
2016-09-18 00:41:21 +02:00
if config.GetUser() == nil {
2017-04-09 01:43:25 +02:00
return nil, newError("user is not specified")
2016-01-31 17:01:28 +01:00
}
2016-10-31 15:24:28 +01:00
2016-11-02 16:18:29 +01:00
rawAccount, err := config.User.GetTypedAccount()
2016-11-02 16:17:57 +01:00
if err != nil {
2017-04-09 01:43:25 +02:00
return nil, newError("failed to get user account").Base(err)
2016-11-02 16:17:57 +01:00
}
2017-11-29 22:57:18 +01:00
account := rawAccount.(*MemoryAccount)
2016-11-02 16:17:57 +01:00
2016-09-18 00:41:21 +02:00
s := &Server{
2018-04-03 17:51:01 +02:00
config: *config,
2016-11-02 16:17:57 +01:00
user: config.GetUser(),
account: account,
2018-02-21 17:05:29 +01:00
v: core.MustFromContext(ctx),
}
2017-11-27 22:09:30 +01:00
2016-09-18 00:41:21 +02:00
return s, nil
2016-01-27 12:46:40 +01:00
}
func (s *Server) Network() net.NetworkList {
2017-01-15 00:57:06 +01:00
list := net.NetworkList{
2018-04-03 17:51:01 +02:00
Network: s.config.Network,
}
if len(list.Network) == 0 {
list.Network = append(list.Network, net.Network_TCP)
2017-01-15 00:57:06 +01:00
}
if s.config.UdpEnabled {
2017-01-15 00:57:06 +01: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-09 01:43:25 +02: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 21:30:05 +01:00
reader := buf.NewReader(conn)
for {
2017-11-09 22:33:15 +01:00
mpayload, err := reader.ReadMultiBuffer()
2016-01-28 21:30:05 +01:00
if err != nil {
break
2016-01-28 21:30:05 +01:00
}
2017-04-15 21:07:23 +02:00
for _, payload := range mpayload {
2017-09-19 23:27:49 +02:00
request, data, err := DecodeUDPPacket(s.user, payload)
2017-04-15 21:07:23 +02: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 21:28:12 +01:00
log.Record(&log.AccessMessage{
From: source,
To: "",
Status: log.AccessRejected,
Reason: err,
})
2017-04-15 21:07:23 +02:00
}
payload.Release()
continue
2017-02-09 22:49:38 +01:00
}
2017-09-19 23:27:49 +02: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 21:07:23 +02:00
payload.Release()
continue
}
2016-01-28 21:30:05 +01:00
2017-09-19 23:27:49 +02: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 21:07:23 +02:00
payload.Release()
continue
}
2016-11-02 16:17:57 +01:00
2017-04-15 21:07:23 +02:00
dest := request.Destination()
if source, ok := proxy.SourceFromContext(ctx); ok {
2017-12-19 21:28:12 +01:00
log.Record(&log.AccessMessage{
From: source,
To: dest,
Status: log.AccessAccepted,
Reason: "",
})
2017-04-15 21:07:23 +02:00
}
newError("tunnelling request to ", dest).WriteToLog(session.ExportIDToError(ctx))
2016-11-02 16:17:57 +01:00
2017-04-15 21:07:23 +02:00
ctx = protocol.ContextWithUser(ctx, request.User)
ctx = protocol.ContextWithRequestHeader(ctx, request)
udpServer.Dispatch(ctx, dest, data)
2017-04-15 21:07:23 +02:00
}
}
2016-01-28 21:30:05 +01:00
return nil
2016-01-28 21:30:05 +01: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))
2018-04-21 00:54:53 +02:00
bufferedReader := buf.BufferedReader{Reader: buf.NewReader(conn)}
request, bodyReader, err := ReadTCPSession(s.user, &bufferedReader)
2016-01-28 12:33:58 +01:00
if err != nil {
2017-12-19 21:28:12 +01:00
log.Record(&log.AccessMessage{
From: conn.RemoteAddr(),
To: "",
Status: log.AccessRejected,
Reason: err,
})
2017-04-09 01:43:25 +02:00
return newError("failed to create request from: ", conn.RemoteAddr()).Base(err)
2016-01-28 12:33:58 +01:00
}
2017-01-31 16:49:59 +01:00
conn.SetReadDeadline(time.Time{})
2016-10-31 15:24:28 +01:00
2018-04-21 00:54:53 +02:00
bufferedReader.Direct = true
2016-01-28 12:33:58 +01:00
2016-10-31 15:24:28 +01:00
dest := request.Destination()
2017-12-19 21:28:12 +01: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 22:57:20 +01:00
ctx = protocol.ContextWithUser(ctx, request.User)
2017-01-31 12:42:05 +01:00
2017-11-15 00:36:14 +01:00
ctx, cancel := context.WithCancel(ctx)
timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle)
2018-05-25 12:08:28 +02:00
ctx = core.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer)
2018-04-17 00:31:10 +02:00
link, err := dispatcher.Dispatch(ctx, dest)
if err != nil {
return err
}
2016-01-28 12:33:58 +01:00
2018-04-11 16:45:09 +02:00
responseDone := func() error {
2018-02-23 17:07:45 +01:00
defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly)
2017-11-09 22:33:15 +01:00
bufferedWriter := buf.NewBufferedWriter(buf.NewWriter(conn))
2016-10-31 15:24:28 +01:00
responseWriter, err := WriteTCPResponse(request, bufferedWriter)
if err != nil {
2017-04-09 01:43:25 +02:00
return newError("failed to write response").Base(err)
2016-10-31 15:24:28 +01:00
}
2016-02-23 18:16:13 +01:00
2018-03-28 22:23:24 +02:00
{
2018-04-17 00:31:10 +02:00
payload, err := link.Reader.ReadMultiBuffer()
2018-03-28 22:23:24 +02:00
if err != nil {
return err
}
if err := responseWriter.WriteMultiBuffer(payload); err != nil {
return err
}
2017-01-06 11:40:49 +01:00
}
if err := bufferedWriter.SetBuffered(false); err != nil {
return err
}
2018-04-17 00:31:10 +02:00
if err := buf.Copy(link.Reader, responseWriter, buf.UpdateActivity(timer)); err != nil {
2017-04-09 01:43:25 +02:00
return newError("failed to transport all TCP response").Base(err)
2016-01-28 12:40:00 +00:00
}
2016-01-28 12:33:58 +01:00
2016-12-30 00:32:20 +01:00
return nil
2018-04-11 16:45:09 +02:00
}
2016-12-30 00:32:20 +01:00
2018-04-11 16:45:09 +02:00
requestDone := func() error {
2018-02-23 17:07:45 +01:00
defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly)
2016-12-30 00:32:20 +01:00
2018-04-17 00:31:10 +02:00
if err := buf.Copy(bodyReader, link.Writer, buf.UpdateActivity(timer)); err != nil {
2017-04-09 01:43:25 +02:00
return newError("failed to transport all TCP request").Base(err)
2016-12-30 00:32:20 +01:00
}
2018-02-23 17:07:45 +01:00
2016-12-30 00:32:20 +01:00
return nil
2018-04-11 16:45:09 +02:00
}
2016-01-28 12:33:58 +01:00
2018-05-27 13:02:29 +02: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-17 00:31:10 +02:00
pipe.CloseError(link.Reader)
pipe.CloseError(link.Writer)
2017-04-09 01:43:25 +02:00
return newError("connection ends").Base(err)
2016-12-30 00:32:20 +01:00
}
return nil
2016-01-28 12:33:58 +01: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 22:54:08 +02:00
}