lzhr
/
v2fly
mirror of https://github.com/v2fly/v2ray-core.git
1
0
Fork 0
Code Issues Releases Wiki Activity

session id

This commit is contained in:
Darien Raymond 2018-02-22 15:26:00 +01:00
parent 80a1e73361
commit 6b872c266c
No known key found for this signature in database
GPG Key ID: 7251FFA14BB18169
19 changed files with 143 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
app/dispatcher/default.go
View File

@ -62,7 +62,7 @@ func (d *DefaultDispatcher) Dispatch(ctx context.Context, destination net.Destin
go func() {
domain, err := snifer(ctx, sniferList, outbound)
if err == nil {
newError("sniffed domain: ", domain).WriteToLog()
newError("sniffed domain: ", domain).WithContext(ctx).WriteToLog()
destination.Address = net.ParseAddress(domain)
ctx = proxy.ContextWithTarget(ctx, destination)
}
@ -107,13 +107,13 @@ func (d *DefaultDispatcher) routedDispatch(ctx context.Context, outbound ray.Out
if d.router != nil {
if tag, err := d.router.PickRoute(ctx); err == nil {
if handler := d.ohm.GetHandler(tag); handler != nil {
newError("taking detour [", tag, "] for [", destination, "]").WriteToLog()
newError("taking detour [", tag, "] for [", destination, "]").WithContext(ctx).WriteToLog()
dispatcher = handler
} else {
newError("nonexisting tag: ", tag).AtWarning().WriteToLog()
newError("nonexisting tag: ", tag).AtWarning().WithContext(ctx).WriteToLog()
}
} else {
newError("default route for ", destination).WriteToLog()
newError("default route for ", destination).WithContext(ctx).WriteToLog()
}
}
dispatcher.Dispatch(ctx, outbound)

14
app/proxyman/inbound/worker.go
View File

@ -7,6 +7,8 @@ import (
"sync/atomic"
"time"
"v2ray.com/core/common/session"
"v2ray.com/core"
"v2ray.com/core/app/proxyman"
"v2ray.com/core/common"
@ -41,10 +43,13 @@ type tcpWorker struct {
func (w *tcpWorker) callback(conn internet.Connection) {
ctx, cancel := context.WithCancel(context.Background())
sid := session.NewID()
ctx = session.ContextWithID(ctx, sid)
if w.recvOrigDest {
dest, err := tcp.GetOriginalDestination(conn)
if err != nil {
newError("failed to get original destination").Base(err).WriteToLog()
newError("failed to get original destination").WithContext(ctx).Base(err).WriteToLog()
}
if dest.IsValid() {
ctx = proxy.ContextWithOriginalTarget(ctx, dest)
@ -59,11 +64,11 @@ func (w *tcpWorker) callback(conn internet.Connection) {
ctx = proxyman.ContextWithProtocolSniffers(ctx, w.sniffers)
}
if err := w.proxy.Process(ctx, net.Network_TCP, conn, w.dispatcher); err != nil {
newError("connection ends").Base(err).WriteToLog()
newError("connection ends").Base(err).WithContext(ctx).WriteToLog()
}
cancel()
if err := conn.Close(); err != nil {
newError("failed to close connection").Base(err).WriteToLog()
newError("failed to close connection").Base(err).WithContext(ctx).WriteToLog()
}
}
@ -220,6 +225,9 @@ func (w *udpWorker) callback(b *buf.Buffer, source net.Destination, originalDest
if !existing {
go func() {
ctx := context.Background()
sid := session.NewID()
ctx = session.ContextWithID(ctx, sid)
if originalDest.IsValid() {
ctx = proxy.ContextWithOriginalTarget(ctx, originalDest)
}

12
app/proxyman/mux/mux.go
View File

@ -149,14 +149,14 @@ func fetchInput(ctx context.Context, s *Session, output buf.Writer) {
defer writer.Close()
defer s.Close()
newError("dispatching request to ", dest).WriteToLog()
newError("dispatching request to ", dest).WithContext(ctx).WriteToLog()
data, _ := s.input.ReadTimeout(time.Millisecond * 500)
if err := writer.WriteMultiBuffer(data); err != nil {
newError("failed to write first payload").Base(err).WriteToLog()
newError("failed to write first payload").Base(err).WithContext(ctx).WriteToLog()
return
}
if err := buf.Copy(s.input, writer); err != nil {
newError("failed to fetch all input").Base(err).WriteToLog()
newError("failed to fetch all input").Base(err).WithContext(ctx).WriteToLog()
}
}
@ -298,7 +298,7 @@ type ServerWorker struct {
func handle(ctx context.Context, s *Session, output buf.Writer) {
writer := NewResponseWriter(s.ID, output, s.transferType)
if err := buf.Copy(s.input, writer); err != nil {
newError("session ", s.ID, " ends.").Base(err).WriteToLog()
newError("session ", s.ID, " ends.").Base(err).WithContext(ctx).WriteToLog()
}
writer.Close()
s.Close()
@ -312,7 +312,7 @@ func (w *ServerWorker) handleStatusKeepAlive(meta *FrameMetadata, reader *buf.Bu
}
func (w *ServerWorker) handleStatusNew(ctx context.Context, meta *FrameMetadata, reader *buf.BufferedReader) error {
newError("received request for ", meta.Target).WriteToLog()
newError("received request for ", meta.Target).WithContext(ctx).WriteToLog()
inboundRay, err := w.dispatcher.Dispatch(ctx, meta.Target)
if err != nil {
if meta.Option.Has(OptionData) {
@ -397,7 +397,7 @@ func (w *ServerWorker) run(ctx context.Context) {
err := w.handleFrame(ctx, reader)
if err != nil {
if errors.Cause(err) != io.EOF {
newError("unexpected EOF").Base(err).WriteToLog()
newError("unexpected EOF").Base(err).WithContext(ctx).WriteToLog()
input.CloseError()
}
return

8
app/proxyman/outbound/handler.go
View File

@ -78,14 +78,14 @@ func (h *Handler) 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()
newError("failed to process outbound traffic").Base(err).WithContext(ctx).WriteToLog()
outboundRay.OutboundOutput().CloseError()
}
} else {
err := h.proxy.Process(ctx, outboundRay, h)
// Ensure outbound ray is properly closed.
if err != nil {
newError("failed to process outbound traffic").Base(err).WriteToLog()
newError("failed to process outbound traffic").Base(err).WithContext(ctx).WriteToLog()
outboundRay.OutboundOutput().CloseError()
} else {
outboundRay.OutboundOutput().Close()
@ -101,14 +101,14 @@ func (h *Handler) Dial(ctx context.Context, dest net.Destination) (internet.Conn
tag := h.senderSettings.ProxySettings.Tag
handler := h.outboundManager.GetHandler(tag)
if handler != nil {
newError("proxying to ", tag, " for dest ", dest).AtDebug().WriteToLog()
newError("proxying to ", tag, " for dest ", dest).AtDebug().WithContext(ctx).WriteToLog()
ctx = proxy.ContextWithTarget(ctx, dest)
stream := ray.NewRay(ctx)
go handler.Dispatch(ctx, stream)
return ray.NewConnection(stream.InboundOutput(), stream.InboundInput()), nil
}
newError("failed to get outbound handler with tag: ", tag).AtWarning().WriteToLog()
newError("failed to get outbound handler with tag: ", tag).AtWarning().WithContext(ctx).WriteToLog()
}
if h.senderSettings.Via != nil {

51
common/errors/errors.go
View File

@ -2,10 +2,12 @@
package errors
import (
"context"
"strings"
"v2ray.com/core/common/log"
"v2ray.com/core/common/serial"
"v2ray.com/core/common/session"
)
type hasInnerError interface {
@ -17,12 +19,17 @@ type hasSeverity interface {
Severity() log.Severity
}
type hasContext interface {
Context() context.Context
}
// Error is an error object with underlying error.
type Error struct {
message []interface{}
inner error
severity log.Severity
path []string
ctx context.Context
}
// Error implements error.Error().
@ -50,6 +57,27 @@ func (v *Error) Base(err error) *Error {
return v
}
func (v *Error) WithContext(ctx context.Context) *Error {
v.ctx = ctx
return v
}
func (v *Error) Context() context.Context {
if v.ctx != nil {
return v.ctx
}
if v.inner == nil {
return nil
}
if c, ok := v.inner.(hasContext); ok {
return c.Context()
}
return nil
}
func (v *Error) atSeverity(s log.Severity) *Error {
v.severity = s
return v
@ -103,9 +131,21 @@ func (v *Error) String() string {
// WriteToLog writes current error into log.
func (v *Error) WriteToLog() {
ctx := v.Context()
var sid session.ID
if ctx != nil {
sid = session.IDFromContext(ctx)
}
var c interface{} = v
if sid > 0 {
c = sessionLog{
id: sid,
content: v,
}
}
log.Record(&log.GeneralMessage{
Severity: GetSeverity(v),
Content: v,
Content: c,
})
}
@ -139,3 +179,12 @@ func GetSeverity(err error) log.Severity {
}
return log.Severity_Info
}
type sessionLog struct {
id session.ID
content interface{}
}
func (s sessionLog) String() string {
return serial.Concat("[", s.id, "] ", s.content)
}

34
common/session/session.go Normal file
View File

@ -0,0 +1,34 @@
package session
import (
"context"
"math/rand"
)
type ID uint32
func NewID() ID {
for {
id := ID(rand.Uint32())
if id != 0 {
return id
}
}
}
type sessionKey int
const (
idSessionKey sessionKey = iota
)
func ContextWithID(ctx context.Context, id ID) context.Context {
return context.WithValue(ctx, idSessionKey, id)
}
func IDFromContext(ctx context.Context) ID {
if id, ok := ctx.Value(idSessionKey).(ID); ok {
return id
}
return 0
}

2
proxy/dokodemo/dokodemo.go
View File

@ -53,7 +53,7 @@ func (d *DokodemoDoor) policy() core.Policy {
}
func (d *DokodemoDoor) Process(ctx context.Context, network net.Network, conn internet.Connection, dispatcher core.Dispatcher) error {
newError("processing connection from: ", conn.RemoteAddr()).AtDebug().WriteToLog()
newError("processing connection from: ", conn.RemoteAddr()).AtDebug().WithContext(ctx).WriteToLog()
dest := net.Destination{
Network: network,
Address: d.address,

6
proxy/freedom/freedom.go
View File

@ -56,7 +56,7 @@ func (h *Handler) resolveIP(ctx context.Context, domain string) net.Address {
ips, err := h.dns.LookupIP(domain)
if err != nil {
newError("failed to get IP address for domain ", domain).Base(err).WriteToLog()
newError("failed to get IP address for domain ", domain).Base(err).WithContext(ctx).WriteToLog()
}
if len(ips) == 0 {
return nil
@ -75,7 +75,7 @@ func (h *Handler) Process(ctx context.Context, outboundRay ray.OutboundRay, dial
Port: net.Port(server.Port),
}
}
newError("opening connection to ", destination).WriteToLog()
newError("opening connection to ", destination).WithContext(ctx).WriteToLog()
input := outboundRay.OutboundInput()
output := outboundRay.OutboundOutput()
@ -88,7 +88,7 @@ func (h *Handler) Process(ctx context.Context, outboundRay ray.OutboundRay, dial
Address: ip,
Port: destination.Port,
}
newError("changing destination to ", destination).WriteToLog()
newError("changing destination to ", destination).WithContext(ctx).WriteToLog()
}
}

4
proxy/http/server.go
View File

@ -121,7 +121,7 @@ Start:
}
}
newError("request to Method [", request.Method, "] Host [", request.Host, "] with URL [", request.URL, "]").WriteToLog()
newError("request to Method [", request.Method, "] Host [", request.Host, "] with URL [", request.URL, "]").WithContext(ctx).WriteToLog()
conn.SetReadDeadline(time.Time{})
defaultPort := net.Port(80)
@ -276,7 +276,7 @@ func (s *Server) handlePlainHTTP(ctx context.Context, request *http.Request, wri
result = nil
}
} else {
newError("failed to read response from ", request.Host).Base(err).AtWarning().WriteToLog()
newError("failed to read response from ", request.Host).Base(err).AtWarning().WithContext(ctx).WriteToLog()
response = &http.Response{
Status: "Service Unavailable",
StatusCode: 503,

12
proxy/shadowsocks/server.go
View File

@ -80,7 +80,7 @@ func (s *Server) handlerUDPPayload(ctx context.Context, conn internet.Connection
request, data, err := DecodeUDPPacket(s.user, payload)
if err != nil {
if source, ok := proxy.SourceFromContext(ctx); ok {
newError("dropping invalid UDP packet from: ", source).Base(err).WriteToLog()
newError("dropping invalid UDP packet from: ", source).Base(err).WithContext(ctx).WriteToLog()
log.Record(&log.AccessMessage{
From: source,
To: "",
@ -93,13 +93,13 @@ func (s *Server) handlerUDPPayload(ctx context.Context, conn internet.Connection
}
if request.Option.Has(RequestOptionOneTimeAuth) && s.account.OneTimeAuth == Account_Disabled {
newError("client payload enables OTA but server doesn't allow it").WriteToLog()
newError("client payload enables OTA but server doesn't allow it").WithContext(ctx).WriteToLog()
payload.Release()
continue
}
if !request.Option.Has(RequestOptionOneTimeAuth) && s.account.OneTimeAuth == Account_Enabled {
newError("client payload disables OTA but server forces it").WriteToLog()
newError("client payload disables OTA but server forces it").WithContext(ctx).WriteToLog()
payload.Release()
continue
}
@ -113,7 +113,7 @@ func (s *Server) handlerUDPPayload(ctx context.Context, conn internet.Connection
Reason: "",
})
}
newError("tunnelling request to ", dest).WriteToLog()
newError("tunnelling request to ", dest).WithContext(ctx).WriteToLog()
ctx = protocol.ContextWithUser(ctx, request.User)
udpServer.Dispatch(ctx, dest, data, func(payload *buf.Buffer) {
@ -121,7 +121,7 @@ func (s *Server) handlerUDPPayload(ctx context.Context, conn internet.Connection
data, err := EncodeUDPPacket(request, payload.Bytes())
if err != nil {
newError("failed to encode UDP packet").Base(err).AtWarning().WriteToLog()
newError("failed to encode UDP packet").Base(err).AtWarning().WithContext(ctx).WriteToLog()
return
}
defer data.Release()
@ -159,7 +159,7 @@ func (s *Server) handleConnection(ctx context.Context, conn internet.Connection,
Status: log.AccessAccepted,
Reason: "",
})
newError("tunnelling request to ", dest).WriteToLog()
newError("tunnelling request to ", dest).WithContext(ctx).WriteToLog()
ctx = protocol.ContextWithUser(ctx, request.User)

4
proxy/socks/client.go
View File

@ -86,7 +86,7 @@ func (c *Client) Process(ctx context.Context, ray ray.OutboundRay, dialer proxy.
}
if err := conn.SetDeadline(time.Now().Add(p.Timeouts.Handshake)); err != nil {
newError("failed to set deadline for handshake").Base(err).WriteToLog()
newError("failed to set deadline for handshake").Base(err).WithContext(ctx).WriteToLog()
}
udpRequest, err := ClientHandshake(request, conn, conn)
if err != nil {
@ -94,7 +94,7 @@ func (c *Client) Process(ctx context.Context, ray ray.OutboundRay, dialer proxy.
}
if err := conn.SetDeadline(time.Time{}); err != nil {
newError("failed to clear deadline after handshake").Base(err).WriteToLog()
newError("failed to clear deadline after handshake").Base(err).WithContext(ctx).WriteToLog()
}
ctx, cancel := context.WithCancel(ctx)

12
proxy/socks/server.go
View File

@ -91,7 +91,7 @@ func (s *Server) processTCP(ctx context.Context, conn internet.Connection, dispa
if request.Command == protocol.RequestCommandTCP {
dest := request.Destination()
newError("TCP Connect request to ", dest).WriteToLog()
newError("TCP Connect request to ", dest).WithContext(ctx).WriteToLog()
if source, ok := proxy.SourceFromContext(ctx); ok {
log.Record(&log.AccessMessage{
From: source,
@ -163,7 +163,7 @@ func (v *Server) handleUDPPayload(ctx context.Context, conn internet.Connection,
udpServer := udp.NewDispatcher(dispatcher)
if source, ok := proxy.SourceFromContext(ctx); ok {
newError("client UDP connection from ", source).WriteToLog()
newError("client UDP connection from ", source).WithContext(ctx).WriteToLog()
}
reader := buf.NewReader(conn)
@ -177,7 +177,7 @@ func (v *Server) handleUDPPayload(ctx context.Context, conn internet.Connection,
request, data, err := DecodeUDPPacket(payload.Bytes())
if err != nil {
newError("failed to parse UDP request").Base(err).WriteToLog()
newError("failed to parse UDP request").Base(err).WithContext(ctx).WriteToLog()
continue
}
@ -185,7 +185,7 @@ func (v *Server) handleUDPPayload(ctx context.Context, conn internet.Connection,
continue
}
newError("send packet to ", request.Destination(), " with ", len(data), " bytes").AtDebug().WriteToLog()
newError("send packet to ", request.Destination(), " with ", len(data), " bytes").AtDebug().WithContext(ctx).WriteToLog()
if source, ok := proxy.SourceFromContext(ctx); ok {
log.Record(&log.AccessMessage{
From: source,
@ -200,12 +200,12 @@ func (v *Server) handleUDPPayload(ctx context.Context, conn internet.Connection,
udpServer.Dispatch(ctx, request.Destination(), dataBuf, func(payload *buf.Buffer) {
defer payload.Release()
newError("writing back UDP response with ", payload.Len(), " bytes").AtDebug().WriteToLog()
newError("writing back UDP response with ", payload.Len(), " bytes").AtDebug().WithContext(ctx).WriteToLog()
udpMessage, err := EncodeUDPPacket(request, payload.Bytes())
defer udpMessage.Release()
if err != nil {
newError("failed to write UDP response").AtWarning().Base(err).WriteToLog()
newError("failed to write UDP response").AtWarning().Base(err).WithContext(ctx).WriteToLog()
}
conn.Write(udpMessage.Bytes())

9
proxy/vmess/encoding/client.go
View File

@ -63,8 +63,7 @@ func (c *ClientSession) EncodeRequestHeader(header *protocol.RequestHeader, writ
timestamp := protocol.NewTimestampGenerator(protocol.NowTime(), 30)()
account, err := header.User.GetTypedAccount()
if err != nil {
newError("failed to get user account: ", err).AtError().WriteToLog()
return nil
return newError("failed to get user account: ", err).AtError()
}
idHash := c.idHash(account.(*vmess.InternalAccount).AnyValidID().Bytes())
common.Must2(idHash.Write(timestamp.Bytes(nil)))
@ -200,8 +199,7 @@ func (c *ClientSession) DecodeResponseHeader(reader io.Reader) (*protocol.Respon
defer buffer.Release()
if err := buffer.AppendSupplier(buf.ReadFullFrom(c.responseReader, 4)); err != nil {
newError("failed to read response header").Base(err).WriteToLog()
return nil, err
return nil, newError("failed to read response header").Base(err)
}
if buffer.Byte(0) != c.responseHeader {
@ -217,8 +215,7 @@ func (c *ClientSession) DecodeResponseHeader(reader io.Reader) (*protocol.Respon
dataLen := int(buffer.Byte(3))
if err := buffer.Reset(buf.ReadFullFrom(c.responseReader, dataLen)); err != nil {
newError("failed to read response command").Base(err).WriteToLog()
return nil, err
return nil, newError("failed to read response command").Base(err)
}
command, err := UnmarshalCommand(cmdID, buffer.Bytes())
if err == nil {

8
proxy/vmess/inbound/inbound.go
View File

@ -242,10 +242,10 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection i
Reason: "",
})
newError("received request for ", request.Destination()).WriteToLog()
newError("received request for ", request.Destination()).WithContext(ctx).WriteToLog()
if err := connection.SetReadDeadline(time.Time{}); err != nil {
newError("unable to set back read deadline").Base(err).WriteToLog()
newError("unable to set back read deadline").Base(err).WithContext(ctx).WriteToLog()
}
sessionPolicy = h.policyManager.ForLevel(request.User.Level)
@ -292,7 +292,7 @@ func (h *Handler) generateCommand(ctx context.Context, request *protocol.Request
if h.inboundHandlerManager != nil {
handler, err := h.inboundHandlerManager.GetHandler(ctx, tag)
if err != nil {
newError("failed to get detour handler: ", tag).Base(err).AtWarning().WriteToLog()
newError("failed to get detour handler: ", tag).Base(err).AtWarning().WithContext(ctx).WriteToLog()
return nil
}
proxyHandler, port, availableMin := handler.GetRandomInboundProxy()
@ -302,7 +302,7 @@ func (h *Handler) generateCommand(ctx context.Context, request *protocol.Request
availableMin = 255
}
newError("pick detour handler for port ", port, " for ", availableMin, " minutes.").AtDebug().WriteToLog()
newError("pick detour handler for port ", port, " for ", availableMin, " minutes.").AtDebug().WithContext(ctx).WriteToLog()
user := inboundHandler.GetUser(request.User.Email)
if user == nil {
return nil

2
proxy/vmess/outbound/outbound.go
View File

@ -65,7 +65,7 @@ func (v *Handler) Process(ctx context.Context, outboundRay ray.OutboundRay, dial
if !ok {
return newError("target not specified").AtError()
}
newError("tunneling request to ", target, " via ", rec.Destination()).WriteToLog()
newError("tunneling request to ", target, " via ", rec.Destination()).WithContext(ctx).WriteToLog()
command := protocol.RequestCommandTCP
if target.Network == net.Network_UDP {

2
transport/internet/tcp/dialer.go
View File

@ -19,7 +19,7 @@ func getTCPSettingsFromContext(ctx context.Context) *Config {
// Dial dials a new TCP connection to the given destination.
func Dial(ctx context.Context, dest net.Destination) (internet.Connection, error) {
newError("dialing TCP to ", dest).WriteToLog()
newError("dialing TCP to ", dest).WithContext(ctx).WriteToLog()
src := internet.DialerSourceFromContext(ctx)
conn, err := internet.DialSystem(ctx, src, dest)

2
transport/internet/tcp/hub.go
View File

@ -29,7 +29,7 @@ func ListenTCP(ctx context.Context, address net.Address, port net.Port, handler
if err != nil {
return nil, err
}
newError("listening TCP on ", address, ":", port).WriteToLog()
newError("listening TCP on ", address, ":", port).WithContext(ctx).WriteToLog()
networkSettings := internet.TransportSettingsFromContext(ctx)
tcpSettings := networkSettings.(*Config)

6
transport/internet/udp/dispatcher.go
View File

@ -72,13 +72,13 @@ func (v *Dispatcher) getInboundRay(dest net.Destination, callback ResponseCallba
func (v *Dispatcher) Dispatch(ctx context.Context, destination net.Destination, payload *buf.Buffer, callback ResponseCallback) {
// TODO: Add user to destString
newError("dispatch request to: ", destination).AtDebug().WriteToLog()
newError("dispatch request to: ", destination).AtDebug().WithContext(ctx).WriteToLog()
conn := v.getInboundRay(destination, callback)
outputStream := conn.inbound.InboundInput()
if outputStream != nil {
if err := outputStream.WriteMultiBuffer(buf.NewMultiBufferValue(payload)); err != nil {
newError("failed to write first UDP payload").Base(err).WriteToLog()
newError("failed to write first UDP payload").Base(err).WithContext(ctx).WriteToLog()
conn.cancel()
return
}
@ -98,7 +98,7 @@ func handleInput(ctx context.Context, conn *connEntry, callback ResponseCallback
mb, err := input.ReadMultiBuffer()
if err != nil {
newError("failed to handle UDP input").Base(err).WriteToLog()
newError("failed to handle UDP input").Base(err).WithContext(ctx).WriteToLog()
conn.cancel()
return
}

2
transport/internet/websocket/dialer.go
View File

@ -13,7 +13,7 @@ import (
// Dial dials a WebSocket connection to the given destination.
func Dial(ctx context.Context, dest net.Destination) (internet.Connection, error) {
newError("creating connection to ", dest).WriteToLog()
newError("creating connection to ", dest).WithContext(ctx).WriteToLog()
conn, err := dialWebsocket(ctx, dest)
if err != nil {