Restrict DC for partition-exports writes

There are two restrictions:
- Writes from the primary DC which explicitly target a secondary DC.
- Writes to a secondary DC that do not explicitly target the primary DC.

The first restriction is because the config entry is not supported in
secondary datacenters.

The second restriction is to prevent the scenario where a user writes
the config entry to a secondary DC, the write gets forwarded to the
primary, but then the config entry does not apply in the secondary.
This makes the scope more explicit.

agent/consul/config_endpoint.go

@@ -54,6 +54,11 @@ func (c *ConfigEntry) Apply(args *structs.ConfigEntryRequest, reply *bool) error
 		return err
 	}
 
+	err := gateWriteToSecondary(args.Datacenter, c.srv.config.Datacenter, c.srv.config.PrimaryDatacenter, args.Entry.GetKind())
+	if err != nil {
+		return err
+	}
+
 	// Ensure that all config entry writes go to the primary datacenter. These will then
 	// be replicated to all the other datacenters.
 	args.Datacenter = c.srv.config.PrimaryDatacenter
@@ -586,6 +591,29 @@ func (c *ConfigEntry) ResolveServiceConfig(args *structs.ServiceConfigRequest, r
 		})
 }
 
+func gateWriteToSecondary(targetDC, localDC, primaryDC, kind string) error {
+	// Partition exports are gated from interactions from secondary DCs
+	// because non-default partitions cannot be created in secondaries
+	// and services cannot be exported to another datacenter.
+	if kind != structs.PartitionExports {
+		return nil
+	}
+
+	if primaryDC == "" {
+		primaryDC = localDC
+	}
+
+	switch {
+	case targetDC == "" && localDC != primaryDC:
+		return fmt.Errorf("partition-exports writes in secondary datacenters must target the primary datacenter explicitly.")
+
+	case targetDC != "" && targetDC != primaryDC:
+		return fmt.Errorf("partition-exports writes must not target secondary datacenters.")
+
+	}
+	return nil
+}
+
 // preflightCheck is meant to have kind-specific system validation outside of
 // content validation. The initial use case is restricting the ability to do
 // writes of service-intentions until the system is finished migration.

agent/consul/config_endpoint_test.go

@@ -1,6 +1,7 @@
 package consul
 
 import (
+	"fmt"
 	"os"
 	"sort"
 	"testing"
@@ -2058,3 +2059,137 @@ func runStep(t *testing.T, name string, fn func(t *testing.T)) {
 		t.FailNow()
 	}
 }
+
+func Test_gateWriteToSecondary(t *testing.T) {
+	type args struct {
+		targetDC  string
+		localDC   string
+		primaryDC string
+		kind      string
+	}
+	type testCase struct {
+		name    string
+		args    args
+		wantErr string
+	}
+
+	run := func(t *testing.T, tc testCase) {
+		err := gateWriteToSecondary(tc.args.targetDC, tc.args.localDC, tc.args.primaryDC, tc.args.kind)
+		if tc.wantErr != "" {
+			require.Error(t, err)
+			require.Contains(t, err.Error(), tc.wantErr)
+			return
+		}
+		require.NoError(t, err)
+	}
+
+	tt := []testCase{
+		{
+			name: "primary to primary with implicit primary and target",
+			args: args{
+				targetDC:  "",
+				localDC:   "dc1",
+				primaryDC: "",
+				kind:      structs.PartitionExports,
+			},
+		},
+		{
+			name: "primary to primary with explicit primary and implicit target",
+			args: args{
+				targetDC:  "",
+				localDC:   "dc1",
+				primaryDC: "dc1",
+				kind:      structs.PartitionExports,
+			},
+		},
+		{
+			name: "primary to primary with all filled in",
+			args: args{
+				targetDC:  "dc1",
+				localDC:   "dc1",
+				primaryDC: "dc1",
+				kind:      structs.PartitionExports,
+			},
+		},
+		{
+			name: "primary to secondary with implicit primary and target",
+			args: args{
+				targetDC:  "dc2",
+				localDC:   "dc1",
+				primaryDC: "",
+				kind:      structs.PartitionExports,
+			},
+			wantErr: "writes must not target secondary datacenters",
+		},
+		{
+			name: "primary to secondary with all filled in",
+			args: args{
+				targetDC:  "dc2",
+				localDC:   "dc1",
+				primaryDC: "dc1",
+				kind:      structs.PartitionExports,
+			},
+			wantErr: "writes must not target secondary datacenters",
+		},
+		{
+			name: "secondary to secondary with all filled in",
+			args: args{
+				targetDC:  "dc2",
+				localDC:   "dc2",
+				primaryDC: "dc1",
+				kind:      structs.PartitionExports,
+			},
+			wantErr: "writes must not target secondary datacenters",
+		},
+		{
+			name: "implicit write to secondary",
+			args: args{
+				targetDC:  "",
+				localDC:   "dc2",
+				primaryDC: "dc1",
+				kind:      structs.PartitionExports,
+			},
+			wantErr: "must target the primary datacenter explicitly",
+		},
+	}
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func Test_gateWriteToSecondary_AllowedKinds(t *testing.T) {
+	type args struct {
+		targetDC  string
+		localDC   string
+		primaryDC string
+		kind      string
+	}
+
+	for _, kind := range structs.AllConfigEntryKinds {
+		if kind == structs.PartitionExports {
+			continue
+		}
+
+		t.Run(fmt.Sprintf("%s-secondary-to-secondary", kind), func(t *testing.T) {
+			tcase := args{
+				targetDC:  "",
+				localDC:   "dc2",
+				primaryDC: "dc1",
+				kind:      kind,
+			}
+			require.NoError(t, gateWriteToSecondary(tcase.targetDC, tcase.localDC, tcase.primaryDC, tcase.kind))
+		})
+
+		t.Run(fmt.Sprintf("%s-primary-to-secondary", kind), func(t *testing.T) {
+			tcase := args{
+				targetDC:  "dc2",
+				localDC:   "dc1",
+				primaryDC: "dc1",
+				kind:      kind,
+			}
+			require.NoError(t, gateWriteToSecondary(tcase.targetDC, tcase.localDC, tcase.primaryDC, tcase.kind))
+		})
+	}
+}