Repository: nats-io/nats-server

Issue:
Messages with internal  _GR_. prefix  rejected unless explicitely allowed in leaf node connections
### Observed behavior

For a request/Reply service running on a leaf node connected to a supercluster replies would fail at the leaf node connection level (permission failure) unless _GR_.>  was explicitely allowed.
Also, when the reply subject in question was used (in a messages) more than 2s after the subcription was made the issue disappeared.

In addition to being an itnernal optimization the effect confusing as the _GR_.  prefix is conditionally applied. Sometimes its there, sometimes not (see details below). 

### Expected behavior

_GR_ prefixing (routed replies) is an internal optimization, which should not be visible to client connections. AND leaf node connections (which should be treated as client connections).

A quick check in the server code showed that the _GR_ prefix is conditionally applied when a matching subscription is in the account subscription cache (which times out after a few seconds). Both _GR_ and the account subscription cache are server internal details.

### Server and client version

2.11.3 and 2.10.29 (and earlier)

### Host environment

Linux and Windows

### Steps to reproduce

![Image](https://github.com/user-attachments/assets/40d17dff-55ac-49ea-a0c8-77ef972abe53)

- Create a topology as per the diagram(US NAT and EU NATS being a supercluster).
- Set permissions on the leaf node connections restricted to the request and reply subjects used.
- The service is a boilerplate R/R server (reply to reply subject)
- The client subscribes to the reply subject. Waits for a configurable period of time. Sends the request.
- Service replies will be silently suppressed by the leaf node.
- Check the leaf node in debug mode and find: "Not permitted to deliver to _GR_......" 
- When the client waits for 2+ seconds before sending the request the reply goes through (no _GR_ prefixing )
- Disabling the account subscriber cache should also make the effect disappear



Hints from issue comments:

Solution patch:
diff --git a/server/client.go b/server/client.go
index 006db041..be734777 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3494,10 +3494,18 @@ func (c *client) deliverMsg(prodIsMQTT bool, sub *subscription, acc *Account, su
 
 	// Check if we are a leafnode and have perms to check.
 	if client.kind == LEAF && client.perms != nil {
-		if !client.pubAllowedFullCheck(string(subject), true, true) {
+		var subjectToCheck []byte
+		if subject[0] == '_' && bytes.HasPrefix(subject, []byte(gwReplyPrefix)) {
+			subjectToCheck = subject[gwSubjectOffset:]
+		} else if subject[0] == '$' && bytes.HasPrefix(subject, []byte(oldGWReplyPrefix)) {
+			subjectToCheck = subject[oldGWReplyStart:]
+		} else {
+			subjectToCheck = subject
+		}
+		if !client.pubAllowedFullCheck(string(subjectToCheck), true, true) {
 			mt.addEgressEvent(client, sub, errMsgTracePubViolation)
 			client.mu.Unlock()
-			client.Debugf("Not permitted to deliver to %q", subject)
+			client.Debugf("Not permitted to deliver to %q", subjectToCheck)
 			return false
 		}
 	}



Test patch:
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index f00a46d8..d5ebc462 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -9975,3 +9975,115 @@ func TestLeafNodePermissionWithLiteralSubjectAndQueueInterest(t *testing.T) {
 	})
 	require_Equal(t, "OK", string(resp.Data))
 }
+
+func TestLeafNodePermissionWithGateways(t *testing.T) {
+	usConf := createConfFile(t, []byte(`
+		server_name: "US"
+		listen: "127.0.0.1:-1"
+		gateway {
+			name: "US"
+			listen: "127.0.0.1:-1"
+		}
+		accounts {
+			sys { users: [{user: sys, password: sys}] }
+			leaf { users: [{user: leaf, password: leaf}] }
+		}
+		system_account: sys
+	`))
+	us, ous := RunServerWithConfig(usConf)
+	defer us.Shutdown()
+
+	euConf := createConfFile(t, fmt.Appendf(nil, `
+		server_name: "EU"
+		listen: "127.0.0.1:-1"
+		gateway {
+			name: "EU"
+			listen: "127.0.0.1:-1"
+			gateways: [
+				{
+					name: "US"
+					urls: ["nats://127.0.0.1:%d"]
+				}
+			]
+		}
+		leafnodes {
+			listen: "127.0.0.1:-1"
+		}
+		accounts {
+			sys { users: [{user: sys, password: sys}] }
+			leaf {
+				users: [
+					{
+						user: leaf
+						password: leaf
+						permissions: {
+							publish: "bar"
+							subscribe: "foo"
+						}
+					}
+				]
+			}
+		}
+		system_account: sys
+		`, ous.Gateway.Port))
+	eu, oeu := RunServerWithConfig(euConf)
+	defer eu.Shutdown()
+
+	waitForOutboundGateways(t, us, 1, 2*time.Second)
+	waitForOutboundGateways(t, eu, 1, 2*time.Second)
+	waitForInboundGateways(t, us, 1, 2*time.Second)
+	waitForInboundGateways(t, eu, 1, 2*time.Second)
+
+	leafConf := createConfFile(t, fmt.Appendf(nil, `
+		server_name: "LEAF"
+		listen: "127.0.0.1:-1"
+		leafnodes {
+			remotes: [
+				{ url: "nats://leaf:leaf@127.0.0.1:%d" }
+			]
+		}
+	`, oeu.LeafNode.Port))
+	leaf, _ := RunServerWithConfig(leafConf)
+	defer leaf.Shutdown()
+
+	checkLeafNodeConnected(t, leaf)
+
+	// Run the service from EU leafnode.
+	ncEU := natsConnect(t, leaf.ClientURL())
+	defer ncEU.Close()
+	natsSub(t, ncEU, "foo", func(m *nats.Msg) {
+		m.Respond([]byte("response"))
+	})
+	natsFlush(t, ncEU)
+
+	// Wait for subject interest to propagate.
+	checkGWInterestOnlyModeInterestOn(t, us, "EU", "leaf", "foo")
+
+	// Connect to the US server
+	ncUS := natsConnect(t, us.ClientURL(), nats.UserInfo("leaf", "leaf"))
+	defer ncUS.Close()
+
+	// Create a subscription on "bar" and send request on "foo"
+	sub := natsSubSync(t, ncUS, "bar")
+	// Wait for subject to propagate so we know that EU server
+	// would know about the "bar" subscription.
+	checkGWInterestOnlyModeInterestOn(t, eu, "US", "leaf", "bar")
+	// Send the request and make sure we receive the reply.
+	natsPubReq(t, ncUS, "foo", "bar", []byte("request"))
+	reply := natsNexMsg(t, sub, time.Second)
+	if string(reply.Data) != "response" {
+		t.Fatalf("Invalid response: %q", reply.Data)
+	}
+	// Make sure that we don't blindly accept any reply because there
+	// is the routing protocol. So create a sub on "baz" that the leaf
+	// would not be allowed to reply to. We should not get the reply
+	// to our request.
+	sub2 := natsSubSync(t, ncUS, "baz")
+	checkGWInterestOnlyModeInterestOn(t, eu, "US", "leaf", "baz")
+	// Send the request. We should not get the message since the
+	// leaf server does not have permission to publish on "baz".
+	natsPubReq(t, ncUS, "foo", "baz", []byte("request2"))
+	if msg, err := sub2.NextMsg(250 * time.Millisecond); err == nil {
+		t.Fatalf("Should not have received the reply, got %q", msg.Data)
+	}
+}