From 58b881c22eb8fecc16f6afa301031b657e659bb1 Mon Sep 17 00:00:00 2001 From: namay26 Date: Tue, 10 Jun 2025 19:31:54 +0530 Subject: [PATCH 1/8] Add initial passthrough implementation --- config/rules.yaml | 3 ++ protocols/protocols.go | 3 ++ protocols/tcp/passthrough.go | 91 ++++++++++++++++++++++++++++++++++++ rules/rules.go | 9 ++-- 4 files changed, 103 insertions(+), 3 deletions(-) create mode 100644 protocols/tcp/passthrough.go diff --git a/config/rules.yaml b/config/rules.yaml index 6b5bb4c..0b9a653 100644 --- a/config/rules.yaml +++ b/config/rules.yaml @@ -35,6 +35,9 @@ rules: - match: tcp dst port 27017 type: conn_handler target: mongodb + - match: tcp dst port 9889 + type: passthrough + target: passthrough # Will switch to host:ip - match: tcp type: conn_handler target: tcp diff --git a/protocols/protocols.go b/protocols/protocols.go index d43b3ca..f41842e 100644 --- a/protocols/protocols.go +++ b/protocols/protocols.go @@ -72,6 +72,9 @@ func MapTCPProtocolHandlers(log interfaces.Logger, h interfaces.Honeypot) map[st protocolHandlers["mongodb"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { return tcp.HandleMongoDB(ctx, conn, md, log, h) } + protocolHandlers["passthrough"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { + return tcp.HandlePassThrough(ctx, conn, md, log, h) + } protocolHandlers["tcp"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { snip, bufConn, err := Peek(conn, 4) if err != nil { diff --git a/protocols/tcp/passthrough.go b/protocols/tcp/passthrough.go new file mode 100644 index 0000000..9e38902 --- /dev/null +++ b/protocols/tcp/passthrough.go @@ -0,0 +1,91 @@ +package tcp + +import ( + "context" + "fmt" + "io" + "log/slog" + "net" + + "github.com/mushorg/glutton/connection" + "github.com/mushorg/glutton/producer" + "github.com/mushorg/glutton/protocols/interfaces" +) + +type parsedPassThrough struct { + Direction string `json:"direction,omitempty"` + Payload []byte `json:"payload,omitempty"` + PayloadHash string `json:"payload_hash,omitempty"` +} + +type passThroughServer struct { + events []parsedPassThrough + target string +} + +// Dial to the source ip, acting as a proxy between the client and real source by piping the data back and forth w/o interfering w it. +func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot) error { + var err error + defer func() { + if err := h.ProduceTCP("passthrough", conn, md, nil, nil); err != nil { + logger.Error("failed to produce passthrough message", producer.ErrAttr(err)) + } + if err := conn.Close(); err != nil { + logger.Error("failed to close incoming connection", slog.String("handler", "passthrough"), producer.ErrAttr(err)) + } + }() + + srcAddr := conn.RemoteAddr().String() + + // Still figuring out on this. + // targetIP := conn.LocalAddr().(*net.TCPAddr).IP.String() + // targetPort := md.TargetPort + // destAddr := fmt.Sprintf("%s:%d", targetIP, targetPort) + + // Hardcoded for now + destAddr := "127.0.0.1:5000" + + fmt.Println("dst", destAddr, " src", srcAddr) + if destAddr == "" { + logger.Error("no target defined", slog.String("handler", "passthrough")) + return nil + } + + targetConn, err := net.Dial("tcp", string(destAddr)) + if err != nil { + logger.Error("failed to connect to the target", slog.String("handler", "passthrough"), slog.String("target", string(destAddr)), producer.ErrAttr(err)) + return nil + } + defer targetConn.Close() + + logger.Info("starting passthrough", slog.String("source", srcAddr), slog.String("target", string(destAddr)), slog.String("handler", "passthrough")) + + errChan := make(chan error, 2) + + // Source to target + go func() { + _, err := io.Copy(targetConn, conn) + errChan <- err + }() + + // Target to source + go func() { + _, err := io.Copy(conn, targetConn) + errChan <- err + }() + + // When either of the error is returned or no more data is left to be sent, the go routines exit. + select { + case err := <-errChan: + if err != nil && err != io.EOF { + logger.Error("transfer error", producer.ErrAttr(err)) + return err + } + case <-ctx.Done(): + logger.Info("context cancelled") + return ctx.Err() + } + + logger.Info("Passthrough completed successfully") + return nil +} diff --git a/rules/rules.go b/rules/rules.go index 0b4d595..13707dd 100644 --- a/rules/rules.go +++ b/rules/rules.go @@ -18,6 +18,7 @@ type RuleType int const ( UserConnHandler RuleType = iota Drop + Passthrough ) type Config struct { @@ -32,7 +33,7 @@ type Rule struct { Name string `yaml:"name,omitempty"` isInit bool - ruleType RuleType + RuleType RuleType index int matcher *pcap.BPF } @@ -59,9 +60,11 @@ func (rule *Rule) init(idx int) error { switch rule.Type { case "conn_handler": - rule.ruleType = UserConnHandler + rule.RuleType = UserConnHandler + case "passthrough": + rule.RuleType = Passthrough case "drop": - rule.ruleType = Drop + rule.RuleType = Drop default: return fmt.Errorf("unknown rule type: %s", rule.Type) } From 84d3ba6f2702a234b451ef5e9f22a2dbc471b96b Mon Sep 17 00:00:00 2001 From: namay26 Date: Thu, 26 Jun 2025 13:37:24 +0530 Subject: [PATCH 2/8] Add destination routing logic --- protocols/tcp/passthrough.go | 46 ++++++++++++++++++++++++++---------- 1 file changed, 34 insertions(+), 12 deletions(-) diff --git a/protocols/tcp/passthrough.go b/protocols/tcp/passthrough.go index 9e38902..a72ef8c 100644 --- a/protocols/tcp/passthrough.go +++ b/protocols/tcp/passthrough.go @@ -37,15 +37,11 @@ func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadat srcAddr := conn.RemoteAddr().String() - // Still figuring out on this. - // targetIP := conn.LocalAddr().(*net.TCPAddr).IP.String() - // targetPort := md.TargetPort - // destAddr := fmt.Sprintf("%s:%d", targetIP, targetPort) + targetIP := conn.LocalAddr() + destAddr := fmt.Sprintf("%s", targetIP) - // Hardcoded for now - destAddr := "127.0.0.1:5000" + fmt.Println("src : ", srcAddr, ", dest : ", destAddr) - fmt.Println("dst", destAddr, " src", srcAddr) if destAddr == "" { logger.Error("no target defined", slog.String("handler", "passthrough")) return nil @@ -64,14 +60,40 @@ func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadat // Source to target go func() { - _, err := io.Copy(targetConn, conn) - errChan <- err + buf := make([]byte, 4096) + for { + n, err := conn.Read(buf) + if err != nil { + errChan <- err + return + } + if n > 0 { + logger.Info("source to target", slog.String("payload", string(buf[:n]))) + if _, err := targetConn.Write(buf[:n]); err != nil { + errChan <- err + return + } + } + } }() - // Target to source go func() { - _, err := io.Copy(conn, targetConn) - errChan <- err + buf := make([]byte, 4096) + for { + n, err := targetConn.Read(buf) + if err != nil { + errChan <- err + return + } + if n > 0 { + logger.Info("target to source", slog.String("payload", string(buf[:n]))) + if _, err := conn.Write(buf[:n]); err != nil { + errChan <- err + return + } + } + + } }() // When either of the error is returned or no more data is left to be sent, the go routines exit. From 50bec0eabb6c1d66ee480736d8cdcfbb74dafbaf Mon Sep 17 00:00:00 2001 From: namay26 Date: Tue, 8 Jul 2025 19:05:27 +0530 Subject: [PATCH 3/8] Add traffic capture config for passthrough and host:port target for dest --- config/config.yaml | 3 +++ config/rules.yaml | 2 +- glutton.go | 23 ++++++++++++----- protocols/protocols.go | 8 +++++- protocols/tcp/passthrough.go | 50 ++++++++++++++++++++++++++++-------- 5 files changed, 67 insertions(+), 19 deletions(-) diff --git a/config/config.yaml b/config/config.yaml index 222deb7..c8c4219 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -24,3 +24,6 @@ producers: conn_timeout: 45 max_tcp_payload: 4096 + +capture_traffic: + enabled: false \ No newline at end of file diff --git a/config/rules.yaml b/config/rules.yaml index 0b9a653..83e598f 100644 --- a/config/rules.yaml +++ b/config/rules.yaml @@ -37,7 +37,7 @@ rules: target: mongodb - match: tcp dst port 9889 type: passthrough - target: passthrough # Will switch to host:ip + target: 127.0.0.1:9889 # Can use hostip:port for the required destination. - match: tcp type: conn_handler target: tcp diff --git a/glutton.go b/glutton.go index c6ef637..619f414 100644 --- a/glutton.go +++ b/glutton.go @@ -222,12 +222,23 @@ func (g *Glutton) tcpListen() { g.Logger.Error("Failed to set connection timeout", producer.ErrAttr(err)) } - if hfunc, ok := g.tcpProtocolHandlers[rule.Target]; ok { - go func() { - if err := hfunc(g.ctx, conn, md); err != nil { - g.Logger.Error("Failed to handle TCP connection", producer.ErrAttr(err), slog.String("handler", rule.Target)) - } - }() + // If hostip:port is used as target, pass on to the passthrough handler. + if host, port, err := net.SplitHostPort(rule.Target); err == nil && host != "" && port != "" { + if hfunc, ok := g.tcpProtocolHandlers["passthrough"]; ok { + go func() { + if err := hfunc(g.ctx, conn, md); err != nil { + g.Logger.Error("Failed to handle TCP passthrough", producer.ErrAttr(err), slog.String("handler", "Passthrough")) + } + }() + } + } else { + if hfunc, ok := g.tcpProtocolHandlers[rule.Target]; ok { + go func() { + if err := hfunc(g.ctx, conn, md); err != nil { + g.Logger.Error("Failed to handle TCP connection", producer.ErrAttr(err), slog.String("handler", rule.Target)) + } + }() + } } } } diff --git a/protocols/protocols.go b/protocols/protocols.go index f41842e..2f0de84 100644 --- a/protocols/protocols.go +++ b/protocols/protocols.go @@ -12,6 +12,7 @@ import ( "github.com/mushorg/glutton/protocols/interfaces" "github.com/mushorg/glutton/protocols/tcp" "github.com/mushorg/glutton/protocols/udp" + "github.com/spf13/viper" ) type TCPHandlerFunc func(ctx context.Context, conn net.Conn, md connection.Metadata) error @@ -72,8 +73,13 @@ func MapTCPProtocolHandlers(log interfaces.Logger, h interfaces.Honeypot) map[st protocolHandlers["mongodb"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { return tcp.HandleMongoDB(ctx, conn, md, log, h) } + var capture bool + if viper.GetBool("capture_traffic.enabled") { + log.Info("Capturing traffic enabled.") + capture = true + } protocolHandlers["passthrough"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { - return tcp.HandlePassThrough(ctx, conn, md, log, h) + return tcp.HandlePassThrough(ctx, conn, md, log, h, capture) } protocolHandlers["tcp"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { snip, bufConn, err := Peek(conn, 4) diff --git a/protocols/tcp/passthrough.go b/protocols/tcp/passthrough.go index a72ef8c..8fd1854 100644 --- a/protocols/tcp/passthrough.go +++ b/protocols/tcp/passthrough.go @@ -2,6 +2,7 @@ package tcp import ( "context" + "crypto/sha256" "fmt" "io" "log/slog" @@ -15,19 +16,51 @@ import ( type parsedPassThrough struct { Direction string `json:"direction,omitempty"` Payload []byte `json:"payload,omitempty"` - PayloadHash string `json:"payload_hash,omitempty"` + PayloadHash string `json:"payload_hash,omitempty"` // Used for easier identification, can remove } type passThroughServer struct { events []parsedPassThrough + conn net.Conn target string + source string +} + +func (srv *passThroughServer) recordEvent(dir string, buf []byte, capture bool) { + if !capture { + return + } + hash := sha256.Sum256(buf) + + payload := append([]byte(nil), buf...) // defensive copy + + srv.events = append(srv.events, parsedPassThrough{ + Direction: dir, + Payload: payload, + PayloadHash: fmt.Sprintf("%x", hash[:]), + }) } // Dial to the source ip, acting as a proxy between the client and real source by piping the data back and forth w/o interfering w it. -func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot) error { +func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot, capture bool) error { var err error + + srcAddr := conn.RemoteAddr().String() + destAddr := md.Rule.Target + + server := &passThroughServer{ + events: []parsedPassThrough{}, + conn: conn, + target: destAddr, + source: srcAddr, + } + defer func() { - if err := h.ProduceTCP("passthrough", conn, md, nil, nil); err != nil { + var events []parsedPassThrough + if capture { + events = server.events + } + if err := h.ProduceTCP("passthrough", conn, md, nil, events); err != nil { logger.Error("failed to produce passthrough message", producer.ErrAttr(err)) } if err := conn.Close(); err != nil { @@ -35,19 +68,12 @@ func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadat } }() - srcAddr := conn.RemoteAddr().String() - - targetIP := conn.LocalAddr() - destAddr := fmt.Sprintf("%s", targetIP) - - fmt.Println("src : ", srcAddr, ", dest : ", destAddr) - if destAddr == "" { logger.Error("no target defined", slog.String("handler", "passthrough")) return nil } - targetConn, err := net.Dial("tcp", string(destAddr)) + targetConn, err := net.Dial("tcp", destAddr) if err != nil { logger.Error("failed to connect to the target", slog.String("handler", "passthrough"), slog.String("target", string(destAddr)), producer.ErrAttr(err)) return nil @@ -69,6 +95,7 @@ func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadat } if n > 0 { logger.Info("source to target", slog.String("payload", string(buf[:n]))) + server.recordEvent("source->target", buf[:n], capture) if _, err := targetConn.Write(buf[:n]); err != nil { errChan <- err return @@ -87,6 +114,7 @@ func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadat } if n > 0 { logger.Info("target to source", slog.String("payload", string(buf[:n]))) + server.recordEvent("target->source", buf[:n], capture) if _, err := conn.Write(buf[:n]); err != nil { errChan <- err return From de76bf41efd5842304212347908751b57d995a21 Mon Sep 17 00:00:00 2001 From: namay26 Date: Mon, 4 Aug 2025 00:07:47 +0530 Subject: [PATCH 4/8] Add tests and improve function structure --- glutton.go | 19 ++- protocols/protocols.go | 8 +- protocols/tcp/passthrough.go | 125 ++++++++++----- protocols/tcp/passthrough_test.go | 255 ++++++++++++++++++++++++++++++ 4 files changed, 349 insertions(+), 58 deletions(-) create mode 100644 protocols/tcp/passthrough_test.go diff --git a/glutton.go b/glutton.go index 619f414..a7c3707 100644 --- a/glutton.go +++ b/glutton.go @@ -222,22 +222,21 @@ func (g *Glutton) tcpListen() { g.Logger.Error("Failed to set connection timeout", producer.ErrAttr(err)) } - // If hostip:port is used as target, pass on to the passthrough handler. - if host, port, err := net.SplitHostPort(rule.Target); err == nil && host != "" && port != "" { + if rule.Type == "passthrough" { if hfunc, ok := g.tcpProtocolHandlers["passthrough"]; ok { go func() { if err := hfunc(g.ctx, conn, md); err != nil { g.Logger.Error("Failed to handle TCP passthrough", producer.ErrAttr(err), slog.String("handler", "Passthrough")) } }() - } - } else { - if hfunc, ok := g.tcpProtocolHandlers[rule.Target]; ok { - go func() { - if err := hfunc(g.ctx, conn, md); err != nil { - g.Logger.Error("Failed to handle TCP connection", producer.ErrAttr(err), slog.String("handler", rule.Target)) - } - }() + } else { + if hfunc, ok := g.tcpProtocolHandlers[rule.Target]; ok { + go func() { + if err := hfunc(g.ctx, conn, md); err != nil { + g.Logger.Error("Failed to handle TCP connection", producer.ErrAttr(err), slog.String("handler", rule.Target)) + } + }() + } } } } diff --git a/protocols/protocols.go b/protocols/protocols.go index 2f0de84..f41842e 100644 --- a/protocols/protocols.go +++ b/protocols/protocols.go @@ -12,7 +12,6 @@ import ( "github.com/mushorg/glutton/protocols/interfaces" "github.com/mushorg/glutton/protocols/tcp" "github.com/mushorg/glutton/protocols/udp" - "github.com/spf13/viper" ) type TCPHandlerFunc func(ctx context.Context, conn net.Conn, md connection.Metadata) error @@ -73,13 +72,8 @@ func MapTCPProtocolHandlers(log interfaces.Logger, h interfaces.Honeypot) map[st protocolHandlers["mongodb"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { return tcp.HandleMongoDB(ctx, conn, md, log, h) } - var capture bool - if viper.GetBool("capture_traffic.enabled") { - log.Info("Capturing traffic enabled.") - capture = true - } protocolHandlers["passthrough"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { - return tcp.HandlePassThrough(ctx, conn, md, log, h, capture) + return tcp.HandlePassThrough(ctx, conn, md, log, h) } protocolHandlers["tcp"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { snip, bufConn, err := Peek(conn, 4) diff --git a/protocols/tcp/passthrough.go b/protocols/tcp/passthrough.go index 8fd1854..8435e9c 100644 --- a/protocols/tcp/passthrough.go +++ b/protocols/tcp/passthrough.go @@ -3,6 +3,7 @@ package tcp import ( "context" "crypto/sha256" + "encoding/hex" "fmt" "io" "log/slog" @@ -11,6 +12,7 @@ import ( "github.com/mushorg/glutton/connection" "github.com/mushorg/glutton/producer" "github.com/mushorg/glutton/protocols/interfaces" + "github.com/spf13/viper" ) type parsedPassThrough struct { @@ -26,6 +28,44 @@ type passThroughServer struct { source string } +// checks whether the payload can be converted to text, to prevent expensive hex coding. +func (srv *passThroughServer) isLikelyText(data []byte) bool { + if len(data) == 0 { + return false + } + + printable := 0 + for _, b := range data { + if b >= 32 && b <= 126 || b == '\n' || b == '\r' || b == '\t' { + printable++ + } + } + + return (printable*100)/len(data) > 80 // threshold value --> 80% +} + +// logs the payload hex or payload text. +func (srv *passThroughServer) logPayload(direction string, data []byte, logger interfaces.Logger) { + if len(data) == 0 { + return + } + + fields := []any{ + slog.String("direction", direction), + slog.Int("length", len(data)), + slog.String("sha256", fmt.Sprintf("%x", sha256.Sum256(data))), + } + + if srv.isLikelyText(data) { + fields = append(fields, slog.String("payload", string(data))) + } else { + fields = append(fields, slog.String("hex", hex.EncodeToString(data))) + } + + logger.Info("payload_transferred", fields...) +} + +// records the events in the server func (srv *passThroughServer) recordEvent(dir string, buf []byte, capture bool) { if !capture { return @@ -41,8 +81,44 @@ func (srv *passThroughServer) recordEvent(dir string, buf []byte, capture bool) }) } +// pipeBidirectional handles data transfer between the two connections +func pipeBidirectional(ctx context.Context, src, dst net.Conn, server *passThroughServer, logger interfaces.Logger, capture bool, errChan chan error) { + buf := make([]byte, 4096) + direction := getDirection(src, dst) + for { + select { + case <-ctx.Done(): + errChan <- ctx.Err() + return + default: + n, err := src.Read(buf) + if err != nil { + errChan <- err + return + } + + if n > 0 { + server.logPayload(direction, buf[:n], logger) + server.recordEvent(direction, buf[:n], capture) + + if _, err := dst.Write(buf[:n]); err != nil { + errChan <- err + return + } + } + } + } +} + +// getDirection returns the direction as a string +func getDirection(src, dst net.Conn) string { + srcAddr := src.RemoteAddr().String() + dstAddr := dst.RemoteAddr().String() + return fmt.Sprintf("%s -> %s", srcAddr, dstAddr) +} + // Dial to the source ip, acting as a proxy between the client and real source by piping the data back and forth w/o interfering w it. -func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot, capture bool) error { +func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot) error { var err error srcAddr := conn.RemoteAddr().String() @@ -55,6 +131,11 @@ func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadat source: srcAddr, } + var capture bool + if viper.GetBool("capture_traffic.enabled") { + capture = true + } + defer func() { var events []parsedPassThrough if capture { @@ -84,47 +165,9 @@ func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadat errChan := make(chan error, 2) - // Source to target - go func() { - buf := make([]byte, 4096) - for { - n, err := conn.Read(buf) - if err != nil { - errChan <- err - return - } - if n > 0 { - logger.Info("source to target", slog.String("payload", string(buf[:n]))) - server.recordEvent("source->target", buf[:n], capture) - if _, err := targetConn.Write(buf[:n]); err != nil { - errChan <- err - return - } - } - } - }() - - go func() { - buf := make([]byte, 4096) - for { - n, err := targetConn.Read(buf) - if err != nil { - errChan <- err - return - } - if n > 0 { - logger.Info("target to source", slog.String("payload", string(buf[:n]))) - server.recordEvent("target->source", buf[:n], capture) - if _, err := conn.Write(buf[:n]); err != nil { - errChan <- err - return - } - } - - } - }() + go pipeBidirectional(ctx, conn, targetConn, server, logger, capture, errChan) // source to target + go pipeBidirectional(ctx, targetConn, conn, server, logger, capture, errChan) // target to source - // When either of the error is returned or no more data is left to be sent, the go routines exit. select { case err := <-errChan: if err != nil && err != io.EOF { diff --git a/protocols/tcp/passthrough_test.go b/protocols/tcp/passthrough_test.go new file mode 100644 index 0000000..c444b19 --- /dev/null +++ b/protocols/tcp/passthrough_test.go @@ -0,0 +1,255 @@ +package tcp + +import ( + "context" + "crypto/rand" + "io" + "net" + "testing" + + "github.com/mushorg/glutton/protocols/interfaces" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +type MockLogger struct { + mock.Mock +} + +func (m *MockLogger) Info(msg string, attrs ...interface{}) { + m.Called(msg, attrs) +} + +func (m *MockLogger) Debug(msg string, attrs ...interface{}) { + m.Called(msg, attrs) +} + +func (m *MockLogger) Error(msg string, attrs ...interface{}) { + m.Called(msg, attrs) +} + +func (m *MockLogger) Warn(msg string, attrs ...interface{}) { + m.Called(msg, attrs) +} + +func TestIsLikelyText(t *testing.T) { + tests := []struct { + name string + input []byte + expected bool + }{ + { + name: "Empty input", + input: []byte(""), + expected: false, + }, + { + name: "Simple ASCII text", + input: []byte("This is plain text"), + expected: true, + }, + { + name: "Text with whitespace", + input: []byte("Text with\nnewlines\tand tabs\r\n"), + expected: true, + }, + { + name: "Binary data", + input: []byte{0x01, 0x02, 0x03, 0x04, 0x05}, + expected: false, + }, + { + name: "Mixed content with few non-printable", + input: []byte("Text\x00with\x01binary"), + expected: true, // checking threshold at 85.7% + }, + { + name: "Exactly 80% printable", + input: []byte("AAAA\x01"), // 4/5 = 80% + expected: false, + }, + { + name: "Just below 80% printable", + input: []byte("AAA\x01\x02"), // 3/5 = 60% + expected: false, + }, + } + + srv := &passThroughServer{} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + result := srv.isLikelyText(test.input) + require.Equal(t, test.expected, result, "unexpected result for test case: %s", test.name) + }) + } +} + +func TestRecording(t *testing.T) { + s := &passThroughServer{} + s.recordEvent("test", []byte("data"), true) + assert.Len(t, s.events, 1) +} + +func TestPipeBidirectional(t *testing.T) { + mockLogger := &MockLogger{} + mockLogger.On("Info", mock.Anything, mock.Anything).Return() + + mockServer := &passThroughServer{ + events: make([]parsedPassThrough, 0), + conn: nil, + target: "test-target:1234", + source: "test-source:5678", + } + + type args struct { + ctx context.Context + src net.Conn + dst net.Conn + server *passThroughServer + logger interfaces.Logger + capture bool + errChan chan error + } + tests := []struct { + name string + args args + setup func() (net.Conn, net.Conn) + wantErr bool + wantErrType error + verify func(t *testing.T, args args) + }{ + { + name: "successful data transfer with capture", + args: args{ + ctx: context.Background(), + server: mockServer, + logger: mockLogger, + capture: true, + errChan: make(chan error, 1), + }, + + setup: func() (net.Conn, net.Conn) { + client, server := net.Pipe() + go func() { + client.Write([]byte("test data")) + client.Close() + }() + return client, server + }, + verify: func(t *testing.T, args args) { + buf := make([]byte, 1024) + n, err := args.dst.Read(buf) + + require.NoError(t, err) + assert.Equal(t, "test data", string(buf[:n])) + + require.True(t, args.capture, "Capture should be enabled") + }, + }, + { + name: "read error from source", + args: args{ + ctx: context.Background(), + server: mockServer, + logger: mockLogger, + capture: false, + errChan: make(chan error, 1), + }, + setup: func() (net.Conn, net.Conn) { + client, server := net.Pipe() + client.Close() + return client, server + }, + wantErr: true, + wantErrType: io.EOF, + }, + { + name: "write error to destination", + args: args{ + ctx: context.Background(), + server: mockServer, + logger: mockLogger, + capture: false, + errChan: make(chan error, 1), + }, + setup: func() (net.Conn, net.Conn) { + client, server := net.Pipe() + server.Close() + return client, server + }, + wantErr: true, + }, + { + name: "zero byte read", + args: args{ + ctx: context.Background(), + server: mockServer, + logger: mockLogger, + capture: true, + errChan: make(chan error, 1), + }, + setup: func() (net.Conn, net.Conn) { + client, server := net.Pipe() + go func() { + client.Write([]byte{}) + client.Close() + }() + return client, server + }, + verify: func(t *testing.T, args args) { + assert.Empty(t, args.server.events) + }, + }, + { + name: "large data transfer", + args: args{ + ctx: context.Background(), + server: mockServer, + logger: mockLogger, + capture: true, + errChan: make(chan error, 1), + }, + setup: func() (net.Conn, net.Conn) { + client, server := net.Pipe() + largeData := make([]byte, 8192) + rand.Read(largeData) + go func() { + client.Write(largeData) + client.Close() + }() + return client, server + }, + verify: func(t *testing.T, args args) { + buf := make([]byte, 8192) + n, err := io.ReadFull(args.dst, buf) + require.NoError(t, err) + assert.Equal(t, 8192, n) + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if tt.setup != nil { + tt.args.src, tt.args.dst = tt.setup() + defer tt.args.src.Close() + defer tt.args.dst.Close() + } + + go pipeBidirectional( + tt.args.ctx, + tt.args.src, + tt.args.dst, + tt.args.server, + tt.args.logger, + tt.args.capture, + tt.args.errChan, + ) + + if tt.verify != nil { + tt.verify(t, tt.args) + } + }) + } +} From 797bb89575d3748088cedb744c52c35132e891df Mon Sep 17 00:00:00 2001 From: namay26 Date: Wed, 20 Aug 2025 23:21:43 +0530 Subject: [PATCH 5/8] Add io.copy and change to tcp_proxy --- config/rules.yaml | 2 +- glutton.go | 22 +++---- protocols/protocols.go | 2 +- protocols/tcp/passthrough.go | 98 ++++++++++++++++++------------- protocols/tcp/passthrough_test.go | 1 - rules/rules.go | 6 +- 6 files changed, 73 insertions(+), 58 deletions(-) diff --git a/config/rules.yaml b/config/rules.yaml index 83e598f..88ab2c2 100644 --- a/config/rules.yaml +++ b/config/rules.yaml @@ -36,7 +36,7 @@ rules: type: conn_handler target: mongodb - match: tcp dst port 9889 - type: passthrough + type: tcp_proxy target: 127.0.0.1:9889 # Can use hostip:port for the required destination. - match: tcp type: conn_handler diff --git a/glutton.go b/glutton.go index a7c3707..61e58e1 100644 --- a/glutton.go +++ b/glutton.go @@ -222,21 +222,21 @@ func (g *Glutton) tcpListen() { g.Logger.Error("Failed to set connection timeout", producer.ErrAttr(err)) } - if rule.Type == "passthrough" { - if hfunc, ok := g.tcpProtocolHandlers["passthrough"]; ok { + if rule.Type == "tcp_proxy" { + if hfunc, ok := g.tcpProtocolHandlers[rule.Type]; ok { go func() { if err := hfunc(g.ctx, conn, md); err != nil { - g.Logger.Error("Failed to handle TCP passthrough", producer.ErrAttr(err), slog.String("handler", "Passthrough")) + g.Logger.Error("Failed to handle TCP passthrough", producer.ErrAttr(err), slog.String("handler", "tcp_proxy")) + } + }() + } + } else { + if hfunc, ok := g.tcpProtocolHandlers[rule.Target]; ok { + go func() { + if err := hfunc(g.ctx, conn, md); err != nil { + g.Logger.Error("Failed to handle TCP connection", producer.ErrAttr(err), slog.String("handler", rule.Target)) } }() - } else { - if hfunc, ok := g.tcpProtocolHandlers[rule.Target]; ok { - go func() { - if err := hfunc(g.ctx, conn, md); err != nil { - g.Logger.Error("Failed to handle TCP connection", producer.ErrAttr(err), slog.String("handler", rule.Target)) - } - }() - } } } } diff --git a/protocols/protocols.go b/protocols/protocols.go index f41842e..7e8f6ba 100644 --- a/protocols/protocols.go +++ b/protocols/protocols.go @@ -72,7 +72,7 @@ func MapTCPProtocolHandlers(log interfaces.Logger, h interfaces.Honeypot) map[st protocolHandlers["mongodb"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { return tcp.HandleMongoDB(ctx, conn, md, log, h) } - protocolHandlers["passthrough"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { + protocolHandlers["tcp_proxy"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { return tcp.HandlePassThrough(ctx, conn, md, log, h) } protocolHandlers["tcp"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { diff --git a/protocols/tcp/passthrough.go b/protocols/tcp/passthrough.go index 8435e9c..232b9a7 100644 --- a/protocols/tcp/passthrough.go +++ b/protocols/tcp/passthrough.go @@ -6,8 +6,10 @@ import ( "encoding/hex" "fmt" "io" + "log" "log/slog" "net" + "time" "github.com/mushorg/glutton/connection" "github.com/mushorg/glutton/producer" @@ -28,6 +30,20 @@ type passThroughServer struct { source string } +type loggingWriter struct { + dst net.Conn + server *passThroughServer + logger interfaces.Logger + capture bool + dir string +} + +func (lw *loggingWriter) Write(p []byte) (int, error) { + lw.server.logPayload(lw.dir, p, lw.logger) + lw.server.recordEvent(lw.dir, p, lw.capture) + return lw.dst.Write(p) +} + // checks whether the payload can be converted to text, to prevent expensive hex coding. func (srv *passThroughServer) isLikelyText(data []byte) bool { if len(data) == 0 { @@ -82,32 +98,24 @@ func (srv *passThroughServer) recordEvent(dir string, buf []byte, capture bool) } // pipeBidirectional handles data transfer between the two connections -func pipeBidirectional(ctx context.Context, src, dst net.Conn, server *passThroughServer, logger interfaces.Logger, capture bool, errChan chan error) { - buf := make([]byte, 4096) +func pipeBidirectional(src, dst net.Conn, server *passThroughServer, logger interfaces.Logger, capture bool, errChan chan error) { direction := getDirection(src, dst) - for { - select { - case <-ctx.Done(): - errChan <- ctx.Err() - return - default: - n, err := src.Read(buf) - if err != nil { - errChan <- err - return - } - - if n > 0 { - server.logPayload(direction, buf[:n], logger) - server.recordEvent(direction, buf[:n], capture) - - if _, err := dst.Write(buf[:n]); err != nil { - errChan <- err - return - } - } - } - } + writer := &loggingWriter{dst: dst, server: server, logger: logger, capture: capture, dir: direction} + + // source to target + go func() { + _, err := io.Copy(writer, src) + errChan <- err + }() + + revDirection := getDirection(dst, src) + revWriter := &loggingWriter{dst: src, server: server, logger: logger, capture: capture, dir: revDirection} + + // target to source + go func() { + _, err := io.Copy(revWriter, dst) + errChan <- err + }() } // getDirection returns the direction as a string @@ -120,10 +128,23 @@ func getDirection(src, dst net.Conn) string { // Dial to the source ip, acting as a proxy between the client and real source by piping the data back and forth w/o interfering w it. func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot) error { var err error + handler := "tcp_proxy" srcAddr := conn.RemoteAddr().String() destAddr := md.Rule.Target + host, _, err := net.SplitHostPort(destAddr) + if err != nil { + logger.Error("invalid address format", producer.ErrAttr(err)) + return nil + } + + if ip := net.ParseIP(host); ip == nil { + if _, err := net.LookupHost(host); err != nil { + return fmt.Errorf("invalid host: %w", err) + } + } + server := &passThroughServer{ events: []parsedPassThrough{}, conn: conn, @@ -145,38 +166,33 @@ func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadat logger.Error("failed to produce passthrough message", producer.ErrAttr(err)) } if err := conn.Close(); err != nil { - logger.Error("failed to close incoming connection", slog.String("handler", "passthrough"), producer.ErrAttr(err)) + logger.Error("failed to close incoming connection", slog.String("handler", handler), producer.ErrAttr(err)) } }() if destAddr == "" { - logger.Error("no target defined", slog.String("handler", "passthrough")) + logger.Error("no target defined", slog.String("handler", handler)) return nil } - targetConn, err := net.Dial("tcp", destAddr) + timeout := 5 * time.Second + + targetConn, err := net.DialTimeout("tcp", destAddr, timeout) if err != nil { - logger.Error("failed to connect to the target", slog.String("handler", "passthrough"), slog.String("target", string(destAddr)), producer.ErrAttr(err)) + logger.Error("failed to connect to the target", slog.String("handler", handler), slog.String("target", string(destAddr)), producer.ErrAttr(err)) return nil } defer targetConn.Close() - logger.Info("starting passthrough", slog.String("source", srcAddr), slog.String("target", string(destAddr)), slog.String("handler", "passthrough")) + logger.Info("starting passthrough", slog.String("source", srcAddr), slog.String("target", string(destAddr)), slog.String("handler", handler)) errChan := make(chan error, 2) - go pipeBidirectional(ctx, conn, targetConn, server, logger, capture, errChan) // source to target - go pipeBidirectional(ctx, targetConn, conn, server, logger, capture, errChan) // target to source + go pipeBidirectional(conn, targetConn, server, logger, capture, errChan) - select { - case err := <-errChan: - if err != nil && err != io.EOF { - logger.Error("transfer error", producer.ErrAttr(err)) - return err - } - case <-ctx.Done(): - logger.Info("context cancelled") - return ctx.Err() + // wait for either side to close + if err := <-errChan; err != nil { + log.Printf("connection closed: %v", err) } logger.Info("Passthrough completed successfully") diff --git a/protocols/tcp/passthrough_test.go b/protocols/tcp/passthrough_test.go index c444b19..4f5569f 100644 --- a/protocols/tcp/passthrough_test.go +++ b/protocols/tcp/passthrough_test.go @@ -238,7 +238,6 @@ func TestPipeBidirectional(t *testing.T) { } go pipeBidirectional( - tt.args.ctx, tt.args.src, tt.args.dst, tt.args.server, diff --git a/rules/rules.go b/rules/rules.go index 13707dd..b28e234 100644 --- a/rules/rules.go +++ b/rules/rules.go @@ -18,7 +18,7 @@ type RuleType int const ( UserConnHandler RuleType = iota Drop - Passthrough + Tcp_Proxy ) type Config struct { @@ -61,8 +61,8 @@ func (rule *Rule) init(idx int) error { switch rule.Type { case "conn_handler": rule.RuleType = UserConnHandler - case "passthrough": - rule.RuleType = Passthrough + case "tcp_proxy": + rule.RuleType = Tcp_Proxy case "drop": rule.RuleType = Drop default: From 20d95ae40cacfa5314fdfe17edfc29648dfaf948 Mon Sep 17 00:00:00 2001 From: furusiyya Date: Sat, 25 Apr 2026 19:44:58 -0700 Subject: [PATCH 6/8] refactor: improve naming, logging, and connection handling --- .gitignore | 1 + config/config.yaml | 5 +- config/rules.yaml | 7 +- docs/configuration.md | 20 +- glutton.go | 4 +- mkdocs.yml | 2 +- protocols/protocols.go | 4 +- protocols/tcp/passthrough.go | 200 ----------- protocols/tcp/passthrough_test.go | 254 -------------- protocols/tcp/proxy_tcp.go | 536 ++++++++++++++++++++++++++++++ protocols/tcp/proxytcp_test.go | 423 +++++++++++++++++++++++ rules/rules.go | 59 +++- rules/rules_test.go | 107 ++++++ rules/test.yaml | 3 + 14 files changed, 1151 insertions(+), 474 deletions(-) delete mode 100644 protocols/tcp/passthrough.go delete mode 100644 protocols/tcp/passthrough_test.go create mode 100644 protocols/tcp/proxy_tcp.go create mode 100644 protocols/tcp/proxytcp_test.go diff --git a/.gitignore b/.gitignore index 16257c2..2646ced 100644 --- a/.gitignore +++ b/.gitignore @@ -37,3 +37,4 @@ poc/ # Dev .vscode +openspec/ diff --git a/config/config.yaml b/config/config.yaml index c8c4219..69c64ee 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -22,8 +22,9 @@ producers: auth: auth channel: test -conn_timeout: 45 -max_tcp_payload: 4096 +conn_timeout: 45 # idle I/O timeout in seconds for established connections. +max_tcp_payload: 4096 # bytes +dial_timeout: 5 # timeout in seconds for proxy target connection. capture_traffic: enabled: false \ No newline at end of file diff --git a/config/rules.yaml b/config/rules.yaml index 88ab2c2..d991241 100644 --- a/config/rules.yaml +++ b/config/rules.yaml @@ -36,8 +36,11 @@ rules: type: conn_handler target: mongodb - match: tcp dst port 9889 - type: tcp_proxy - target: 127.0.0.1:9889 # Can use hostip:port for the required destination. + type: proxy_tcp + target: 127.0.0.1:9889 + - match: tcp dst port 3306 + type: proxy_tcp + target: 127.0.0.1:3306 - match: tcp type: conn_handler target: tcp diff --git a/docs/configuration.md b/docs/configuration.md index 7d036a1..c5d711c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -13,8 +13,10 @@ This file holds the core settings for Glutton. Key configuration options include - **udp:** The UDP port for intercepted packets (default: `5001`). - **ssh:** Typically excluded from redirection to avoid interfering with SSH (default: `22`). - **interface:** The network interface Glutton listens on (default: `eth0`). -- **max_tcp_payload:** Maximum TCP payload size in bytes (default: `4096`). -- **conn_timeout:** The connection timeout duration in seconds (default: `45`). +- **conn_timeout:** Idle I/O timeout, in seconds, for established connections (default: `45`). +- **max_tcp_payload:** Maximum TCP payload size in bytes (default: `4096`). Proxy TCP uses this as the per-direction captured payload cap. +- **dial_timeout:** Timeout, in seconds, for opening outbound proxy TCP target connections (default: `5`). +- **capture_traffic.enabled:** Enables raw payload capture in logs and produced decoded events. When disabled, proxy TCP still forwards traffic and logs metadata, but raw payload bytes are omitted from decoded events. - **confpath:** The directory path where the configuration file resides. - **producers:** - **enabled**: Boolean flag to enable or disable logging/producer functionality. @@ -55,6 +57,10 @@ producers: conn_timeout: 45 max_tcp_payload: 4096 +dial_timeout: 5 + +capture_traffic: + enabled: false ``` ### config/rules.yaml @@ -63,8 +69,8 @@ This file defines the rules that Glutton uses to determine which protocol handle Key elements include: -- **type**: `conn_handler` to pass off to the appropriate protocol handler or `drop` to ignore packets. -- **target**: Indicates the protocol handler (e.g., "http", "ftp") to be used. +- **type**: `conn_handler` to pass off to the appropriate protocol handler, `proxy_tcp` to forward the TCP connection to an upstream target, or `drop` to ignore packets. +- **target**: For `conn_handler`, indicates the protocol handler (e.g., `http`, `ftp`) to use. For `proxy_tcp`, this must be the upstream target in `host:port` form. - **match**: Define criteria such as source IP ranges or destination ports to match incoming traffic, according to [BPF syntax](https://biot.com/capstats/bpf.html). Example rule: @@ -80,8 +86,14 @@ rules: - match: tcp dst port 6969 type: drop # drops any matching packets target: bittorrent + - name: Proxy TCP example + match: tcp dst port 9889 + type: proxy_tcp + target: 127.0.0.1:9889 ``` +`proxy_tcp` dials the configured `target` and forwards bytes in both directions between the incoming connection and the upstream service. Produced decoded events use the `proxy_tcp` protocol name and can include one captured payload entry per direction. Captured payloads are capped by `max_tcp_payload`; when a direction transfers more bytes than the cap, the decoded event is marked as truncated. + ## Configuration Loading Process Glutton uses the [Viper](https://github.com/spf13/viper) library to load configuration settings. The process works as follows: diff --git a/glutton.go b/glutton.go index 61e58e1..41fdf91 100644 --- a/glutton.go +++ b/glutton.go @@ -222,11 +222,11 @@ func (g *Glutton) tcpListen() { g.Logger.Error("Failed to set connection timeout", producer.ErrAttr(err)) } - if rule.Type == "tcp_proxy" { + if rule.Type == "proxy_tcp" { if hfunc, ok := g.tcpProtocolHandlers[rule.Type]; ok { go func() { if err := hfunc(g.ctx, conn, md); err != nil { - g.Logger.Error("Failed to handle TCP passthrough", producer.ErrAttr(err), slog.String("handler", "tcp_proxy")) + g.Logger.Error("Failed to handle proxy TCP", producer.ErrAttr(err), slog.String("handler", "proxy_tcp")) } }() } diff --git a/mkdocs.yml b/mkdocs.yml index ecb8baf..75ea170 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -5,7 +5,7 @@ nav: - Setup: setup.md - Configuration: configuration.md - Extension: extension.md + - Engineering Guidelines: engineering-guidelines.md - FAQs: faq.md theme: name: readthedocs - diff --git a/protocols/protocols.go b/protocols/protocols.go index 7e8f6ba..e9a8b2a 100644 --- a/protocols/protocols.go +++ b/protocols/protocols.go @@ -72,8 +72,8 @@ func MapTCPProtocolHandlers(log interfaces.Logger, h interfaces.Honeypot) map[st protocolHandlers["mongodb"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { return tcp.HandleMongoDB(ctx, conn, md, log, h) } - protocolHandlers["tcp_proxy"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { - return tcp.HandlePassThrough(ctx, conn, md, log, h) + protocolHandlers["proxy_tcp"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { + return tcp.HandleProxyTCP(ctx, conn, md, log, h) } protocolHandlers["tcp"] = func(ctx context.Context, conn net.Conn, md connection.Metadata) error { snip, bufConn, err := Peek(conn, 4) diff --git a/protocols/tcp/passthrough.go b/protocols/tcp/passthrough.go deleted file mode 100644 index 232b9a7..0000000 --- a/protocols/tcp/passthrough.go +++ /dev/null @@ -1,200 +0,0 @@ -package tcp - -import ( - "context" - "crypto/sha256" - "encoding/hex" - "fmt" - "io" - "log" - "log/slog" - "net" - "time" - - "github.com/mushorg/glutton/connection" - "github.com/mushorg/glutton/producer" - "github.com/mushorg/glutton/protocols/interfaces" - "github.com/spf13/viper" -) - -type parsedPassThrough struct { - Direction string `json:"direction,omitempty"` - Payload []byte `json:"payload,omitempty"` - PayloadHash string `json:"payload_hash,omitempty"` // Used for easier identification, can remove -} - -type passThroughServer struct { - events []parsedPassThrough - conn net.Conn - target string - source string -} - -type loggingWriter struct { - dst net.Conn - server *passThroughServer - logger interfaces.Logger - capture bool - dir string -} - -func (lw *loggingWriter) Write(p []byte) (int, error) { - lw.server.logPayload(lw.dir, p, lw.logger) - lw.server.recordEvent(lw.dir, p, lw.capture) - return lw.dst.Write(p) -} - -// checks whether the payload can be converted to text, to prevent expensive hex coding. -func (srv *passThroughServer) isLikelyText(data []byte) bool { - if len(data) == 0 { - return false - } - - printable := 0 - for _, b := range data { - if b >= 32 && b <= 126 || b == '\n' || b == '\r' || b == '\t' { - printable++ - } - } - - return (printable*100)/len(data) > 80 // threshold value --> 80% -} - -// logs the payload hex or payload text. -func (srv *passThroughServer) logPayload(direction string, data []byte, logger interfaces.Logger) { - if len(data) == 0 { - return - } - - fields := []any{ - slog.String("direction", direction), - slog.Int("length", len(data)), - slog.String("sha256", fmt.Sprintf("%x", sha256.Sum256(data))), - } - - if srv.isLikelyText(data) { - fields = append(fields, slog.String("payload", string(data))) - } else { - fields = append(fields, slog.String("hex", hex.EncodeToString(data))) - } - - logger.Info("payload_transferred", fields...) -} - -// records the events in the server -func (srv *passThroughServer) recordEvent(dir string, buf []byte, capture bool) { - if !capture { - return - } - hash := sha256.Sum256(buf) - - payload := append([]byte(nil), buf...) // defensive copy - - srv.events = append(srv.events, parsedPassThrough{ - Direction: dir, - Payload: payload, - PayloadHash: fmt.Sprintf("%x", hash[:]), - }) -} - -// pipeBidirectional handles data transfer between the two connections -func pipeBidirectional(src, dst net.Conn, server *passThroughServer, logger interfaces.Logger, capture bool, errChan chan error) { - direction := getDirection(src, dst) - writer := &loggingWriter{dst: dst, server: server, logger: logger, capture: capture, dir: direction} - - // source to target - go func() { - _, err := io.Copy(writer, src) - errChan <- err - }() - - revDirection := getDirection(dst, src) - revWriter := &loggingWriter{dst: src, server: server, logger: logger, capture: capture, dir: revDirection} - - // target to source - go func() { - _, err := io.Copy(revWriter, dst) - errChan <- err - }() -} - -// getDirection returns the direction as a string -func getDirection(src, dst net.Conn) string { - srcAddr := src.RemoteAddr().String() - dstAddr := dst.RemoteAddr().String() - return fmt.Sprintf("%s -> %s", srcAddr, dstAddr) -} - -// Dial to the source ip, acting as a proxy between the client and real source by piping the data back and forth w/o interfering w it. -func HandlePassThrough(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot) error { - var err error - handler := "tcp_proxy" - - srcAddr := conn.RemoteAddr().String() - destAddr := md.Rule.Target - - host, _, err := net.SplitHostPort(destAddr) - if err != nil { - logger.Error("invalid address format", producer.ErrAttr(err)) - return nil - } - - if ip := net.ParseIP(host); ip == nil { - if _, err := net.LookupHost(host); err != nil { - return fmt.Errorf("invalid host: %w", err) - } - } - - server := &passThroughServer{ - events: []parsedPassThrough{}, - conn: conn, - target: destAddr, - source: srcAddr, - } - - var capture bool - if viper.GetBool("capture_traffic.enabled") { - capture = true - } - - defer func() { - var events []parsedPassThrough - if capture { - events = server.events - } - if err := h.ProduceTCP("passthrough", conn, md, nil, events); err != nil { - logger.Error("failed to produce passthrough message", producer.ErrAttr(err)) - } - if err := conn.Close(); err != nil { - logger.Error("failed to close incoming connection", slog.String("handler", handler), producer.ErrAttr(err)) - } - }() - - if destAddr == "" { - logger.Error("no target defined", slog.String("handler", handler)) - return nil - } - - timeout := 5 * time.Second - - targetConn, err := net.DialTimeout("tcp", destAddr, timeout) - if err != nil { - logger.Error("failed to connect to the target", slog.String("handler", handler), slog.String("target", string(destAddr)), producer.ErrAttr(err)) - return nil - } - defer targetConn.Close() - - logger.Info("starting passthrough", slog.String("source", srcAddr), slog.String("target", string(destAddr)), slog.String("handler", handler)) - - errChan := make(chan error, 2) - - go pipeBidirectional(conn, targetConn, server, logger, capture, errChan) - - // wait for either side to close - if err := <-errChan; err != nil { - log.Printf("connection closed: %v", err) - } - - logger.Info("Passthrough completed successfully") - return nil -} diff --git a/protocols/tcp/passthrough_test.go b/protocols/tcp/passthrough_test.go deleted file mode 100644 index 4f5569f..0000000 --- a/protocols/tcp/passthrough_test.go +++ /dev/null @@ -1,254 +0,0 @@ -package tcp - -import ( - "context" - "crypto/rand" - "io" - "net" - "testing" - - "github.com/mushorg/glutton/protocols/interfaces" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/mock" - "github.com/stretchr/testify/require" -) - -type MockLogger struct { - mock.Mock -} - -func (m *MockLogger) Info(msg string, attrs ...interface{}) { - m.Called(msg, attrs) -} - -func (m *MockLogger) Debug(msg string, attrs ...interface{}) { - m.Called(msg, attrs) -} - -func (m *MockLogger) Error(msg string, attrs ...interface{}) { - m.Called(msg, attrs) -} - -func (m *MockLogger) Warn(msg string, attrs ...interface{}) { - m.Called(msg, attrs) -} - -func TestIsLikelyText(t *testing.T) { - tests := []struct { - name string - input []byte - expected bool - }{ - { - name: "Empty input", - input: []byte(""), - expected: false, - }, - { - name: "Simple ASCII text", - input: []byte("This is plain text"), - expected: true, - }, - { - name: "Text with whitespace", - input: []byte("Text with\nnewlines\tand tabs\r\n"), - expected: true, - }, - { - name: "Binary data", - input: []byte{0x01, 0x02, 0x03, 0x04, 0x05}, - expected: false, - }, - { - name: "Mixed content with few non-printable", - input: []byte("Text\x00with\x01binary"), - expected: true, // checking threshold at 85.7% - }, - { - name: "Exactly 80% printable", - input: []byte("AAAA\x01"), // 4/5 = 80% - expected: false, - }, - { - name: "Just below 80% printable", - input: []byte("AAA\x01\x02"), // 3/5 = 60% - expected: false, - }, - } - - srv := &passThroughServer{} - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - result := srv.isLikelyText(test.input) - require.Equal(t, test.expected, result, "unexpected result for test case: %s", test.name) - }) - } -} - -func TestRecording(t *testing.T) { - s := &passThroughServer{} - s.recordEvent("test", []byte("data"), true) - assert.Len(t, s.events, 1) -} - -func TestPipeBidirectional(t *testing.T) { - mockLogger := &MockLogger{} - mockLogger.On("Info", mock.Anything, mock.Anything).Return() - - mockServer := &passThroughServer{ - events: make([]parsedPassThrough, 0), - conn: nil, - target: "test-target:1234", - source: "test-source:5678", - } - - type args struct { - ctx context.Context - src net.Conn - dst net.Conn - server *passThroughServer - logger interfaces.Logger - capture bool - errChan chan error - } - tests := []struct { - name string - args args - setup func() (net.Conn, net.Conn) - wantErr bool - wantErrType error - verify func(t *testing.T, args args) - }{ - { - name: "successful data transfer with capture", - args: args{ - ctx: context.Background(), - server: mockServer, - logger: mockLogger, - capture: true, - errChan: make(chan error, 1), - }, - - setup: func() (net.Conn, net.Conn) { - client, server := net.Pipe() - go func() { - client.Write([]byte("test data")) - client.Close() - }() - return client, server - }, - verify: func(t *testing.T, args args) { - buf := make([]byte, 1024) - n, err := args.dst.Read(buf) - - require.NoError(t, err) - assert.Equal(t, "test data", string(buf[:n])) - - require.True(t, args.capture, "Capture should be enabled") - }, - }, - { - name: "read error from source", - args: args{ - ctx: context.Background(), - server: mockServer, - logger: mockLogger, - capture: false, - errChan: make(chan error, 1), - }, - setup: func() (net.Conn, net.Conn) { - client, server := net.Pipe() - client.Close() - return client, server - }, - wantErr: true, - wantErrType: io.EOF, - }, - { - name: "write error to destination", - args: args{ - ctx: context.Background(), - server: mockServer, - logger: mockLogger, - capture: false, - errChan: make(chan error, 1), - }, - setup: func() (net.Conn, net.Conn) { - client, server := net.Pipe() - server.Close() - return client, server - }, - wantErr: true, - }, - { - name: "zero byte read", - args: args{ - ctx: context.Background(), - server: mockServer, - logger: mockLogger, - capture: true, - errChan: make(chan error, 1), - }, - setup: func() (net.Conn, net.Conn) { - client, server := net.Pipe() - go func() { - client.Write([]byte{}) - client.Close() - }() - return client, server - }, - verify: func(t *testing.T, args args) { - assert.Empty(t, args.server.events) - }, - }, - { - name: "large data transfer", - args: args{ - ctx: context.Background(), - server: mockServer, - logger: mockLogger, - capture: true, - errChan: make(chan error, 1), - }, - setup: func() (net.Conn, net.Conn) { - client, server := net.Pipe() - largeData := make([]byte, 8192) - rand.Read(largeData) - go func() { - client.Write(largeData) - client.Close() - }() - return client, server - }, - verify: func(t *testing.T, args args) { - buf := make([]byte, 8192) - n, err := io.ReadFull(args.dst, buf) - require.NoError(t, err) - assert.Equal(t, 8192, n) - }, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if tt.setup != nil { - tt.args.src, tt.args.dst = tt.setup() - defer tt.args.src.Close() - defer tt.args.dst.Close() - } - - go pipeBidirectional( - tt.args.src, - tt.args.dst, - tt.args.server, - tt.args.logger, - tt.args.capture, - tt.args.errChan, - ) - - if tt.verify != nil { - tt.verify(t, tt.args) - } - }) - } -} diff --git a/protocols/tcp/proxy_tcp.go b/protocols/tcp/proxy_tcp.go new file mode 100644 index 0000000..686d0cb --- /dev/null +++ b/protocols/tcp/proxy_tcp.go @@ -0,0 +1,536 @@ +package tcp + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "log/slog" + "net" + "syscall" + "time" + + "github.com/mushorg/glutton/connection" + "github.com/mushorg/glutton/producer" + "github.com/mushorg/glutton/protocols/interfaces" + "github.com/spf13/viper" +) + +const handlerName = "proxy_tcp" + +// a proxy connection event sent to producers +type event struct { + Direction string `json:"direction,omitempty"` + Payload []byte `json:"payload,omitempty"` + PayloadHash string `json:"payload_hash,omitempty"` // Used for easier identification, can remove + Bytes int64 `json:"bytes,omitempty"` + Truncated bool `json:"truncated,omitempty"` +} + +// holds a proxy connection metadata +type session struct { + source string + target string + producer bool + idleTimeout time.Duration + payloadSize int +} + +// reader wrapps the source connection for idle deadline. +type reader struct { + conn net.Conn + idle time.Duration + name string +} + +func (r reader) Read(p []byte) (int, error) { + + if r.idle > 0 { + if err := r.conn.SetReadDeadline(time.Now().Add(r.idle)); err != nil { + return 0, fmt.Errorf("%s set read deadline: %w", r.name, err) + } + } + + n, err := r.conn.Read(p) + // EOF is the normal way a stream says there are no more bytes to read. + if err != nil && !errors.Is(err, io.EOF) { + return n, fmt.Errorf("%s read: %w", r.name, err) + } + return n, err +} + +// writer wraps the destination connection for logging and idle deadline +type writer struct { + conn net.Conn + session *session + logger interfaces.Logger + dir string + name string + + written int64 + payload []byte +} + +func (w *writer) Write(p []byte) (int, error) { + + if w.session.idleTimeout > 0 { + if err := w.conn.SetWriteDeadline(time.Now().Add(w.session.idleTimeout)); err != nil { + return 0, fmt.Errorf("%s set write deadline: %w", w.name, err) + } + } + + n, err := w.conn.Write(p) + + // A Write can partially succeed. Only p[:n] reached the destination, so only + // those bytes should affect logs, byte counts, and capture samples. + if n > 0 { + written := p[:n] + w.written += int64(n) + w.session.logPayload(w.dir, written, w.logger) + w.storePayload(written) + } + + if err != nil { + w.logger.Debug("proxy writer returned error", logAttrs( + slog.String("function", "writer.Write"), + slog.String("direction", w.dir), + slog.Int("bytes_written", n), + producer.ErrAttr(err), + )...) + return n, fmt.Errorf("%s write: %w", w.name, err) + } + + if n != len(p) { + return n, fmt.Errorf("%s short write: %w", w.name, io.ErrShortWrite) + } + + return n, nil +} + +// emits a structured log for connection metadata including raw payload +func (s *session) logPayload(direction string, data []byte, logger interfaces.Logger) { + + if len(data) == 0 { + return + } + + // always log transfer metadata excluding raw service data. + fields := logAttrs( + slog.String("direction", direction), + slog.Int("length", len(data)), + slog.String("payload_hash", fmt.Sprintf("%x", sha256.Sum256(data))), + ) + + // when caputure enabled then include raw payload data to logs + if s.producer && s.payloadSize > 0 { + sample := data + truncated := false + if len(sample) > s.payloadSize { + sample = sample[:s.payloadSize] + truncated = true + } + + if isLikelyText(sample) { + fields = append(fields, slog.String("payload", string(sample))) + } else { + fields = append(fields, slog.String("hex", hex.EncodeToString(sample))) + } + fields = append(fields, slog.Bool("payload_truncated", truncated)) + } + + logger.Info("proxy_tcp payload_transferred", fields...) +} + +// storePayload stores a bounded sample of written bytes for producer output. +func (w *writer) storePayload(p []byte) { + + if !w.session.producer || w.session.payloadSize <= 0 { + return + } + + if len(w.payload) >= w.session.payloadSize { + return + } + + remaining := w.session.payloadSize - len(w.payload) + if len(p) > remaining { + p = p[:remaining] + } + w.payload = append(w.payload, p...) +} + +// event converts the stored payload writer-owned capture sample into a producer event. +func (w *writer) event() *event { + if !w.session.producer || len(w.payload) == 0 { + return nil + } + + payload := append([]byte(nil), w.payload...) + hash := sha256.Sum256(payload) + return &event{ + Direction: w.dir, + Payload: payload, + PayloadHash: fmt.Sprintf("%x", hash[:]), + Bytes: w.written, + Truncated: w.written > int64(len(payload)), + } +} + +// logAttrs adds common structured log fields to every proxy_tcp log. +func logAttrs(fields ...any) []any { + base := []any{ + slog.String("handler", handlerName), + } + return append(base, fields...) +} + +// isLikelyText checks whether a byte slice is mostly printable text. +func isLikelyText(data []byte) bool { + + if len(data) == 0 { + return false + } + + printable := 0 + for _, b := range data { + if b >= 32 && b <= 126 || b == '\n' || b == '\r' || b == '\t' { + printable++ + } + } + + // Require more than 80% printable bytes so mostly-binary payloads stay as hex. + return (printable*100)/len(data) > 80 +} + +// pipeResult is the completion report from one directional pipe. +type pipeResult struct { + dir string + bytes int64 + event *event + err error +} + +// pipe copies bytes in one direction and sends a pipeResult when that direction ends. +func pipe(done chan<- pipeResult, dst, src net.Conn, session *session, logger interfaces.Logger) { + dir := getDirection(src, dst) + writer := &writer{ + conn: dst, + session: session, + logger: logger, + dir: dir, + name: dir + " dst", + } + + reader := reader{ + conn: src, + idle: session.idleTimeout, + name: dir + " src", + } + + _, err := io.Copy(writer, reader) + + // Tell the destination peer there will be no more bytes from this direction. + if closeErr := finishWriteSide(dst, logger, dir); closeErr != nil && err == nil { + err = fmt.Errorf("%s close write: %w", dir, closeErr) + } + + // Stop reading from the source side after this direction has completed. + if closeErr := finishReadSide(src, logger, dir); closeErr != nil && err == nil { + err = fmt.Errorf("%s close read: %w", dir, closeErr) + } + + done <- pipeResult{ + dir: dir, + bytes: writer.written, + event: writer.event(), + err: err, + } +} + +// pipeBothWays starts proxy connection between client and target +func pipeBothWays(client, target net.Conn, session *session, logger interfaces.Logger) []pipeResult { + + logger.Debug("starting proxy bidirectional copy", logAttrs( + slog.String("function", "pipeBothWays"), + slog.String("source", session.source), + slog.String("target", session.target), + )...) + + done := make(chan pipeResult, 2) + go pipe(done, target, client, session, logger) + go pipe(done, client, target, session, logger) + + // Wait for both directions before returning + return []pipeResult{<-done, <-done} +} + +// eventsFromResults extracts captured producer events from completed pipe results. +func eventsFromResults(results []pipeResult) []event { + events := make([]event, 0, len(results)) + for _, result := range results { + if result.event != nil { + events = append(events, *result.event) + } + } + return events +} + +func getDirection(src, dst net.Conn) string { + return fmt.Sprintf("%s -> %s", src.RemoteAddr().String(), dst.RemoteAddr().String()) +} + +// logResult records the outcome from one directional pipe. +func logResult(logger interfaces.Logger, result pipeResult) { + fields := logAttrs( + slog.String("function", "logResult"), + slog.String("direction", result.dir), + slog.Int64("bytes", result.bytes), + ) + + // Include capture metadata without logging raw bytes again. + if result.event != nil { + fields = append(fields, + slog.Int("captured_bytes", len(result.event.Payload)), + slog.String("payload_hash", result.event.PayloadHash), + slog.Bool("truncated", result.event.Truncated), + ) + } + + // Attach the copy error before deciding the log level. + if result.err != nil { + fields = append(fields, producer.ErrAttr(result.err)) + } + + if expectedPipeError(result.err) { + logger.Debug("proxy pipe completed", fields...) + } else { + logger.Error("proxy pipe failed", fields...) + } +} + +// expectedPipeError classifies normal network shutdown results from io.Copy. +func expectedPipeError(err error) bool { + + if err == nil { + return true + } + + // EOF and "use of closed network connection" are expected when scanners or + // targets close sockets during normal request/response exchanges. + if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || errors.Is(err, syscall.ENOTCONN) { + return true + } + + // Deadline timeouts can be created by reader or shutdown fallbacks + var nerr net.Error + return errors.As(err, &nerr) && nerr.Timeout() +} + +type closeWriter interface { + CloseWrite() error +} + +type closeReader interface { + CloseRead() error +} + +// Use for used for half-closing a write side, if unavailable, it uses a deadline +func finishWriteSide(conn net.Conn, logger interfaces.Logger, dir string) error { + + if cw, ok := conn.(closeWriter); ok { + logger.Debug("closing proxy write side", logAttrs( + slog.String("function", "finishWriteSide"), + slog.String("direction", dir), + slog.String("address", conn.RemoteAddr().String()), + )...) + if err := cw.CloseWrite(); err != nil && !errors.Is(err, net.ErrClosed) { + return err + } + return nil + } + + // Non-TCP net.Conn implementations may not support half-close. + logger.Debug("setting proxy write shutdown deadline", logAttrs( + slog.String("function", "finishWriteSide"), + slog.String("direction", dir), + slog.String("address", conn.RemoteAddr().String()), + )...) + return conn.SetDeadline(time.Now().Add(2 * time.Second)) +} + +// Use for used for half-closing a read side, if unavailable, it uses a deadline +func finishReadSide(conn net.Conn, logger interfaces.Logger, dir string) error { + + if cr, ok := conn.(closeReader); ok { + logger.Debug("closing proxy read side", logAttrs( + slog.String("function", "finishReadSide"), + slog.String("direction", dir), + slog.String("address", conn.RemoteAddr().String()), + )...) + if err := cr.CloseRead(); err != nil && !errors.Is(err, net.ErrClosed) { + return err + } + return nil + } + + // Non-TCP net.Conn implementations may not support CloseRead. + logger.Debug("setting proxy read shutdown deadline", logAttrs( + slog.String("function", "finishReadSide"), + slog.String("direction", dir), + slog.String("address", conn.RemoteAddr().String()), + )...) + return conn.SetReadDeadline(time.Now().Add(2 * time.Second)) +} + +// It is best-effort to enables TCP keepalive for real TCP connections +func setKeepAlive(conn net.Conn, logger interfaces.Logger, name string) { + tcpConn, ok := conn.(*net.TCPConn) + if !ok { + return + } + if err := tcpConn.SetKeepAlive(true); err != nil { + logger.Debug("failed to enable proxy keepalive", logAttrs( + slog.String("function", "setKeepAlive"), + slog.String("name", name), + producer.ErrAttr(err), + )...) + } +} + +func stopProxyOnCancel(ctx context.Context, client, target net.Conn, logger interfaces.Logger) func() { + done := make(chan struct{}) + + go func() { + select { + case <-ctx.Done(): + logger.Debug("closing proxy connections after context cancellation", logAttrs( + slog.String("function", "stopProxyOnCancel"), + producer.ErrAttr(ctx.Err()), + )...) + closeProxyConn(client, logger, "client") + closeProxyConn(target, logger, "target") + case <-done: + } + }() + + return func() { + close(done) + } +} + +func closeProxyConn(conn net.Conn, logger interfaces.Logger, name string) { + if err := conn.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + logger.Debug("failed to close proxy connection", logAttrs( + slog.String("function", "closeProxyConn"), + slog.String("name", name), + producer.ErrAttr(err), + )...) + } +} + +func HandleProxyTCP(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot) error { + + srcAddr := conn.RemoteAddr().String() + + logger.Debug("entered proxy handler", logAttrs( + slog.String("function", "HandleProxyTCP"), + slog.String("source", srcAddr), + )...) + defer logger.Debug("leaving proxy handler", logAttrs( + slog.String("function", "HandleProxyTCP"), + slog.String("source", srcAddr), + )...) + + defer func() { + if err := conn.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + logger.Error("failed to close incoming connection", logAttrs( + slog.String("function", "HandleProxyTCP"), + producer.ErrAttr(err), + )...) + } + }() + + // If missing metadata, close without panick + if md.Rule == nil { + logger.Error("missing proxy_tcp rule metadata", logAttrs( + slog.String("function", "HandleProxyTCP"), + )...) + return nil + } + + // If it is missing here, the handler cannot safely dial anything + if md.Rule.ProxyTarget == nil || md.Rule.ProxyTarget.DialAddress == "" { + logger.Error("missing proxy_tcp target metadata", logAttrs( + slog.String("function", "HandleProxyTCP"), + )...) + return nil + } + destAddr := md.Rule.ProxyTarget.DialAddress + + session := &session{ + source: srcAddr, + target: destAddr, + producer: viper.GetBool("capture_traffic.enabled"), + idleTimeout: time.Duration(viper.GetInt("conn_timeout")) * time.Second, + payloadSize: viper.GetInt("max_tcp_payload"), + } + + var results []pipeResult + + // capture is enabled, produces one final proxy_tcp event after connection closes. + defer func() { + var events []event + if session.producer { + events = eventsFromResults(results) + } + if err := h.ProduceTCP("proxy_tcp", conn, md, nil, events); err != nil { + logger.Error("failed to produce proxy_tcp message", logAttrs( + slog.String("function", "HandleProxyTCP"), + producer.ErrAttr(err), + )...) + } + }() + + // Enable keepalive for the client side + setKeepAlive(conn, logger, "client") + + dialerTimeout := time.Duration(viper.GetInt("dial_timeout")) * time.Second + dialer := net.Dialer{Timeout: dialerTimeout} + targetConn, err := dialer.DialContext(ctx, "tcp", destAddr) + if err != nil { + logger.Error("failed to connect to the target", logAttrs( + slog.String("function", "HandleProxyTCP"), + slog.String("target", destAddr), + producer.ErrAttr(err), + )...) + return nil + } + defer targetConn.Close() + stopCancelWatcher := stopProxyOnCancel(ctx, conn, targetConn, logger) + defer stopCancelWatcher() + + // Enable keepalive for the target side + setKeepAlive(targetConn, logger, "target") + + // At this point both sockets are open. The next step is full-duplex copying. + logger.Debug("starting proxy tcp", logAttrs( + slog.String("function", "HandleProxyTCP"), + slog.String("source", srcAddr), + slog.String("target", destAddr), + slog.Duration("idle_timeout", session.idleTimeout), + slog.Int("payload_size", session.payloadSize), + )...) + + // pipeBothWays waits for the success / failure of proxy session and return results + results = pipeBothWays(conn, targetConn, session, logger) + for _, result := range results { + logResult(logger, result) + } + + logger.Debug("proxy tcp completed successfully", logAttrs( + slog.String("function", "HandleProxyTCP"), + )...) + return nil +} diff --git a/protocols/tcp/proxytcp_test.go b/protocols/tcp/proxytcp_test.go new file mode 100644 index 0000000..68d22d6 --- /dev/null +++ b/protocols/tcp/proxytcp_test.go @@ -0,0 +1,423 @@ +package tcp + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "net" + "sync" + "syscall" + "testing" + "time" + + "github.com/mushorg/glutton/connection" + "github.com/mushorg/glutton/rules" + "github.com/spf13/viper" + "github.com/stretchr/testify/require" +) + +type recordingLogger struct { + mtx sync.Mutex + infos []string + debugs []string + errs []string + warns []string + fields []any +} + +func (l *recordingLogger) Info(msg string, fields ...any) { + l.record(&l.infos, msg, fields...) +} + +func (l *recordingLogger) Debug(msg string, fields ...any) { + l.record(&l.debugs, msg, fields...) +} + +func (l *recordingLogger) Error(msg string, fields ...any) { + l.record(&l.errs, msg, fields...) +} + +func (l *recordingLogger) Warn(msg string, fields ...any) { + l.record(&l.warns, msg, fields...) +} + +func (l *recordingLogger) record(target *[]string, msg string, fields ...any) { + l.mtx.Lock() + defer l.mtx.Unlock() + + *target = append(*target, msg) + l.fields = append(l.fields, fields...) +} + +func (l *recordingLogger) hasAttr(key, value string) bool { + l.mtx.Lock() + defer l.mtx.Unlock() + + for _, field := range l.fields { + attr, ok := field.(slog.Attr) + if !ok { + continue + } + if attr.Key == key && attr.Value.String() == value { + return true + } + } + return false +} + +type producedTCP struct { + protocol string + decoded interface{} +} + +type fakeHoneypot struct { + produced chan producedTCP +} + +func newFakeHoneypot() *fakeHoneypot { + return &fakeHoneypot{produced: make(chan producedTCP, 4)} +} + +func (h *fakeHoneypot) ProduceTCP(protocol string, conn net.Conn, md connection.Metadata, payload []byte, decoded interface{}) error { + h.produced <- producedTCP{protocol: protocol, decoded: decoded} + return nil +} + +func (h *fakeHoneypot) ProduceUDP(handler string, srcAddr, dstAddr *net.UDPAddr, md connection.Metadata, payload []byte, decoded interface{}) error { + return nil +} + +func (h *fakeHoneypot) ConnectionByFlow([2]uint64) connection.Metadata { + return connection.Metadata{} +} + +func (h *fakeHoneypot) UpdateConnectionTimeout(context.Context, net.Conn) error { + return nil +} + +func (h *fakeHoneypot) MetadataByConnection(net.Conn) (connection.Metadata, error) { + return connection.Metadata{}, nil +} + +func setCapture(t *testing.T, enabled bool) { + t.Helper() + + previousCapture := viper.Get("capture_traffic.enabled") + previousMaxPayload := viper.Get("max_tcp_payload") + previousTimeout := viper.Get("conn_timeout") + viper.Set("capture_traffic.enabled", enabled) + viper.Set("max_tcp_payload", 4096) + viper.Set("conn_timeout", 0) + t.Cleanup(func() { + viper.Set("capture_traffic.enabled", previousCapture) + viper.Set("max_tcp_payload", previousMaxPayload) + viper.Set("conn_timeout", previousTimeout) + }) +} + +func proxyMetadata(target string) connection.Metadata { + return connection.Metadata{ + Rule: &rules.Rule{ + Type: "proxy_tcp", + ProxyTarget: &rules.ProxyTarget{ + Host: "127.0.0.1", + Port: 0, + DialAddress: target, + }, + }, + } +} + +func startTCPServer(t *testing.T, handle func(net.Conn)) string { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { + _ = listener.Close() + }) + + done := make(chan struct{}) + t.Cleanup(func() { + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatalf("server at %s did not finish", listener.Addr().String()) + } + }) + + go func() { + defer close(done) + + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + handle(conn) + }() + + return listener.Addr().String() +} + +func startProxyServer(t *testing.T, target string, hp *fakeHoneypot, logger *recordingLogger) string { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { + _ = listener.Close() + }) + + done := make(chan error, 1) + t.Cleanup(func() { + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatalf("proxy at %s did not finish", listener.Addr().String()) + } + }) + + go func() { + conn, err := listener.Accept() + if err != nil { + done <- nil + return + } + done <- HandleProxyTCP(context.Background(), conn, proxyMetadata(target), logger, hp) + }() + + return listener.Addr().String() +} + +func waitProduced(t *testing.T, hp *fakeHoneypot) producedTCP { + t.Helper() + + select { + case event := <-hp.produced: + return event + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for produced TCP event") + } + return producedTCP{} +} + +func TestHandleProxyAfterCloseWrite(t *testing.T) { + setCapture(t, false) + + targetAddr := startTCPServer(t, func(conn net.Conn) { + request, err := io.ReadAll(conn) + require.NoError(t, err) + require.Equal(t, "request", string(request)) + + _, err = conn.Write([]byte("response")) + require.NoError(t, err) + }) + + hp := newFakeHoneypot() + logger := &recordingLogger{} + proxyAddr := startProxyServer(t, targetAddr, hp, logger) + + client, err := net.Dial("tcp", proxyAddr) + require.NoError(t, err) + defer client.Close() + + _, err = client.Write([]byte("request")) + require.NoError(t, err) + require.NoError(t, client.(*net.TCPConn).CloseWrite()) + + response, err := io.ReadAll(client) + require.NoError(t, err) + require.Equal(t, "response", string(response)) + + event := waitProduced(t, hp) + require.Equal(t, "proxy_tcp", event.protocol) + require.Empty(t, event.decoded) + require.True(t, logger.hasAttr("handler", "proxy_tcp")) +} + +func TestHandleProxyTCP(t *testing.T) { + setCapture(t, true) + + targetAddr := startTCPServer(t, func(conn net.Conn) { + request, err := io.ReadAll(conn) + require.NoError(t, err) + require.Equal(t, "client-payload", string(request)) + + _, err = conn.Write([]byte("target-response")) + require.NoError(t, err) + }) + + hp := newFakeHoneypot() + logger := &recordingLogger{} + proxyAddr := startProxyServer(t, targetAddr, hp, logger) + + client, err := net.Dial("tcp", proxyAddr) + require.NoError(t, err) + defer client.Close() + + _, err = client.Write([]byte("client-payload")) + require.NoError(t, err) + require.NoError(t, client.(*net.TCPConn).CloseWrite()) + + response, err := io.ReadAll(client) + require.NoError(t, err) + require.Equal(t, "target-response", string(response)) + + produced := waitProduced(t, hp) + events, ok := produced.decoded.([]event) + require.True(t, ok) + require.Len(t, events, 2) + + payloads := map[string]bool{} + for _, captured := range events { + payloads[string(captured.Payload)] = true + require.NotEmpty(t, captured.Direction) + require.NotEmpty(t, captured.PayloadHash) + } + require.True(t, payloads["client-payload"]) + require.True(t, payloads["target-response"]) +} + +func TestWriter(t *testing.T) { + writeErr := errors.New("short write") + logger := &recordingLogger{} + session := &session{producer: true, payloadSize: 4096} + writer := &writer{ + conn: partialWriteConn{written: 3, err: writeErr}, + session: session, + logger: logger, + dir: "client->target", + name: "client->target dst", + } + + n, err := writer.Write([]byte("abcdef")) + require.ErrorIs(t, err, writeErr) + require.Equal(t, 3, n) + + event := writer.event() + require.NotNil(t, event) + require.Equal(t, "abc", string(event.Payload)) + require.Equal(t, int64(3), event.Bytes) + require.False(t, event.Truncated) + require.Equal(t, int64(3), writer.written) +} + +// test capture writer skipped failed writer +func TestFailedWrites(t *testing.T) { + writeErr := errors.New("write failed") + logger := &recordingLogger{} + session := &session{producer: true, payloadSize: 4096} + writer := &writer{ + conn: partialWriteConn{written: 0, err: writeErr}, + session: session, + logger: logger, + dir: "client->target", + name: "client->target dst", + } + + n, err := writer.Write([]byte("abcdef")) + require.ErrorIs(t, err, writeErr) + require.Zero(t, n) + require.Nil(t, writer.event()) + require.Zero(t, writer.written) +} + +// test capture writer returns short write error +func TestShortWriteError(t *testing.T) { + logger := &recordingLogger{} + session := &session{producer: true, payloadSize: 4096} + writer := &writer{ + conn: partialWriteConn{written: 3}, + session: session, + logger: logger, + dir: "client->target", + name: "client->target dst", + } + + n, err := writer.Write([]byte("abcdef")) + require.ErrorIs(t, err, io.ErrShortWrite) + require.Equal(t, 3, n) + event := writer.event() + require.Equal(t, "abc", string(event.Payload)) + require.Equal(t, int64(3), event.Bytes) + require.False(t, event.Truncated) +} + +// test capture writer caps captured bytes +func TestWriterByteCaps(t *testing.T) { + logger := &recordingLogger{} + session := &session{producer: true, payloadSize: 4} + writer := &writer{ + conn: partialWriteConn{written: 6}, + session: session, + logger: logger, + dir: "client->target", + name: "client->target dst", + } + + n, err := writer.Write([]byte("abcdef")) + require.NoError(t, err) + require.Equal(t, 6, n) + event := writer.event() + require.Equal(t, "abcd", string(event.Payload)) + require.Equal(t, int64(6), event.Bytes) + require.True(t, event.Truncated) +} + +func TestExpectedPipeErrorAllowsNotConnected(t *testing.T) { + err := fmt.Errorf("close read: %w", syscall.ENOTCONN) + + require.True(t, expectedPipeError(err)) +} + +// test proxy handler closes connection on missing metadata +func TestMissingMetadata(t *testing.T) { + setCapture(t, true) + + client, server := net.Pipe() + defer client.Close() + + logger := &recordingLogger{} + err := HandleProxyTCP(context.Background(), server, connection.Metadata{}, logger, newFakeHoneypot()) + require.NoError(t, err) + require.True(t, logger.hasAttr("handler", "proxy_tcp")) + require.True(t, logger.hasAttr("function", "HandleProxyTCP")) + + err = client.SetWriteDeadline(time.Now().Add(100 * time.Millisecond)) + if err == nil { + _, err = client.Write([]byte("x")) + } + require.Error(t, err) +} + +type partialWriteConn struct { + net.Conn + written int + err error +} + +func (c partialWriteConn) Write(p []byte) (int, error) { + return c.written, c.err +} + +func (c partialWriteConn) RemoteAddr() net.Addr { + return testAddr("partial-remote") +} + +func (c partialWriteConn) LocalAddr() net.Addr { + return testAddr("partial-local") +} + +type testAddr string + +func (a testAddr) Network() string { + return "test" +} + +func (a testAddr) String() string { + return string(a) +} diff --git a/rules/rules.go b/rules/rules.go index b28e234..0c367ac 100644 --- a/rules/rules.go +++ b/rules/rules.go @@ -5,6 +5,7 @@ import ( "io" "net" "strconv" + "strings" "time" "github.com/google/gopacket" @@ -17,8 +18,8 @@ type RuleType int const ( UserConnHandler RuleType = iota + ProxyTCP Drop - Tcp_Proxy ) type Config struct { @@ -32,10 +33,17 @@ type Rule struct { Target string `yaml:"target,omitempty"` Name string `yaml:"name,omitempty"` - isInit bool - RuleType RuleType - index int - matcher *pcap.BPF + isInit bool + RuleType RuleType + ProxyTarget *ProxyTarget `yaml:"-"` + index int + matcher *pcap.BPF +} + +type ProxyTarget struct { + Host string + Port uint16 + DialAddress string } func (r *Rule) String() string { @@ -61,14 +69,22 @@ func (rule *Rule) init(idx int) error { switch rule.Type { case "conn_handler": rule.RuleType = UserConnHandler - case "tcp_proxy": - rule.RuleType = Tcp_Proxy + case "proxy_tcp": + rule.RuleType = ProxyTCP case "drop": rule.RuleType = Drop default: return fmt.Errorf("unknown rule type: %s", rule.Type) } + if rule.RuleType == ProxyTCP { + target, err := parseProxyTarget(rule.Target) + if err != nil { + return fmt.Errorf("invalid proxy_tcp target: %w", err) + } + rule.ProxyTarget = target + } + var err error if len(rule.Match) > 0 { rule.matcher, err = pcap.NewBPF(layers.LinkTypeEthernet, 65535, rule.Match) @@ -83,6 +99,35 @@ func (rule *Rule) init(idx int) error { return nil } +func parseProxyTarget(raw string) (*ProxyTarget, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, fmt.Errorf("target is required") + } + + host, portValue, err := net.SplitHostPort(raw) + if err != nil { + return nil, err + } + if strings.TrimSpace(host) == "" { + return nil, fmt.Errorf("host is required") + } + + port, err := strconv.Atoi(portValue) + if err != nil { + return nil, fmt.Errorf("invalid port %q: %w", portValue, err) + } + if port < 1 || port > 65535 { + return nil, fmt.Errorf("port out of range: %d", port) + } + + return &ProxyTarget{ + Host: host, + Port: uint16(port), + DialAddress: net.JoinHostPort(host, strconv.Itoa(port)), + }, nil +} + func splitAddr(addr string) (string, uint16, error) { ip, port, err := net.SplitHostPort(addr) if err != nil { diff --git a/rules/rules_test.go b/rules/rules_test.go index c885de9..3c83ec2 100644 --- a/rules/rules_test.go +++ b/rules/rules_test.go @@ -3,6 +3,7 @@ package rules import ( "net" "os" + "strings" "testing" "time" @@ -43,6 +44,97 @@ func TestSplitAddr(t *testing.T) { require.Equal(t, uint16(8080), port) } +func TestParseProxyTarget(t *testing.T) { + tests := []struct { + name string + raw string + wantAddress string + wantErr string + }{ + { + name: "plain target", + raw: "127.0.0.1:9889", + wantAddress: "127.0.0.1:9889", + }, + { + name: "hostname target", + raw: "localhost:9889", + wantAddress: "localhost:9889", + }, + { + name: "empty target", + raw: "", + wantErr: "target is required", + }, + { + name: "scheme not supported", + raw: "tcp://127.0.0.1:9889", + wantErr: "too many colons", + }, + { + name: "missing port", + raw: "127.0.0.1", + wantErr: "missing port", + }, + { + name: "missing host", + raw: ":9889", + wantErr: "host is required", + }, + { + name: "non numeric port", + raw: "127.0.0.1:http", + wantErr: "invalid port", + }, + { + name: "port out of range", + raw: "127.0.0.1:70000", + wantErr: "port out of range", + }, + { + name: "path not supported", + raw: "127.0.0.1:9889/path", + wantErr: "invalid port", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + target, err := parseProxyTarget(tt.raw) + if tt.wantErr != "" { + require.Error(t, err) + require.Contains(t, err.Error(), tt.wantErr) + return + } + + require.NoError(t, err) + require.Equal(t, tt.wantAddress, target.DialAddress) + }) + } +} + +func TestInitProxyTCPRuleParsesTarget(t *testing.T) { + rules, err := Init(strings.NewReader(`rules: + - match: tcp dst port 9889 + type: proxy_tcp + target: 127.0.0.1:9889 +`)) + require.NoError(t, err) + require.Len(t, rules, 1) + require.NotNil(t, rules[0].ProxyTarget) + require.Equal(t, "127.0.0.1:9889", rules[0].ProxyTarget.DialAddress) +} + +func TestInitProxyTCPRuleRejectsInvalidTarget(t *testing.T) { + _, err := Init(strings.NewReader(`rules: + - match: tcp dst port 9889 + type: proxy_tcp + target: tcp://127.0.0.1:9889 +`)) + require.Error(t, err) + require.Contains(t, err.Error(), "too many colons") +} + func testConn(t *testing.T) (net.Conn, net.Listener) { ln, err := net.Listen("tcp", "127.0.0.1:1234") require.NoError(t, err) @@ -102,6 +194,21 @@ func TestRunMatchUDP(t *testing.T) { require.Equal(t, "test", match.Target) } +func TestRunMatchProxyTCP(t *testing.T) { + rules := parseRules(t) + require.NotEmpty(t, rules) + + srcAddr := &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 50000} + dstAddr := &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 80} + + match, err := rules.Match("tcp", srcAddr, dstAddr) + require.NoError(t, err) + require.NotNil(t, match) + require.Equal(t, "proxy_tcp", match.Type) + require.NotNil(t, match.ProxyTarget) + require.Equal(t, "127.0.0.1:9889", match.ProxyTarget.DialAddress) +} + func TestBPF(t *testing.T) { buf := make([]byte, 65535) bpfi, err := pcap.NewBPF(layers.LinkTypeEthernet, 65535, "icmp") diff --git a/rules/test.yaml b/rules/test.yaml index 38396f2..b8608dc 100644 --- a/rules/test.yaml +++ b/rules/test.yaml @@ -9,6 +9,9 @@ rules: - match: udp dst port 1234 type: conn_handler target: test + - match: tcp dst port 80 + type: proxy_tcp + target: 127.0.0.1:9889 - match: tcp type: conn_handler target: tcp From 04e17af308396c787b2571d26ed2e772bf179628 Mon Sep 17 00:00:00 2001 From: furusiyya Date: Fri, 8 May 2026 12:01:41 -0700 Subject: [PATCH 7/8] (fixes) proxytcp code and tests cleaning --- .gitignore | 6 ++++++ glutton.go | 30 ++++++++++++-------------- protocols/tcp/proxy_tcp.go | 30 ++++++++------------------ protocols/tcp/proxytcp_test.go | 39 ++++++++-------------------------- 4 files changed, 38 insertions(+), 67 deletions(-) diff --git a/.gitignore b/.gitignore index 2646ced..a3f4a9b 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,9 @@ poc/ # Dev .vscode openspec/ +.cache +.codex +openspec/ +docs/roadmap.md +docs/engineering-guidelines.md + diff --git a/glutton.go b/glutton.go index 41fdf91..74edea3 100644 --- a/glutton.go +++ b/glutton.go @@ -222,22 +222,20 @@ func (g *Glutton) tcpListen() { g.Logger.Error("Failed to set connection timeout", producer.ErrAttr(err)) } - if rule.Type == "proxy_tcp" { - if hfunc, ok := g.tcpProtocolHandlers[rule.Type]; ok { - go func() { - if err := hfunc(g.ctx, conn, md); err != nil { - g.Logger.Error("Failed to handle proxy TCP", producer.ErrAttr(err), slog.String("handler", "proxy_tcp")) - } - }() - } - } else { - if hfunc, ok := g.tcpProtocolHandlers[rule.Target]; ok { - go func() { - if err := hfunc(g.ctx, conn, md); err != nil { - g.Logger.Error("Failed to handle TCP connection", producer.ErrAttr(err), slog.String("handler", rule.Target)) - } - }() - } + var handlerName string + switch rule.Type { + case "proxy_tcp": + handlerName = rule.Type + default: + handlerName = rule.Target + } + + if hfunc, ok := g.tcpProtocolHandlers[handlerName]; ok { + go func() { + if err := hfunc(g.ctx, conn, md); err != nil { + g.Logger.Error("Failed to handle ", producer.ErrAttr(err), slog.String("handler", handlerName)) + } + }() } } } diff --git a/protocols/tcp/proxy_tcp.go b/protocols/tcp/proxy_tcp.go index 686d0cb..50e40da 100644 --- a/protocols/tcp/proxy_tcp.go +++ b/protocols/tcp/proxy_tcp.go @@ -46,7 +46,6 @@ type reader struct { } func (r reader) Read(p []byte) (int, error) { - if r.idle > 0 { if err := r.conn.SetReadDeadline(time.Now().Add(r.idle)); err != nil { return 0, fmt.Errorf("%s set read deadline: %w", r.name, err) @@ -74,7 +73,6 @@ type writer struct { } func (w *writer) Write(p []byte) (int, error) { - if w.session.idleTimeout > 0 { if err := w.conn.SetWriteDeadline(time.Now().Add(w.session.idleTimeout)); err != nil { return 0, fmt.Errorf("%s set write deadline: %w", w.name, err) @@ -82,16 +80,6 @@ func (w *writer) Write(p []byte) (int, error) { } n, err := w.conn.Write(p) - - // A Write can partially succeed. Only p[:n] reached the destination, so only - // those bytes should affect logs, byte counts, and capture samples. - if n > 0 { - written := p[:n] - w.written += int64(n) - w.session.logPayload(w.dir, written, w.logger) - w.storePayload(written) - } - if err != nil { w.logger.Debug("proxy writer returned error", logAttrs( slog.String("function", "writer.Write"), @@ -102,6 +90,15 @@ func (w *writer) Write(p []byte) (int, error) { return n, fmt.Errorf("%s write: %w", w.name, err) } + // A Write can partially succeed. Only p[:n] reached the destination, so only + // those bytes should affect logs, byte counts, and capture samples. + if n > 0 { + written := p[:n] + w.written += int64(n) + w.session.logPayload(w.dir, written, w.logger) + w.storePayload(written) + } + if n != len(p) { return n, fmt.Errorf("%s short write: %w", w.name, io.ErrShortWrite) } @@ -111,7 +108,6 @@ func (w *writer) Write(p []byte) (int, error) { // emits a structured log for connection metadata including raw payload func (s *session) logPayload(direction string, data []byte, logger interfaces.Logger) { - if len(data) == 0 { return } @@ -145,7 +141,6 @@ func (s *session) logPayload(direction string, data []byte, logger interfaces.Lo // storePayload stores a bounded sample of written bytes for producer output. func (w *writer) storePayload(p []byte) { - if !w.session.producer || w.session.payloadSize <= 0 { return } @@ -188,7 +183,6 @@ func logAttrs(fields ...any) []any { // isLikelyText checks whether a byte slice is mostly printable text. func isLikelyText(data []byte) bool { - if len(data) == 0 { return false } @@ -251,7 +245,6 @@ func pipe(done chan<- pipeResult, dst, src net.Conn, session *session, logger in // pipeBothWays starts proxy connection between client and target func pipeBothWays(client, target net.Conn, session *session, logger interfaces.Logger) []pipeResult { - logger.Debug("starting proxy bidirectional copy", logAttrs( slog.String("function", "pipeBothWays"), slog.String("source", session.source), @@ -312,7 +305,6 @@ func logResult(logger interfaces.Logger, result pipeResult) { // expectedPipeError classifies normal network shutdown results from io.Copy. func expectedPipeError(err error) bool { - if err == nil { return true } @@ -338,7 +330,6 @@ type closeReader interface { // Use for used for half-closing a write side, if unavailable, it uses a deadline func finishWriteSide(conn net.Conn, logger interfaces.Logger, dir string) error { - if cw, ok := conn.(closeWriter); ok { logger.Debug("closing proxy write side", logAttrs( slog.String("function", "finishWriteSide"), @@ -362,7 +353,6 @@ func finishWriteSide(conn net.Conn, logger interfaces.Logger, dir string) error // Use for used for half-closing a read side, if unavailable, it uses a deadline func finishReadSide(conn net.Conn, logger interfaces.Logger, dir string) error { - if cr, ok := conn.(closeReader); ok { logger.Debug("closing proxy read side", logAttrs( slog.String("function", "finishReadSide"), @@ -401,7 +391,6 @@ func setKeepAlive(conn net.Conn, logger interfaces.Logger, name string) { func stopProxyOnCancel(ctx context.Context, client, target net.Conn, logger interfaces.Logger) func() { done := make(chan struct{}) - go func() { select { case <-ctx.Done(): @@ -431,7 +420,6 @@ func closeProxyConn(conn net.Conn, logger interfaces.Logger, name string) { } func HandleProxyTCP(ctx context.Context, conn net.Conn, md connection.Metadata, logger interfaces.Logger, h interfaces.Honeypot) error { - srcAddr := conn.RemoteAddr().String() logger.Debug("entered proxy handler", logAttrs( diff --git a/protocols/tcp/proxytcp_test.go b/protocols/tcp/proxytcp_test.go index 68d22d6..97e4311 100644 --- a/protocols/tcp/proxytcp_test.go +++ b/protocols/tcp/proxytcp_test.go @@ -282,12 +282,12 @@ func TestHandleProxyTCP(t *testing.T) { require.True(t, payloads["target-response"]) } -func TestWriter(t *testing.T) { +func TestPartialWriter(t *testing.T) { writeErr := errors.New("short write") logger := &recordingLogger{} session := &session{producer: true, payloadSize: 4096} writer := &writer{ - conn: partialWriteConn{written: 3, err: writeErr}, + conn: testWriteConn{written: 3, err: writeErr}, session: session, logger: logger, dir: "client->target", @@ -299,11 +299,8 @@ func TestWriter(t *testing.T) { require.Equal(t, 3, n) event := writer.event() - require.NotNil(t, event) - require.Equal(t, "abc", string(event.Payload)) - require.Equal(t, int64(3), event.Bytes) - require.False(t, event.Truncated) - require.Equal(t, int64(3), writer.written) + require.Nil(t, event) + require.Zero(t, writer.written) } // test capture writer skipped failed writer @@ -312,7 +309,7 @@ func TestFailedWrites(t *testing.T) { logger := &recordingLogger{} session := &session{producer: true, payloadSize: 4096} writer := &writer{ - conn: partialWriteConn{written: 0, err: writeErr}, + conn: testWriteConn{written: 0, err: writeErr}, session: session, logger: logger, dir: "client->target", @@ -331,7 +328,7 @@ func TestShortWriteError(t *testing.T) { logger := &recordingLogger{} session := &session{producer: true, payloadSize: 4096} writer := &writer{ - conn: partialWriteConn{written: 3}, + conn: testWriteConn{written: 3}, session: session, logger: logger, dir: "client->target", @@ -352,7 +349,7 @@ func TestWriterByteCaps(t *testing.T) { logger := &recordingLogger{} session := &session{producer: true, payloadSize: 4} writer := &writer{ - conn: partialWriteConn{written: 6}, + conn: testWriteConn{written: 6}, session: session, logger: logger, dir: "client->target", @@ -394,30 +391,12 @@ func TestMissingMetadata(t *testing.T) { require.Error(t, err) } -type partialWriteConn struct { +type testWriteConn struct { net.Conn written int err error } -func (c partialWriteConn) Write(p []byte) (int, error) { +func (c testWriteConn) Write(p []byte) (int, error) { return c.written, c.err } - -func (c partialWriteConn) RemoteAddr() net.Addr { - return testAddr("partial-remote") -} - -func (c partialWriteConn) LocalAddr() net.Addr { - return testAddr("partial-local") -} - -type testAddr string - -func (a testAddr) Network() string { - return "test" -} - -func (a testAddr) String() string { - return string(a) -} From b64eba90f0bdaadc67268908d7c98ef1ec4c1683 Mon Sep 17 00:00:00 2001 From: furusiyya Date: Fri, 8 May 2026 12:10:30 -0700 Subject: [PATCH 8/8] (docs) minor fix, removed untracked file --- mkdocs.yml | 1 - 1 file changed, 1 deletion(-) diff --git a/mkdocs.yml b/mkdocs.yml index 75ea170..c7b691f 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -5,7 +5,6 @@ nav: - Setup: setup.md - Configuration: configuration.md - Extension: extension.md - - Engineering Guidelines: engineering-guidelines.md - FAQs: faq.md theme: name: readthedocs