[ci_runner] redact secrets across log chunks

enterprise/server/cmd/ci_runner/BUILD

@@ -1,6 +1,6 @@
 load("@io_bazel_rules_docker//container:container.bzl", "container_image")
 load("@io_bazel_rules_docker//go:image.bzl", "go_image")
-load("@io_bazel_rules_go//go:def.bzl", "go_binary", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_binary", "go_library", "go_test")
 
 # gazelle:default_visibility //enterprise:__subpackages__
 package(
@@ -72,3 +72,10 @@ go_image(
     binary = ":ci_runner",
     tags = ["manual"],
 )
+
+go_test(
+    name = "ci_runner_test",
+    srcs = ["main_test.go"],
+    embed = [":main"],  # keep
+    deps = ["@com_github_stretchr_testify//require"],
+)

enterprise/server/cmd/ci_runner/main.go

@@ -486,6 +486,9 @@ func (r *buildEventReporter) Stop() error {
 		r.cancelBackgroundFlush()
 		r.cancelBackgroundFlush = nil
 	}
+	if err := r.log.Flush(); err != nil {
+		return err
+	}
 	r.FlushProgress()
 
 	elapsedTimeSeconds := float64(time.Since(r.startTime)) / float64(time.Second)
@@ -938,6 +941,9 @@ type invocationLog struct {
 	lockingbuffer.LockingBuffer
 	writer        io.Writer
 	writeListener func(s string)
+
+	mu          sync.Mutex
+	partialLine bytes.Buffer
 }
 
 func newInvocationLog() *invocationLog {
@@ -947,16 +953,69 @@ func newInvocationLog() *invocationLog {
 }
 
 func (invLog *invocationLog) Write(b []byte) (int, error) {
-	output := string(b)
+	invLog.mu.Lock()
+	defer invLog.mu.Unlock()
+
+	if _, err := invLog.partialLine.Write(b); err != nil {
+		return len(b), err
+	}
 
-	redacted := redact.RedactText(output)
+	var writeErr error
 
-	invLog.writeListener(redacted)
-	_, err := invLog.writer.Write([]byte(redacted))
+	for {
+		data := invLog.partialLine.Bytes()
+		if len(data) == 0 {
+			break
+		}
+
+		idx := bytes.IndexAny(data, "\n\r")
+		if idx < 0 {
+			break
+		}
+
+		newlineLen := 1
+		newlineSuffix := data[idx : idx+newlineLen]
+
+		if data[idx] == '\r' && idx+1 < len(data) && data[idx+1] == '\n' {
+			newlineLen = 2
+			newlineSuffix = data[idx : idx+newlineLen]
+		}
+
+		lineContent := string(data[:idx])
+		newlineStr := string(newlineSuffix)
+
+		invLog.partialLine.Next(idx + newlineLen)
+
+		redacted := redact.RedactText(lineContent)
+		output := redacted + newlineStr
+
+		invLog.writeListener(output)
+		if _, err := invLog.writer.Write([]byte(output)); err != nil && writeErr == nil {
+			writeErr = err
+		}
+	}
 
 	// Return the size of the original buffer even if a redacted size was written,
 	// or clients will return a short write error
-	return len(b), err
+	return len(b), writeErr
+}
+
+func (invLog *invocationLog) Flush() error {
+	invLog.mu.Lock()
+	defer invLog.mu.Unlock()
+
+	if invLog.partialLine.Len() == 0 {
+		return nil
+	}
+
+	line := invLog.partialLine.String()
+	invLog.partialLine.Reset()
+
+	redacted := redact.RedactText(line)
+
+	invLog.writeListener(redacted)
+	_, err := invLog.writer.Write([]byte(redacted))
+	return err
 }
 
 func (invLog *invocationLog) Println(vals ...interface{}) {

enterprise/server/cmd/ci_runner/main_test.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"bytes"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func newTestInvocationLog() (*invocationLog, *bytes.Buffer) {
+	invLog := &invocationLog{}
+	invLog.writeListener = func(string) {}
+	buf := &bytes.Buffer{}
+	invLog.writer = io.MultiWriter(&invLog.LockingBuffer, buf)
+	return invLog, buf
+}
+
+func TestInvocationLogRedactsRemoteExecHeaderAcrossWrites(t *testing.T) {
+	invLog, buf := newTestInvocationLog()
+
+	firstChunk := "common --remote_exec_header=secret-token"
+	n, err := invLog.Write([]byte(firstChunk))
+	require.NoError(t, err)
+	require.Equal(t, len(firstChunk), n)
+	require.Equal(t, "", buf.String(), "should not flush before newline")
+
+	secondChunk := "-continued\n"
+	n, err = invLog.Write([]byte(secondChunk))
+	require.NoError(t, err)
+	require.Equal(t, len(secondChunk), n)
+	require.Equal(t, "common --remote_exec_header=<REDACTED>\n", buf.String())
+}
+
+func TestInvocationLogRedactsMultipleLines(t *testing.T) {
+	invLog, buf := newTestInvocationLog()
+
+	chunk := "line1 --remote_exec_header=first-secret\nline2 --remote_exec_header=second-secret\n"
+	n, err := invLog.Write([]byte(chunk))
+	require.NoError(t, err)
+	require.Equal(t, len(chunk), n)
+	require.Equal(t,
+		"line1 --remote_exec_header=<REDACTED>\nline2 --remote_exec_header=<REDACTED>\n",
+		buf.String())
+}
+
+func TestInvocationLogFlushesPartialLine(t *testing.T) {
+	invLog, buf := newTestInvocationLog()
+
+	chunk := "common --remote_exec_header=secret-token"
+	n, err := invLog.Write([]byte(chunk))
+	require.NoError(t, err)
+	require.Equal(t, len(chunk), n)
+	require.Equal(t, "", buf.String(), "should not flush before newline")
+
+	require.NoError(t, invLog.Flush())
+	require.Equal(t, "common --remote_exec_header=<REDACTED>", buf.String())
+}