Message tracing requires client publish permission for Nats-Trace-Dest

Signed-off-by: Neil Twigg <neil@nats.io>

Author: neilalexander

server/client.go

@@ -2739,8 +2739,8 @@ func (c *client) processHeaderPub(arg, remaining []byte) error {
 		// Do this only for CLIENT connections.
 		if c.kind == CLIENT && c.pa.hdr > 0 && len(remaining) > 0 {
 			hdr := remaining[:min(len(remaining), c.pa.hdr)]
-			if td := getHeader(MsgTraceDest, hdr); len(td) > 0 {
-				c.initAndSendIngressErrEvent(hdr, string(td), ErrMaxPayload)
+			if td, ok := c.allowedMsgTraceDest(hdr, false); ok && td != _EMPTY_ {
+				c.initAndSendIngressErrEvent(hdr, td, ErrMaxPayload)
 			}
 		}
 		c.maxPayloadViolation(c.pa.size, maxPayload)
@@ -3980,6 +3980,41 @@ func (c *client) pubAllowed(subject string) bool {
 	return c.pubAllowedFullCheck(subject, true, false)
 }
 
+// allowedMsgTraceDest returns the trace destination if present and authorized.
+// It only considers static publish permissions and does not consume dynamic
+// reply permissions because the client is not publishing the trace event itself.
+func (c *client) allowedMsgTraceDest(hdr []byte, hasLock bool) (string, bool) {
+	if len(hdr) == 0 {
+		return _EMPTY_, true
+	}
+	td := sliceHeader(MsgTraceDest, hdr)
+	if len(td) == 0 {
+		return _EMPTY_, true
+	}
+	dest := bytesToString(td)
+	if c.kind == CLIENT {
+		if hasGWRoutedReplyPrefix(td) {
+			return dest, false
+		}
+		var acc *Account
+		var srv *Server
+		if !hasLock {
+			c.mu.Lock()
+		}
+		acc, srv = c.acc, c.srv
+		if !hasLock {
+			c.mu.Unlock()
+		}
+		if bytes.HasPrefix(td, clientNRGPrefix) && srv != nil && acc != srv.SystemAccount() {
+			return dest, false
+		}
+	}
+	if c.perms != nil && (c.perms.pub.allow != nil || c.perms.pub.deny != nil) && !c.pubAllowedFullCheck(dest, false, hasLock) {
+		return dest, false
+	}
+	return dest, true
+}
+
 // pubAllowedFullCheck checks on all publish permissioning depending
 // on the flag for dynamic reply permissions.
 func (c *client) pubAllowedFullCheck(subject string, fullCheck, hasLock bool) bool {
@@ -4115,10 +4150,19 @@ func (c *client) processInboundClientMsg(msg []byte) (bool, bool) {
 	genidAddr := &acc.sl.genid
 
 	// Check pub permissions
-	if c.perms != nil && (c.perms.pub.allow != nil || c.perms.pub.deny != nil) && !c.pubAllowedFullCheck(string(c.pa.subject), true, true) {
-		c.mu.Unlock()
-		c.pubPermissionViolation(c.pa.subject)
-		return false, true
+	if c.perms != nil && (c.perms.pub.allow != nil || c.perms.pub.deny != nil) {
+		if !c.pubAllowedFullCheck(string(c.pa.subject), true, true) {
+			c.mu.Unlock()
+			c.pubPermissionViolation(c.pa.subject)
+			return false, true
+		}
+	}
+	if c.pa.hdr > 0 {
+		if td, ok := c.allowedMsgTraceDest(msg[:c.pa.hdr], true); !ok {
+			c.mu.Unlock()
+			c.pubPermissionViolation(stringToBytes(td))
+			return false, true
+		}
 	}
 	c.mu.Unlock()
 

server/msgtrace.go

@@ -367,6 +367,13 @@ func (c *client) initMsgTrace() *msgTrace {
 			}
 		}
 		dest = getHdrVal(MsgTraceDest)
+		if c.kind == CLIENT {
+			if td, ok := c.allowedMsgTraceDest(hdr, false); !ok {
+				return nil
+			} else if td != _EMPTY_ {
+				dest = td
+			}
+		}
 		// Check the destination to see if this is a valid public subject.
 		if !IsValidPublishSubject(dest) {
 			// We still have to return a msgTrace object (if traceOnly is set)

server/msgtrace_test.go

@@ -395,6 +395,58 @@ func TestMsgTraceIngressMaxPayloadErrorDoesNotScanPayloadForTraceDest(t *testing
 	}
 }
 
+func TestMsgTraceIngressMaxPayloadErrorRequiresPublishPermissionForTraceDest(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: 127.0.0.1:-1
+		max_payload: 1024
+		accounts {
+			A {
+				users: [
+					{
+						user: tracer
+						password: pwd
+						permissions {
+							subscribe: ["my.trace.subj"]
+							publish: ["my.trace.subj"]
+						}
+					},
+					{
+						user: pub
+						password: pwd
+						permissions {
+							publish: ["foo"]
+						}
+					}
+				]
+			}
+		}
+	`))
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nct := natsConnect(t, s.ClientURL(), nats.UserInfo("tracer", "pwd"))
+	defer nct.Close()
+	traceSub := natsSubSync(t, nct, "my.trace.subj")
+	natsFlush(t, nct)
+	checkSubInterest(t, s, "A", "my.trace.subj", time.Second)
+
+	c, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", o.Port))
+	require_NoError(t, err)
+	defer c.Close()
+
+	_, err = c.Write([]byte("CONNECT {\"user\":\"pub\",\"pass\":\"pwd\",\"protocol\":1,\"headers\":true,\"no_responders\":true}\r\n"))
+	require_NoError(t, err)
+
+	hdr := fmt.Sprintf("%s%s:%s\r\n\r\n", hdrLine, MsgTraceDest, traceSub.Subject)
+	hPub := fmt.Sprintf("HPUB foo %d 2048\r\n%sAAAAAAAAAAAAAAAAAA...", len(hdr), hdr)
+	_, err = c.Write([]byte(hPub))
+	require_NoError(t, err)
+
+	if traceMsg, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
+		t.Fatalf("Should not have received trace message: %s", traceMsg.Data)
+	}
+}
+
 func TestMsgTraceIngressErrors(t *testing.T) {
 	conf := createConfFile(t, []byte(`
 		port: -1
@@ -407,7 +459,7 @@ func TestMsgTraceIngressErrors(t *testing.T) {
 						permissions {
 							subscribe: ["my.trace.subj", "foo"]
 							publish {
-								allow: ["foo", "bar.>"]
+								allow: ["foo", "bar.>", "my.trace.subj"]
 								deny: ["bar.baz"]
 							}
 						}
@@ -487,7 +539,7 @@ func TestMsgTraceEgressErrors(t *testing.T) {
 								deny: "bar.bat"
 							}
 							publish {
-								allow: ["foo", "bar.>"]
+								allow: ["foo", "bar.>", "my.trace.subj"]
 								deny: ["bar.baz"]
 							}
 						}
@@ -633,6 +685,132 @@ func TestMsgTraceEgressErrors(t *testing.T) {
 	}
 }
 
+func TestMsgTraceIngressRequiresPublishPermissionForTraceDest(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		port: -1
+		accounts {
+			A {
+				users: [
+					{
+						user: tracer
+						password: pwd
+						permissions {
+							subscribe: ["my.trace.subj", "foo"]
+							publish: ["my.trace.subj"]
+						}
+					},
+					{
+						user: pub
+						password: pwd
+						permissions {
+							publish: ["foo"]
+						}
+					}
+				]
+			}
+		}
+	`))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc := natsConnect(t, s.ClientURL(), nats.UserInfo("tracer", "pwd"))
+	defer nc.Close()
+
+	tsub := natsSubSync(t, nc, "my.trace.subj")
+	asub := natsSubSync(t, nc, "foo")
+	natsFlush(t, nc)
+
+	ncp, err := nats.Connect(
+		s.ClientURL(),
+		nats.UserInfo("pub", "pwd"),
+		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, _ error) {}),
+	)
+	require_NoError(t, err)
+	defer ncp.Close()
+
+	msg := nats.NewMsg("foo")
+	msg.Header.Set(MsgTraceDest, tsub.Subject)
+	msg.Data = []byte("hello")
+
+	require_NoError(t, ncp.PublishMsg(msg))
+	natsFlush(t, ncp)
+
+	err = ncp.LastError()
+	require_Error(t, err)
+	require_Contains(t, err.Error(), fmt.Sprintf("Permissions Violation for Publish to %q", tsub.Subject))
+
+	if m, err := asub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
+		t.Fatalf("Did not expect application message, got %v / %v", m, err)
+	}
+	if tm, err := tsub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
+		t.Fatalf("Did not expect trace message, got %v / %v", tm, err)
+	}
+}
+
+func TestMsgTraceIngressRejectsReservedTraceDest(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		port: -1
+		accounts {
+			A {
+				users: [
+					{
+						user: sub
+						password: pwd
+						permissions {
+							subscribe: ["foo"]
+						}
+					},
+					{
+						user: pub
+						password: pwd
+						permissions {
+							publish: [">"]
+						}
+					},
+					{
+						user: pub2
+						password: pwd
+					}
+				]
+			}
+		}
+	`))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	ncs := natsConnect(t, s.ClientURL(), nats.UserInfo("sub", "pwd"))
+	defer ncs.Close()
+	asub := natsSubSync(t, ncs, "foo")
+	natsFlush(t, ncs)
+
+	for _, user := range []string{"pub", "pub2"} {
+		ncp, err := nats.Connect(
+			s.ClientURL(),
+			nats.UserInfo(user, "pwd"),
+			nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, _ error) {}),
+		)
+		require_NoError(t, err)
+		defer ncp.Close()
+
+		for _, traceDest := range []string{gwReplyPrefix + "trace", "$NRG.trace"} {
+			msg := nats.NewMsg("foo")
+			msg.Header.Set(MsgTraceDest, traceDest)
+			msg.Data = []byte("hello")
+
+			require_NoError(t, ncp.PublishMsg(msg))
+			natsFlush(t, ncp)
+
+			err = ncp.LastError()
+			require_Error(t, err)
+			require_Contains(t, err.Error(), fmt.Sprintf("Permissions Violation for Publish to %q", traceDest))
+
+			if m, err := asub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
+				t.Fatalf("Did not expect application message, got %v / %v", m, err)
+			}
+		}
+	}
+}
+
 func TestMsgTraceWithQueueSub(t *testing.T) {
 	o := DefaultOptions()
 	s := RunServer(o)