Fix data races across session/forwarder/registry, resolve POST/PUT hang, harden port allocation, add frontendURL env (#157)

Reviewed-on: #157
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 {
-		return f.listener.Close()
+	if listener := f.Listener(); listener != nil {
+		return listener.Close()
 	}
 	return nil
 }

internal/session/forwarder/forwarder_test.go

@@ -24,6 +24,7 @@ type mockConfig struct {
 }
 
 func (m *mockConfig) Domain() string            { return m.Called().String(0) }
+func (m *mockConfig) FrontendURL() string       { return m.Called().String(0) }
 func (m *mockConfig) SSHPort() string           { return m.Called().String(0) }
 func (m *mockConfig) HTTPPort() string          { return m.Called().String(0) }
 func (m *mockConfig) HTTPSPort() string         { return m.Called().String(0) }

internal/session/interaction/interaction_test.go

@@ -32,6 +32,7 @@ type MockConfig struct {
 }
 
 func (m *MockConfig) Domain() string            { return m.Called().String(0) }
+func (m *MockConfig) FrontendURL() string       { return m.Called().String(0) }
 func (m *MockConfig) SSHPort() string           { return m.Called().String(0) }
 func (m *MockConfig) HTTPPort() string          { return m.Called().String(0) }
 func (m *MockConfig) HTTPSPort() string         { return m.Called().String(0) }

internal/session/session.go

@@ -26,7 +26,7 @@ type Session interface {
 	HandleGlobalRequest(ch <-chan *ssh.Request) error
 	HandleTCPIPForward(req *ssh.Request) error
 	HandleHTTPForward(req *ssh.Request, port uint16) error
-	HandleTCPForward(req *ssh.Request, addr string, port uint16) error
+	HandleTCPForward(req *ssh.Request, addr string, port uint16, reserved bool) error
 	Lifecycle() lifecycle.Lifecycle
 	Interaction() interaction.Interaction
 	Forwarder() forwarder.Forwarder
@@ -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 {
-	case req, ok := <-s.initialReq:
-		if !ok {
-			log.Println("Forwarding request channel closed")
+	for {
+		select {
+		case req, ok := <-s.initialReq:
+			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
 	}
 }
 
@@ -255,35 +254,35 @@ func (s *session) HandleGlobalRequest(GlobalRequest <-chan *ssh.Request) error {
 	return nil
 }
 
-func (s *session) parseForwardPayload(payload []byte) (address string, port uint16, err error) {
+func (s *session) parseForwardPayload(payload []byte) (address string, port uint16, reserved bool, err error) {
 	var forwardPayload struct {
 		BindAddr string
 		BindPort uint32
 	}
 
 	if err = ssh.Unmarshal(payload, &forwardPayload); err != nil {
-		return "", 0, fmt.Errorf("failed to unmarshal forward payload: %w", err)
+		return "", 0, false, fmt.Errorf("failed to unmarshal forward payload: %w", err)
 	}
 
 	if forwardPayload.BindPort > 65535 {
-		return "", 0, fmt.Errorf("port is larger than allowed port of 65535")
+		return "", 0, false, fmt.Errorf("port is larger than allowed port of 65535")
 	}
 
 	port = uint16(forwardPayload.BindPort)
 
 	if isBlockedPort(port) {
-		return "", 0, fmt.Errorf("port is blocked")
+		return "", 0, false, fmt.Errorf("port is blocked")
 	}
 
 	if port == 0 {
 		unassigned, ok := s.lifecycle.PortRegistry().Unassigned()
 		if !ok {
-			return "", 0, fmt.Errorf("no available port")
+			return "", 0, false, fmt.Errorf("no available port")
 		}
-		return forwardPayload.BindAddr, unassigned, nil
+		return forwardPayload.BindAddr, unassigned, true, nil
 	}
 
-	return forwardPayload.BindAddr, port, nil
+	return forwardPayload.BindAddr, port, false, nil
 }
 
 func (s *session) denyForwardingRequest(req *ssh.Request, key *types.SessionKey, listener io.Closer, msg string) error {
@@ -327,7 +326,7 @@ func (s *session) finalizeForwarding(req *ssh.Request, portToBind uint16, listen
 }
 
 func (s *session) HandleTCPIPForward(req *ssh.Request) error {
-	address, port, err := s.parseForwardPayload(req.Payload)
+	address, port, reserved, err := s.parseForwardPayload(req.Payload)
 	if err != nil {
 		return s.denyForwardingRequest(req, nil, nil, fmt.Sprintf("cannot parse forwarded payload: %s", err.Error()))
 	}
@@ -336,7 +335,7 @@ func (s *session) HandleTCPIPForward(req *ssh.Request) error {
 	case 80, 443:
 		return s.HandleHTTPForward(req, port)
 	default:
-		return s.HandleTCPForward(req, address, port)
+		return s.HandleTCPForward(req, address, port, reserved)
 	}
 }
 
@@ -357,30 +356,40 @@ func (s *session) HandleHTTPForward(req *ssh.Request, portToBind uint16) error {
 	return nil
 }
 
-func (s *session) HandleTCPForward(req *ssh.Request, addr string, portToBind uint16) error {
-	if claimed := s.lifecycle.PortRegistry().Claim(portToBind); !claimed {
-		return s.denyForwardingRequest(req, nil, nil, fmt.Sprintf("Port %d is already in use or restricted", portToBind))
+func (s *session) HandleTCPForward(req *ssh.Request, addr string, portToBind uint16, reserved bool) error {
+	if !reserved {
+		if claimed := s.lifecycle.PortRegistry().Claim(portToBind); !claimed {
+			return s.denyForwardingRequest(req, nil, nil, fmt.Sprintf("Port %d is already in use or restricted", portToBind))
+		}
+	}
+
+	releasePort := func() {
+		if err := s.lifecycle.PortRegistry().SetStatus(portToBind, false); err != nil {
+			log.Printf("failed to release port %d: %v", portToBind, err)
+		}
 	}
 
 	tcpServer := transport.NewTCPServer(portToBind, s.forwarder)
 	listener, err := tcpServer.Listen()
 	if err != nil {
+		releasePort()
 		return s.denyForwardingRequest(req, nil, listener, fmt.Sprintf("Port %d is already in use or restricted", portToBind))
 	}
 
 	key := types.SessionKey{Id: fmt.Sprintf("%d", portToBind), Type: types.TunnelTypeTCP}
 	if !s.registry.Register(key, s) {
+		releasePort()
 		return s.denyForwardingRequest(req, nil, listener, fmt.Sprintf("Failed to register TunnelTypeTCP client with id: %s", key.Id))
 	}
 
 	err = s.finalizeForwarding(req, portToBind, listener, types.TunnelTypeTCP, key.Id)
 	if err != nil {
+		releasePort()
 		return s.denyForwardingRequest(req, &key, listener, fmt.Sprintf("Failed to finalize forwarding: %s", err))
 	}
 
 	go func() {
-		err = tcpServer.Serve(listener)
-		if err != nil {
+		if err := tcpServer.Serve(listener); err != nil {
 			log.Printf("Failed serving tcp server: %s\n", err)
 		}
 	}()

internal/session/session_test.go

@@ -38,8 +38,9 @@ type mockConfig struct {
 	config.Config
 }
 
-func (m *mockConfig) Domain() string  { return m.Called().String(0) }
-func (m *mockConfig) SSHPort() string { return m.Called().String(0) }
+func (m *mockConfig) Domain() string            { return m.Called().String(0) }
+func (m *mockConfig) FrontendURL() string       { return m.Called().String(0) }
+func (m *mockConfig) SSHPort() string           { return m.Called().String(0) }
 func (m *mockConfig) Mode() types.ServerMode {
 	args := m.Called()
 	if args.Get(0) == nil {
@@ -396,6 +397,12 @@ func TestHandleTCPIPForward_Table(t *testing.T) {
 		err := s.HandleTCPIPForward(req)
 		assert.NoError(t, err)
 		assert.Equal(t, uint16(12345), s.forwarder.ForwardedPort())
+
+		defer func() {
+			if l := s.forwarder.Listener(); l != nil {
+				_ = l.Close()
+			}
+		}()
 	})
 
 	t.Run("Invalid Payload", func(t *testing.T) {
@@ -699,6 +706,7 @@ func TestForwardingFailures(t *testing.T) {
 		s, mRegistry, mPort, _, _, sReqs, cConn, cleanup := setup(t)
 		defer cleanup()
 		mPort.On("Claim", mock.Anything).Return(true)
+		mPort.On("SetStatus", uint16(1234), false).Return(nil)
 		mRegistry.On("Register", mock.Anything, mock.Anything).Return(false)
 
 		payload := make([]byte, 4+9+4)
@@ -717,7 +725,7 @@ func TestForwardingFailures(t *testing.T) {
 	})
 
 	t.Run("Finalize Forwarding Failure", func(t *testing.T) {
-		s, mRegistry, _, mRandom, _, sReqs, cConn, cleanup := setup(t)
+		s, mRegistry, _, mRandom, sConn, sReqs, cConn, cleanup := setup(t)
 		defer cleanup()
 		mRandom.On("String", 20).Return("test-slug", nil)
 		mRegistry.On("Register", mock.Anything, mock.Anything).Return(true)
@@ -736,7 +744,7 @@ func TestForwardingFailures(t *testing.T) {
 		err := cConn.Close()
 		assert.NoError(t, err)
 
-		time.Sleep(50 * time.Millisecond)
+		_ = sConn.Wait()
 
 		err = s.HandleTCPIPForward(req)
 		assert.Error(t, err)
@@ -758,6 +766,7 @@ func TestForwardingFailures(t *testing.T) {
 		}(l)
 		_, portStr, _ := net.SplitHostPort(l.Addr().String())
 		port, _ := strconv.Atoi(portStr)
+		mPort.On("SetStatus", uint16(port), false).Return(nil)
 
 		payload := make([]byte, 4+9+4)
 		binary.BigEndian.PutUint32(payload[0:4], 9)
@@ -807,7 +816,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 +826,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) {
@@ -982,7 +1046,7 @@ func TestParseForwardPayload_Errors(t *testing.T) {
 	s := &session{}
 
 	t.Run("Short Address", func(t *testing.T) {
-		_, _, err := s.parseForwardPayload([]byte{0, 0, 0, 4})
+		_, _, _, err := s.parseForwardPayload([]byte{0, 0, 0, 4})
 		if err == nil {
 			t.Error("expected error, got nil")
 		}
@@ -990,7 +1054,7 @@ func TestParseForwardPayload_Errors(t *testing.T) {
 
 	t.Run("Short Port", func(t *testing.T) {
 		payload := append([]byte{0, 0, 0, 4}, []byte("addr")...)
-		_, _, err := s.parseForwardPayload(payload)
+		_, _, _, err := s.parseForwardPayload(payload)
 		if err == nil {
 			t.Error("expected error, got nil")
 		}
@@ -1001,7 +1065,7 @@ func TestParseForwardPayload_Errors(t *testing.T) {
 		portBuf := make([]byte, 4)
 		binary.BigEndian.PutUint32(portBuf, 22)
 		payload = append(payload, portBuf...)
-		_, _, err := s.parseForwardPayload(payload)
+		_, _, _, err := s.parseForwardPayload(payload)
 		if err == nil {
 			t.Error("expected error, got nil")
 		} else if !strings.Contains(err.Error(), "port is block") {
@@ -1165,7 +1229,7 @@ func TestHandleTCPForward_Failures(t *testing.T) {
 		s, _, mPort, _, sReqs, cConn, cleanup := setup(t)
 		defer cleanup()
 		mPort.On("Claim", mock.Anything).Return(false)
-		err := s.HandleTCPForward(getReq(t, cConn, sReqs), "localhost", 1234)
+		err := s.HandleTCPForward(getReq(t, cConn, sReqs), "localhost", 1234, false)
 		if err == nil {
 			t.Error("expected error, got nil")
 		} else if !strings.Contains(err.Error(), "already in use") {
@@ -1186,44 +1250,50 @@ func TestHandleTCPForward_Failures(t *testing.T) {
 			assert.NoError(t, err)
 		}(l)
 		port := uint16(l.Addr().(*net.TCPAddr).Port)
+		mPort.On("SetStatus", port, false).Return(nil)
 
-		err = s.HandleTCPForward(getReq(t, cConn, sReqs), "localhost", port)
+		err = s.HandleTCPForward(getReq(t, cConn, sReqs), "localhost", port, false)
 		if err == nil {
 			t.Error("expected error, got nil")
 		} else if !strings.Contains(err.Error(), "already in use") {
 			t.Errorf("expected error to contain %q, got %q", "already in use", err.Error())
 		}
+		mPort.AssertExpectations(t)
 	})
 
 	t.Run("Registry Register fail", func(t *testing.T) {
 		s, mRegistry, mPort, _, sReqs, cConn, cleanup := setup(t)
 		defer cleanup()
 		mPort.On("Claim", mock.Anything).Return(true)
+		mPort.On("SetStatus", mock.AnythingOfType("uint16"), false).Return(nil)
 		mRegistry.On("Register", mock.Anything, mock.Anything).Return(false)
-		err := s.HandleTCPForward(getReq(t, cConn, sReqs), "localhost", 0)
+		err := s.HandleTCPForward(getReq(t, cConn, sReqs), "localhost", 0, false)
 		if err == nil {
 			t.Error("expected error, got nil")
 		} else if !strings.Contains(err.Error(), "Failed to register") {
 			t.Errorf("expected error to contain %q, got %q", "Failed to register", err.Error())
 		}
+		mPort.AssertExpectations(t)
 	})
 
 	t.Run("Finalize fail (Reply fail)", func(t *testing.T) {
-		s, mRegistry, mPort, _, sReqs, cConn, cleanup := setup(t)
+		s, mRegistry, mPort, sConn, sReqs, cConn, cleanup := setup(t)
 		defer cleanup()
 		mPort.On("Claim", mock.Anything).Return(true)
+		mPort.On("SetStatus", mock.AnythingOfType("uint16"), false).Return(nil)
 		mRegistry.On("Register", mock.Anything, mock.Anything).Return(true)
 		req := getReq(t, cConn, sReqs)
 		err := cConn.Close()
 		assert.NoError(t, err)
-		time.Sleep(100 * time.Millisecond)
+		_ = sConn.Wait()
 
-		err = s.HandleTCPForward(req, "localhost", 0)
+		err = s.HandleTCPForward(req, "localhost", 0, false)
 		if err == nil {
 			t.Error("expected error, got nil")
 		} else if !strings.Contains(err.Error(), "Failed to finalize forwarding") {
 			t.Errorf("expected error to contain %q, got %q", "Failed to finalize forwarding", err.Error())
 		}
+		mPort.AssertExpectations(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
 }