--- /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, + ) +} + +// NewConnectionEx receives TCP connections from the listener and feeds them to the HTTP server. +func (h *Inbound) NewConnectionEx(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 NewConnectionEx) +// 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)