omv-dijiang/works/patch/protocol/obfhttp/inbound.go.patch

--- /dev/null
+++ b/protocol/obfhttp/inbound.go
@@ -0,0 +1,474 @@
+// OMV
+package obfhttp
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/sagernet/sing-box/adapter"
+	"github.com/sagernet/sing-box/adapter/inbound"
+	"github.com/sagernet/sing-box/common/listener"
+	boxTLS "github.com/sagernet/sing-box/common/tls"
+	"github.com/sagernet/sing-box/common/uot"
+	C "github.com/sagernet/sing-box/constant"
+	"github.com/sagernet/sing-box/log"
+	"github.com/sagernet/sing-box/option"
+	"github.com/sagernet/sing/common"
+	E "github.com/sagernet/sing/common/exceptions"
+	"github.com/sagernet/sing/common/logger"
+	M "github.com/sagernet/sing/common/metadata"
+	N "github.com/sagernet/sing/common/network"
+)
+
+func RegisterInbound(registry *inbound.Registry) {
+	inbound.Register[option.ObfHTTPInboundOptions](registry, C.TypeObfHTTP, NewInbound)
+}
+
+var _ adapter.TCPInjectableInbound = (*Inbound)(nil)
+
+type Inbound struct {
+	inbound.Adapter
+	ctx             context.Context
+	router          adapter.ConnectionRouterEx
+	logger          logger.ContextLogger
+	listener        *listener.Listener
+	tlsConfig       boxTLS.ServerConfig
+	httpServer      *http.Server
+	encryptor       Encryptor
+	codec           TextCodec
+	users           map[string]string // user → password
+	path            string
+	sessions        sync.Map // sessionID → *session
+	sessionTimeout  time.Duration
+	longPollTimeout time.Duration
+	connChan        chan net.Conn
+	closeOnce       sync.Once
+}
+
+func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.ObfHTTPInboundOptions) (adapter.Inbound, error) {
+	ib := &Inbound{
+		Adapter:  inbound.NewAdapter(C.TypeObfHTTP, tag),
+		ctx:      ctx,
+		router:   uot.NewRouter(router, logger),
+		logger:   logger,
+		codec:    newTextCodec(options.Encoding),
+		users:    make(map[string]string),
+		path:     options.Path,
+		connChan: make(chan net.Conn, 64),
+	}
+
+	if ib.path == "" {
+		ib.path = "/"
+	}
+
+	for _, user := range options.Users {
+		ib.users[user.Username] = user.Password
+	}
+
+	// Encryption
+	if options.Encryption != nil {
+		enc, err := NewEncryptor(options.Encryption.Method, options.Encryption.Password)
+		if err != nil {
+			return nil, E.Cause(err, "create encryptor")
+		}
+		ib.encryptor = enc
+	} else {
+		ib.encryptor = &noneEncryptor{}
+	}
+
+	// Timeouts
+	ib.sessionTimeout = time.Duration(options.SessionTimeout)
+	if ib.sessionTimeout == 0 {
+		ib.sessionTimeout = 60 * time.Second
+	}
+	ib.longPollTimeout = time.Duration(options.LongPollTimeout)
+	if ib.longPollTimeout == 0 {
+		ib.longPollTimeout = 30 * time.Second
+	}
+
+	// Optional TLS
+	if options.TLS != nil && options.TLS.Enabled {
+		tlsConfig, err := boxTLS.NewServer(ctx, logger, common.PtrValueOrDefault(options.TLS))
+		if err != nil {
+			return nil, err
+		}
+		ib.tlsConfig = tlsConfig
+	}
+
+	ib.httpServer = &http.Server{
+		Handler: ib,
+	}
+
+	ib.listener = listener.New(listener.Options{
+		Context:           ctx,
+		Logger:            logger,
+		Network:           []string{N.NetworkTCP},
+		Listen:            options.ListenOptions,
+		ConnectionHandler: ib,
+	})
+
+	return ib, nil
+}
+
+func (h *Inbound) Start(stage adapter.StartStage) error {
+	if stage != adapter.StartStateStart {
+		return nil
+	}
+	if h.tlsConfig != nil {
+		err := h.tlsConfig.Start()
+		if err != nil {
+			return E.Cause(err, "create TLS config")
+		}
+	}
+	err := h.listener.Start()
+	if err != nil {
+		return err
+	}
+	// Start HTTP server consuming connections from connChan
+	go h.httpServer.Serve(&connListener{ch: h.connChan, addr: h.listener.TCPListener().Addr()})
+	// Start session cleanup goroutine
+	go h.cleanupSessions()
+	return nil
+}
+
+func (h *Inbound) Close() error {
+	h.closeOnce.Do(func() {
+		close(h.connChan)
+	})
+	// Close all sessions
+	h.sessions.Range(func(key, value any) bool {
+		if s, ok := value.(*session); ok {
+			s.conn.Close()
+		}
+		h.sessions.Delete(key)
+		return true
+	})
+	return common.Close(
+		h.httpServer,
+		h.listener,
+		h.tlsConfig,
+	)
+}
+
+// NewConnection receives TCP connections from the listener and feeds them to the HTTP server.
+func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	if h.tlsConfig != nil {
+		tlsConn, err := boxTLS.ServerHandshake(ctx, conn, h.tlsConfig)
+		if err != nil {
+			N.CloseOnHandshakeFailure(conn, onClose, err)
+			h.logger.ErrorContext(ctx, E.Cause(err, "TLS handshake from ", metadata.Source))
+			return
+		}
+		conn = tlsConn
+	}
+	select {
+	case h.connChan <- conn:
+	default:
+		conn.Close()
+		if onClose != nil {
+			onClose(E.New("connection channel full"))
+		}
+	}
+}
+
+// ServeHTTP handles all HTTP requests for the obfhttp protocol.
+func (h *Inbound) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost || r.URL.Path != h.path {
+		h.writeErrorResponse(w, "", "not found")
+		return
+	}
+
+	// Read and decompress body
+	body, err := h.readBody(r)
+	if err != nil {
+		h.writeErrorResponse(w, "", "bad request")
+		return
+	}
+
+	var req request
+	if err := json.Unmarshal(body, &req); err != nil {
+		h.writeErrorResponse(w, "", "bad request")
+		return
+	}
+
+	// Authenticate
+	if !h.authenticate(r) {
+		h.writeErrorResponse(w, req.Session, "unauthorized")
+		return
+	}
+
+	switch req.Action {
+	case actionOpen:
+		h.handleOpen(w, r, &req)
+	case actionData:
+		h.handleData(w, &req)
+	case actionRecv:
+		h.handleRecv(w, &req)
+	case actionClose:
+		h.handleClose(w, &req)
+	default:
+		h.writeErrorResponse(w, req.Session, "unknown action")
+	}
+}
+
+func (h *Inbound) authenticate(r *http.Request) bool {
+	if len(h.users) == 0 {
+		return true
+	}
+	username, password, ok := r.BasicAuth()
+	if !ok {
+		return false
+	}
+	expected, exists := h.users[username]
+	return exists && expected == password
+}
+
+func (h *Inbound) handleOpen(w http.ResponseWriter, r *http.Request, req *request) {
+	if req.Destination == "" || req.Network == "" {
+		h.writeErrorResponse(w, req.Session, "missing destination")
+		return
+	}
+
+	sessionID := generateSessionID()
+	dest := M.ParseSocksaddr(req.Destination)
+
+	remoteAddr := r.RemoteAddr
+	source := M.ParseSocksaddr(remoteAddr)
+
+	sc := newSessionConn(
+		&simpleAddr{network: "tcp", address: "obfhttp-server"},
+		&simpleAddr{network: "tcp", address: remoteAddr},
+	)
+
+	user := ""
+	if username, _, ok := r.BasicAuth(); ok {
+		user = username
+	}
+
+	sess := &session{
+		conn:       sc,
+		lastActive: time.Now(),
+		user:       user,
+	}
+	h.sessions.Store(sessionID, sess)
+
+	// Push any initial data
+	if req.Payload != "" {
+		data, err := h.decodePayload(req.Payload)
+		if err != nil {
+			h.sessions.Delete(sessionID)
+			sc.Close()
+			h.writeErrorResponse(w, sessionID, "decode error")
+			return
+		}
+		sc.pushUpstream(data)
+	}
+
+	// Route the connection
+	var metadata adapter.InboundContext
+	metadata.Inbound = h.Tag()
+	metadata.InboundType = h.Type()
+	metadata.Source = source
+	metadata.Destination = dest
+	metadata.User = user
+
+	go h.router.RouteConnectionEx(h.ctx, sc, metadata, func(err error) {
+		h.sessions.Delete(sessionID)
+	})
+
+	// Return session ID with piggybacked downstream data
+	downData := sc.pullDownstream(500 * time.Millisecond)
+	h.writeResponse(w, sessionID, downData)
+}
+
+func (h *Inbound) handleData(w http.ResponseWriter, req *request) {
+	sess, err := h.getSession(req.Session)
+	if err != nil {
+		h.writeErrorResponse(w, req.Session, "session not found")
+		return
+	}
+	sess.lastActive = time.Now()
+
+	// Push upstream data
+	if req.Payload != "" {
+		data, err := h.decodePayload(req.Payload)
+		if err != nil {
+			h.writeErrorResponse(w, req.Session, "decode error")
+			return
+		}
+		sess.conn.pushUpstream(data)
+	}
+
+	// Piggyback downstream data
+	downData := sess.conn.pullDownstream(500 * time.Millisecond)
+	h.writeResponse(w, req.Session, downData)
+}
+
+func (h *Inbound) handleRecv(w http.ResponseWriter, req *request) {
+	sess, err := h.getSession(req.Session)
+	if err != nil {
+		h.writeErrorResponse(w, req.Session, "session not found")
+		return
+	}
+	sess.lastActive = time.Now()
+
+	// Long poll: wait for downstream data
+	downData := sess.conn.pullDownstream(h.longPollTimeout)
+	h.writeResponse(w, req.Session, downData)
+}
+
+func (h *Inbound) handleClose(w http.ResponseWriter, req *request) {
+	if val, loaded := h.sessions.LoadAndDelete(req.Session); loaded {
+		if s, ok := val.(*session); ok {
+			s.conn.Close()
+		}
+	}
+	h.writeResponse(w, req.Session, nil)
+}
+
+func (h *Inbound) getSession(id string) (*session, error) {
+	val, ok := h.sessions.Load(id)
+	if !ok {
+		return nil, errSessionNotFound
+	}
+	return val.(*session), nil
+}
+
+func (h *Inbound) decodePayload(payload string) ([]byte, error) {
+	decoded, err := h.codec.Decode(payload)
+	if err != nil {
+		return nil, err
+	}
+	return h.encryptor.Decrypt(decoded)
+}
+
+func (h *Inbound) encodePayload(data []byte) (string, error) {
+	encrypted, err := h.encryptor.Encrypt(data)
+	if err != nil {
+		return "", err
+	}
+	return h.codec.Encode(encrypted), nil
+}
+
+func (h *Inbound) writeResponse(w http.ResponseWriter, sessionID string, downData []byte) {
+	resp := response{
+		Session: sessionID,
+		OK:      true,
+		Padding: h.codec.GeneratePadding(randomPaddingLength()),
+	}
+
+	if len(downData) > 0 {
+		encoded, err := h.encodePayload(downData)
+		if err != nil {
+			h.writeErrorResponse(w, sessionID, "encode error")
+			return
+		}
+		resp.Payload = encoded
+	}
+
+	h.writeJSON(w, http.StatusOK, resp)
+}
+
+func (h *Inbound) writeErrorResponse(w http.ResponseWriter, sessionID string, errMsg string) {
+	resp := response{
+		Session: sessionID,
+		OK:      false,
+		Error:   errMsg,
+		Padding: h.codec.GeneratePadding(randomPaddingLength()),
+	}
+	h.writeJSON(w, http.StatusOK, resp)
+}
+
+func (h *Inbound) writeJSON(w http.ResponseWriter, status int, v any) {
+	data, err := json.Marshal(v)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	w.Write(data)
+}
+
+func (h *Inbound) readBody(r *http.Request) ([]byte, error) {
+	var reader io.Reader = r.Body
+	if strings.Contains(r.Header.Get("Content-Encoding"), "gzip") {
+		gr, err := gzip.NewReader(r.Body)
+		if err != nil {
+			return nil, err
+		}
+		defer gr.Close()
+		reader = gr
+	}
+	return io.ReadAll(reader)
+}
+
+func (h *Inbound) cleanupSessions() {
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-h.ctx.Done():
+			return
+		case <-ticker.C:
+			now := time.Now()
+			h.sessions.Range(func(key, value any) bool {
+				s := value.(*session)
+				if now.Sub(s.lastActive) > h.sessionTimeout {
+					h.logger.Debug("cleaning up expired session: ", key)
+					s.conn.Close()
+					h.sessions.Delete(key)
+				}
+				return true
+			})
+		}
+	}
+}
+
+func generateSessionID() string {
+	b := make([]byte, 16)
+	_, _ = rand.Read(b)
+	return hex.EncodeToString(b)
+}
+
+// connListener bridges the listener.Listener (which pushes connections via NewConnection)
+// with http.Server.Serve() (which expects a net.Listener).
+type connListener struct {
+	ch   chan net.Conn
+	addr net.Addr
+}
+
+func (l *connListener) Accept() (net.Conn, error) {
+	conn, ok := <-l.ch
+	if !ok {
+		return nil, net.ErrClosed
+	}
+	return conn, nil
+}
+
+func (l *connListener) Close() error {
+	return nil
+}
+
+func (l *connListener) Addr() net.Addr {
+	return l.addr
+}
+
+// compile-time checks
+var (
+	_ net.Listener = (*connListener)(nil)
+	_ io.Closer    = (*Inbound)(nil)
+)
+
+// Ensure the request body is drained and closed, even if gzip.
+var _ = (*bytes.Buffer)(nil)