Expand time formats supported by chtimes

SOCKS5 Stability Improvements for HTTP/mTLS Transports

client/command/filesystem/chtimes.go

@@ -19,6 +19,8 @@ package filesystem
 
 import (
 	"context"
+	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/bishopfox/sliver/client/console"
@@ -28,14 +30,77 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+const chtimesDefaultLayout = "2006-01-02 15:04:05"
+
+type chtimesTimeFormat struct {
+	name  string
+	parse func(string) (int64, error)
+}
+
+func chtimesParseUnixSeconds(value string) (int64, error) {
+	parsed, err := strconv.ParseInt(value, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return parsed, nil
+}
+
+func chtimesParseUnixMillis(value string) (int64, error) {
+	parsed, err := strconv.ParseInt(value, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return time.UnixMilli(parsed).Unix(), nil
+}
+
+func chtimesParseLayout(layout string) func(string) (int64, error) {
+	return func(value string) (int64, error) {
+		parsed, err := time.Parse(layout, value)
+		if err != nil {
+			return 0, err
+		}
+		return parsed.Unix(), nil
+	}
+}
+
+func chtimesFormatFromFlags(cmd *cobra.Command) (chtimesTimeFormat, error) {
+	formatFlags := []struct {
+		flag   string
+		format chtimesTimeFormat
+	}{
+		{flag: "unix", format: chtimesTimeFormat{name: "unix", parse: chtimesParseUnixSeconds}},
+		{flag: "unix-ms", format: chtimesTimeFormat{name: "unix-ms", parse: chtimesParseUnixMillis}},
+		{flag: "rfc3339", format: chtimesTimeFormat{name: "rfc3339", parse: chtimesParseLayout(time.RFC3339)}},
+		{flag: "rfc1123", format: chtimesTimeFormat{name: "rfc1123", parse: chtimesParseLayout(time.RFC1123)}},
+	}
+
+	var selected *chtimesTimeFormat
+	selectedFlag := ""
+	for _, candidate := range formatFlags {
+		enabled, _ := cmd.Flags().GetBool(candidate.flag)
+		if !enabled {
+			continue
+		}
+		if selected != nil {
+			return chtimesTimeFormat{}, fmt.Errorf("only one time format flag can be used (--%s and --%s are both set)", selectedFlag, candidate.flag)
+		}
+		selected = &candidate.format
+		selectedFlag = candidate.flag
+	}
+
+	if selected == nil {
+		return chtimesTimeFormat{name: "datetime", parse: chtimesParseLayout(chtimesDefaultLayout)}, nil
+	}
+
+	return *selected, nil
+}
+
 // ChtimesCmd - Change the access and modified time of a file on the remote file system.
 func ChtimesCmd(cmd *cobra.Command, con *console.SliverClient, args []string) {
 	session, beacon := con.ActiveTarget.GetInteractive()
 	if session == nil && beacon == nil {
 		return
 	}
-	// DateTime layout (https://pkg.go.dev/time)
-	layout := "2006-01-02 15:04:05"
 	filePath := args[0]
 
 	if filePath == "" {
@@ -43,33 +108,35 @@ func ChtimesCmd(cmd *cobra.Command, con *console.SliverClient, args []string) {
 		return
 	}
 
-	atime := args[1]
+	format, err := chtimesFormatFromFlags(cmd)
+	if err != nil {
+		con.PrintErrorf("%s\n", err)
+		return
+	}
 
+	atime := args[1]
 	if atime == "" {
 		con.PrintErrorf("Missing parameter: Last accessed time id\n")
 		return
 	}
 
-	t_a, err := time.Parse(layout, atime)
+	unixAtime, err := format.parse(atime)
 	if err != nil {
-		con.PrintErrorf("%s\n", err)
+		con.PrintErrorf("Invalid access time (%s): %s\n", format.name, err)
 		return
 	}
-	unixAtime := t_a.Unix()
 
 	mtime := args[2]
-
 	if mtime == "" {
 		con.PrintErrorf("Missing parameter: Last modified time id\n")
 		return
 	}
 
-	t_b, err := time.Parse(layout, mtime)
+	unixMtime, err := format.parse(mtime)
 	if err != nil {
-		con.PrintErrorf("%s\n", err)
+		con.PrintErrorf("Invalid modified time (%s): %s\n", format.name, err)
 		return
 	}
-	unixMtime := t_b.Unix()
 
 	chtimes, err := con.Rpc.Chtimes(context.Background(), &sliverpb.ChtimesReq{
 		Request: con.ActiveTarget.Request(cmd),

client/command/help/long-help.go

@@ -64,6 +64,7 @@ var (
 		consts.UploadStr:           uploadHelp,
 		consts.MkdirStr:            mkdirHelp,
 		consts.RmStr:               rmHelp,
+		consts.ChtimesStr:          chtimesHelp,
 		consts.ProcdumpStr:         procdumpHelp,
 		consts.ElevateStr:          elevateHelp,
 		consts.RunAsStr:            runAsHelp,
@@ -299,6 +300,24 @@ On Windows, escaping is disabled. Instead, '\\' is treated as path separator.
 	rmHelp = `[[.Bold]]Command:[[.Normal]] rm [remote path]
 [[.Bold]]About:[[.Normal]] Delete a remote file or directory.`
 
+	chtimesHelp = `[[.Bold]]Command:[[.Normal]] chtimes [--unix | --unix-ms | --rfc3339 | --rfc1123] <path> <atime> <mtime>
+[[.Bold]]About:[[.Normal]] Change access and modification times on a file (timestomp).
+
+[[.Bold]][[.Underline]]Time Formats[[.Normal]]
+Default (datetime): "2006-01-02 15:04:05"
+--unix: 1704067200
+--unix-ms: 1704067200000
+--rfc3339: 2024-01-01T00:00:00Z
+--rfc1123: Mon, 02 Jan 2006 15:04:05 MST
+
+[[.Bold]][[.Underline]]Examples[[.Normal]]
+chtimes /tmp/file "2024-01-01 12:34:56" "2024-01-01 12:35:56"
+chtimes --unix /tmp/file 1704112496 1704112556
+chtimes --unix-ms /tmp/file 1704112496000 1704112556000
+chtimes --rfc3339 /tmp/file 2024-01-01T12:34:56Z 2024-01-01T12:35:56Z
+chtimes --rfc1123 /tmp/file "Mon, 02 Jan 2006 15:04:05 MST" "Mon, 02 Jan 2006 16:04:05 MST"
+`
+
 	catHelp = `[[.Bold]]Command:[[.Normal]] cat <remote path> 
 [[.Bold]]About:[[.Normal]] Cat a remote file to stdout.`
 

client/command/privilege/commands.go

@@ -150,11 +150,15 @@ func Commands(con *console.SliverClient) []*cobra.Command {
 	}
 	flags.Bind("", false, chtimesCmd, func(f *pflag.FlagSet) {
 		f.Int64P("timeout", "t", flags.DefaultTimeout, "grpc timeout in seconds")
+		f.Bool("unix", false, "interpret timestamps as Unix seconds")
+		f.Bool("unix-ms", false, "interpret timestamps as Unix milliseconds")
+		f.Bool("rfc3339", false, "interpret timestamps as RFC3339")
+		f.Bool("rfc1123", false, "interpret timestamps as RFC1123")
 	})
 	carapace.Gen(chtimesCmd).PositionalCompletion(
 		carapace.ActionValues().Usage("path to file to change access timestamps"),
-		carapace.ActionValues().Usage("last accessed time in DateTime format, i.e. 2006-01-02 15:04:05"),
-		carapace.ActionValues().Usage("last modified time in DateTime format, i.e. 2006-01-02 15:04:05"),
+		carapace.ActionValues().Usage("last accessed time (default: 2006-01-02 15:04:05; see --help for format flags)"),
+		carapace.ActionValues().Usage("last modified time (default: 2006-01-02 15:04:05; see --help for format flags)"),
 	)
 
 	getprivsCmd := &cobra.Command{

client/core/socks.go

@@ -31,6 +31,7 @@ import (
 	"github.com/bishopfox/sliver/protobuf/rpcpb"
 	"github.com/bishopfox/sliver/protobuf/sliverpb"
 	"github.com/bishopfox/sliver/util/leaky"
+	"golang.org/x/time/rate"
 )
 
 var (
@@ -210,6 +211,8 @@ const leakyBufSize = 4108 // data.len(2) + hmacsha1(10) + data(4096)
 var leakyBuf = leaky.NewLeakyBuf(2048, leakyBufSize)
 
 func connect(conn net.Conn, stream rpcpb.SliverRPC_SocksProxyClient, frame *sliverpb.SocksData) {
+	// Client Rate Limiter: 10 operations per second, burst of 1
+	limiter := rate.NewLimiter(rate.Limit(10), 1)
 
 	SocksConnPool.Store(frame.TunnelID, conn)
 
@@ -241,6 +244,11 @@ func connect(conn net.Conn, stream rpcpb.SliverRPC_SocksProxyClient, frame *sliv
 			return
 		}
 		if n > 0 {
+			if err := limiter.Wait(context.Background()); err != nil {
+				log.Printf("[socks] rate limiter error: %s", err)
+				return
+			}
+
 			frame.Data = buff[:n]
 			frame.Sequence = ToImplantSequence
 			log.Printf("[socks] (User to Client) to Server to agent  Data Sequence %d , Data Size %d \n", ToImplantSequence, len(frame.Data))

implant/sliver/handlers/tunnel_handlers/socks_handler.go

@@ -36,16 +36,29 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+const (
+	inactivityCheckInterval = 4 * time.Second
+	inactivityTimeout       = 15 * time.Second
+)
+
 type socksTunnelPool struct {
-	tunnels *sync.Map // map[uint64]chan []byte
+	tunnels      *sync.Map // map[uint64]chan []byte
+	lastActivity *sync.Map // map[uint64]time.Time
 }
 
 var socksTunnels = socksTunnelPool{
-	tunnels: &sync.Map{},
+	tunnels:      &sync.Map{},
+	lastActivity: &sync.Map{},
 }
 
 var socksServer *socks5.Server
 
+// Initialize socks server
+func init() {
+	socksServer = socks5.NewServer()
+	socksTunnels.startCleanupMonitor()
+}
+
 func SocksReqHandler(envelope *sliverpb.Envelope, connection *transports.Connection) {
 	socksData := &sliverpb.SocksData{}
 	err := proto.Unmarshal(envelope.Data, socksData)
@@ -55,9 +68,26 @@ func SocksReqHandler(envelope *sliverpb.Envelope, connection *transports.Connect
 		// {{end}}
 		return
 	}
+	time.Sleep(10 * time.Millisecond) // Necessary delay
+
+	// Check early to see if this is a close request from server
+	if socksData.CloseConn {
+		if tunnel, ok := socksTunnels.tunnels.LoadAndDelete(socksData.TunnelID); ok {
+			if ch, ok := tunnel.(chan []byte); ok {
+				close(ch)
+			}
+		}
+		socksTunnels.lastActivity.Delete(socksData.TunnelID)
+		return
+	}
+
 	if socksData.Data == nil {
 		return
 	}
+
+	// Record activity as soon as we get data for this tunnel
+	socksTunnels.recordActivity(socksData.TunnelID)
+
 	// {{if .Config.Debug}}
 	log.Printf("[socks] User to Client to (server to implant) Data Sequence %d, Data Size %d\n", socksData.Sequence, len(socksData.Data))
 	// {{end}}
@@ -70,8 +100,6 @@ func SocksReqHandler(envelope *sliverpb.Envelope, connection *transports.Connect
 		socksServer = socks5.NewServer(
 			socks5.WithAuthMethods([]socks5.Authenticator{auth}),
 		)
-	} else {
-		socksServer = socks5.NewServer()
 	}
 
 	// {{if .Config.Debug}}
@@ -80,7 +108,7 @@ func SocksReqHandler(envelope *sliverpb.Envelope, connection *transports.Connect
 
 	// init tunnel
 	if tunnel, ok := socksTunnels.tunnels.Load(socksData.TunnelID); !ok {
-		tunnelChan := make(chan []byte, 10)
+		tunnelChan := make(chan []byte, 100) // Buffered channel for 100 messages
 		socksTunnels.tunnels.Store(socksData.TunnelID, tunnelChan)
 		tunnelChan <- socksData.Data
 		err := socksServer.ServeConn(&socks{stream: socksData, conn: connection})
@@ -88,9 +116,12 @@ func SocksReqHandler(envelope *sliverpb.Envelope, connection *transports.Connect
 			// {{if .Config.Debug}}
 			log.Printf("[socks] Failed to serve connection: %v", err)
 			// {{end}}
+			// Cleanup on serve failure
+			socksTunnels.tunnels.Delete(socksData.TunnelID)
 			return
 		}
 	} else {
+		// Will block when channel is full
 		tunnel.(chan []byte) <- socksData.Data
 	}
 }
@@ -105,16 +136,22 @@ type socks struct {
 }
 
 func (s *socks) Read(b []byte) (n int, err error) {
+	time.Sleep(10 * time.Millisecond) // Necessary delay
+
 	channel, ok := socksTunnels.tunnels.Load(s.stream.TunnelID)
 	if !ok {
 		return 0, errors.New("[socks] invalid tunnel id")
 	}
 
+	socksTunnels.recordActivity(s.stream.TunnelID)
 	data := <-channel.(chan []byte)
 	return copy(b, data), nil
 }
 
 func (s *socks) Write(b []byte) (n int, err error) {
+	time.Sleep(10 * time.Millisecond) // Necessary delay
+
+	socksTunnels.recordActivity(s.stream.TunnelID)
 	data, err := proto.Marshal(&sliverpb.SocksData{
 		TunnelID: s.stream.TunnelID,
 		Data:     b,
@@ -136,12 +173,15 @@ func (s *socks) Write(b []byte) (n int, err error) {
 }
 
 func (s *socks) Close() error {
+	time.Sleep(10 * time.Millisecond) // Necessary delay
+
 	channel, ok := socksTunnels.tunnels.LoadAndDelete(s.stream.TunnelID)
 	if !ok {
 		return errors.New("[socks] can't close unknown channel")
 	}
 	close(channel.(chan []byte))
 
+	// Signal to server that we need to close this tunnel
 	data, err := proto.Marshal(&sliverpb.SocksData{
 		TunnelID:  s.stream.TunnelID,
 		CloseConn: true,
@@ -181,3 +221,38 @@ func (c *socks) SetReadDeadline(t time.Time) error {
 func (c *socks) SetWriteDeadline(t time.Time) error {
 	return nil
 }
+
+func (s *socksTunnelPool) recordActivity(tunnelID uint64) {
+	s.lastActivity.Store(tunnelID, time.Now())
+}
+
+// Periodically check for inactive tunnels and clean up
+func (s *socksTunnelPool) startCleanupMonitor() {
+	go func() {
+		ticker := time.NewTicker(inactivityCheckInterval)
+		defer ticker.Stop()
+
+		for range ticker.C {
+			s.tunnels.Range(func(key, value interface{}) bool {
+				tunnelID := key.(uint64)
+				lastActivityI, exists := s.lastActivity.Load(tunnelID)
+				if !exists {
+					// If no activity record exists, create one
+					s.recordActivity(tunnelID)
+					return true
+				}
+
+				lastActivity := lastActivityI.(time.Time)
+				if time.Since(lastActivity) > inactivityTimeout {
+					// Clean up the inactive tunnel
+					if ch, ok := value.(chan []byte); ok {
+						close(ch)
+					}
+					s.tunnels.Delete(tunnelID)
+					s.lastActivity.Delete(tunnelID)
+				}
+				return true
+			})
+		}
+	}()
+}

server/rpc/rpc-socks.go

@@ -20,8 +20,11 @@ package rpc
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/bishopfox/sliver/protobuf/commonpb"
 	"github.com/bishopfox/sliver/protobuf/rpcpb"
@@ -32,15 +35,16 @@ import (
 
 var (
 	// SessionID->Tunnels[TunnelID]->Tunnel->Cache map[uint64]*sliverpb.SocksData{}
-	toImplantCacheSocks = socksDataCache{mutex: &sync.RWMutex{}, cache: map[uint64]map[uint64]*sliverpb.SocksData{}}
+	toImplantCacheSocks = socksDataCache{mutex: &sync.RWMutex{}, cache: map[uint64]map[uint64]*sliverpb.SocksData{}, lastActivity: map[uint64]time.Time{}}
 
 	// SessionID->Tunnels[TunnelID]->Tunnel->Cache
-	fromImplantCacheSocks = socksDataCache{mutex: &sync.RWMutex{}, cache: map[uint64]map[uint64]*sliverpb.SocksData{}}
+	fromImplantCacheSocks = socksDataCache{mutex: &sync.RWMutex{}, cache: map[uint64]map[uint64]*sliverpb.SocksData{}, lastActivity: map[uint64]time.Time{}}
 )
 
 type socksDataCache struct {
-	mutex *sync.RWMutex
-	cache map[uint64]map[uint64]*sliverpb.SocksData
+	mutex        *sync.RWMutex
+	cache        map[uint64]map[uint64]*sliverpb.SocksData
+	lastActivity map[uint64]time.Time
 }
 
 func (c *socksDataCache) Add(tunnelID uint64, sequence uint64, tunnelData *sliverpb.SocksData) {
@@ -72,6 +76,7 @@ func (c *socksDataCache) DeleteTun(tunnelID uint64) {
 	defer c.mutex.Unlock()
 
 	delete(c.cache, tunnelID)
+	delete(c.lastActivity, tunnelID)
 }
 
 func (c *socksDataCache) DeleteSeq(tunnelID uint64, sequence uint64) {
@@ -85,69 +90,299 @@ func (c *socksDataCache) DeleteSeq(tunnelID uint64, sequence uint64) {
 	delete(c.cache[tunnelID], sequence)
 }
 
+func (c *socksDataCache) recordActivity(tunnelID uint64) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	c.lastActivity[tunnelID] = time.Now()
+}
+
 // Socks - Open an in-band port forward
+
+const (
+	writeTimeout            = 5 * time.Second
+	batchSize               = 100 // Maximum number of sequences to batch
+	inactivityCheckInterval = 4 * time.Second
+	inactivityTimeout       = 15 * time.Second
+	ToImplantTickerInterval = 10 * time.Millisecond // data going towards implant is usually smaller request data
+	ToClientTickerInterval  = 5 * time.Millisecond  // data coming back from implant is usually larger response data
+)
+
 func (s *Server) SocksProxy(stream rpcpb.SliverRPC_SocksProxyServer) error {
+	errChan := make(chan error, 2)
+	defer close(errChan)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	connDone := make(chan struct{})
+	defer close(connDone)
+
+	// Track all goroutines spawned for this session
+	var wg sync.WaitGroup
+	defer wg.Wait()
+
+	// Track all tunnels created for this session
+	activeTunnels := make(map[uint64]bool)
+	var tunnelMutex sync.Mutex
+
+	// Cleanup all tunnels on SocksProxy closure
+	defer func() {
+		tunnelMutex.Lock()
+		for tunnelID := range activeTunnels {
+			if tunnel := core.SocksTunnels.Get(tunnelID); tunnel != nil {
+				rpcLog.Infof("Cleaning up tunnel %d on proxy closure", tunnelID)
+				close(tunnel.FromImplant)
+				tunnel.Client = nil
+				s.CloseSocks(context.Background(), &sliverpb.Socks{TunnelID: tunnelID})
+			}
+		}
+		tunnelMutex.Unlock()
+	}()
+
 	for {
+		select {
+		case err := <-errChan:
+			rpcLog.Errorf("SocksProxy error: %v", err)
+			return err
+		default:
+		}
+
 		fromClient, err := stream.Recv()
 		if err == io.EOF {
-			break
+			return nil
 		}
-		//fmt.Println("Send Agent 1 ",fromClient.TunnelID,len(fromClient.Data))
 		if err != nil {
 			rpcLog.Warnf("Error on stream recv %s", err)
 			return rpcError(err)
 		}
-		tunnelLog.Debugf("Tunnel %d: From client %d byte(s)",
-			fromClient.TunnelID, len(fromClient.Data))
-		socks := core.SocksTunnels.Get(fromClient.TunnelID)
-		if socks == nil {
-			return nil
+
+		tunnelMutex.Lock()
+		activeTunnels[fromClient.TunnelID] = true // Mark this as an active tunnel
+		tunnelMutex.Unlock()
+
+		tunnel := core.SocksTunnels.Get(fromClient.TunnelID)
+		if tunnel == nil {
+			continue
 		}
-		if socks.Client == nil {
-			socks.Client = stream // Bind client to tunnel
-			// Send Client
-			go func() {
-				for tunnelData := range socks.FromImplant {
 
-					fromImplantCacheSocks.Add(fromClient.TunnelID, tunnelData.Sequence, tunnelData)
+		if tunnel.Client == nil {
+			tunnel.Client = stream
+			tunnel.FromImplant = make(chan *sliverpb.SocksData, 100) // Buffered channel for 100 messages
 
-					for recv, ok := fromImplantCacheSocks.Get(fromClient.TunnelID, socks.FromImplantSequence); ok; recv, ok = fromImplantCacheSocks.Get(fromClient.TunnelID, socks.FromImplantSequence) {
-						rpcLog.Debugf("[socks] agent to (Server To Client)  Data Sequence %d , Data Size %d ,Data %v\n", socks.FromImplantSequence, len(recv.Data), recv.Data)
-						socks.Client.Send(&sliverpb.SocksData{
-							CloseConn: recv.CloseConn,
-							TunnelID:  recv.TunnelID,
-							Data:      recv.Data,
-						})
-
-						fromImplantCacheSocks.DeleteSeq(fromClient.TunnelID, socks.FromImplantSequence)
-						socks.FromImplantSequence++
+			// Monitor tunnel goroutines for inactivity and cleanup
+			wg.Add(1)
+			go func(tunnelID uint64) {
+				defer wg.Done()
+				defer func() {
+					if r := recover(); r != nil {
+						rpcLog.Errorf("Recovered from panic in monitor: %v", r)
+						errChan <- fmt.Errorf("monitor goroutine panic: %v", r)
+						cancel() // Cancel context in case of a panic
 					}
+				}()
 
+				ticker := time.NewTicker(inactivityCheckInterval)
+				defer ticker.Stop()
+
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case <-connDone:
+						return
+					case <-ticker.C:
+						tunnel := core.SocksTunnels.Get(tunnelID)
+						if tunnel == nil || tunnel.Client == nil {
+							return
+						}
+						session := core.Sessions.Get(tunnel.SessionID)
+
+						// Check both caches for activity
+						toImplantCacheSocks.mutex.RLock()
+						fromImplantCacheSocks.mutex.RLock()
+						toLastActivity := toImplantCacheSocks.lastActivity[tunnelID]
+						fromLastActivity := fromImplantCacheSocks.lastActivity[tunnelID]
+						toImplantCacheSocks.mutex.RUnlock()
+						fromImplantCacheSocks.mutex.RUnlock()
+
+						// Clean up goroutine if both directions have hit the idle threshold
+						if time.Since(toLastActivity) > inactivityTimeout && time.Since(fromLastActivity) > inactivityTimeout {
+							s.CloseSocks(context.Background(), &sliverpb.Socks{TunnelID: tunnelID})
+							return
+						}
+
+						// Clean up goroutine if the client has disconnected early
+						if tunnel.Client == nil || session == nil {
+							s.CloseSocks(context.Background(), &sliverpb.Socks{TunnelID: tunnelID})
+							return
+						}
+
+					}
+				}
+			}(fromClient.TunnelID)
+
+			// Send Client
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				defer func() {
+					if r := recover(); r != nil {
+						rpcLog.Errorf("Recovered from panic in client sender: %v", r)
+						errChan <- fmt.Errorf("client sender panic: %v", r)
+						cancel() // Cancel context in case of a panic
+					}
+				}()
+
+				pendingData := make(map[uint64]*sliverpb.SocksData)
+				ticker := time.NewTicker(ToClientTickerInterval)
+				defer ticker.Stop()
+
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case <-connDone:
+						return
+					case tunnelData, ok := <-tunnel.FromImplant:
+						if !ok {
+							return
+						}
+
+						// Check if implant is requesting to close the tunnel
+						if tunnelData.CloseConn {
+							// Clean up the tunnel
+							s.CloseSocks(context.Background(), &sliverpb.Socks{TunnelID: fromClient.TunnelID})
+							return
+						}
+
+						sequence := tunnelData.Sequence
+						fromImplantCacheSocks.Add(fromClient.TunnelID, sequence, tunnelData)
+						pendingData[sequence] = tunnelData
+						fromImplantCacheSocks.recordActivity(fromClient.TunnelID)
+
+					case <-ticker.C:
+						if tunnel.Client == nil {
+							return
+						}
+						if len(pendingData) == 0 {
+							continue
+						}
+
+						expectedSequence := atomic.LoadUint64(&tunnel.FromImplantSequence)
+						processed := 0
+
+						// Perform Batching
+						for i := 0; i < batchSize && processed < len(pendingData); i++ {
+							data, exists := pendingData[expectedSequence]
+							if !exists {
+								break // Stop batching if we don't have the next expected sequence
+							}
+
+							func() {
+								defer func() {
+									if r := recover(); r != nil {
+										errChan <- fmt.Errorf("Client sender panic: %v", r)
+									}
+								}()
+
+								err := stream.Send(&sliverpb.SocksData{
+									CloseConn: data.CloseConn,
+									TunnelID:  data.TunnelID,
+									Data:      data.Data,
+								})
+
+								if err != nil {
+									rpcLog.Errorf("Send error: %v", err)
+									return
+								}
+
+								delete(pendingData, expectedSequence)
+								fromImplantCacheSocks.DeleteSeq(fromClient.TunnelID, expectedSequence)
+								atomic.AddUint64(&tunnel.FromImplantSequence, 1)
+								expectedSequence++
+								processed++
+							}()
+						}
+
+					}
+				}
+			}()
+
+			// Send Agent
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				defer func() {
+					if r := recover(); r != nil {
+						rpcLog.Errorf("Recovered from panic in agent sender: %v", r)
+						errChan <- fmt.Errorf("agent sender panic: %v", r)
+						cancel() // Cancel context in case of a panic
+					}
+				}()
+
+				pendingData := make(map[uint64]*sliverpb.SocksData)
+				ticker := time.NewTicker(ToImplantTickerInterval)
+				defer ticker.Stop()
+
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case <-connDone:
+						return
+					case <-ticker.C:
+						if tunnel.Client == nil {
+							return
+						}
+						sequence := atomic.LoadUint64(&tunnel.ToImplantSequence)
+
+						func() {
+							defer func() {
+								if r := recover(); r != nil {
+									rpcLog.Errorf("Recovered from processing panic: %v", r)
+								}
+							}()
+
+							for {
+								recv, ok := toImplantCacheSocks.Get(fromClient.TunnelID, sequence)
+								if !ok {
+									break
+								}
+
+								session := core.Sessions.Get(fromClient.Request.SessionID)
+								if session == nil {
+									rpcLog.Error("Session not found")
+									break
+								}
+
+								data, err := proto.Marshal(recv)
+								if err != nil {
+									rpcLog.Errorf("Failed to marshal data: %v", err)
+									continue
+								}
+
+								select {
+								case session.Connection.Send <- &sliverpb.Envelope{
+									Type: sliverpb.MsgSocksData,
+									Data: data,
+								}:
+									toImplantCacheSocks.DeleteSeq(fromClient.TunnelID, sequence)
+									atomic.AddUint64(&tunnel.ToImplantSequence, 1)
+									sequence++
+								case <-time.After(writeTimeout):
+									rpcLog.Error("Write timeout to implant")
+									pendingData[sequence] = recv
+									break
+								}
+							}
+						}()
+					}
 				}
 			}()
 		}
 
-		// Send Agent
-		go func() {
-			toImplantCacheSocks.Add(fromClient.TunnelID, fromClient.Sequence, fromClient)
-
-			for recv, ok := toImplantCacheSocks.Get(fromClient.TunnelID, socks.ToImplantSequence); ok; recv, ok = toImplantCacheSocks.Get(fromClient.TunnelID, socks.ToImplantSequence) {
-				rpcLog.Debugf("[socks] Client to (Server To Agent) Data Sequence %d ,  Data Size %d \n", socks.ToImplantSequence, len(fromClient.Data))
-				data, _ := proto.Marshal(recv)
-
-				session := core.Sessions.Get(fromClient.Request.SessionID)
-				session.Connection.Send <- &sliverpb.Envelope{
-					Type: sliverpb.MsgSocksData,
-					Data: data,
-				}
-
-				toImplantCacheSocks.DeleteSeq(fromClient.TunnelID, socks.ToImplantSequence)
-				socks.ToImplantSequence++
-			}
-
-		}()
+		toImplantCacheSocks.Add(fromClient.TunnelID, fromClient.Sequence, fromClient)
 	}
-	return nil
 }
 
 // CreateSocks5 - Create requests we close a Socks
@@ -169,11 +404,48 @@ func (s *Server) CreateSocks(ctx context.Context, req *sliverpb.Socks) (*sliverp
 
 // CloseSocks - Client requests we close a Socks
 func (s *Server) CloseSocks(ctx context.Context, req *sliverpb.Socks) (*commonpb.Empty, error) {
-	err := core.SocksTunnels.Close(req.TunnelID)
+	defer func() {
+		if r := recover(); r != nil {
+			rpcLog.Errorf("Recovered from panic in CloseSocks for tunnel %d: %v", req.TunnelID, r)
+		}
+	}()
+
+	tunnel := core.SocksTunnels.Get(req.TunnelID)
+	if tunnel != nil {
+		// Signal close to implant first
+		if session := core.Sessions.Get(tunnel.SessionID); session != nil {
+			data, _ := proto.Marshal(&sliverpb.SocksData{
+				TunnelID:  req.TunnelID,
+				CloseConn: true,
+			})
+			session.Connection.Send <- &sliverpb.Envelope{
+				Type: sliverpb.MsgSocksData,
+				Data: data,
+			}
+		}
+		time.Sleep(100 * time.Millisecond) // Delay to allow close message to be sent
+		tunnel.Client = nil                // Cleanup the tunnel
+		if tunnel.FromImplant != nil {
+			select {
+			case _, ok := <-tunnel.FromImplant:
+				if ok {
+					close(tunnel.FromImplant)
+				}
+			default:
+				close(tunnel.FromImplant)
+			}
+			tunnel.FromImplant = nil
+		}
+	}
+
+	// Clean up caches
 	toImplantCacheSocks.DeleteTun(req.TunnelID)
 	fromImplantCacheSocks.DeleteTun(req.TunnelID)
-	if err != nil {
-		return nil, rpcError(err)
+
+	// Remove from core tunnels last
+	if err := core.SocksTunnels.Close(req.TunnelID); err != nil {
+		rpcLog.Errorf("Error closing tunnel %d: %v", req.TunnelID, err)
 	}
+
 	return &commonpb.Empty{}, nil
 }