br: redact ak/sk in logging (#55622)

br/pkg/streamhelper/BUILD.bazel

@@ -30,6 +30,7 @@ go_library(
         "//pkg/util/codec",
         "//pkg/util/engine",
         "//pkg/util/mathutil",
+        "//pkg/util/redact",
         "@com_github_gogo_protobuf//proto",
         "@com_github_golang_protobuf//proto",
         "@com_github_google_uuid//:uuid",
@@ -68,7 +69,7 @@ go_test(
     ],
     flaky = True,
     race = "on",
-    shard_count = 32,
+    shard_count = 33,
     deps = [
         ":streamhelper",
         "//br/pkg/errors",
@@ -81,6 +82,7 @@ go_test(
         "//pkg/kv",
         "//pkg/tablecodec",
         "//pkg/util/codec",
+        "//pkg/util/redact",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_pingcap_kvproto//pkg/brpb",

br/pkg/streamhelper/advancer.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/metrics"
+	"github.com/pingcap/tidb/pkg/util/redact"
 	tikvstore "github.com/tikv/client-go/v2/kv"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
@@ -437,7 +438,7 @@ func (c *CheckpointAdvancer) onTaskEvent(ctx context.Context, e TaskEvent) error
 		if err != nil {
 			log.Warn("failed to upload service GC safepoint, skipping.", logutil.ShortError(err))
 		}
-		log.Info("added event", zap.Stringer("task", e.Info),
+		log.Info("added event", zap.Stringer("task", redact.TaskInfoRedacted{Info: e.Info}),
 			zap.Stringer("ranges", logutil.StringifyKeys(c.taskRange)), zap.Uint64("current-checkpoint", p))
 	case EventDel:
 		utils.LogBackupTaskCountDec()

br/pkg/streamhelper/advancer_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/pingcap/tidb/br/pkg/streamhelper/config"
 	"github.com/pingcap/tidb/br/pkg/streamhelper/spans"
 	"github.com/pingcap/tidb/pkg/kv"
+	"github.com/pingcap/tidb/pkg/util/redact"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/oracle"
@@ -822,3 +823,53 @@ func TestSubscriptionPanic(t *testing.T) {
 	cancel()
 	wg.Wait()
 }
+
+func TestRedactBackend(t *testing.T) {
+	info := new(backup.StreamBackupTaskInfo)
+	info.Name = "test"
+	info.Storage = &backup.StorageBackend{
+		Backend: &backup.StorageBackend_S3{
+			S3: &backup.S3{
+				Endpoint:        "http://",
+				Bucket:          "test",
+				Prefix:          "test",
+				AccessKey:       "12abCD!@#[]{}?/\\",
+				SecretAccessKey: "12abCD!@#[]{}?/\\",
+			},
+		},
+	}
+
+	redacted := redact.TaskInfoRedacted{Info: info}
+	require.Equal(t, "storage:<s3:<endpoint:\"http://\" bucket:\"test\" prefix:\"test\" access_key:\"[REDACTED]\" secret_access_key:\"[REDACTED]\" > > name:\"test\" ", redacted.String())
+
+	info.Storage = &backup.StorageBackend{
+		Backend: &backup.StorageBackend_Gcs{
+			Gcs: &backup.GCS{
+				Endpoint:        "http://",
+				Bucket:          "test",
+				Prefix:          "test",
+				CredentialsBlob: "12abCD!@#[]{}?/\\",
+			},
+		},
+	}
+	redacted = redact.TaskInfoRedacted{Info: info}
+	require.Equal(t, "storage:<gcs:<endpoint:\"http://\" bucket:\"test\" prefix:\"test\" CredentialsBlob:\"[REDACTED]\" > > name:\"test\" ", redacted.String())
+
+	info.Storage = &backup.StorageBackend{
+		Backend: &backup.StorageBackend_AzureBlobStorage{
+			AzureBlobStorage: &backup.AzureBlobStorage{
+				Endpoint:  "http://",
+				Bucket:    "test",
+				Prefix:    "test",
+				SharedKey: "12abCD!@#[]{}?/\\",
+				AccessSig: "12abCD!@#[]{}?/\\",
+				EncryptionKey: &backup.AzureCustomerKey{
+					EncryptionKey:       "12abCD!@#[]{}?/\\",
+					EncryptionKeySha256: "12abCD!@#[]{}?/\\",
+				},
+			},
+		},
+	}
+	redacted = redact.TaskInfoRedacted{Info: info}
+	require.Equal(t, "storage:<azure_blob_storage:<endpoint:\"http://\" bucket:\"test\" prefix:\"test\" shared_key:\"[REDACTED]\" access_sig:\"[REDACTED]\" encryption_key:<[REDACTED]> > > name:\"test\" ", redacted.String())
+}

pkg/util/redact/BUILD.bazel

@@ -8,5 +8,6 @@ go_library(
     deps = [
         "//pkg/util/intest",
         "@com_github_pingcap_errors//:errors",
+        "@com_github_pingcap_kvproto//pkg/brpb",
     ],
 )

pkg/util/redact/redact.go

@@ -22,14 +22,22 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"regexp"
 	"strings"
 
 	"github.com/pingcap/errors"
+	backup "github.com/pingcap/kvproto/pkg/brpb"
 	"github.com/pingcap/tidb/pkg/util/intest"
 )
 
 var (
-	_ fmt.Stringer = redactStringer{}
+	_                 fmt.Stringer = redactStringer{}
+	reAccessKey                    = regexp.MustCompile(`access_key:\"[^\"]*\"`)
+	reSecretAccessKey              = regexp.MustCompile(`secret_access_key:\"[^\"]*\"`)
+	reSharedKey                    = regexp.MustCompile(`shared_key:\"[^\"]*\"`)
+	reCredentialsBlob              = regexp.MustCompile(`credentials_blob:\"[^\"]*\"`)
+	reAccessSig                    = regexp.MustCompile(`access_sig:\"[^\"]*\"`)
+	reEncryptKey                   = regexp.MustCompile(`encryption_key:<.*?>`)
 )
 
 // String will redact the input string according to 'mode'. Check 'tidb_redact_log': https://github.com/pingcap/tidb/blob/acf9e3128693a5a13f31027f05f4de41edf8d7b2/pkg/sessionctx/variable/sysvar.go#L2154.
@@ -218,3 +226,25 @@ func WriteRedact(build *strings.Builder, v string, redact bool) {
 	}
 	build.WriteString(v)
 }
+
+// TaskInfoRedacted is a wrapper of backup.StreamBackupTaskInfo to redact sensitive information
+type TaskInfoRedacted struct {
+	Info *backup.StreamBackupTaskInfo
+}
+
+func (TaskInfoRedacted) redact(input string) string {
+	// Replace the matched fields with redacted versions
+	output := reAccessKey.ReplaceAllString(input, `access_key:"[REDACTED]"`)
+	output = reSecretAccessKey.ReplaceAllString(output, `secret_access_key:"[REDACTED]"`)
+	output = reSharedKey.ReplaceAllString(output, `shared_key:"[REDACTED]"`)
+	output = reCredentialsBlob.ReplaceAllString(output, `CredentialsBlob:"[REDACTED]"`)
+	output = reAccessSig.ReplaceAllString(output, `access_sig:"[REDACTED]"`)
+	output = reEncryptKey.ReplaceAllString(output, `encryption_key:<[REDACTED]>`)
+
+	return output
+}
+
+// String returns the redacted string of the task info
+func (t TaskInfoRedacted) String() string {
+	return t.redact(t.Info.String())
+}