go-gost/x
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

use http.transport for non-connect http tunnel

Browse Source
This commit is contained in:
ginuerzh
2024-10-17 21:57:20 +08:00
parent 7e51404ae5
commit a0cbee8817
15 changed files with 213 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
handler/http/handler.go
+172 -85
View File
@@ -40,20 +40,37 @@ import (
stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
xrecorder "github.com/go-gost/x/recorder"
"github.com/go-gost/x/registry"
"golang.org/x/net/http/httpguts"
)
type recorderObjectCtxKey struct{}
var (
ctxKeyRecorderObject = &recorderObjectCtxKey{}
)
func contextWithRecorderObject(ctx context.Context, ro *xrecorder.HandlerRecorderObject) context.Context {
return context.WithValue(ctx, ctxKeyRecorderObject, ro)
}
func recorderObjectFromContext(ctx context.Context) *xrecorder.HandlerRecorderObject {
v, _ := ctx.Value(ctxKeyRecorderObject).(*xrecorder.HandlerRecorderObject)
return v
}
func init() {
registry.HandlerRegistry().Register("http", NewHandler)
}
type httpHandler struct {
md metadata
options handler.Options
stats *stats_util.HandlerStats
limiter traffic.TrafficLimiter
cancel context.CancelFunc
recorder recorder.RecorderObject
certPool tls_util.CertPool
md metadata
options handler.Options
stats *stats_util.HandlerStats
limiter traffic.TrafficLimiter
cancel context.CancelFunc
recorder recorder.RecorderObject
certPool tls_util.CertPool
transport http.RoundTripper
}
func NewHandler(opts ...handler.Option) handler.Handler {
@@ -95,6 +112,13 @@ func (h *httpHandler) Init(md md.Metadata) error {
h.certPool = tls_util.NewMemoryCertPool()
}
h.transport = &http.Transport{
DialContext: h.dial,
IdleConnTimeout: 30 * time.Second,
ResponseHeaderTimeout: h.md.readTimeout,
DisableKeepAlives: !h.md.keepalive,
}
return nil
}
@@ -153,7 +177,8 @@ func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler
return rate_limiter.ErrRateLimit
}
req, err := http.ReadRequest(bufio.NewReader(conn))
br := bufio.NewReader(conn)
req, err := http.ReadRequest(br)
if err != nil {
log.Error(err)
return err
@@ -164,6 +189,8 @@ func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler
ro.ClientIP = clientIP.String()
}
conn = xnet.NewReadWriteConn(br, conn, conn)
return h.handleRequest(ctx, conn, req, ro, log)
}
@@ -205,7 +232,8 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt
ro.Host = addr
fields := map[string]any{
"dst": addr,
"dst": addr,
"host": addr,
}
if u, _, _ := h.basicProxyAuth(req.Header.Get("Proxy-Authorization")); u != "" {
@@ -249,6 +277,18 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt
ro.HTTP.Response.Header = resp.Header
}()
if req.Method == "PRI" ||
(req.Method != http.MethodConnect && req.URL.Scheme != "http") {
resp.StatusCode = http.StatusBadRequest
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpResponse(resp, false)
log.Trace(string(dump))
}
return resp.Write(conn)
}
clientID, ok := h.authenticate(ctx, conn, req, resp, log)
if !ok {
return errors.New("authentication failed")
@@ -271,38 +311,6 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt
return h.handleUDP(ctx, conn, ro, log)
}
if req.Method == "PRI" ||
(req.Method != http.MethodConnect && req.URL.Scheme != "http") {
resp.StatusCode = http.StatusBadRequest
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpResponse(resp, false)
log.Trace(string(dump))
}
return resp.Write(conn)
}
switch h.md.hash {
case "host":
ctx = ctxvalue.ContextWithHash(ctx, &ctxvalue.Hash{Source: addr})
}
var buf bytes.Buffer
cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), network, addr)
ro.Route = buf.String()
if err != nil {
resp.StatusCode = http.StatusServiceUnavailable
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpResponse(resp, false)
log.Trace(string(dump))
}
resp.Write(conn)
return err
}
defer cc.Close()
{
rw := traffic_wrapper.WrapReadWriter(
h.limiter,
@@ -327,12 +335,25 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt
}
if req.Method != http.MethodConnect {
ro2 := &xrecorder.HandlerRecorderObject{}
*ro2 = *ro
ro.Time = time.Time{}
return h.handleProxy(ctx, conn, cc, req, ro2, log)
return h.handleProxy(ctx, conn, req, ro, log)
}
ctx = contextWithRecorderObject(ctx, ro)
ctx = ctxvalue.ContextWithLogger(ctx, log)
cc, err := h.dial(ctx, "tcp", addr)
if err != nil {
resp.StatusCode = http.StatusServiceUnavailable
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpResponse(resp, false)
log.Trace(string(dump))
}
resp.Write(conn)
return err
}
defer cc.Close()
resp.StatusCode = http.StatusOK
resp.Status = "200 Connection established"
@@ -404,13 +425,21 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt
return nil
}
func (h *httpHandler) handleProxy(ctx context.Context, rw io.ReadWriter, cc net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
if shouldClose, err := h.proxyRoundTrip(ctx, rw, cc, req, ro, log); err != nil || shouldClose {
func (h *httpHandler) handleProxy(ctx context.Context, conn net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
pStats := stats.Stats{}
conn = stats_wrapper.WrapConn(conn, &pStats)
ro.Time = time.Time{}
if err := h.proxyRoundTrip(ctx, conn, req, ro, &pStats, log); err != nil {
return err
}
br := bufio.NewReader(conn)
for {
req, err := http.ReadRequest(bufio.NewReader(rw))
pStats.Reset()
req, err := http.ReadRequest(br)
if err != nil {
if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
return nil
@@ -423,40 +452,47 @@ func (h *httpHandler) handleProxy(ctx context.Context, rw io.ReadWriter, cc net.
log.Trace(string(dump))
}
if shouldClose, err := h.proxyRoundTrip(ctx, rw, cc, req, ro, log); err != nil || shouldClose {
if err := h.proxyRoundTrip(ctx, conn, req, ro, &pStats, log); err != nil {
return err
}
}
}
func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (close bool, err error) {
close = true
func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats *stats.Stats, log logger.Logger) (err error) {
ro2 := &xrecorder.HandlerRecorderObject{}
*ro2 = *ro
ro = ro2
if req == nil {
return
host := req.Host
if _, port, _ := net.SplitHostPort(host); port == "" {
host = net.JoinHostPort(strings.Trim(host, "[]"), "80")
}
ro.Host = host
ro.Time = time.Now()
start := time.Now()
ro.Time = start
log = log.WithFields(map[string]any{
"host": host,
})
log.Infof("%s <-> %s", ro.RemoteAddr, req.Host)
defer func() {
ro.Duration = time.Since(start)
if err != nil {
ro.Err = err.Error()
}
ro.InputBytes = pStats.Get(stats.KindInputBytes)
ro.OutputBytes = pStats.Get(stats.KindOutputBytes)
ro.Duration = time.Since(ro.Time)
if err := ro.Record(ctx, h.recorder.Recorder); err != nil {
log.Errorf("record: %v", err)
}
log.WithFields(map[string]any{
"duration": time.Since(start),
"duration": time.Since(ro.Time),
"inputBytes": ro.InputBytes,
"outputBytes": ro.OutputBytes,
}).Infof("%s >-< %s", ro.RemoteAddr, req.Host)
}()
if clientIP := xhttp.GetClientIP(req); clientIP != nil {
ro.ClientIP = clientIP.String()
}
ro.HTTP = &xrecorder.HTTPRecorderObject{
Host: req.Host,
Proto: req.Proto,
@@ -468,6 +504,16 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n
Header: req.Header.Clone(),
},
}
if clientIP := xhttp.GetClientIP(req); clientIP != nil {
ro.ClientIP = clientIP.String()
}
clientAddr := ro.RemoteAddr
if ro.ClientIP != "" {
if _, port, _ := net.SplitHostPort(ro.RemoteAddr); port != "" {
clientAddr = net.JoinHostPort(ro.ClientIP, port)
}
}
// HTTP/1.0
if req.ProtoMajor == 1 && req.ProtoMinor == 0 {
@@ -478,15 +524,19 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n
}
}
if !h.md.keepalive {
req.Header.Set("Connection", "close")
}
req.Header.Del("Proxy-Authorization")
req.Header.Del("Proxy-Connection")
req.Header.Del("Gost-Target")
req.Header.Del("X-Gost-Target")
res := &http.Response{
ProtoMajor: req.ProtoMajor,
ProtoMinor: req.ProtoMinor,
Header: http.Header{},
StatusCode: http.StatusServiceUnavailable,
}
ro.HTTP.StatusCode = res.StatusCode
var reqBody *xhttp.Body
if opts := h.recorder.Options; opts != nil && opts.HTTPBody {
if req.Body != nil {
@@ -499,34 +549,22 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n
}
}
res := &http.Response{
ProtoMajor: req.ProtoMajor,
ProtoMinor: req.ProtoMinor,
Header: http.Header{},
StatusCode: http.StatusServiceUnavailable,
}
ro.HTTP.StatusCode = res.StatusCode
ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(clientAddr))
ctx = contextWithRecorderObject(ctx, ro)
ctx = ctxvalue.ContextWithLogger(ctx, log)
if err = req.Write(cc); err != nil {
resp, err := h.transport.RoundTrip(req.WithContext(ctx))
if err != nil {
res.Write(rw)
return
}
defer resp.Body.Close()
if reqBody != nil {
ro.HTTP.Request.Body = reqBody.Content()
ro.HTTP.Request.ContentLength = reqBody.Length()
}
cc.SetReadDeadline(time.Now().Add(h.md.readTimeout))
resp, err := http.ReadResponse(bufio.NewReader(cc), req)
if err != nil {
log.Errorf("read response: %v", err)
res.Write(rw)
return
}
defer resp.Body.Close()
cc.SetReadDeadline(time.Time{})
ro.HTTP.StatusCode = resp.StatusCode
ro.HTTP.Response.Header = resp.Header.Clone()
ro.HTTP.Response.ContentLength = resp.ContentLength
@@ -545,6 +583,10 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n
resp.ProtoMinor = req.ProtoMinor
}
if resp.StatusCode == http.StatusSwitchingProtocols {
return h.handleUpgradeResponse(rw, req, resp)
}
var respBody *xhttp.Body
if opts := h.recorder.Options; opts != nil && opts.HTTPBody {
maxSize := opts.MaxBodySize
@@ -556,8 +598,7 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n
}
if err = resp.Write(rw); err != nil {
log.Errorf("write response: %v", err)
return
return fmt.Errorf("write response: %v", err)
}
if respBody != nil {
@@ -565,7 +606,53 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n
ro.HTTP.Response.ContentLength = respBody.Length()
}
return resp.Close, nil
return
}
func (h *httpHandler) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) {
switch h.md.hash {
case "host":
ctx = ctxvalue.ContextWithHash(ctx, &ctxvalue.Hash{Source: addr})
}
if log := ctxvalue.LoggerFromContext(ctx); log != nil {
log.Debugf("dial: new connection to host %s", addr)
}
var buf bytes.Buffer
conn, err = h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), network, addr)
if ro := recorderObjectFromContext(ctx); ro != nil {
ro.Route = buf.String()
}
return
}
func upgradeType(h http.Header) string {
if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") {
return ""
}
return h.Get("Upgrade")
}
func (h *httpHandler) handleUpgradeResponse(rw io.ReadWriter, req *http.Request, res *http.Response) error {
reqUpType := upgradeType(req.Header)
resUpType := upgradeType(res.Header)
if !strings.EqualFold(reqUpType, resUpType) {
return fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType)
}
backConn, ok := res.Body.(io.ReadWriteCloser)
if !ok {
return fmt.Errorf("internal error: 101 switching protocols response with non-writable body")
}
res.Body = nil
if err := res.Write(rw); err != nil {
return fmt.Errorf("response write: %v", err)
}
return xnet.Transport(rw, backConn)
}
func (h *httpHandler) decodeServerName(s string) (string, error) {
handler/http/metadata.go
+4 -1
View File
@@ -53,7 +53,10 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error {
h.md.header = hd
}
h.md.keepalive = mdutil.GetBool(md, "http.keepalive", "keepalive")
h.md.keepalive = true
if mdutil.IsExists(md, "http.keepalive", "keepalive") {
h.md.keepalive = mdutil.GetBool(md, "http.keepalive", "keepalive")
}
if pr := mdutil.GetString(md, "probeResist", "probe_resist"); pr != "" {
if ss := strings.SplitN(pr, ":", 2); len(ss) == 2 {
handler/http2/handler.go
+2 -1
View File
@@ -178,7 +178,8 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
ro.Host = host
fields := map[string]any{
"dst": host,
"dst": host,
"host": host,
}
if u, _, _ := h.basicProxyAuth(req.Header.Get("Proxy-Authorization")); u != "" {
fields["user"] = u
handler/http3/handler.go
+2 -1
View File
@@ -124,7 +124,8 @@ func (h *http3Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
}
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", target.Addr, "tcp"),
"dst": fmt.Sprintf("%s/%s", target.Addr, "tcp"),
"host": target.Addr,
})
log.Debugf("%s >> %s", req.RemoteAddr, addr)
handler/redirect/tcp/handler.go
+1
View File
@@ -131,6 +131,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", dstAddr, dstAddr.Network()),
"host": dstAddr.String(),
})
if h.md.sniffing {
handler/redirect/udp/handler.go
+2 -1
View File
@@ -108,7 +108,8 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han
ro.Host = dstAddr.String()
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", dstAddr, dstAddr.Network()),
"dst": fmt.Sprintf("%s/%s", dstAddr, dstAddr.Network()),
"host": dstAddr.String(),
})
log.Debugf("%s >> %s", conn.RemoteAddr(), dstAddr)
handler/relay/connect.go
+3 -2
View File
@@ -33,8 +33,9 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network
}
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", address, network),
"cmd": "connect",
"dst": fmt.Sprintf("%s/%s", address, network),
"cmd": "connect",
"host": address,
})
log.Debugf("%s >> %s/%s", conn.RemoteAddr(), address, network)
handler/socks/v4/handler.go
+2 -1
View File
@@ -196,7 +196,8 @@ func (h *socks4Handler) handleConnect(ctx context.Context, conn net.Conn, req *g
addr := req.Addr.String()
log = log.WithFields(map[string]any{
"dst": addr,
"dst": addr,
"host": addr,
})
log.Debugf("%s >> %s", conn.RemoteAddr(), addr)
handler/socks/v5/connect.go
+3 -2
View File
@@ -23,8 +23,9 @@ import (
func (h *socks5Handler) handleConnect(ctx context.Context, conn net.Conn, network, address string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", address, network),
"cmd": "connect",
"dst": fmt.Sprintf("%s/%s", address, network),
"cmd": "connect",
"host": address,
})
log.Debugf("%s >> %s", conn.RemoteAddr(), address)
handler/ss/handler.go
+2 -1
View File
@@ -148,7 +148,8 @@ func (h *ssHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.H
conn.SetReadDeadline(time.Time{})
log = log.WithFields(map[string]any{
"dst": addr.String(),
"dst": addr.String(),
"host": addr.String(),
})
log.Debugf("%s >> %s", conn.RemoteAddr(), addr)
handler/sshd/handler.go
+3 -2
View File
@@ -140,8 +140,9 @@ func (h *forwardHandler) handleDirectForward(ctx context.Context, conn *sshd_uti
ro.Host = targetAddr
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", targetAddr, "tcp"),
"cmd": "connect",
"dst": fmt.Sprintf("%s/%s", targetAddr, "tcp"),
"cmd": "connect",
"host": targetAddr,
})
log.Debugf("%s >> %s", conn.RemoteAddr(), targetAddr)
handler/tunnel/connect.go
+1
View File
@@ -16,6 +16,7 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, c
"dst": fmt.Sprintf("%s/%s", dstAddr, network),
"cmd": "connect",
"tunnel": tunnelID.String(),
"host": dstAddr,
})
resp := relay.Response{
handler/tunnel/entrypoint.go
+11 -2
View File
@@ -180,7 +180,16 @@ func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *
*ro2 = *ro
ro = ro2
host := req.Host
if _, port, _ := net.SplitHostPort(host); port == "" {
host = net.JoinHostPort(strings.Trim(host, "[]"), "80")
}
ro.Host = host
ro.Time = time.Now()
log = log.WithFields(map[string]any{
"host": host,
})
log.Infof("%s <-> %s", ro.RemoteAddr, req.Host)
defer func() {
if err != nil {
@@ -344,6 +353,7 @@ func (h *entrypoint) dial(ctx context.Context, network, addr string) (conn net.C
if log == nil {
log = h.log
}
log.Debugf("dial: new connection to host %s", addr)
if tunnelID.IsZero() {
return nil, fmt.Errorf("%w %s", ErrTunnelRoute, addr)
@@ -358,7 +368,6 @@ func (h *entrypoint) dial(ctx context.Context, network, addr string) (conn net.C
}
log = log.WithFields(map[string]any{
"host": addr,
"tunnel": tunnelID.String(),
})
@@ -374,7 +383,7 @@ func (h *entrypoint) dial(ctx context.Context, network, addr string) (conn net.C
if err != nil {
return
}
log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid)
log.Debugf("dial: connected to host %s, tunnel: %s, connector: %s", addr, tunnelID, cid)
if node == h.node {
clientAddr := ctxvalue.ClientAddrFromContext(ctx)
handler/unix/handler.go
+1
View File
@@ -126,6 +126,7 @@ func (h *unixHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler
log = log.WithFields(map[string]any{
"node": target.Name,
"dst": target.Addr,
"host": target.Addr,
})
ro.Host = target.Addr
metadata/util/util.go
+4
View File
@@ -9,6 +9,10 @@ import (
)
func IsExists(md metadata.Metadata, keys ...string) bool {
if md == nil {
return false
}
for _, key := range keys {
if md.IsExists(key) {
return true