[ADDED] Import/Export.AllowTrace and Account.TraceDest (#216)

This is needed for the NATS Server distributed message tracing
feature.

This PR updates validation of a Subject to check for `.` at the
first or last position and consecutive ones.

The validation of the Account.TraceDest also checks that the
destination does not have wildcards since this is supposed to
be a publish subject.

The Import/Export validation ensures that AllowTrace is used
by the correct type (service or stream)

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

* Added container for message trace options

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

---------

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

v2/account_claims.go

@@ -17,6 +17,7 @@ package jwt
 
 import (
 	"errors"
+	"fmt"
 	"sort"
 	"time"
 
@@ -229,10 +230,16 @@ type Account struct {
 	DefaultPermissions Permissions           `json:"default_permissions,omitempty"`
 	Mappings           Mapping               `json:"mappings,omitempty"`
 	Authorization      ExternalAuthorization `json:"authorization,omitempty"`
+	Trace              *MsgTrace             `json:"trace,omitempty"`
 	Info
 	GenericFields
 }
 
+// MsgTrace holds distributed message tracing configuration
+type MsgTrace struct {
+	Destination Subject `json:"dest,omitempty"`
+}
+
 // Validate checks if the account is valid, based on the wrapper
 func (a *Account) Validate(acct *AccountClaims, vr *ValidationResults) {
 	a.Imports.Validate(acct.Subject, vr)
@@ -241,6 +248,16 @@ func (a *Account) Validate(acct *AccountClaims, vr *ValidationResults) {
 	a.DefaultPermissions.Validate(vr)
 	a.Mappings.Validate(vr)
 	a.Authorization.Validate(vr)
+	if a.Trace != nil {
+		tvr := CreateValidationResults()
+		a.Trace.Destination.Validate(tvr)
+		if !tvr.IsEmpty() {
+			vr.AddError(fmt.Sprintf("the account Trace.Destination %s", tvr.Issues[0].Description))
+		}
+		if a.Trace.Destination.HasWildCards() {
+			vr.AddError("the account Trace.Destination subject %q is not a valid publish subject", a.Trace.Destination)
+		}
+	}
 
 	if !a.Limits.IsEmpty() && a.Limits.Imports >= 0 && int64(len(a.Imports)) > a.Limits.Imports {
 		vr.AddError("the account contains more imports than allowed by the operator")

v2/account_claims_test.go

@@ -876,3 +876,37 @@ func TestAccountClaims_GetTags(t *testing.T) {
 		t.Fatal("expected tag bar")
 	}
 }
+
+func TestAccountClaimsTraceDest(t *testing.T) {
+	akp := createAccountNKey(t)
+	apk := publicKey(akp, t)
+
+	account := NewAccountClaims(apk)
+	for i, test := range []struct {
+		name        string
+		invalidSubj Subject
+		expectErr   bool
+	}{
+		{"trace not specified", "", false},
+		{"trace created but with empty destination", "", true},
+		{"trace dest has spaces", "invalid dest", true},
+		{"trace dest start with a dot", ".invalid.dest", true},
+		{"trace dest ends with a dot", "invalid.dest.", true},
+		{"trace dest has consecutive dots", "invalid..dest", true},
+		{"trace dest invalid publish dest", "invalid.publish.*.dest", true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			if i > 0 {
+				account.Trace = &MsgTrace{Destination: test.invalidSubj}
+			}
+			vr := CreateValidationResults()
+			account.Validate(vr)
+
+			if test.expectErr && vr.IsEmpty() {
+				t.Fatal("account validation should have failed")
+			} else if !test.expectErr && !vr.IsEmpty() {
+				t.Fatalf("account validation should not have failed, got %+v", vr.Issues)
+			}
+		})
+	}
+}

v2/exports.go

@@ -119,6 +119,7 @@ type Export struct {
 	Latency              *ServiceLatency `json:"service_latency,omitempty"`
 	AccountTokenPosition uint            `json:"account_token_position,omitempty"`
 	Advertise            bool            `json:"advertise,omitempty"`
+	AllowTrace           bool            `json:"allow_trace,omitempty"`
 	Info
 }
 
@@ -160,8 +161,13 @@ func (e *Export) Validate(vr *ValidationResults) {
 	if e.IsService() && !e.IsSingleResponse() && !e.IsChunkedResponse() && !e.IsStreamResponse() {
 		vr.AddError("invalid response type for service: %q", e.ResponseType)
 	}
-	if e.IsStream() && e.ResponseType != "" {
-		vr.AddError("invalid response type for stream: %q", e.ResponseType)
+	if e.IsStream() {
+		if e.ResponseType != "" {
+			vr.AddError("invalid response type for stream: %q", e.ResponseType)
+		}
+		if e.AllowTrace {
+			vr.AddError("AllowTrace only valid for service export")
+		}
 	}
 	if e.Latency != nil {
 		if !e.IsService() {

v2/exports_test.go

@@ -17,6 +17,7 @@ package jwt
 
 import (
 	"sort"
+	"strings"
 	"testing"
 	"time"
 
@@ -440,3 +441,24 @@ func TestExport_ResponseThreshold(t *testing.T) {
 		t.Fatal("expected this to fail due to negative duration")
 	}
 }
+
+func TestExportAllowTrace(t *testing.T) {
+	// AllowTrace is only applicable to ServiceExport
+	e := &Export{Subject: "foo", Type: Stream, AllowTrace: true}
+	vr := CreateValidationResults()
+	e.Validate(vr)
+	if vr.IsEmpty() {
+		t.Fatalf("AllowTrace on stream should have an validation issue")
+	}
+	issue := vr.Issues[0]
+	if !strings.Contains(issue.Description, "AllowTrace only valid for service export") {
+		t.Fatalf("AllowTrace should be valid only for service export, got %q", issue.Description)
+	}
+
+	e.Type = Service
+	vr = CreateValidationResults()
+	e.Validate(vr)
+	if !vr.IsEmpty() {
+		t.Fatalf("validation should have been ok, got %+v", vr.Issues)
+	}
+}

v2/imports.go

@@ -40,6 +40,7 @@ type Import struct {
 	LocalSubject RenamingSubject `json:"local_subject,omitempty"`
 	Type         ExportType      `json:"type,omitempty"`
 	Share        bool            `json:"share,omitempty"`
+	AllowTrace   bool            `json:"allow_trace,omitempty"`
 }
 
 // IsService returns true if the import is of type service
@@ -66,6 +67,9 @@ func (i *Import) Validate(actPubKey string, vr *ValidationResults) {
 	if !i.IsService() && !i.IsStream() {
 		vr.AddError("invalid import type: %q", i.Type)
 	}
+	if i.IsService() && i.AllowTrace {
+		vr.AddError("AllowTrace only valid for stream import")
+	}
 
 	if i.Account == "" {
 		vr.AddError("account to import from is not specified")

v2/imports_test.go

@@ -498,3 +498,27 @@ func TestImports_Validate(t *testing.T) {
 		}
 	}
 }
+
+func TestImportAllowTrace(t *testing.T) {
+	ak2 := createAccountNKey(t)
+	akp2 := publicKey(ak2, t)
+
+	// AllowTrace is only applicable to StreamImport
+	i := &Import{Subject: "foo", Account: akp2, Type: Service, AllowTrace: true}
+	vr := CreateValidationResults()
+	i.Validate("", vr)
+	if vr.IsEmpty() {
+		t.Fatalf("AllowTrace on service should have an validation issue")
+	}
+	issue := vr.Issues[0]
+	if !strings.Contains(issue.Description, "AllowTrace only valid for stream import") {
+		t.Fatalf("AllowTrace should be valid only for stream import, got %q", issue.Description)
+	}
+
+	i.Type = Stream
+	vr = CreateValidationResults()
+	i.Validate("", vr)
+	if !vr.IsEmpty() {
+		t.Fatalf("validation should have been ok, got %+v", vr.Issues)
+	}
+}

v2/types.go

@@ -177,10 +177,18 @@ func (s Subject) Validate(vr *ValidationResults) {
 	v := string(s)
 	if v == "" {
 		vr.AddError("subject cannot be empty")
+		// No other checks after that make sense
+		return
 	}
 	if strings.Contains(v, " ") {
 		vr.AddError("subject %q cannot have spaces", v)
 	}
+	if v[0] == '.' || v[len(v)-1] == '.' {
+		vr.AddError("subject %q cannot start or end with a `.`", v)
+	}
+	if strings.Contains(v, "..") {
+		vr.AddError("subject %q cannot contain consecutive `.`", v)
+	}
 }
 
 func (s Subject) countTokenWildcards() int {

v2/types_test.go

@@ -186,6 +186,27 @@ func TestSubjectValid(t *testing.T) {
 		t.Fatalf("Subjects cannot have spaces")
 	}
 
+	s = ".start.with.dot"
+	vr = CreateValidationResults()
+	s.Validate(vr)
+	if vr.IsEmpty() || !strings.Contains(vr.Issues[0].Description, "start or end with a `.`") {
+		t.Fatalf("Did not get expected failure: %+v", vr.Issues)
+	}
+
+	s = "end.with.dot."
+	vr = CreateValidationResults()
+	s.Validate(vr)
+	if vr.IsEmpty() || !strings.Contains(vr.Issues[0].Description, "start or end with a `.`") {
+		t.Fatalf("Did not get expected failure: %+v", vr.Issues)
+	}
+
+	s = "consecutive..dot"
+	vr = CreateValidationResults()
+	s.Validate(vr)
+	if vr.IsEmpty() || !strings.Contains(vr.Issues[0].Description, "consecutive `.`") {
+		t.Fatalf("Did not get expected failure: %+v", vr.Issues)
+	}
+
 	s = "one"
 	vr = CreateValidationResults()
 	s.Validate(vr)