Fix metadata parsing

common/fe-ste-models.go

@@ -23,6 +23,7 @@ package common
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math"
 	"os"
@@ -674,9 +675,11 @@ type TransferStatus int32 // Must be 32-bit for atomic operations; negative #s r
 func (TransferStatus) NotStarted() TransferStatus { return TransferStatus(0) }
 
 // TODO confirm whether this is actually needed
-//   Outdated:
-//     Transfer started & at least 1 chunk has successfully been transferred.
-//     Used to resume a transfer that started to avoid transferring all chunks thereby improving performance
+//
+//	Outdated:
+//	  Transfer started & at least 1 chunk has successfully been transferred.
+//	  Used to resume a transfer that started to avoid transferring all chunks thereby improving performance
+//
 // Update(Jul 2020): This represents the state of transfer as soon as the file is scheduled.
 func (TransferStatus) Started() TransferStatus { return TransferStatus(1) }
 
@@ -1075,13 +1078,53 @@ func UnMarshalToCommonMetadata(metadataString string) (Metadata, error) {
 func StringToMetadata(metadataString string) (Metadata, error) {
 	metadataMap := Metadata{}
 	if len(metadataString) > 0 {
-		for _, keyAndValue := range strings.Split(metadataString, ";") { // key/value pairs are separated by ';'
-			kv := strings.Split(keyAndValue, "=") // key/value are separated by '='
-			// what if '=' not present?
-			if len(kv) != 2 {
-				return metadataMap, fmt.Errorf("invalid metadata string passed")
+		cKey := ""
+		cVal := ""
+		keySet := false
+		ignoreRules := false
+
+		addchar := func(c rune) {
+			if !keySet {
+				cKey += string(c)
+			} else {
+				cVal += string(c)
+			}
+		}
+		for _, c := range metadataString {
+			if ignoreRules {
+				addchar(c)
+				ignoreRules = false
+			} else {
+				switch c {
+				case '=':
+					if keySet {
+						addchar(c)
+					} else {
+						keySet = true
+					}
+
+				case ';':
+					if !keySet {
+						return Metadata{}, errors.New("metadata names must conform to C# naming rules (https://learn.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#metadata-names)")
+					}
+
+					metadataMap[cKey] = cVal
+					cKey = ""
+					cVal = ""
+					keySet = false
+					ignoreRules = false
+
+				case '\\':
+					ignoreRules = true // ignore the rules on the next character
+
+				default:
+					addchar(c)
+				}
 			}
-			metadataMap[kv[0]] = kv[1]
+		}
+
+		if cKey != "" {
+			metadataMap[cKey] = cVal
 		}
 	}
 	return metadataMap, nil

e2etest/zt_preserve_properties_test.go

@@ -42,10 +42,10 @@ func TestProperties_NameValueMetadataIsPreservedS2S(t *testing.T) {
 }
 
 func TestProperties_NameValueMetadataCanBeUploaded(t *testing.T) {
-	expectedMap := map[string]string{"foo": "abc", "bar": "def"}
+	expectedMap := map[string]string{"foo": "abc", "bar": "def", "baz": "te=s;t"}
 	RunScenarios(t, eOperation.Copy(), eTestFromTo.AllUploads(), eValidate.Auto(), anonymousAuthOnly, anonymousAuthOnly, params{
 		recursive: true,
-		metadata:  "foo=abc;bar=def",
+		metadata:  "foo=abc;bar=def;baz=te=s\\;t",
 	}, nil, testFiles{
 		defaultSize: "1K",
 		shouldTransfer: []interface{}{

ste/mgr-JobPartMgr.go

@@ -377,9 +377,10 @@ func (jpm *jobPartMgr) ScheduleTransfers(jobCtx context.Context) {
 	metadataString := string(dstData.Metadata[:dstData.MetadataLength])
 	jpm.metadata = common.Metadata{}
 	if len(metadataString) > 0 {
-		for _, keyAndValue := range strings.Split(metadataString, ";") { // key/value pairs are separated by ';'
-			kv := strings.Split(keyAndValue, "=") // key/value are separated by '='
-			jpm.metadata[kv[0]] = kv[1]
+		var err error
+		jpm.metadata, err = common.StringToMetadata(metadataString)
+		if err != nil {
+			panic("sanity check: metadata string should be valid at this point: " + metadataString)
 		}
 	}
 	blobTagsStr := string(dstData.BlobTags[:dstData.BlobTagsLength])