fix(httphandler): use readSlice for reading http header

## Summary

Comprehensive fixes for race conditions, concurrency bugs, and a critical session handling bug across the session package and its child packages.

## Changes

### Bug Fix: waitForTCPIPForward fails on non-tcpip-forward global requests

- Changed `waitForTCPIPForward()` from a single select to a for loop that drains non-tcpip-forward requests gracefully
- Non-tcpip-forward requests now receive a false reply and the function continues waiting for the actual forward request
- The 500ms timeout resets after each request to handle slow clients with multiple early requests
- Updated and added tests to cover the new behavior: Wrong_Request_Type_Then_Timeout, Multiple_Non-Forward_Requests_Then_Success, Timeout_After_Non-Forward_Requests

### Race Condition Fixes

- Added sync.RWMutex to the forwarder struct to protect concurrent field access
- Fixed data race in forwarder setters (`SetType()`, `SetForwardedPort()`, `SetListener()`) by adding `Lock()`/`Unlock()` protection
- Fixed data race in forwarder getters (`TunnelType()`, `ForwardedPort()`, `Listener()`) by adding `RLock()`/`RUnlock()` protection
- Added sync.RWMutex to the slug struct to protect concurrent field access
- Fixed data race in `slug.Set()` by adding `Lock()`/`Unlock()` protection
- Fixed data race in `slug.String()` by adding `RLock()`/`RUnlock()` protection

### Concurrency Safety Improvements

- Changed `OpenForwardedChannel()` to use `f.ForwardedPort()` getter instead of direct field access
- Fixed `Close()` in forwarder to use `f.Listener()` getter consistently instead of accessing `f.listener` directly
- Fixed data race in `HandleTCPForward()` where the goroutine running `tcpServer.Serve(listener)` wrote to the outer err variable from the enclosing function scope
- Changed err = `tcpServer.Serve(listener)` to `if err := tcpServer.Serve(listener)` so the goroutine uses a local variable

Reviewed-on: #151
Co-authored-by: Bagas <bagas@fossy.my.id>
Co-committed-by: Bagas <bagas@fossy.my.id>

internal/session/forwarder/forwarder.go

@@ -28,6 +28,7 @@ type Forwarder interface {
 	Close() error
 }
 type forwarder struct {
+	mu            sync.RWMutex
 	listener      net.Listener
 	tunnelType    types.TunnelType
 	forwardedPort uint16
@@ -60,7 +61,7 @@ func (f *forwarder) copyWithBuffer(dst io.Writer, src io.Reader) (written int64,
 }
 
 func (f *forwarder) OpenForwardedChannel(ctx context.Context, origin net.Addr) (ssh.Channel, <-chan *ssh.Request, error) {
-	payload := createForwardedTCPIPPayload(origin, f.forwardedPort)
+	payload := createForwardedTCPIPPayload(origin, f.ForwardedPort())
 	type channelResult struct {
 		channel ssh.Channel
 		reqs    <-chan *ssh.Request
@@ -141,32 +142,44 @@ func (f *forwarder) HandleConnection(dst io.ReadWriter, src ssh.Channel) {
 }
 
 func (f *forwarder) SetType(tunnelType types.TunnelType) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.tunnelType = tunnelType
 }
 
 func (f *forwarder) TunnelType() types.TunnelType {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	return f.tunnelType
 }
 
 func (f *forwarder) ForwardedPort() uint16 {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	return f.forwardedPort
 }
 
 func (f *forwarder) SetForwardedPort(port uint16) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.forwardedPort = port
 }
 
 func (f *forwarder) SetListener(listener net.Listener) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	f.listener = listener
 }
 
 func (f *forwarder) Listener() net.Listener {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	return f.listener
 }
 
 func (f *forwarder) Close() error {
-	if f.Listener() != nil {
+	if listener := f.Listener(); listener != nil {
-		return f.listener.Close()
+		return listener.Close()
 	}
 	return nil
 }

internal/session/session.go

@@ -158,7 +158,7 @@ func (s *session) setupInteractiveMode(channel ssh.NewChannel) error {
 	}
 
 	go func() {
-		err = s.HandleGlobalRequest(reqs)
+		err := s.HandleGlobalRequest(reqs)
 		if err != nil {
 			log.Printf("global request handler error: %v", err)
 		}
@@ -200,23 +200,22 @@ func (s *session) waitForSessionEnd() error {
 }
 
 func (s *session) waitForTCPIPForward() *ssh.Request {
-	select {
+	for {
-	case req, ok := <-s.initialReq:
+		select {
-		if !ok {
+		case req, ok := <-s.initialReq:
-			log.Println("Forwarding request channel closed")
+			if !ok {
+				log.Println("Forwarding request channel closed")
+				return nil
+			}
+			if req.Type == "tcpip-forward" {
+				return req
+			}
+			log.Printf("Ignoring unexpected global request: %s", req.Type)
+			_ = req.Reply(false, nil)
+		case <-time.After(500 * time.Millisecond):
+			log.Println("No tcpip-forward request received within timeout")
 			return nil
 		}
-		if req.Type == "tcpip-forward" {
-			return req
-		}
-		if err := req.Reply(false, nil); err != nil {
-			log.Printf("Failed to reply to request: %v", err)
-		}
-		log.Printf("Expected tcpip-forward request, got: %s", req.Type)
-		return nil
-	case <-time.After(500 * time.Millisecond):
-		log.Println("No forwarding request received")
-		return nil
 	}
 }
 
@@ -379,8 +378,7 @@ func (s *session) HandleTCPForward(req *ssh.Request, addr string, portToBind uin
 	}
 
 	go func() {
-		err = tcpServer.Serve(listener)
+		if err := tcpServer.Serve(listener); err != nil {
-		if err != nil {
 			log.Printf("Failed serving tcp server: %s\n", err)
 		}
 	}()

internal/session/session_test.go

@@ -807,7 +807,7 @@ func (m *mockNewChanFail) Accept() (ssh.Channel, <-chan *ssh.Request, error) {
 }
 
 func TestWaitForTCPIPForward_EdgeCases(t *testing.T) {
-	t.Run("Wrong Request Type", func(t *testing.T) {
+	t.Run("Wrong Request Type Then Timeout", func(t *testing.T) {
 		_, sReqs, _, cConn, cleanup := setupSSH(t)
 		defer cleanup()
 
@@ -817,10 +817,65 @@ func TestWaitForTCPIPForward_EdgeCases(t *testing.T) {
 			_, _, _ = cConn.SendRequest("not-tcpip-forward", true, nil)
 		}()
 
+		start := time.Now()
 		req := s.waitForTCPIPForward()
+		elapsed := time.Since(start)
+
 		if req != nil {
 			t.Error("expected nil request")
 		}
+		if elapsed < 400*time.Millisecond {
+			t.Errorf("expected timeout ~500ms, got %v", elapsed)
+		}
+	})
+
+	t.Run("Multiple Non-Forward Requests Then Success", func(t *testing.T) {
+		_, sReqs, _, cConn, cleanup := setupSSH(t)
+		defer cleanup()
+
+		s := &session{initialReq: sReqs}
+
+		go func() {
+			time.Sleep(100 * time.Millisecond)
+			_, _, _ = cConn.SendRequest("keepalive@openssh.com", false, nil)
+			time.Sleep(100 * time.Millisecond)
+			_, _, _ = cConn.SendRequest("hostkeys-00@openssh.com", false, nil)
+			time.Sleep(100 * time.Millisecond)
+			_, _, _ = cConn.SendRequest("tcpip-forward", true, nil)
+		}()
+
+		req := s.waitForTCPIPForward()
+		if req == nil {
+			t.Error("expected tcpip-forward request, got nil")
+		}
+		if req != nil && req.Type != "tcpip-forward" {
+			t.Errorf("expected tcpip-forward, got %s", req.Type)
+		}
+	})
+
+	t.Run("Timeout After Non-Forward Requests", func(t *testing.T) {
+		_, sReqs, _, cConn, cleanup := setupSSH(t)
+		defer cleanup()
+
+		s := &session{initialReq: sReqs}
+
+		go func() {
+			time.Sleep(100 * time.Millisecond)
+			_, _, _ = cConn.SendRequest("keepalive@openssh.com", false, nil)
+			time.Sleep(100 * time.Millisecond)
+			_, _, _ = cConn.SendRequest("hostkeys-00@openssh.com", false, nil)
+		}()
+
+		start := time.Now()
+		req := s.waitForTCPIPForward()
+		elapsed := time.Since(start)
+
+		if req != nil {
+			t.Error("expected nil request after timeout")
+		}
+		if elapsed < 400*time.Millisecond {
+			t.Errorf("expected timeout ~500ms after last request, got %v", elapsed)
+		}
 	})
 
 	t.Run("Channel Closed", func(t *testing.T) {

internal/session/slug/slug.go

@@ -1,11 +1,14 @@
 package slug
 
+import "sync"
+
 type Slug interface {
 	String() string
 	Set(slug string)
 }
 
 type slug struct {
+	mu   sync.RWMutex
 	slug string
 }
 
@@ -16,9 +19,13 @@ func New() Slug {
 }
 
 func (s *slug) String() string {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.slug
 }
 
 func (s *slug) Set(slug string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	s.slug = slug
 }

internal/transport/httphandler.go

@@ -1,6 +1,7 @@
 package transport
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"errors"
@@ -52,25 +53,43 @@ func (hh *httpHandler) badRequest(conn net.Conn) error {
 	return nil
 }
 
+func readHTTPHeader(br *bufio.Reader, limit int) ([]byte, error) {
+	var headerBuf []byte
+	for {
+		line, err := br.ReadSlice('\n')
+		headerBuf = append(headerBuf, line...)
+		if errors.Is(err, bufio.ErrBufferFull) {
+			if len(headerBuf) > limit {
+				return nil, fmt.Errorf("headers too large")
+			}
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+		if bytes.HasSuffix(headerBuf, []byte("\r\n\r\n")) {
+			return headerBuf, nil
+		}
+		if len(headerBuf) > limit {
+			return nil, fmt.Errorf("headers too large")
+		}
+	}
+}
+
 func (hh *httpHandler) Handler(conn net.Conn, isTLS bool) {
 	defer hh.closeConnection(conn)
 
 	_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
-	buf := make([]byte, hh.config.HeaderSize())
+	br := bufio.NewReaderSize(conn, hh.config.HeaderSize())
-	n, err := conn.Read(buf)
+	headerBuf, err := readHTTPHeader(br, hh.config.HeaderSize())
 	if err != nil {
 		_ = hh.badRequest(conn)
 		return
 	}
 
-	if idx := bytes.Index(buf[:n], []byte("\r\n\r\n")); idx == -1 {
-		_ = hh.badRequest(conn)
-		return
-	}
-
 	_ = conn.SetReadDeadline(time.Time{})
 
-	reqhf, err := header.NewRequest(buf[:n])
+	reqhf, err := header.NewRequest(headerBuf)
 	if err != nil {
 		log.Printf("Error creating request header: %v", err)
 		_ = hh.badRequest(conn)
@@ -101,7 +120,7 @@ func (hh *httpHandler) Handler(conn net.Conn, isTLS bool) {
 		return
 	}
 
-	hw := stream.New(conn, conn, conn.RemoteAddr())
+	hw := stream.New(conn, br, conn.RemoteAddr())
 	defer func(hw stream.HTTP) {
 		err = hw.Close()
 		if err != nil {

internal/transport/httphandler_test.go

@@ -321,8 +321,14 @@ func TestHandler(t *testing.T) {
 			isTLS:       false,
 			redirectTLS: false,
 			request:     []byte(""),
-			expected:    []byte("HTTP/1.1 400 Bad Request\r\n\r\n"),
+			expected:    []byte(""),
-			setupMocks: func(msr *MockSessionRegistry) {
+			setupConn: func() (net.Conn, net.Conn) {
+				mc := new(MockConn)
+				mc.ReadBuffer = bytes.NewBuffer(nil)
+				mc.On("SetReadDeadline", mock.Anything).Return(nil)
+				mc.On("Write", []byte("HTTP/1.1 400 Bad Request\r\n\r\n")).Return(0, nil)
+				mc.On("Close").Return(nil)
+				return mc, nil
 			},
 		},
 		{
@@ -715,3 +721,113 @@ func TestHandler(t *testing.T) {
 		})
 	}
 }
+
+func TestHandlerForwardsPostBody(t *testing.T) {
+	mockSessionRegistry := new(MockSessionRegistry)
+	mockConfig := &MockConfig{}
+	mockConfig.On("Domain").Return("example.com")
+	mockConfig.On("HTTPPort").Return("0")
+	mockConfig.On("HeaderSize").Return(4096)
+	mockConfig.On("TLSRedirect").Return(true)
+	hh := &httpHandler{
+		sessionRegistry: mockSessionRegistry,
+		config:          mockConfig,
+	}
+
+	mockSession := new(MockSession)
+	mockForwarder := new(MockForwarder)
+	mockSSHChannel := new(MockSSHChannel)
+
+	mockSessionRegistry.On("Get", types.SessionKey{
+		Id:   "test",
+		Type: types.TunnelTypeHTTP,
+	}).Return(mockSession, nil)
+	mockSession.On("Forwarder").Return(mockForwarder)
+
+	reqCh := make(chan *ssh.Request)
+	mockForwarder.On("OpenForwardedChannel", mock.Anything, mock.Anything).Return(mockSSHChannel, (<-chan *ssh.Request)(reqCh), nil)
+
+	var mu sync.Mutex
+	var capturedHeaders []byte
+	mockSSHChannel.On("Write", mock.Anything).Run(func(args mock.Arguments) {
+		mu.Lock()
+		capturedHeaders = append(capturedHeaders, args.Get(0).([]byte)...)
+		mu.Unlock()
+	}).Return(0, nil)
+	mockSSHChannel.On("Close").Return(nil)
+
+	bodyChan := make(chan string, 1)
+	mockForwarder.On("HandleConnection", mock.Anything, mockSSHChannel).Run(func(args mock.Arguments) {
+		w := args.Get(0).(io.ReadWriter)
+		buf := make([]byte, len("hello=world"))
+		if _, err := io.ReadFull(w, buf); err != nil {
+			bodyChan <- ""
+		} else {
+			bodyChan <- string(buf)
+		}
+		_, _ = w.Write([]byte("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"))
+	})
+
+	go func() {
+		for range reqCh {
+		}
+	}()
+
+	serverConn, clientConn := net.Pipe()
+	defer func() {
+		_ = clientConn.Close()
+	}()
+
+	remoteAddr, _ := net.ResolveTCPAddr("tcp", "127.0.0.1:12345")
+	wrappedServerConn := &wrappedConn{Conn: serverConn, remoteAddr: remoteAddr}
+
+	go hh.Handler(wrappedServerConn, true)
+
+	request := []byte("POST / HTTP/1.1\r\nHost: test.domain\r\nContent-Type: application/x-www-form-urlencoded\r\nContent-Length: 11\r\n\r\nhello=world")
+	go func() {
+		_, _ = clientConn.Write(request)
+	}()
+
+	var response []byte
+	respDone := make(chan struct{})
+	go func() {
+		defer close(respDone)
+		buf := make([]byte, 4096)
+		for {
+			n, err := clientConn.Read(buf)
+			if err != nil {
+				break
+			}
+			response = append(response, buf[:n]...)
+			if bytes.Contains(response, []byte("\r\n\r\nok")) {
+				break
+			}
+		}
+	}()
+
+	select {
+	case body := <-bodyChan:
+		assert.Equal(t, "hello=world", body)
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for forwarded body")
+	}
+
+	select {
+	case <-respDone:
+		resStr := string(response)
+		assert.True(t, strings.HasPrefix(resStr, "HTTP/1.1 200 OK\r\n"))
+		assert.Contains(t, resStr, "Server: Tunnel Please\r\n")
+		assert.True(t, strings.HasSuffix(resStr, "\r\n\r\nok"))
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for response")
+	}
+
+	mu.Lock()
+	hdrStr := string(capturedHeaders)
+	mu.Unlock()
+	assert.Contains(t, hdrStr, "POST / HTTP/1.1\r\n")
+	assert.Contains(t, hdrStr, "Content-Length: 11\r\n")
+	assert.Contains(t, hdrStr, "X-Forwarded-For: 127.0.0.1\r\n")
+
+	mockSessionRegistry.AssertExpectations(t)
+}