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

add http3 reverse proxy

Browse Source
This commit is contained in:
ginuerzh 2022-11-12 17:14:11 +08:00
parent f87183b652
commit c1b2d3b086
13 changed files with 542 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
handler/http2/conn.go
View File

@ -6,19 +6,6 @@ import (
"net/http"
)
type readWriter struct {
r io.Reader
w io.Writer
}
func (rw *readWriter) Read(p []byte) (n int, err error) {
return rw.r.Read(p)
}
func (rw *readWriter) Write(p []byte) (n int, err error) {
return rw.w.Write(p)
}
type flushWriter struct {
w io.Writer
}

3
handler/http2/handler.go
View File

@ -22,6 +22,7 @@ import (
"github.com/go-gost/core/handler"
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
xio "github.com/go-gost/x/internal/io"
netpkg "github.com/go-gost/x/internal/net"
sx "github.com/go-gost/x/internal/util/selector"
"github.com/go-gost/x/registry"
@ -200,7 +201,7 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
start := time.Now()
log.Debugf("%s <-> %s", req.RemoteAddr, addr)
netpkg.Transport(&readWriter{r: req.Body, w: flushWriter{w}}, cc)
netpkg.Transport(xio.NewReadWriter(req.Body, flushWriter{w}), cc)
log.WithFields(map[string]any{
"duration": time.Since(start),
}).Debugf("%s >-< %s", req.RemoteAddr, addr)

178
handler/http3/handler.go Normal file
View File

@ -0,0 +1,178 @@
package http3
import (
"context"
"errors"
"fmt"
"net"
"net/http"
"net/http/httputil"
"time"
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
sx "github.com/go-gost/x/internal/util/selector"
"github.com/go-gost/x/registry"
)
func init() {
registry.HandlerRegistry().Register("http3", NewHandler)
}
type http3Handler struct {
hop chain.Hop
router *chain.Router
md metadata
options handler.Options
}
func NewHandler(opts ...handler.Option) handler.Handler {
options := handler.Options{}
for _, opt := range opts {
opt(&options)
}
return &http3Handler{
options: options,
}
}
func (h *http3Handler) Init(md md.Metadata) error {
if err := h.parseMetadata(md); err != nil {
return err
}
h.router = h.options.Router
if h.router == nil {
h.router = chain.NewRouter(chain.LoggerRouterOption(h.options.Logger))
}
return nil
}
// Forward implements handler.Forwarder.
func (h *http3Handler) Forward(hop chain.Hop) {
h.hop = hop
}
func (h *http3Handler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error {
defer conn.Close()
start := time.Now()
log := h.options.Logger.WithFields(map[string]any{
"remote": conn.RemoteAddr().String(),
"local": conn.LocalAddr().String(),
})
log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
defer func() {
log.WithFields(map[string]any{
"duration": time.Since(start),
}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
}()
if !h.checkRateLimit(conn.RemoteAddr()) {
return nil
}
v, ok := conn.(md.Metadatable)
if !ok || v == nil {
err := errors.New("wrong connection type")
log.Error(err)
return err
}
md := v.Metadata()
return h.roundTrip(ctx,
md.Get("w").(http.ResponseWriter),
md.Get("r").(*http.Request),
log,
)
}
func (h *http3Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req *http.Request, log logger.Logger) error {
addr := req.Host
if _, port, _ := net.SplitHostPort(addr); port == "" {
addr = net.JoinHostPort(addr, "80")
}
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpRequest(req, false)
log.Trace(string(dump))
}
for k := range h.md.header {
w.Header().Set(k, h.md.header.Get(k))
}
if h.options.Bypass != nil && h.options.Bypass.Contains(addr) {
w.WriteHeader(http.StatusForbidden)
log.Debug("bypass: ", addr)
return nil
}
switch h.md.hash {
case "host":
ctx = sx.ContextWithHash(ctx, &sx.Hash{Source: addr})
}
var target *chain.Node
if h.hop != nil {
target = h.hop.Select(ctx, chain.HostSelectOption(addr))
}
if target == nil {
err := errors.New("target not available")
log.Error(err)
return err
}
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", target.Addr, "tcp"),
})
log.Debugf("%s >> %s", req.RemoteAddr, addr)
rp := &httputil.ReverseProxy{
Director: func(r *http.Request) {
r.URL.Scheme = "http"
r.URL.Host = req.Host
dump, _ := httputil.DumpRequest(r, false)
log.Debug(string(dump))
},
Transport: &http.Transport{
ForceAttemptHTTP2: true,
MaxIdleConns: 100,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
conn, err := h.router.Dial(ctx, network, target.Addr)
if err != nil {
log.Error(err)
// TODO: the router itself may be failed due to the failed node in the router,
// the dead marker may be a wrong operation.
if marker := target.Marker(); marker != nil {
marker.Mark()
}
}
return conn, err
},
},
}
rp.ServeHTTP(w, req)
return nil
}
func (h *http3Handler) checkRateLimit(addr net.Addr) bool {
if h.options.RateLimiter == nil {
return true
}
host, _, _ := net.SplitHostPort(addr.String())
if limiter := h.options.RateLimiter.Limiter(host); limiter != nil {
return limiter.Allow(1)
}
return true
}

56
handler/http3/metadata.go Normal file
View File

@ -0,0 +1,56 @@
package http3
import (
"net/http"
"strings"
mdata "github.com/go-gost/core/metadata"
mdutil "github.com/go-gost/core/metadata/util"
)
type metadata struct {
probeResistance *probeResistance
header http.Header
hash string
}
func (h *http3Handler) parseMetadata(md mdata.Metadata) error {
const (
header = "header"
probeResistKey = "probeResistance"
probeResistKeyX = "probe_resist"
knock = "knock"
hash = "hash"
)
if m := mdutil.GetStringMapString(md, header); len(m) > 0 {
hd := http.Header{}
for k, v := range m {
hd.Add(k, v)
}
h.md.header = hd
}
pr := mdutil.GetString(md, probeResistKey)
if pr == "" {
pr = mdutil.GetString(md, probeResistKeyX)
}
if pr != "" {
if ss := strings.SplitN(pr, ":", 2); len(ss) == 2 {
h.md.probeResistance = &probeResistance{
Type: ss[0],
Value: ss[1],
Knock: mdutil.GetString(md, knock),
}
}
}
h.md.hash = mdutil.GetString(md, hash)
return nil
}
type probeResistance struct {
Type string
Value string
Knock string
}

8
handler/redirect/tcp/conn.go
View File

@ -1,8 +0,0 @@
package redirect
import "io"
type readWriter struct {
io.Reader
io.Writer
}

16
handler/redirect/tcp/handler.go
View File

@ -18,6 +18,7 @@ import (
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
dissector "github.com/go-gost/tls-dissector"
xio "github.com/go-gost/x/internal/io"
netpkg "github.com/go-gost/x/internal/net"
"github.com/go-gost/x/registry"
)
@ -99,10 +100,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han
// try to sniff TLS traffic
var hdr [dissector.RecordHeaderLen]byte
_, err := io.ReadFull(rw, hdr[:])
rw = &readWriter{
Reader: io.MultiReader(bytes.NewReader(hdr[:]), rw),
Writer: rw,
}
rw = xio.NewReadWriter(io.MultiReader(bytes.NewReader(hdr[:]), rw), rw)
if err == nil &&
hdr[0] == dissector.Handshake &&
binary.BigEndian.Uint16(hdr[1:3]) == tls.VersionTLS10 {
@ -112,10 +110,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han
// try to sniff HTTP traffic
buf := new(bytes.Buffer)
_, err = http.ReadRequest(bufio.NewReader(io.TeeReader(rw, buf)))
rw = &readWriter{
Reader: io.MultiReader(buf, rw),
Writer: rw,
}
rw = xio.NewReadWriter(io.MultiReader(buf, rw), rw)
if err == nil {
return h.handleHTTP(ctx, rw, conn.RemoteAddr(), log)
}
@ -241,10 +236,7 @@ func (h *redirectHandler) handleHTTPS(ctx context.Context, rw io.ReadWriter, rad
t := time.Now()
log.Debugf("%s <-> %s", raddr, host)
netpkg.Transport(&readWriter{
Reader: io.MultiReader(buf, rw),
Writer: rw,
}, cc)
netpkg.Transport(xio.NewReadWriter(io.MultiReader(buf, rw), rw), cc)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Debugf("%s >-< %s", raddr, host)

10
handler/sni/conn.go
View File

@ -1,10 +0,0 @@
package sni
import (
"io"
)
type readWriter struct {
io.Reader
io.Writer
}

12
handler/sni/handler.go
View File

@ -20,6 +20,7 @@ import (
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
dissector "github.com/go-gost/tls-dissector"
xio "github.com/go-gost/x/internal/io"
netpkg "github.com/go-gost/x/internal/net"
sx "github.com/go-gost/x/internal/util/selector"
"github.com/go-gost/x/registry"
@ -87,10 +88,7 @@ func (h *sniHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
return err
}
rw := &readWriter{
Reader: io.MultiReader(bytes.NewReader(hdr[:]), conn),
Writer: conn,
}
rw := xio.NewReadWriter(io.MultiReader(bytes.NewReader(hdr[:]), conn), conn)
if hdr[0] == dissector.Handshake &&
binary.BigEndian.Uint16(hdr[1:3]) == tls.VersionTLS10 {
return h.handleHTTPS(ctx, rw, conn.RemoteAddr(), log)
@ -199,16 +197,12 @@ func (h *sniHandler) handleHTTPS(ctx context.Context, rw io.ReadWriter, raddr ne
t := time.Now()
log.Debugf("%s <-> %s", raddr, host)
netpkg.Transport(&readWriter{
Reader: io.MultiReader(buf, rw),
Writer: rw,
}, cc)
netpkg.Transport(xio.NewReadWriter(io.MultiReader(buf, rw), rw), cc)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Debugf("%s >-< %s", raddr, host)
return nil
}
func (h *sniHandler) decodeHost(r io.Reader) (host string, err error) {

66
listener/http3/conn.go Normal file
View File

@ -0,0 +1,66 @@
package http3
import (
"errors"
"net"
"net/http"
"time"
mdata "github.com/go-gost/core/metadata"
)
// a dummy HTTP3 server conn used by HTTP3 handler
type conn struct {
md mdata.Metadata
r *http.Request
w http.ResponseWriter
laddr net.Addr
raddr net.Addr
closed chan struct{}
}
func (c *conn) Read(b []byte) (n int, err error) {
return 0, &net.OpError{Op: "read", Net: "http3", Source: nil, Addr: nil, Err: errors.New("read not supported")}
}
func (c *conn) Write(b []byte) (n int, err error) {
return 0, &net.OpError{Op: "write", Net: "http3", Source: nil, Addr: nil, Err: errors.New("write not supported")}
}
func (c *conn) Close() error {
select {
case <-c.closed:
default:
close(c.closed)
}
return nil
}
func (c *conn) LocalAddr() net.Addr {
return c.laddr
}
func (c *conn) RemoteAddr() net.Addr {
return c.raddr
}
func (c *conn) SetDeadline(t time.Time) error {
return &net.OpError{Op: "set", Net: "http3", Source: nil, Addr: nil, Err: errors.New("deadline not supported")}
}
func (c *conn) SetReadDeadline(t time.Time) error {
return &net.OpError{Op: "set", Net: "http3", Source: nil, Addr: nil, Err: errors.New("deadline not supported")}
}
func (c *conn) SetWriteDeadline(t time.Time) error {
return &net.OpError{Op: "set", Net: "http3", Source: nil, Addr: nil, Err: errors.New("deadline not supported")}
}
func (c *conn) Done() <-chan struct{} {
return c.closed
}
// Metadata implements metadata.Metadatable interface.
func (c *conn) Metadata() mdata.Metadata {
return c.md
}

99
listener/http3/h3/listener.go Normal file
View File

@ -0,0 +1,99 @@
package h3
import (
"net"
"github.com/go-gost/core/listener"
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
admission "github.com/go-gost/x/admission/wrapper"
xnet "github.com/go-gost/x/internal/net"
pht_util "github.com/go-gost/x/internal/util/pht"
limiter "github.com/go-gost/x/limiter/traffic/wrapper"
metrics "github.com/go-gost/x/metrics/wrapper"
"github.com/go-gost/x/registry"
"github.com/lucas-clemente/quic-go"
)
func init() {
registry.ListenerRegistry().Register("h3", NewListener)
}
type http3Listener struct {
addr net.Addr
server *pht_util.Server
logger logger.Logger
md metadata
options listener.Options
}
func NewListener(opts ...listener.Option) listener.Listener {
options := listener.Options{}
for _, opt := range opts {
opt(&options)
}
return &http3Listener{
logger: options.Logger,
options: options,
}
}
func (l *http3Listener) Init(md md.Metadata) (err error) {
if err = l.parseMetadata(md); err != nil {
return
}
network := "udp"
if xnet.IsIPv4(l.options.Addr) {
network = "udp4"
}
l.addr, err = net.ResolveUDPAddr(network, l.options.Addr)
if err != nil {
return
}
l.server = pht_util.NewHTTP3Server(
l.options.Addr,
&quic.Config{
KeepAlivePeriod: l.md.keepAlivePeriod,
HandshakeIdleTimeout: l.md.handshakeTimeout,
MaxIdleTimeout: l.md.maxIdleTimeout,
Versions: []quic.VersionNumber{
quic.Version1,
},
MaxIncomingStreams: int64(l.md.maxStreams),
},
pht_util.TLSConfigServerOption(l.options.TLSConfig),
pht_util.BacklogServerOption(l.md.backlog),
pht_util.PathServerOption(l.md.authorizePath, l.md.pushPath, l.md.pullPath),
pht_util.LoggerServerOption(l.options.Logger),
)
go func() {
if err := l.server.ListenAndServe(); err != nil {
l.logger.Error(err)
}
}()
return
}
func (l *http3Listener) Accept() (conn net.Conn, err error) {
conn, err = l.server.Accept()
if err != nil {
return
}
conn = metrics.WrapConn(l.options.Service, conn)
conn = admission.WrapConn(l.options.Admission, conn)
conn = limiter.WrapConn(l.options.TrafficLimiter, conn)
return conn, nil
}
func (l *http3Listener) Addr() net.Addr {
return l.addr
}
func (l *http3Listener) Close() (err error) {
return l.server.Close()
}

75
listener/http3/h3/metadata.go Normal file
View File

@ -0,0 +1,75 @@
package h3
import (
"strings"
"time"
mdata "github.com/go-gost/core/metadata"
mdutil "github.com/go-gost/core/metadata/util"
)
const (
defaultAuthorizePath = "/authorize"
defaultPushPath = "/push"
defaultPullPath = "/pull"
defaultBacklog = 128
)
type metadata struct {
authorizePath string
pushPath string
pullPath string
backlog int
// QUIC config options
keepAlivePeriod time.Duration
maxIdleTimeout time.Duration
handshakeTimeout time.Duration
maxStreams int
}
func (l *http3Listener) parseMetadata(md mdata.Metadata) (err error) {
const (
authorizePath = "authorizePath"
pushPath = "pushPath"
pullPath = "pullPath"
keepAlive = "keepAlive"
keepAlivePeriod = "ttl"
handshakeTimeout = "handshakeTimeout"
maxIdleTimeout = "maxIdleTimeout"
maxStreams = "maxStreams"
backlog = "backlog"
)
l.md.authorizePath = mdutil.GetString(md, authorizePath)
if !strings.HasPrefix(l.md.authorizePath, "/") {
l.md.authorizePath = defaultAuthorizePath
}
l.md.pushPath = mdutil.GetString(md, pushPath)
if !strings.HasPrefix(l.md.pushPath, "/") {
l.md.pushPath = defaultPushPath
}
l.md.pullPath = mdutil.GetString(md, pullPath)
if !strings.HasPrefix(l.md.pullPath, "/") {
l.md.pullPath = defaultPullPath
}
l.md.backlog = mdutil.GetInt(md, backlog)
if l.md.backlog <= 0 {
l.md.backlog = defaultBacklog
}
if mdutil.GetBool(md, keepAlive) {
l.md.keepAlivePeriod = mdutil.GetDuration(md, keepAlivePeriod)
if l.md.keepAlivePeriod <= 0 {
l.md.keepAlivePeriod = 10 * time.Second
}
}
l.md.handshakeTimeout = mdutil.GetDuration(md, handshakeTimeout)
l.md.maxIdleTimeout = mdutil.GetDuration(md, maxIdleTimeout)
l.md.maxStreams = mdutil.GetInt(md, maxStreams)
return
}

80
listener/http3/listener.go
View File

@ -2,30 +2,32 @@ package http3
import (
"net"
"net/http"
"sync"
"github.com/go-gost/core/listener"
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
admission "github.com/go-gost/x/admission/wrapper"
xnet "github.com/go-gost/x/internal/net"
pht_util "github.com/go-gost/x/internal/util/pht"
limiter "github.com/go-gost/x/limiter/traffic/wrapper"
metrics "github.com/go-gost/x/metrics/wrapper"
mdx "github.com/go-gost/x/metadata"
"github.com/go-gost/x/registry"
"github.com/lucas-clemente/quic-go"
"github.com/lucas-clemente/quic-go/http3"
)
func init() {
registry.ListenerRegistry().Register("http3", NewListener)
registry.ListenerRegistry().Register("h3", NewListener)
}
type http3Listener struct {
server *http3.Server
addr net.Addr
server *pht_util.Server
cqueue chan net.Conn
errChan chan error
logger logger.Logger
md metadata
options listener.Options
mu sync.Mutex
}
func NewListener(opts ...listener.Option) listener.Listener {
@ -53,9 +55,10 @@ func (l *http3Listener) Init(md md.Metadata) (err error) {
return
}
l.server = pht_util.NewHTTP3Server(
l.options.Addr,
&quic.Config{
l.server = &http3.Server{
Addr: l.options.Addr,
TLSConfig: l.options.TLSConfig,
QuicConfig: &quic.Config{
KeepAlivePeriod: l.md.keepAlivePeriod,
HandshakeIdleTimeout: l.md.handshakeTimeout,
MaxIdleTimeout: l.md.maxIdleTimeout,
@ -64,11 +67,11 @@ func (l *http3Listener) Init(md md.Metadata) (err error) {
},
MaxIncomingStreams: int64(l.md.maxStreams),
},
pht_util.TLSConfigServerOption(l.options.TLSConfig),
pht_util.BacklogServerOption(l.md.backlog),
pht_util.PathServerOption(l.md.authorizePath, l.md.pushPath, l.md.pullPath),
pht_util.LoggerServerOption(l.options.Logger),
)
Handler: http.HandlerFunc(l.handleFunc),
}
l.cqueue = make(chan net.Conn, l.md.backlog)
l.errChan = make(chan error, 1)
go func() {
if err := l.server.ListenAndServe(); err != nil {
@ -80,15 +83,15 @@ func (l *http3Listener) Init(md md.Metadata) (err error) {
}
func (l *http3Listener) Accept() (conn net.Conn, err error) {
conn, err = l.server.Accept()
if err != nil {
return
var ok bool
select {
case conn = <-l.cqueue:
case err, ok = <-l.errChan:
if !ok {
err = listener.ErrClosed
}
conn = metrics.WrapConn(l.options.Service, conn)
conn = admission.WrapConn(l.options.Admission, conn)
conn = limiter.WrapConn(l.options.TrafficLimiter, conn)
return conn, nil
}
return
}
func (l *http3Listener) Addr() net.Addr {
@ -96,5 +99,36 @@ func (l *http3Listener) Addr() net.Addr {
}
func (l *http3Listener) Close() (err error) {
return l.server.Close()
l.mu.Lock()
defer l.mu.Unlock()
select {
case <-l.errChan:
default:
err = l.server.Close()
l.errChan <- err
close(l.errChan)
}
return nil
}
func (l *http3Listener) handleFunc(w http.ResponseWriter, r *http.Request) {
raddr, _ := net.ResolveTCPAddr("tcp", r.RemoteAddr)
conn := &conn{
laddr: l.addr,
raddr: raddr,
closed: make(chan struct{}),
md: mdx.NewMetadata(map[string]any{
"r": r,
"w": w,
}),
}
select {
case l.cqueue <- conn:
default:
l.logger.Warnf("connection queue is full, client %s discarded", r.RemoteAddr)
return
}
<-conn.Done()
}

24
listener/http3/metadata.go
View File

@ -1,7 +1,6 @@
package http3
import (
"strings"
"time"
mdata "github.com/go-gost/core/metadata"
@ -9,16 +8,10 @@ import (
)
const (
defaultAuthorizePath = "/authorize"
defaultPushPath = "/push"
defaultPullPath = "/pull"
defaultBacklog = 128
)
type metadata struct {
authorizePath string
pushPath string
pullPath string
backlog int
// QUIC config options
@ -30,10 +23,6 @@ type metadata struct {
func (l *http3Listener) parseMetadata(md mdata.Metadata) (err error) {
const (
authorizePath = "authorizePath"
pushPath = "pushPath"
pullPath = "pullPath"
keepAlive = "keepAlive"
keepAlivePeriod = "ttl"
handshakeTimeout = "handshakeTimeout"
@ -43,19 +32,6 @@ func (l *http3Listener) parseMetadata(md mdata.Metadata) (err error) {
backlog = "backlog"
)
l.md.authorizePath = mdutil.GetString(md, authorizePath)
if !strings.HasPrefix(l.md.authorizePath, "/") {
l.md.authorizePath = defaultAuthorizePath
}
l.md.pushPath = mdutil.GetString(md, pushPath)
if !strings.HasPrefix(l.md.pushPath, "/") {
l.md.pushPath = defaultPushPath
}
l.md.pullPath = mdutil.GetString(md, pullPath)
if !strings.HasPrefix(l.md.pullPath, "/") {
l.md.pullPath = defaultPullPath
}
l.md.backlog = mdutil.GetInt(md, backlog)
if l.md.backlog <= 0 {
l.md.backlog = defaultBacklog