consul: correctly check consul acl token namespace when using consul oss

This PR fixes the Nomad Object Namespace <-> Consul ACL Token relationship
check when using Consul OSS (or Consul ENT without namespace support).

Nomad v1.1.0 introduced a regression where Nomad would fail the validation
when submitting Connect jobs and allow_unauthenticated set to true, with
Consul OSS - because it would do the namespace check against the Consul ACL
token assuming the "default" namespace, which does not work because Consul OSS
does not have namespaces.

Instead of making the bad assumption, expand the namespace check to handle
each special case explicitly.

Fixes #10718

nomad/consul.go

@@ -215,13 +215,6 @@ func (c *consulACLsAPI) CheckPermissions(ctx context.Context, namespace string,
 		return nil
 	}
 
-	// If namespace is not declared on nomad jobs, assume default consul namespace
-	// when comparing with the consul ACL token. This maintains backwards compatibility
-	// with existing connect jobs, which may already be authorized with Consul tokens.
-	if namespace == "" {
-		namespace = "default"
-	}
-
 	// lookup the token from consul
 	token, readErr := c.readToken(ctx, secretID)
 	if readErr != nil {

nomad/consul_policy.go

@@ -67,17 +67,43 @@ func (c *consulACLsAPI) isManagementToken(token *api.ACLToken) bool {
 	return false
 }
 
-// namespaceCheck is used to verify the namespace of the object matches the
-// namespace of the ACL token provided.
+// namespaceCheck is used to fail the request if the namespace of the object does
+// not match the namespace of the ACL token provided.
 //
-// exception: iff token is in the default namespace, it may contain policies
+// *exception*: if token is in the default namespace, it may contain policies
 // that extend into other namespaces using namespace_prefix, which must bypass
 // this early check and validate in the service/keystore helpers
+//
+// *exception*: if token is not in a namespace, consul namespaces are not enabled
+// and there is nothing to validate
+//
+// If the namespaces match, whether the token is allowed to perform an operation
+// is checked later.
 func namespaceCheck(namespace string, token *api.ACLToken) error {
-	if token.Namespace != "default" && token.Namespace != namespace {
+
+	switch {
+	case namespace == "" && token.Namespace == "":
+		// ACLs not enabled (if Consul ACLs are enabled, the token response from
+		// consul will always set the namespace (to "default" if otherwise unset))
+		return nil
+
+	case token.Namespace == "default":
+		// ACLs enabled, must defer to per-object checking, since the token could
+		// have namespace_prefix blocks allowing the operation
+		return nil
+
+	case namespace == token.Namespace:
+		// ACLs enabled, namespaces are the same
+		return nil
+
+	case namespace == "" && token.Namespace != "default":
+		// ACLs enabled with non-default token, but namespace on job not set, so
+		// provide a more informative error message.
+		return errors.Errorf("consul ACL token requires using namespace %q", token.Namespace)
+
+	default:
 		return errors.Errorf("consul ACL token cannot use namespace %q", namespace)
 	}
-	return nil
 }
 
 func (c *consulACLsAPI) canReadKeystore(namespace string, token *api.ACLToken) (bool, error) {

nomad/consul_policy_test.go

@@ -652,3 +652,64 @@ func TestConsulPolicy_isManagementToken(t *testing.T) {
 		require.True(t, result)
 	})
 }
+
+func TestConsulPolicy_namespaceCheck(t *testing.T) {
+	withoutNS := &api.ACLToken{Namespace: ""}
+	withDefault := &api.ACLToken{Namespace: "default"}
+	withOther := &api.ACLToken{Namespace: "other"}
+
+	// ACLs not enabled
+
+	t.Run("acl:disable ns:unset", func(t *testing.T) {
+		err := namespaceCheck("", withoutNS)
+		require.NoError(t, err)
+	})
+
+	t.Run("acl:disable ns:default", func(t *testing.T) {
+		err := namespaceCheck("default", withoutNS)
+		require.EqualError(t, err, `consul ACL token cannot use namespace "default"`)
+	})
+
+	t.Run("acl:disable ns:other", func(t *testing.T) {
+		err := namespaceCheck("other", withoutNS)
+		require.EqualError(t, err, `consul ACL token cannot use namespace "other"`)
+	})
+
+	// ACLs with "default" token
+
+	t.Run("acl:enable token:default ns:unset", func(t *testing.T) {
+		// the bypass case where a legacy job (with no namespace set) should work
+		// with the a token in the "default" consul namespace
+		err := namespaceCheck("", withDefault)
+		require.NoError(t, err)
+	})
+
+	t.Run("acl:enable token:default ns:default", func(t *testing.T) {
+		err := namespaceCheck("default", withDefault)
+		require.NoError(t, err)
+	})
+
+	t.Run("acl:enable token:default ns:other", func(t *testing.T) {
+		// the bypass case where a default token could have namespace_prefix
+		// blocks
+		err := namespaceCheck("other", withDefault)
+		require.NoError(t, err)
+	})
+
+	// ACLs with non-"default" token
+
+	t.Run("acl:enable token:other ns:unset", func(t *testing.T) {
+		err := namespaceCheck("", withOther)
+		require.EqualError(t, err, `consul ACL token requires using namespace "other"`)
+	})
+
+	t.Run("acl:enable token:other ns:default", func(t *testing.T) {
+		err := namespaceCheck("default", withOther)
+		require.EqualError(t, err, `consul ACL token cannot use namespace "default"`)
+	})
+
+	t.Run("acl:enable token:other ns:other", func(t *testing.T) {
+		err := namespaceCheck("other", withOther)
+		require.NoError(t, err)
+	})
+}