From 7586fc1368f25ffcb51ac200ebbec6647977e8d0 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Thu, 21 May 2026 06:54:37 +0000 Subject: [PATCH] fix: bound ipc event size --- internal/ipc/ipc.go | 25 +++++++++-------- internal/ipc/ipc_test.go | 60 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 11 deletions(-) diff --git a/internal/ipc/ipc.go b/internal/ipc/ipc.go index b5028a6..c44eb5b 100644 --- a/internal/ipc/ipc.go +++ b/internal/ipc/ipc.go @@ -7,7 +7,6 @@ import ( "encoding/json" "errors" "fmt" - "io" "net" "os" "path/filepath" @@ -22,6 +21,7 @@ const ( SocketDirMode os.FileMode = 0o700 SocketFileMode os.FileMode = 0o600 DefaultSocket = "daemon.sock" + MaxEventBytes = 1 << 20 ) var sendRetryDelays = []time.Duration{ @@ -95,20 +95,23 @@ func (s Server) ListenAndServe(ctx context.Context) error { func (s Server) handleConn(ctx context.Context, conn net.Conn) { defer conn.Close() - reader := bufio.NewReader(conn) - for { - line, err := reader.ReadBytes('\n') - if len(bytes.TrimSpace(line)) > 0 { - if err := s.handleLine(ctx, line); err != nil { - s.handleError(err) - } + scanner := bufio.NewScanner(conn) + scanner.Buffer(make([]byte, 0, 64*1024), MaxEventBytes) + for scanner.Scan() { + line := scanner.Bytes() + if len(bytes.TrimSpace(line)) == 0 { + continue } - if errors.Is(err, io.EOF) { - return + if err := s.handleLine(ctx, line); err != nil { + s.handleError(err) } - if err != nil { + } + if err := scanner.Err(); err != nil { + if strings.Contains(err.Error(), "token too long") { + s.handleError(fmt.Errorf("read event: event line exceeds %d bytes", MaxEventBytes)) return } + s.handleError(fmt.Errorf("read event: %w", err)) } } diff --git a/internal/ipc/ipc_test.go b/internal/ipc/ipc_test.go index 0ad0f19..76d4e4f 100644 --- a/internal/ipc/ipc_test.go +++ b/internal/ipc/ipc_test.go @@ -6,6 +6,7 @@ import ( "net" "os" "path/filepath" + "strings" "testing" "time" @@ -260,6 +261,65 @@ func TestServerReportsInvalidLines(t *testing.T) { } } +func TestServerRejectsOversizedEventLines(t *testing.T) { + socketPath := filepath.Join(t.TempDir(), "daemon.sock") + rejected := make(chan error, 1) + received := make(chan core.Event, 1) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + server := Server{ + SocketPath: socketPath, + Handler: HandlerFunc(func(_ context.Context, event core.Event) error { + received <- event + return nil + }), + ErrorHandler: func(err error) { + rejected <- err + }, + } + + errCh := make(chan error, 1) + go func() { + errCh <- server.ListenAndServe(ctx) + }() + waitForSocket(t, socketPath) + + conn, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatalf("dial socket: %v", err) + } + oversizedLine := strings.Repeat("a", MaxEventBytes+1) + "\n" + if _, err := conn.Write([]byte(oversizedLine)); err != nil { + t.Fatalf("write oversized event: %v", err) + } + _ = conn.Close() + + select { + case err := <-rejected: + if err == nil { + t.Fatal("error handler received nil") + } + if !strings.Contains(err.Error(), "event line exceeds") { + t.Fatalf("rejected error = %v, want event size error", err) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for oversized event rejection") + } + + select { + case event := <-received: + t.Fatalf("handler received oversized event: %s", event.EventID) + default: + } + + cancel() + if err := <-errCh; err != nil { + t.Fatalf("ListenAndServe returned error after cancellation: %v", err) + } +} + func TestPrepareSocketRejectsNonSocket(t *testing.T) { path := filepath.Join(t.TempDir(), "daemon.sock") if err := os.WriteFile(path, []byte("not a socket"), 0o600); err != nil {