acl: check ACL against object namespace

Fix a bug where a millicious user can access or manipulate an alloc in a
namespace they don't have access to.  The allocation endpoints perform
ACL checks against the request namespace, not the allocation namespace,
and performs the allocation lookup independently from namespaces.

Here, we check that the requested can access the alloc namespace
regardless of the declared request namespace.

Ideally, we'd enforce that the declared request namespace matches
the actual allocation namespace.  Unfortunately, we haven't documented
alloc endpoints as namespaced functions; we suspect starting to enforce
this will be very disruptive and inappropriate for a nomad point
release.  As such, we maintain current behavior that doesn't require
passing the proper namespace in request.  A future major release may
start enforcing checking declared namespace.

acl/acl.go

@@ -481,3 +481,25 @@ func (a *ACL) AllowQuotaWrite() bool {
 func (a *ACL) IsManagement() bool {
 	return a.management
 }
+
+// NamespaceValidator returns a func that wraps ACL.AllowNamespaceOperation in
+// a list of operations. Returns true (allowed) if acls are disabled or if
+// *any* capabilities match.
+func NamespaceValidator(ops ...string) func(*ACL, string) bool {
+	return func(acl *ACL, ns string) bool {
+		// Always allow if ACLs are disabled.
+		if acl == nil {
+			return true
+		}
+
+		for _, op := range ops {
+			if acl.AllowNamespaceOperation(ns, op) {
+				// An operation is allowed, return true
+				return true
+			}
+		}
+
+		// No operations are allowed by this ACL, return false
+		return false
+	}
+}

client/alloc_endpoint.go

@@ -49,10 +49,15 @@ func (a *Allocations) GarbageCollectAll(args *nstructs.NodeSpecificRequest, repl
 func (a *Allocations) GarbageCollect(args *nstructs.AllocSpecificRequest, reply *nstructs.GenericResponse) error {
 	defer metrics.MeasureSince([]string{"client", "allocations", "garbage_collect"}, time.Now())
 
-	// Check submit job permissions
+	alloc, err := a.c.GetAlloc(args.AllocID)
+	if err != nil {
+		return err
+	}
+
+	// Check namespace submit job permission.
 	if aclObj, err := a.c.ResolveToken(args.AuthToken); err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilitySubmitJob) {
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilitySubmitJob) {
 		return nstructs.ErrPermissionDenied
 	}
 
@@ -68,10 +73,15 @@ func (a *Allocations) GarbageCollect(args *nstructs.AllocSpecificRequest, reply
 func (a *Allocations) Signal(args *nstructs.AllocSignalRequest, reply *nstructs.GenericResponse) error {
 	defer metrics.MeasureSince([]string{"client", "allocations", "signal"}, time.Now())
 
-	// Check alloc-lifecycle permissions
+	alloc, err := a.c.GetAlloc(args.AllocID)
+	if err != nil {
+		return err
+	}
+
+	// Check namespace alloc-lifecycle permission.
 	if aclObj, err := a.c.ResolveToken(args.AuthToken); err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
 		return nstructs.ErrPermissionDenied
 	}
 
@@ -82,9 +92,15 @@ func (a *Allocations) Signal(args *nstructs.AllocSignalRequest, reply *nstructs.
 func (a *Allocations) Restart(args *nstructs.AllocRestartRequest, reply *nstructs.GenericResponse) error {
 	defer metrics.MeasureSince([]string{"client", "allocations", "restart"}, time.Now())
 
+	alloc, err := a.c.GetAlloc(args.AllocID)
+	if err != nil {
+		return err
+	}
+
+	// Check namespace alloc-lifecycle permission.
 	if aclObj, err := a.c.ResolveToken(args.AuthToken); err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
 		return nstructs.ErrPermissionDenied
 	}
 
@@ -95,10 +111,15 @@ func (a *Allocations) Restart(args *nstructs.AllocRestartRequest, reply *nstruct
 func (a *Allocations) Stats(args *cstructs.AllocStatsRequest, reply *cstructs.AllocStatsResponse) error {
 	defer metrics.MeasureSince([]string{"client", "allocations", "stats"}, time.Now())
 
-	// Check read job permissions
+	alloc, err := a.c.GetAlloc(args.AllocID)
+	if err != nil {
+		return err
+	}
+
+	// Check read-job permission.
 	if aclObj, err := a.c.ResolveToken(args.AuthToken); err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityReadJob) {
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadJob) {
 		return nstructs.ErrPermissionDenied
 	}
 
@@ -148,6 +169,20 @@ func (a *Allocations) execImpl(encoder *codec.Encoder, decoder *codec.Decoder, e
 		return nil, structs.ErrPermissionDenied
 	}
 
+	if req.AllocID == "" {
+		return helper.Int64ToPtr(400), allocIDNotPresentErr
+	}
+	ar, err := a.c.getAllocRunner(req.AllocID)
+	if err != nil {
+		code := helper.Int64ToPtr(500)
+		if structs.IsErrUnknownAllocation(err) {
+			code = helper.Int64ToPtr(404)
+		}
+
+		return code, err
+	}
+	alloc := ar.Alloc()
+
 	aclObj, token, err := a.c.resolveTokenAndACL(req.QueryOptions.AuthToken)
 	{
 		// log access
@@ -167,20 +202,14 @@ func (a *Allocations) execImpl(encoder *codec.Encoder, decoder *codec.Decoder, e
 		)
 	}
 
-	// Check read permissions
+	// Check alloc-exec permission.
 	if err != nil {
 		return nil, err
-	} else if aclObj != nil {
-		exec := aclObj.AllowNsOp(req.QueryOptions.Namespace, acl.NamespaceCapabilityAllocExec)
-		if !exec {
-			return nil, structs.ErrPermissionDenied
-		}
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityAllocExec) {
+		return nil, structs.ErrPermissionDenied
 	}
 
 	// Validate the arguments
-	if req.AllocID == "" {
-		return helper.Int64ToPtr(400), allocIDNotPresentErr
-	}
 	if req.Task == "" {
 		return helper.Int64ToPtr(400), taskNotPresentErr
 	}
@@ -188,16 +217,6 @@ func (a *Allocations) execImpl(encoder *codec.Encoder, decoder *codec.Decoder, e
 		return helper.Int64ToPtr(400), errors.New("command is not present")
 	}
 
-	ar, err := a.c.getAllocRunner(req.AllocID)
-	if err != nil {
-		code := helper.Int64ToPtr(500)
-		if structs.IsErrUnknownAllocation(err) {
-			code = helper.Int64ToPtr(404)
-		}
-
-		return code, err
-	}
-
 	capabilities, err := ar.GetTaskDriverCapabilities(req.Task)
 	if err != nil {
 		code := helper.Int64ToPtr(500)
@@ -210,7 +229,7 @@ func (a *Allocations) execImpl(encoder *codec.Encoder, decoder *codec.Decoder, e
 
 	// check node access
 	if aclObj != nil && capabilities.FSIsolation == drivers.FSIsolationNone {
-		exec := aclObj.AllowNsOp(req.QueryOptions.Namespace, acl.NamespaceCapabilityAllocNodeExec)
+		exec := aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityAllocNodeExec)
 		if !exec {
 			return nil, structs.ErrPermissionDenied
 		}

client/alloc_endpoint_test.go

@@ -78,9 +78,19 @@ func TestAllocations_Restart_ACL(t *testing.T) {
 	})
 	defer cleanup()
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, server.RPC, job, root.SecretID)[0]
+
 	// Try request without a token and expect failure
 	{
 		req := &nstructs.AllocRestartRequest{}
+		req.AllocID = alloc.ID
 		var resp nstructs.GenericResponse
 		err := client.ClientRPC("Allocations.Restart", &req, &resp)
 		require.NotNil(err)
@@ -91,6 +101,7 @@ func TestAllocations_Restart_ACL(t *testing.T) {
 	{
 		token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{}))
 		req := &nstructs.AllocRestartRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = token.SecretID
 
 		var resp nstructs.GenericResponse
@@ -106,20 +117,27 @@ func TestAllocations_Restart_ACL(t *testing.T) {
 		token := mock.CreatePolicyAndToken(t, server.State(), 1007, "valid", policyHCL)
 		require.NotNil(token)
 		req := &nstructs.AllocRestartRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = token.SecretID
 		req.Namespace = nstructs.DefaultNamespace
 		var resp nstructs.GenericResponse
 		err := client.ClientRPC("Allocations.Restart", &req, &resp)
-		require.True(nstructs.IsErrUnknownAllocation(err), "Expected unknown alloc, found: %v", err)
+		require.NoError(err)
+		//require.True(nstructs.IsErrUnknownAllocation(err), "Expected unknown alloc, found: %v", err)
 	}
 
 	// Try request with a management token
 	{
 		req := &nstructs.AllocRestartRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = root.SecretID
 		var resp nstructs.GenericResponse
 		err := client.ClientRPC("Allocations.Restart", &req, &resp)
-		require.True(nstructs.IsErrUnknownAllocation(err), "Expected unknown alloc, found: %v", err)
+		// Depending on how quickly the alloc restarts there may be no
+		// error *or* a task not running error; either is fine.
+		if err != nil {
+			require.Contains(err.Error(), "Task not running", err)
+		}
 	}
 }
 
@@ -239,9 +257,19 @@ func TestAllocations_GarbageCollect_ACL(t *testing.T) {
 	})
 	defer cleanup()
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, server.RPC, job, root.SecretID)[0]
+
 	// Try request without a token and expect failure
 	{
 		req := &nstructs.AllocSpecificRequest{}
+		req.AllocID = alloc.ID
 		var resp nstructs.GenericResponse
 		err := client.ClientRPC("Allocations.GarbageCollect", &req, &resp)
 		require.NotNil(err)
@@ -252,6 +280,7 @@ func TestAllocations_GarbageCollect_ACL(t *testing.T) {
 	{
 		token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NodePolicy(acl.PolicyDeny))
 		req := &nstructs.AllocSpecificRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = token.SecretID
 
 		var resp nstructs.GenericResponse
@@ -266,6 +295,7 @@ func TestAllocations_GarbageCollect_ACL(t *testing.T) {
 		token := mock.CreatePolicyAndToken(t, server.State(), 1005, "test-valid",
 			mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{acl.NamespaceCapabilitySubmitJob}))
 		req := &nstructs.AllocSpecificRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = token.SecretID
 		req.Namespace = nstructs.DefaultNamespace
 
@@ -323,9 +353,19 @@ func TestAllocations_Signal_ACL(t *testing.T) {
 	})
 	defer cleanup()
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, server.RPC, job, root.SecretID)[0]
+
 	// Try request without a token and expect failure
 	{
 		req := &nstructs.AllocSignalRequest{}
+		req.AllocID = alloc.ID
 		var resp nstructs.GenericResponse
 		err := client.ClientRPC("Allocations.Signal", &req, &resp)
 		require.NotNil(err)
@@ -336,6 +376,7 @@ func TestAllocations_Signal_ACL(t *testing.T) {
 	{
 		token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NodePolicy(acl.PolicyDeny))
 		req := &nstructs.AllocSignalRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = token.SecretID
 
 		var resp nstructs.GenericResponse
@@ -350,22 +391,24 @@ func TestAllocations_Signal_ACL(t *testing.T) {
 		token := mock.CreatePolicyAndToken(t, server.State(), 1005, "test-valid",
 			mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{acl.NamespaceCapabilityAllocLifecycle}))
 		req := &nstructs.AllocSignalRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = token.SecretID
 		req.Namespace = nstructs.DefaultNamespace
 
 		var resp nstructs.GenericResponse
 		err := client.ClientRPC("Allocations.Signal", &req, &resp)
-		require.True(nstructs.IsErrUnknownAllocation(err))
+		require.NoError(err)
 	}
 
 	// Try request with a management token
 	{
 		req := &nstructs.AllocSignalRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = root.SecretID
 
 		var resp nstructs.GenericResponse
 		err := client.ClientRPC("Allocations.Signal", &req, &resp)
-		require.True(nstructs.IsErrUnknownAllocation(err))
+		require.NoError(err)
 	}
 }
 
@@ -414,9 +457,19 @@ func TestAllocations_Stats_ACL(t *testing.T) {
 	})
 	defer cleanup()
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, server.RPC, job, root.SecretID)[0]
+
 	// Try request without a token and expect failure
 	{
 		req := &cstructs.AllocStatsRequest{}
+		req.AllocID = alloc.ID
 		var resp cstructs.AllocStatsResponse
 		err := client.ClientRPC("Allocations.Stats", &req, &resp)
 		require.NotNil(err)
@@ -427,6 +480,7 @@ func TestAllocations_Stats_ACL(t *testing.T) {
 	{
 		token := mock.CreatePolicyAndToken(t, server.State(), 1005, "invalid", mock.NodePolicy(acl.PolicyDeny))
 		req := &cstructs.AllocStatsRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = token.SecretID
 
 		var resp cstructs.AllocStatsResponse
@@ -441,22 +495,24 @@ func TestAllocations_Stats_ACL(t *testing.T) {
 		token := mock.CreatePolicyAndToken(t, server.State(), 1005, "test-valid",
 			mock.NamespacePolicy(nstructs.DefaultNamespace, "", []string{acl.NamespaceCapabilityReadJob}))
 		req := &cstructs.AllocStatsRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = token.SecretID
 		req.Namespace = nstructs.DefaultNamespace
 
 		var resp cstructs.AllocStatsResponse
 		err := client.ClientRPC("Allocations.Stats", &req, &resp)
-		require.True(nstructs.IsErrUnknownAllocation(err))
+		require.NoError(err)
 	}
 
 	// Try request with a management token
 	{
 		req := &cstructs.AllocStatsRequest{}
+		req.AllocID = alloc.ID
 		req.AuthToken = root.SecretID
 
 		var resp cstructs.AllocStatsResponse
 		err := client.ClientRPC("Allocations.Stats", &req, &resp)
-		require.True(nstructs.IsErrUnknownAllocation(err))
+		require.NoError(err)
 	}
 }
 
@@ -677,7 +733,6 @@ func TestAlloc_ExecStreaming_DisableRemoteExec(t *testing.T) {
 
 func TestAlloc_ExecStreaming_ACL_Basic(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server and client
 	s, root := nomad.TestACLServer(t, nil)
@@ -698,6 +753,15 @@ func TestAlloc_ExecStreaming_ACL_Basic(t *testing.T) {
 		[]string{acl.NamespaceCapabilityAllocExec, acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)[0]
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -711,12 +775,12 @@ func TestAlloc_ExecStreaming_ACL_Basic(t *testing.T) {
 		{
 			Name:          "good token",
 			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: "task not found",
 		},
 		{
 			Name:          "root token",
 			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: "task not found",
 		},
 	}
 
@@ -725,7 +789,7 @@ func TestAlloc_ExecStreaming_ACL_Basic(t *testing.T) {
 
 			// Make the request
 			req := &cstructs.AllocExecRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				Task:    "testtask",
 				Tty:     true,
 				Cmd:     []string{"placeholder command"},
@@ -738,7 +802,7 @@ func TestAlloc_ExecStreaming_ACL_Basic(t *testing.T) {
 
 			// Get the handler
 			handler, err := client.StreamingRpcHandler("Allocations.Exec")
-			require.Nil(err)
+			require.Nil(t, err)
 
 			// Create a pipe
 			p1, p2 := net.Pipe()
@@ -754,15 +818,15 @@ func TestAlloc_ExecStreaming_ACL_Basic(t *testing.T) {
 
 			// Send the request
 			encoder := codec.NewEncoder(p1, nstructs.MsgpackHandle)
-			require.Nil(encoder.Encode(req))
+			require.Nil(t, encoder.Encode(req))
 
 			select {
 			case <-time.After(3 * time.Second):
-				require.FailNow("timed out")
+				require.FailNow(t, "timed out")
 			case err := <-errCh:
-				require.Contains(err.Error(), c.ExpectedError)
+				require.Contains(t, err.Error(), c.ExpectedError)
 			case f := <-frames:
-				require.Fail("received unexpected frame", "frame: %#v", f)
+				require.Fail(t, "received unexpected frame", "frame: %#v", f)
 			}
 		})
 	}

client/client.go

@@ -731,6 +731,16 @@ func (c *Client) Stats() map[string]map[string]string {
 	return stats
 }
 
+// GetAlloc returns an allocation or an error.
+func (c *Client) GetAlloc(allocID string) (*structs.Allocation, error) {
+	ar, err := c.getAllocRunner(allocID)
+	if err != nil {
+		return nil, err
+	}
+
+	return ar.Alloc(), nil
+}
+
 // SignalAllocation sends a signal to the tasks within an allocation.
 // If the provided task is empty, then every allocation will be signalled.
 // If a task is provided, then only an exactly matching task will be signalled.
@@ -778,6 +788,8 @@ func (c *Client) Node() *structs.Node {
 	return c.configCopy.Node
 }
 
+// getAllocRunner returns an AllocRunner or an UnknownAllocation error if the
+// client has no runner for the given alloc ID.
 func (c *Client) getAllocRunner(allocID string) (AllocRunner, error) {
 	c.allocLock.RLock()
 	defer c.allocLock.RUnlock()
@@ -882,7 +894,6 @@ func (c *Client) GetAllocFS(allocID string) (allocdir.AllocDirFS, error) {
 	if err != nil {
 		return nil, err
 	}
-
 	return ar.GetAllocDir(), nil
 }
 
@@ -1878,6 +1889,7 @@ func (c *Client) watchAllocations(updates chan *allocUpdates) {
 		QueryOptions: structs.QueryOptions{
 			Region:     c.Region(),
 			AllowStale: true,
+			AuthToken:  c.secretNodeID(),
 		},
 	}
 	var allocsResp structs.AllocsGetResponse

client/fs_endpoint.go

@@ -99,10 +99,15 @@ func handleStreamResultError(err error, code *int64, encoder *codec.Encoder) {
 func (f *FileSystem) List(args *cstructs.FsListRequest, reply *cstructs.FsListResponse) error {
 	defer metrics.MeasureSince([]string{"client", "file_system", "list"}, time.Now())
 
-	// Check read permissions
+	alloc, err := f.c.GetAlloc(args.AllocID)
+	if err != nil {
+		return err
+	}
+
+	// Check namespace read-fs permission.
 	if aclObj, err := f.c.ResolveToken(args.QueryOptions.AuthToken); err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityReadFS) {
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadFS) {
 		return structs.ErrPermissionDenied
 	}
 
@@ -123,10 +128,15 @@ func (f *FileSystem) List(args *cstructs.FsListRequest, reply *cstructs.FsListRe
 func (f *FileSystem) Stat(args *cstructs.FsStatRequest, reply *cstructs.FsStatResponse) error {
 	defer metrics.MeasureSince([]string{"client", "file_system", "stat"}, time.Now())
 
-	// Check read permissions
+	alloc, err := f.c.GetAlloc(args.AllocID)
+	if err != nil {
+		return err
+	}
+
+	// Check namespace read-fs permission.
 	if aclObj, err := f.c.ResolveToken(args.QueryOptions.AuthToken); err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityReadFS) {
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadFS) {
 		return structs.ErrPermissionDenied
 	}
 
@@ -159,20 +169,26 @@ func (f *FileSystem) stream(conn io.ReadWriteCloser) {
 		return
 	}
 
+	if req.AllocID == "" {
+		handleStreamResultError(allocIDNotPresentErr, helper.Int64ToPtr(400), encoder)
+		return
+	}
+	alloc, err := f.c.GetAlloc(req.AllocID)
+	if err != nil {
+		handleStreamResultError(structs.NewErrUnknownAllocation(req.AllocID), helper.Int64ToPtr(404), encoder)
+		return
+	}
+
 	// Check read permissions
 	if aclObj, err := f.c.ResolveToken(req.QueryOptions.AuthToken); err != nil {
 		handleStreamResultError(err, helper.Int64ToPtr(403), encoder)
 		return
-	} else if aclObj != nil && !aclObj.AllowNsOp(req.Namespace, acl.NamespaceCapabilityReadFS) {
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadFS) {
 		handleStreamResultError(structs.ErrPermissionDenied, helper.Int64ToPtr(403), encoder)
 		return
 	}
 
 	// Validate the arguments
-	if req.AllocID == "" {
-		handleStreamResultError(allocIDNotPresentErr, helper.Int64ToPtr(400), encoder)
-		return
-	}
 	if req.Path == "" {
 		handleStreamResultError(pathNotPresentErr, helper.Int64ToPtr(400), encoder)
 		return
@@ -332,13 +348,23 @@ func (f *FileSystem) logs(conn io.ReadWriteCloser) {
 		return
 	}
 
+	if req.AllocID == "" {
+		handleStreamResultError(allocIDNotPresentErr, helper.Int64ToPtr(400), encoder)
+		return
+	}
+	alloc, err := f.c.GetAlloc(req.AllocID)
+	if err != nil {
+		handleStreamResultError(structs.NewErrUnknownAllocation(req.AllocID), helper.Int64ToPtr(404), encoder)
+		return
+	}
+
 	// Check read permissions
 	if aclObj, err := f.c.ResolveToken(req.QueryOptions.AuthToken); err != nil {
 		handleStreamResultError(err, nil, encoder)
 		return
 	} else if aclObj != nil {
-		readfs := aclObj.AllowNsOp(req.QueryOptions.Namespace, acl.NamespaceCapabilityReadFS)
-		logs := aclObj.AllowNsOp(req.QueryOptions.Namespace, acl.NamespaceCapabilityReadLogs)
+		readfs := aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadFS)
+		logs := aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadLogs)
 		if !readfs && !logs {
 			handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
 			return
@@ -346,10 +372,6 @@ func (f *FileSystem) logs(conn io.ReadWriteCloser) {
 	}
 
 	// Validate the arguments
-	if req.AllocID == "" {
-		handleStreamResultError(allocIDNotPresentErr, helper.Int64ToPtr(400), encoder)
-		return
-	}
 	if req.Task == "" {
 		handleStreamResultError(taskNotPresentErr, helper.Int64ToPtr(400), encoder)
 		return

client/fs_endpoint_test.go

@@ -114,7 +114,6 @@ func TestFS_Stat(t *testing.T) {
 
 func TestFS_Stat_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := nomad.TestACLServer(t, nil)
@@ -135,6 +134,15 @@ func TestFS_Stat_ACL(t *testing.T) {
 		[]string{acl.NamespaceCapabilityReadLogs, acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)[0]
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -146,22 +154,19 @@ func TestFS_Stat_ACL(t *testing.T) {
 			ExpectedError: structs.ErrPermissionDenied.Error(),
 		},
 		{
-			Name:          "good token",
-			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			Name:  "good token",
+			Token: tokenGood.SecretID,
 		},
 		{
-			Name:          "root token",
-			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			Name:  "root token",
+			Token: root.SecretID,
 		},
 	}
 
 	for _, c := range cases {
 		t.Run(c.Name, func(t *testing.T) {
-			// Make the request with bad allocation id
 			req := &cstructs.FsStatRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				Path:    "/",
 				QueryOptions: structs.QueryOptions{
 					Region:    "global",
@@ -172,8 +177,12 @@ func TestFS_Stat_ACL(t *testing.T) {
 
 			var resp cstructs.FsStatResponse
 			err := client.ClientRPC("FileSystem.Stat", req, &resp)
-			require.NotNil(err)
-			require.Contains(err.Error(), c.ExpectedError)
+			if c.ExpectedError == "" {
+				require.NoError(t, err)
+			} else {
+				require.NotNil(t, err)
+				require.Contains(t, err.Error(), c.ExpectedError)
+			}
 		})
 	}
 }
@@ -238,7 +247,6 @@ func TestFS_List(t *testing.T) {
 
 func TestFS_List_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := nomad.TestACLServer(t, nil)
@@ -259,6 +267,15 @@ func TestFS_List_ACL(t *testing.T) {
 		[]string{acl.NamespaceCapabilityReadLogs, acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)[0]
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -270,14 +287,12 @@ func TestFS_List_ACL(t *testing.T) {
 			ExpectedError: structs.ErrPermissionDenied.Error(),
 		},
 		{
-			Name:          "good token",
-			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			Name:  "good token",
+			Token: tokenGood.SecretID,
 		},
 		{
-			Name:          "root token",
-			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			Name:  "root token",
+			Token: root.SecretID,
 		},
 	}
 
@@ -285,7 +300,7 @@ func TestFS_List_ACL(t *testing.T) {
 		t.Run(c.Name, func(t *testing.T) {
 			// Make the request with bad allocation id
 			req := &cstructs.FsListRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				Path:    "/",
 				QueryOptions: structs.QueryOptions{
 					Region:    "global",
@@ -296,8 +311,11 @@ func TestFS_List_ACL(t *testing.T) {
 
 			var resp cstructs.FsListResponse
 			err := client.ClientRPC("FileSystem.List", req, &resp)
-			require.NotNil(err)
-			require.Contains(err.Error(), c.ExpectedError)
+			if c.ExpectedError == "" {
+				require.NoError(t, err)
+			} else {
+				require.EqualError(t, err, c.ExpectedError)
+			}
 		})
 	}
 }
@@ -379,7 +397,6 @@ OUTER:
 
 func TestFS_Stream_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := nomad.TestACLServer(t, nil)
@@ -400,6 +417,15 @@ func TestFS_Stream_ACL(t *testing.T) {
 		[]string{acl.NamespaceCapabilityReadLogs, acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)[0]
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -411,14 +437,12 @@ func TestFS_Stream_ACL(t *testing.T) {
 			ExpectedError: structs.ErrPermissionDenied.Error(),
 		},
 		{
-			Name:          "good token",
-			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			Name:  "good token",
+			Token: tokenGood.SecretID,
 		},
 		{
-			Name:          "root token",
-			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			Name:  "root token",
+			Token: root.SecretID,
 		},
 	}
 
@@ -426,7 +450,7 @@ func TestFS_Stream_ACL(t *testing.T) {
 		t.Run(c.Name, func(t *testing.T) {
 			// Make the request with bad allocation id
 			req := &cstructs.FsStreamRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				Path:    "foo",
 				Origin:  "start",
 				QueryOptions: structs.QueryOptions{
@@ -438,7 +462,7 @@ func TestFS_Stream_ACL(t *testing.T) {
 
 			// Get the handler
 			handler, err := client.StreamingRpcHandler("FileSystem.Stream")
-			require.Nil(err)
+			require.Nil(t, err)
 
 			// Create a pipe
 			p1, p2 := net.Pipe()
@@ -457,10 +481,8 @@ func TestFS_Stream_ACL(t *testing.T) {
 				for {
 					var msg cstructs.StreamErrWrapper
 					if err := decoder.Decode(&msg); err != nil {
-						if err == io.EOF || strings.Contains(err.Error(), "closed") {
-							return
-						}
-						errCh <- fmt.Errorf("error decoding: %v", err)
+						errCh <- err
+						return
 					}
 
 					streamMsg <- &msg
@@ -469,7 +491,7 @@ func TestFS_Stream_ACL(t *testing.T) {
 
 			// Send the request
 			encoder := codec.NewEncoder(p1, structs.MsgpackHandle)
-			require.Nil(encoder.Encode(req))
+			require.NoError(t, encoder.Encode(req))
 
 			timeout := time.After(5 * time.Second)
 
@@ -479,6 +501,11 @@ func TestFS_Stream_ACL(t *testing.T) {
 				case <-timeout:
 					t.Fatal("timeout")
 				case err := <-errCh:
+					eof := err == io.EOF || strings.Contains(err.Error(), "closed")
+					if c.ExpectedError == "" && eof {
+						// No error was expected!
+						return
+					}
 					t.Fatal(err)
 				case msg := <-streamMsg:
 					if msg.Error == nil {
@@ -1019,6 +1046,15 @@ func TestFS_Logs_ACL(t *testing.T) {
 		[]string{acl.NamespaceCapabilityReadLogs, acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	job := mock.BatchJob()
+	job.TaskGroups[0].Count = 1
+	job.TaskGroups[0].Tasks[0].Config = map[string]interface{}{
+		"run_for": "20s",
+	}
+
+	// Wait for client to be running job
+	alloc := testutil.WaitForRunningWithToken(t, s.RPC, job, root.SecretID)[0]
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -1030,14 +1066,12 @@ func TestFS_Logs_ACL(t *testing.T) {
 			ExpectedError: structs.ErrPermissionDenied.Error(),
 		},
 		{
-			Name:          "good token",
-			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			Name:  "good token",
+			Token: tokenGood.SecretID,
 		},
 		{
-			Name:          "root token",
-			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			Name:  "root token",
+			Token: root.SecretID,
 		},
 	}
 
@@ -1045,8 +1079,8 @@ func TestFS_Logs_ACL(t *testing.T) {
 		t.Run(c.Name, func(t *testing.T) {
 			// Make the request with bad allocation id
 			req := &cstructs.FsLogsRequest{
-				AllocID: uuid.Generate(),
-				Task:    "foo",
+				AllocID: alloc.ID,
+				Task:    job.TaskGroups[0].Tasks[0].Name,
 				LogType: "stdout",
 				Origin:  "start",
 				QueryOptions: structs.QueryOptions{
@@ -1077,10 +1111,8 @@ func TestFS_Logs_ACL(t *testing.T) {
 				for {
 					var msg cstructs.StreamErrWrapper
 					if err := decoder.Decode(&msg); err != nil {
-						if err == io.EOF || strings.Contains(err.Error(), "closed") {
-							return
-						}
-						errCh <- fmt.Errorf("error decoding: %v", err)
+						errCh <- err
+						return
 					}
 
 					streamMsg <- &msg
@@ -1099,6 +1131,11 @@ func TestFS_Logs_ACL(t *testing.T) {
 				case <-timeout:
 					t.Fatal("timeout")
 				case err := <-errCh:
+					eof := err == io.EOF || strings.Contains(err.Error(), "closed")
+					if c.ExpectedError == "" && eof {
+						// No error was expected!
+						return
+					}
 					t.Fatal(err)
 				case msg := <-streamMsg:
 					if msg.Error == nil {
@@ -1106,6 +1143,7 @@ func TestFS_Logs_ACL(t *testing.T) {
 					}
 
 					if strings.Contains(msg.Error.Error(), c.ExpectedError) {
+						// Ok! Error matched expectation.
 						break OUTER
 					} else {
 						t.Fatalf("Bad error: %v", msg.Error)

client/testutil/rpc.go

@@ -0,0 +1,86 @@
+package testutil
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	cstructs "github.com/hashicorp/nomad/client/structs"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/stretchr/testify/require"
+	"github.com/ugorji/go/codec"
+)
+
+// StreamingRPC may be satisfied by client.Client or server.Server.
+type StreamingRPC interface {
+	StreamingRpcHandler(method string) (structs.StreamingRpcHandler, error)
+}
+
+// StreamingRPCErrorTestCase is a test case to be passed to the
+// assertStreamingRPCError func.
+type StreamingRPCErrorTestCase struct {
+	Name   string
+	RPC    string
+	Req    interface{}
+	Assert func(error) bool
+}
+
+// AssertStreamingRPCError asserts a streaming RPC's error matches the given
+// assertion in the test case.
+func AssertStreamingRPCError(t *testing.T, s StreamingRPC, tc StreamingRPCErrorTestCase) {
+	handler, err := s.StreamingRpcHandler(tc.RPC)
+	require.NoError(t, err)
+
+	// Create a pipe
+	p1, p2 := net.Pipe()
+	defer p1.Close()
+	defer p2.Close()
+
+	errCh := make(chan error, 1)
+	streamMsg := make(chan *cstructs.StreamErrWrapper, 1)
+
+	// Start the handler
+	go handler(p2)
+
+	// Start the decoder
+	go func() {
+		decoder := codec.NewDecoder(p1, structs.MsgpackHandle)
+		for {
+			var msg cstructs.StreamErrWrapper
+			if err := decoder.Decode(&msg); err != nil {
+				if err == io.EOF || strings.Contains(err.Error(), "closed") {
+					return
+				}
+				errCh <- fmt.Errorf("error decoding: %v", err)
+			}
+
+			streamMsg <- &msg
+		}
+	}()
+
+	// Send the request
+	encoder := codec.NewEncoder(p1, structs.MsgpackHandle)
+	require.NoError(t, encoder.Encode(tc.Req))
+
+	timeout := time.After(5 * time.Second)
+
+	for {
+		select {
+		case <-timeout:
+			t.Fatal("timeout")
+		case err := <-errCh:
+			require.NoError(t, err)
+		case msg := <-streamMsg:
+			// Convert RpcError to error
+			var err error
+			if msg.Error != nil {
+				err = msg.Error
+			}
+			require.True(t, tc.Assert(err), "(%T) %s", msg.Error, msg.Error)
+			return
+		}
+	}
+}

command/agent/alloc_endpoint_test.go

@@ -346,7 +346,7 @@ func TestHTTP_AllocRestart_ACL(t *testing.T) {
 			respW := httptest.NewRecorder()
 			_, err = s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.Equal(err.Error(), structs.ErrPermissionDenied.Error())
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with an invalid token and expect it to fail
@@ -360,7 +360,7 @@ func TestHTTP_AllocRestart_ACL(t *testing.T) {
 			setToken(req, token)
 			_, err = s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.Equal(err.Error(), structs.ErrPermissionDenied.Error())
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with a valid token
@@ -376,7 +376,7 @@ func TestHTTP_AllocRestart_ACL(t *testing.T) {
 			setToken(req, token)
 			_, err = s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.True(structs.IsErrUnknownAllocation(err))
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with a management token
@@ -523,7 +523,7 @@ func TestHTTP_AllocStats_ACL(t *testing.T) {
 			respW := httptest.NewRecorder()
 			_, err := s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.Equal(err.Error(), structs.ErrPermissionDenied.Error())
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with an invalid token and expect failure
@@ -533,7 +533,7 @@ func TestHTTP_AllocStats_ACL(t *testing.T) {
 			setToken(req, token)
 			_, err := s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.Equal(err.Error(), structs.ErrPermissionDenied.Error())
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with a valid token
@@ -545,7 +545,7 @@ func TestHTTP_AllocStats_ACL(t *testing.T) {
 			setToken(req, token)
 			_, err := s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.True(structs.IsErrUnknownAllocation(err))
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with a management token
@@ -812,7 +812,7 @@ func TestHTTP_AllocGC_ACL(t *testing.T) {
 			respW := httptest.NewRecorder()
 			_, err := s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.Equal(err.Error(), structs.ErrPermissionDenied.Error())
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with an invalid token and expect failure
@@ -822,7 +822,7 @@ func TestHTTP_AllocGC_ACL(t *testing.T) {
 			setToken(req, token)
 			_, err := s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.Equal(err.Error(), structs.ErrPermissionDenied.Error())
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with a valid token
@@ -834,7 +834,7 @@ func TestHTTP_AllocGC_ACL(t *testing.T) {
 			setToken(req, token)
 			_, err := s.Server.ClientAllocRequest(respW, req)
 			require.NotNil(err)
-			require.True(structs.IsErrUnknownAllocation(err))
+			require.True(structs.IsErrUnknownAllocation(err), "(%T) %v", err, err)
 		}
 
 		// Try request with a management token

nomad/alloc_endpoint.go

@@ -86,8 +86,10 @@ func (a *Alloc) GetAlloc(args *structs.AllocSpecificRequest,
 	}
 	defer metrics.MeasureSince([]string{"nomad", "alloc", "get_alloc"}, time.Now())
 
-	// Check namespace read-job permissions
-	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+	// Check namespace read-job permissions before performing blocking query.
+	allowNsOp := acl.NamespaceValidator(acl.NamespaceCapabilityReadJob)
+	aclObj, err := a.srv.ResolveToken(args.AuthToken)
+	if err != nil {
 		// If ResolveToken had an unexpected error return that
 		if err != structs.ErrTokenNotFound {
 			return err
@@ -107,8 +109,6 @@ func (a *Alloc) GetAlloc(args *structs.AllocSpecificRequest,
 		if node == nil {
 			return structs.ErrTokenNotFound
 		}
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilityReadJob) {
-		return structs.ErrPermissionDenied
 	}
 
 	// Setup the blocking query
@@ -125,6 +125,11 @@ func (a *Alloc) GetAlloc(args *structs.AllocSpecificRequest,
 			// Setup the output
 			reply.Alloc = out
 			if out != nil {
+				// Re-check namespace in case it differs from request.
+				if !allowNsOp(aclObj, out.Namespace) {
+					return structs.NewErrUnknownAllocation(args.AllocID)
+				}
+
 				reply.Index = out.ModifyIndex
 			} else {
 				// Use the last index that affected the allocs table
@@ -214,25 +219,18 @@ func (a *Alloc) Stop(args *structs.AllocStopRequest, reply *structs.AllocStopRes
 	}
 	defer metrics.MeasureSince([]string{"nomad", "alloc", "stop"}, time.Now())
 
-	// Check that it is a management token.
-	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
-		return structs.ErrPermissionDenied
-	}
-
-	if args.AllocID == "" {
-		return fmt.Errorf("must provide an alloc id")
-	}
-
-	ws := memdb.NewWatchSet()
-	alloc, err := a.srv.State().AllocByID(ws, args.AllocID)
+	alloc, err := getAlloc(a.srv.State(), args.AllocID)
 	if err != nil {
 		return err
 	}
 
-	if alloc == nil {
-		return fmt.Errorf(structs.ErrUnknownAllocationPrefix)
+	// Check for namespace alloc-lifecycle permissions.
+	allowNsOp := acl.NamespaceValidator(acl.NamespaceCapabilityAllocLifecycle)
+	aclObj, err := a.srv.ResolveToken(args.AuthToken)
+	if err != nil {
+		return err
+	} else if !allowNsOp(aclObj, alloc.Namespace) {
+		return structs.ErrPermissionDenied
 	}
 
 	now := time.Now().UTC().UnixNano()

nomad/alloc_endpoint_test.go

@@ -276,54 +276,90 @@ func TestAllocEndpoint_GetAlloc_ACL(t *testing.T) {
 	invalidToken := mock.CreatePolicyAndToken(t, state, 1003, "test-invalid",
 		mock.NamespacePolicy(structs.DefaultNamespace, "", []string{acl.NamespaceCapabilityListJobs}))
 
-	get := &structs.AllocSpecificRequest{
-		AllocID:      alloc.ID,
-		QueryOptions: structs.QueryOptions{Region: "global"},
+	getReq := func() *structs.AllocSpecificRequest {
+		return &structs.AllocSpecificRequest{
+			AllocID: alloc.ID,
+			QueryOptions: structs.QueryOptions{
+				Region: "global",
+			},
+		}
 	}
 
-	// Lookup the alloc without a token and expect failure
-	{
-		var resp structs.SingleAllocResponse
-		err := msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp)
-		assert.Equal(structs.ErrPermissionDenied.Error(), err.Error())
+	cases := []struct {
+		Name string
+		F    func(t *testing.T)
+	}{
+		// Lookup the alloc without a token and expect failure
+		{
+			Name: "no-token",
+			F: func(t *testing.T) {
+				var resp structs.SingleAllocResponse
+				err := msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", getReq(), &resp)
+				require.True(t, structs.IsErrUnknownAllocation(err), "expected unknown alloc but found: %v", err)
+			},
+		},
+
+		// Try with a valid ACL token
+		{
+			Name: "valid-token",
+			F: func(t *testing.T) {
+				get := getReq()
+				get.AuthToken = validToken.SecretID
+				get.AllocID = alloc.ID
+				var resp structs.SingleAllocResponse
+				require.NoError(t, msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp), "RPC")
+				require.EqualValues(t, resp.Index, 1000, "resp.Index")
+				require.Equal(t, alloc, resp.Alloc, "Returned alloc not equal")
+			},
+		},
+
+		// Try with a valid Node.SecretID
+		{
+			Name: "valid-node-secret",
+			F: func(t *testing.T) {
+				node := mock.Node()
+				assert.Nil(state.UpsertNode(1005, node))
+				get := getReq()
+				get.AuthToken = node.SecretID
+				get.AllocID = alloc.ID
+				var resp structs.SingleAllocResponse
+				require.NoError(t, msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp), "RPC")
+				require.EqualValues(t, resp.Index, 1000, "resp.Index")
+				require.Equal(t, alloc, resp.Alloc, "Returned alloc not equal")
+			},
+		},
+
+		// Try with a invalid token
+		{
+			Name: "invalid-token",
+			F: func(t *testing.T) {
+				get := getReq()
+				get.AuthToken = invalidToken.SecretID
+				get.AllocID = alloc.ID
+				var resp structs.SingleAllocResponse
+				err := msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp)
+				require.NotNil(t, err, "RPC")
+				require.True(t, structs.IsErrUnknownAllocation(err), "expected unknown alloc but found: %v", err)
+			},
+		},
+
+		// Try with a root token
+		{
+			Name: "root-token",
+			F: func(t *testing.T) {
+				get := getReq()
+				get.AuthToken = root.SecretID
+				get.AllocID = alloc.ID
+				var resp structs.SingleAllocResponse
+				require.NoError(t, msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp), "RPC")
+				require.EqualValues(t, resp.Index, 1000, "resp.Index")
+				require.Equal(t, alloc, resp.Alloc, "Returned alloc not equal")
+			},
+		},
 	}
 
-	// Try with a valid ACL token
-	{
-		get.AuthToken = validToken.SecretID
-		var resp structs.SingleAllocResponse
-		assert.Nil(msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp), "RPC")
-		assert.EqualValues(resp.Index, 1000, "resp.Index")
-		assert.Equal(alloc, resp.Alloc, "Returned alloc not equal")
-	}
-
-	// Try with a valid Node.SecretID
-	{
-		node := mock.Node()
-		assert.Nil(state.UpsertNode(1005, node))
-		get.AuthToken = node.SecretID
-		var resp structs.SingleAllocResponse
-		assert.Nil(msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp), "RPC")
-		assert.EqualValues(resp.Index, 1000, "resp.Index")
-		assert.Equal(alloc, resp.Alloc, "Returned alloc not equal")
-	}
-
-	// Try with a invalid token
-	{
-		get.AuthToken = invalidToken.SecretID
-		var resp structs.SingleAllocResponse
-		err := msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp)
-		assert.NotNil(err, "RPC")
-		assert.Equal(err.Error(), structs.ErrPermissionDenied.Error())
-	}
-
-	// Try with a root token
-	{
-		get.AuthToken = root.SecretID
-		var resp structs.SingleAllocResponse
-		assert.Nil(msgpackrpc.CallWithCodec(codec, "Alloc.GetAlloc", get, &resp), "RPC")
-		assert.EqualValues(resp.Index, 1000, "resp.Index")
-		assert.Equal(alloc, resp.Alloc, "Returned alloc not equal")
+	for _, tc := range cases {
+		t.Run(tc.Name, tc.F)
 	}
 }
 

nomad/client_alloc_endpoint.go

@@ -87,13 +87,6 @@ func (a *ClientAllocations) Signal(args *structs.AllocSignalRequest, reply *stru
 	}
 	defer metrics.MeasureSince([]string{"nomad", "client_allocations", "signal"}, time.Now())
 
-	// Check node read permissions
-	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
-		return structs.ErrPermissionDenied
-	}
-
 	// Verify the arguments.
 	if args.AllocID == "" {
 		return errors.New("missing AllocID")
@@ -105,13 +98,16 @@ func (a *ClientAllocations) Signal(args *structs.AllocSignalRequest, reply *stru
 		return err
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
 	if err != nil {
 		return err
 	}
 
-	if alloc == nil {
-		return structs.NewErrUnknownAllocation(args.AllocID)
+	// Check namespace alloc-lifecycle permission.
+	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
+		return structs.ErrPermissionDenied
 	}
 
 	// Make sure Node is valid and new enough to support RPC
@@ -143,13 +139,6 @@ func (a *ClientAllocations) GarbageCollect(args *structs.AllocSpecificRequest, r
 	}
 	defer metrics.MeasureSince([]string{"nomad", "client_allocations", "garbage_collect"}, time.Now())
 
-	// Check node read permissions
-	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilitySubmitJob) {
-		return structs.ErrPermissionDenied
-	}
-
 	// Verify the arguments.
 	if args.AllocID == "" {
 		return errors.New("missing AllocID")
@@ -161,13 +150,16 @@ func (a *ClientAllocations) GarbageCollect(args *structs.AllocSpecificRequest, r
 		return err
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
 	if err != nil {
 		return err
 	}
 
-	if alloc == nil {
-		return structs.NewErrUnknownAllocation(args.AllocID)
+	// Check namespace submit-job permission.
+	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilitySubmitJob) {
+		return structs.ErrPermissionDenied
 	}
 
 	// Make sure Node is valid and new enough to support RPC
@@ -199,30 +191,22 @@ func (a *ClientAllocations) Restart(args *structs.AllocRestartRequest, reply *st
 	}
 	defer metrics.MeasureSince([]string{"nomad", "client_allocations", "restart"}, time.Now())
 
-	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
-		return structs.ErrPermissionDenied
-	}
-
-	// Verify the arguments.
-	if args.AllocID == "" {
-		return errors.New("missing AllocID")
-	}
-
 	// Find the allocation
 	snap, err := a.srv.State().Snapshot()
 	if err != nil {
 		return err
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
 	if err != nil {
 		return err
 	}
 
-	if alloc == nil {
-		return structs.NewErrUnknownAllocation(args.AllocID)
+	// Check for namespace alloc-lifecycle permissions.
+	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityAllocLifecycle) {
+		return structs.ErrPermissionDenied
 	}
 
 	// Make sure Node is valid and new enough to support RPC
@@ -254,31 +238,22 @@ func (a *ClientAllocations) Stats(args *cstructs.AllocStatsRequest, reply *cstru
 	}
 	defer metrics.MeasureSince([]string{"nomad", "client_allocations", "stats"}, time.Now())
 
-	// Check node read permissions
-	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityReadJob) {
-		return structs.ErrPermissionDenied
-	}
-
-	// Verify the arguments.
-	if args.AllocID == "" {
-		return errors.New("missing AllocID")
-	}
-
 	// Find the allocation
 	snap, err := a.srv.State().Snapshot()
 	if err != nil {
 		return err
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
 	if err != nil {
 		return err
 	}
 
-	if alloc == nil {
-		return structs.NewErrUnknownAllocation(args.AllocID)
+	// Check for namespace read-job permissions.
+	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadJob) {
+		return structs.ErrPermissionDenied
 	}
 
 	// Make sure Node is valid and new enough to support RPC
@@ -319,19 +294,6 @@ func (a *ClientAllocations) exec(conn io.ReadWriteCloser) {
 		return
 	}
 
-	// Check node read permissions
-	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
-		handleStreamResultError(err, nil, encoder)
-		return
-	} else if aclObj != nil {
-		// client ultimately checks if AllocNodeExec is required
-		exec := aclObj.AllowNsOp(args.QueryOptions.Namespace, acl.NamespaceCapabilityAllocExec)
-		if !exec {
-			handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
-			return
-		}
-	}
-
 	// Verify the arguments.
 	if args.AllocID == "" {
 		handleStreamResultError(errors.New("missing AllocID"), helper.Int64ToPtr(400), encoder)
@@ -345,15 +307,26 @@ func (a *ClientAllocations) exec(conn io.ReadWriteCloser) {
 		return
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
+	if structs.IsErrUnknownAllocation(err) {
+		handleStreamResultError(err, helper.Int64ToPtr(404), encoder)
+		return
+	}
 	if err != nil {
 		handleStreamResultError(err, nil, encoder)
 		return
 	}
-	if alloc == nil {
-		handleStreamResultError(structs.NewErrUnknownAllocation(args.AllocID), helper.Int64ToPtr(404), encoder)
+
+	// Check node read permissions
+	if aclObj, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		handleStreamResultError(err, nil, encoder)
+		return
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityAllocExec) {
+		// client ultimately checks if AllocNodeExec is required
+		handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
 		return
 	}
+
 	nodeID := alloc.NodeID
 
 	// Make sure Node is valid and new enough to support RPC

nomad/client_alloc_endpoint_test.go

@@ -363,7 +363,6 @@ func TestClientAllocations_GarbageCollect_Local(t *testing.T) {
 
 func TestClientAllocations_GarbageCollect_Local_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := TestACLServer(t, nil)
@@ -378,6 +377,12 @@ func TestClientAllocations_GarbageCollect_Local_ACL(t *testing.T) {
 	policyGood := mock.NamespacePolicy(structs.DefaultNamespace, "", []string{acl.NamespaceCapabilitySubmitJob})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	// Upsert the allocation
+	state := s.State()
+	alloc := mock.Alloc()
+	require.NoError(t, state.UpsertJob(1010, alloc.Job))
+	require.NoError(t, state.UpsertAllocs(1011, []*structs.Allocation{alloc}))
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -391,12 +396,12 @@ func TestClientAllocations_GarbageCollect_Local_ACL(t *testing.T) {
 		{
 			Name:          "good token",
 			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 		{
 			Name:          "root token",
 			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 	}
 
@@ -405,7 +410,7 @@ func TestClientAllocations_GarbageCollect_Local_ACL(t *testing.T) {
 
 			// Make the request without having a node-id
 			req := &structs.AllocSpecificRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				QueryOptions: structs.QueryOptions{
 					AuthToken: c.Token,
 					Region:    "global",
@@ -416,8 +421,8 @@ func TestClientAllocations_GarbageCollect_Local_ACL(t *testing.T) {
 			// Fetch the response
 			var resp structs.GenericResponse
 			err := msgpackrpc.CallWithCodec(codec, "ClientAllocations.GarbageCollect", req, &resp)
-			require.NotNil(err)
-			require.Contains(err.Error(), c.ExpectedError)
+			require.NotNil(t, err)
+			require.Contains(t, err.Error(), c.ExpectedError)
 		})
 	}
 }
@@ -634,7 +639,7 @@ func TestClientAllocations_Stats_Local(t *testing.T) {
 	var resp cstructs.AllocStatsResponse
 	err := msgpackrpc.CallWithCodec(codec, "ClientAllocations.Stats", req, &resp)
 	require.NotNil(err)
-	require.Contains(err.Error(), "missing")
+	require.EqualError(err, structs.ErrMissingAllocID.Error(), "(%T) %v")
 
 	// Fetch the response setting the node id
 	req.AllocID = a.ID
@@ -646,7 +651,6 @@ func TestClientAllocations_Stats_Local(t *testing.T) {
 
 func TestClientAllocations_Stats_Local_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := TestACLServer(t, nil)
@@ -661,6 +665,12 @@ func TestClientAllocations_Stats_Local_ACL(t *testing.T) {
 	policyGood := mock.NamespacePolicy(structs.DefaultNamespace, "", []string{acl.NamespaceCapabilityReadJob})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	// Upsert the allocation
+	state := s.State()
+	alloc := mock.Alloc()
+	require.NoError(t, state.UpsertJob(1010, alloc.Job))
+	require.NoError(t, state.UpsertAllocs(1011, []*structs.Allocation{alloc}))
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -674,12 +684,12 @@ func TestClientAllocations_Stats_Local_ACL(t *testing.T) {
 		{
 			Name:          "good token",
 			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 		{
 			Name:          "root token",
 			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 	}
 
@@ -688,7 +698,7 @@ func TestClientAllocations_Stats_Local_ACL(t *testing.T) {
 
 			// Make the request without having a node-id
 			req := &structs.AllocSpecificRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				QueryOptions: structs.QueryOptions{
 					AuthToken: c.Token,
 					Region:    "global",
@@ -699,8 +709,8 @@ func TestClientAllocations_Stats_Local_ACL(t *testing.T) {
 			// Fetch the response
 			var resp cstructs.AllocStatsResponse
 			err := msgpackrpc.CallWithCodec(codec, "ClientAllocations.Stats", req, &resp)
-			require.NotNil(err)
-			require.Contains(err.Error(), c.ExpectedError)
+			require.NotNil(t, err)
+			require.Contains(t, err.Error(), c.ExpectedError)
 		})
 	}
 }
@@ -867,7 +877,7 @@ func TestClientAllocations_Restart_Local(t *testing.T) {
 	var resp structs.GenericResponse
 	err := msgpackrpc.CallWithCodec(codec, "ClientAllocations.Restart", req, &resp)
 	require.NotNil(err)
-	require.Contains(err.Error(), "missing")
+	require.EqualError(err, structs.ErrMissingAllocID.Error(), "(%T) %v")
 
 	// Fetch the response setting the alloc id - This should not error because the
 	// alloc is running.
@@ -981,14 +991,14 @@ func TestClientAllocations_Restart_Remote(t *testing.T) {
 	var resp structs.GenericResponse
 	err := msgpackrpc.CallWithCodec(codec, "ClientAllocations.Restart", req, &resp)
 	require.NotNil(err)
-	require.Contains(err.Error(), "missing")
+	require.EqualError(err, structs.ErrMissingAllocID.Error(), "(%T) %v")
 
 	// Fetch the response setting the alloc id - This should succeed because the
 	// alloc is running
 	req.AllocID = a.ID
 	var resp2 structs.GenericResponse
 	err = msgpackrpc.CallWithCodec(codec, "ClientAllocations.Restart", req, &resp2)
-	require.Nil(err)
+	require.NoError(err)
 }
 
 func TestClientAllocations_Restart_ACL(t *testing.T) {
@@ -1005,6 +1015,12 @@ func TestClientAllocations_Restart_ACL(t *testing.T) {
 	policyGood := mock.NamespacePolicy(structs.DefaultNamespace, acl.PolicyWrite, nil)
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	// Upsert the allocation
+	state := s.State()
+	alloc := mock.Alloc()
+	require.NoError(t, state.UpsertJob(1010, alloc.Job))
+	require.NoError(t, state.UpsertAllocs(1011, []*structs.Allocation{alloc}))
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -1018,12 +1034,12 @@ func TestClientAllocations_Restart_ACL(t *testing.T) {
 		{
 			Name:          "good token",
 			Token:         tokenGood.SecretID,
-			ExpectedError: "Unknown alloc",
+			ExpectedError: "Unknown node",
 		},
 		{
 			Name:          "root token",
 			Token:         root.SecretID,
-			ExpectedError: "Unknown alloc",
+			ExpectedError: "Unknown node",
 		},
 	}
 
@@ -1032,7 +1048,7 @@ func TestClientAllocations_Restart_ACL(t *testing.T) {
 
 			// Make the request without having a node-id
 			req := &structs.AllocRestartRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				QueryOptions: structs.QueryOptions{
 					Namespace: structs.DefaultNamespace,
 					AuthToken: c.Token,

nomad/client_fs_endpoint.go

@@ -108,13 +108,6 @@ func (f *FileSystem) List(args *cstructs.FsListRequest, reply *cstructs.FsListRe
 	}
 	defer metrics.MeasureSince([]string{"nomad", "file_system", "list"}, time.Now())
 
-	// Check filesystem read permissions
-	if aclObj, err := f.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityReadFS) {
-		return structs.ErrPermissionDenied
-	}
-
 	// Verify the arguments.
 	if args.AllocID == "" {
 		return errors.New("missing allocation ID")
@@ -126,12 +119,18 @@ func (f *FileSystem) List(args *cstructs.FsListRequest, reply *cstructs.FsListRe
 		return err
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
 	if err != nil {
 		return err
 	}
-	if alloc == nil {
-		return structs.NewErrUnknownAllocation(args.AllocID)
+
+	// Check namespace filesystem read permissions
+	allowNsOp := acl.NamespaceValidator(acl.NamespaceCapabilityReadFS)
+	aclObj, err := f.srv.ResolveToken(args.AuthToken)
+	if err != nil {
+		return err
+	} else if !allowNsOp(aclObj, alloc.Namespace) {
+		return structs.ErrPermissionDenied
 	}
 
 	// Make sure Node is valid and new enough to support RPC
@@ -163,13 +162,6 @@ func (f *FileSystem) Stat(args *cstructs.FsStatRequest, reply *cstructs.FsStatRe
 	}
 	defer metrics.MeasureSince([]string{"nomad", "file_system", "stat"}, time.Now())
 
-	// Check filesystem read permissions
-	if aclObj, err := f.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityReadFS) {
-		return structs.ErrPermissionDenied
-	}
-
 	// Verify the arguments.
 	if args.AllocID == "" {
 		return errors.New("missing allocation ID")
@@ -181,12 +173,16 @@ func (f *FileSystem) Stat(args *cstructs.FsStatRequest, reply *cstructs.FsStatRe
 		return err
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
 	if err != nil {
 		return err
 	}
-	if alloc == nil {
-		return structs.NewErrUnknownAllocation(args.AllocID)
+
+	// Check filesystem read permissions
+	if aclObj, err := f.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadFS) {
+		return structs.ErrPermissionDenied
 	}
 
 	// Make sure Node is valid and new enough to support RPC
@@ -228,15 +224,6 @@ func (f *FileSystem) stream(conn io.ReadWriteCloser) {
 		return
 	}
 
-	// Check node read permissions
-	if aclObj, err := f.srv.ResolveToken(args.AuthToken); err != nil {
-		handleStreamResultError(err, nil, encoder)
-		return
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.Namespace, acl.NamespaceCapabilityReadFS) {
-		handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
-		return
-	}
-
 	// Verify the arguments.
 	if args.AllocID == "" {
 		handleStreamResultError(errors.New("missing AllocID"), helper.Int64ToPtr(400), encoder)
@@ -250,15 +237,25 @@ func (f *FileSystem) stream(conn io.ReadWriteCloser) {
 		return
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
+	if structs.IsErrUnknownAllocation(err) {
+		handleStreamResultError(structs.NewErrUnknownAllocation(args.AllocID), helper.Int64ToPtr(404), encoder)
+		return
+	}
 	if err != nil {
 		handleStreamResultError(err, nil, encoder)
 		return
 	}
-	if alloc == nil {
-		handleStreamResultError(structs.NewErrUnknownAllocation(args.AllocID), helper.Int64ToPtr(404), encoder)
+
+	// Check namespace read-fs permissions.
+	if aclObj, err := f.srv.ResolveToken(args.AuthToken); err != nil {
+		handleStreamResultError(err, nil, encoder)
+		return
+	} else if aclObj != nil && !aclObj.AllowNsOp(alloc.Namespace, acl.NamespaceCapabilityReadFS) {
+		handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
 		return
 	}
+
 	nodeID := alloc.NodeID
 
 	// Make sure Node is valid and new enough to support RPC
@@ -346,22 +343,9 @@ func (f *FileSystem) logs(conn io.ReadWriteCloser) {
 		return
 	}
 
-	// Check node read permissions
-	if aclObj, err := f.srv.ResolveToken(args.AuthToken); err != nil {
-		handleStreamResultError(err, nil, encoder)
-		return
-	} else if aclObj != nil {
-		readfs := aclObj.AllowNsOp(args.QueryOptions.Namespace, acl.NamespaceCapabilityReadFS)
-		logs := aclObj.AllowNsOp(args.QueryOptions.Namespace, acl.NamespaceCapabilityReadLogs)
-		if !readfs && !logs {
-			handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
-			return
-		}
-	}
-
 	// Verify the arguments.
 	if args.AllocID == "" {
-		handleStreamResultError(errors.New("missing AllocID"), helper.Int64ToPtr(400), encoder)
+		handleStreamResultError(structs.ErrMissingAllocID, helper.Int64ToPtr(400), encoder)
 		return
 	}
 
@@ -372,15 +356,28 @@ func (f *FileSystem) logs(conn io.ReadWriteCloser) {
 		return
 	}
 
-	alloc, err := snap.AllocByID(nil, args.AllocID)
+	alloc, err := getAlloc(snap, args.AllocID)
+	if structs.IsErrUnknownAllocation(err) {
+		handleStreamResultError(structs.NewErrUnknownAllocation(args.AllocID), helper.Int64ToPtr(404), encoder)
+		return
+	}
 	if err != nil {
 		handleStreamResultError(err, nil, encoder)
 		return
 	}
-	if alloc == nil {
-		handleStreamResultError(structs.NewErrUnknownAllocation(args.AllocID), helper.Int64ToPtr(404), encoder)
+
+	// Check namespace read-logs *or* read-fs permissions.
+	allowNsOp := acl.NamespaceValidator(
+		acl.NamespaceCapabilityReadFS, acl.NamespaceCapabilityReadLogs)
+	aclObj, err := f.srv.ResolveToken(args.AuthToken)
+	if err != nil {
+		handleStreamResultError(err, nil, encoder)
+		return
+	} else if !allowNsOp(aclObj, alloc.Namespace) {
+		handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
 		return
 	}
+
 	nodeID := alloc.NodeID
 
 	// Make sure Node is valid and new enough to support RPC

nomad/client_fs_endpoint_test.go

@@ -107,7 +107,6 @@ func TestClientFS_List_Local(t *testing.T) {
 
 func TestClientFS_List_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := TestACLServer(t, nil)
@@ -122,6 +121,12 @@ func TestClientFS_List_ACL(t *testing.T) {
 	policyGood := mock.NamespacePolicy(structs.DefaultNamespace, "", []string{acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	// Upsert the allocation
+	state := s.State()
+	alloc := mock.Alloc()
+	require.NoError(t, state.UpsertJob(1010, alloc.Job))
+	require.NoError(t, state.UpsertAllocs(1011, []*structs.Allocation{alloc}))
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -135,12 +140,12 @@ func TestClientFS_List_ACL(t *testing.T) {
 		{
 			Name:          "good token",
 			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 		{
 			Name:          "root token",
 			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 	}
 
@@ -149,7 +154,7 @@ func TestClientFS_List_ACL(t *testing.T) {
 
 			// Make the request
 			req := &cstructs.FsListRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				Path:    "/",
 				QueryOptions: structs.QueryOptions{
 					Region:    "global",
@@ -161,8 +166,8 @@ func TestClientFS_List_ACL(t *testing.T) {
 			// Fetch the response
 			var resp cstructs.FsListResponse
 			err := msgpackrpc.CallWithCodec(codec, "FileSystem.List", req, &resp)
-			require.NotNil(err)
-			require.Contains(err.Error(), c.ExpectedError)
+			require.NotNil(t, err)
+			require.Contains(t, err.Error(), c.ExpectedError)
 		})
 	}
 }
@@ -376,7 +381,6 @@ func TestClientFS_Stat_Local(t *testing.T) {
 
 func TestClientFS_Stat_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := TestACLServer(t, nil)
@@ -391,6 +395,12 @@ func TestClientFS_Stat_ACL(t *testing.T) {
 	policyGood := mock.NamespacePolicy(structs.DefaultNamespace, "", []string{acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	// Upsert the allocation
+	state := s.State()
+	alloc := mock.Alloc()
+	require.NoError(t, state.UpsertJob(1010, alloc.Job))
+	require.NoError(t, state.UpsertAllocs(1011, []*structs.Allocation{alloc}))
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -404,12 +414,12 @@ func TestClientFS_Stat_ACL(t *testing.T) {
 		{
 			Name:          "good token",
 			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 		{
 			Name:          "root token",
 			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 	}
 
@@ -418,7 +428,7 @@ func TestClientFS_Stat_ACL(t *testing.T) {
 
 			// Make the request
 			req := &cstructs.FsStatRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				Path:    "/",
 				QueryOptions: structs.QueryOptions{
 					Region:    "global",
@@ -430,8 +440,8 @@ func TestClientFS_Stat_ACL(t *testing.T) {
 			// Fetch the response
 			var resp cstructs.FsStatResponse
 			err := msgpackrpc.CallWithCodec(codec, "FileSystem.Stat", req, &resp)
-			require.NotNil(err)
-			require.Contains(err.Error(), c.ExpectedError)
+			require.NotNil(t, err)
+			require.Contains(t, err.Error(), c.ExpectedError)
 		})
 	}
 }
@@ -601,7 +611,6 @@ OUTER:
 
 func TestClientFS_Streaming_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := TestACLServer(t, nil)
@@ -616,6 +625,12 @@ func TestClientFS_Streaming_ACL(t *testing.T) {
 		[]string{acl.NamespaceCapabilityReadLogs, acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	// Upsert the allocation
+	state := s.State()
+	alloc := mock.Alloc()
+	require.NoError(t, state.UpsertJob(1010, alloc.Job))
+	require.NoError(t, state.UpsertAllocs(1011, []*structs.Allocation{alloc}))
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -629,12 +644,12 @@ func TestClientFS_Streaming_ACL(t *testing.T) {
 		{
 			Name:          "good token",
 			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 		{
 			Name:          "root token",
 			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 	}
 
@@ -642,7 +657,7 @@ func TestClientFS_Streaming_ACL(t *testing.T) {
 		t.Run(c.Name, func(t *testing.T) {
 			// Make the request with bad allocation id
 			req := &cstructs.FsStreamRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				QueryOptions: structs.QueryOptions{
 					Namespace: structs.DefaultNamespace,
 					Region:    "global",
@@ -652,7 +667,7 @@ func TestClientFS_Streaming_ACL(t *testing.T) {
 
 			// Get the handler
 			handler, err := s.StreamingRpcHandler("FileSystem.Stream")
-			require.Nil(err)
+			require.NoError(t, err)
 
 			// Create a pipe
 			p1, p2 := net.Pipe()
@@ -683,7 +698,7 @@ func TestClientFS_Streaming_ACL(t *testing.T) {
 
 			// Send the request
 			encoder := codec.NewEncoder(p1, structs.MsgpackHandle)
-			require.Nil(encoder.Encode(req))
+			require.NoError(t, encoder.Encode(req))
 
 			timeout := time.After(5 * time.Second)
 
@@ -1423,7 +1438,6 @@ OUTER:
 
 func TestClientFS_Logs_ACL(t *testing.T) {
 	t.Parallel()
-	require := require.New(t)
 
 	// Start a server
 	s, root := TestACLServer(t, nil)
@@ -1438,6 +1452,12 @@ func TestClientFS_Logs_ACL(t *testing.T) {
 		[]string{acl.NamespaceCapabilityReadLogs, acl.NamespaceCapabilityReadFS})
 	tokenGood := mock.CreatePolicyAndToken(t, s.State(), 1009, "valid2", policyGood)
 
+	// Upsert the allocation
+	state := s.State()
+	alloc := mock.Alloc()
+	require.NoError(t, state.UpsertJob(1010, alloc.Job))
+	require.NoError(t, state.UpsertAllocs(1011, []*structs.Allocation{alloc}))
+
 	cases := []struct {
 		Name          string
 		Token         string
@@ -1451,12 +1471,12 @@ func TestClientFS_Logs_ACL(t *testing.T) {
 		{
 			Name:          "good token",
 			Token:         tokenGood.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 		{
 			Name:          "root token",
 			Token:         root.SecretID,
-			ExpectedError: structs.ErrUnknownAllocationPrefix,
+			ExpectedError: structs.ErrUnknownNodePrefix,
 		},
 	}
 
@@ -1464,7 +1484,7 @@ func TestClientFS_Logs_ACL(t *testing.T) {
 		t.Run(c.Name, func(t *testing.T) {
 			// Make the request with bad allocation id
 			req := &cstructs.FsLogsRequest{
-				AllocID: uuid.Generate(),
+				AllocID: alloc.ID,
 				QueryOptions: structs.QueryOptions{
 					Namespace: structs.DefaultNamespace,
 					Region:    "global",
@@ -1474,7 +1494,7 @@ func TestClientFS_Logs_ACL(t *testing.T) {
 
 			// Get the handler
 			handler, err := s.StreamingRpcHandler("FileSystem.Logs")
-			require.Nil(err)
+			require.NoError(t, err)
 
 			// Create a pipe
 			p1, p2 := net.Pipe()
@@ -1505,7 +1525,7 @@ func TestClientFS_Logs_ACL(t *testing.T) {
 
 			// Send the request
 			encoder := codec.NewEncoder(p1, structs.MsgpackHandle)
-			require.Nil(encoder.Encode(req))
+			require.NoError(t, encoder.Encode(req))
 
 			timeout := time.After(5 * time.Second)
 

nomad/deployment_endpoint.go

@@ -28,9 +28,11 @@ func (d *Deployment) GetDeployment(args *structs.DeploymentSpecificRequest,
 	defer metrics.MeasureSince([]string{"nomad", "deployment", "get_deployment"}, time.Now())
 
 	// Check namespace read-job permissions
-	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
+	allowNsOp := acl.NamespaceValidator(acl.NamespaceCapabilityReadJob)
+	aclObj, err := d.srv.ResolveToken(args.AuthToken)
+	if err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilityReadJob) {
+	} else if !allowNsOp(aclObj, args.RequestNamespace()) {
 		return structs.ErrPermissionDenied
 	}
 
@@ -53,6 +55,11 @@ func (d *Deployment) GetDeployment(args *structs.DeploymentSpecificRequest,
 			// Setup the output
 			reply.Deployment = out
 			if out != nil {
+				// Re-check namespace in case it differs from request.
+				if !allowNsOp(aclObj, out.Namespace) {
+					return structs.NewErrUnknownAllocation(args.DeploymentID)
+				}
+
 				reply.Index = out.ModifyIndex
 			} else {
 				// Use the last index that affected the deployments table
@@ -77,13 +84,6 @@ func (d *Deployment) Fail(args *structs.DeploymentFailRequest, reply *structs.De
 	}
 	defer metrics.MeasureSince([]string{"nomad", "deployment", "fail"}, time.Now())
 
-	// Check namespace submit-job permissions
-	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilitySubmitJob) {
-		return structs.ErrPermissionDenied
-	}
-
 	// Validate the arguments
 	if args.DeploymentID == "" {
 		return fmt.Errorf("missing deployment ID")
@@ -104,6 +104,13 @@ func (d *Deployment) Fail(args *structs.DeploymentFailRequest, reply *structs.De
 		return fmt.Errorf("deployment not found")
 	}
 
+	// Check namespace submit-job permissions
+	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(deploy.Namespace, acl.NamespaceCapabilitySubmitJob) {
+		return structs.ErrPermissionDenied
+	}
+
 	if !deploy.Active() {
 		return fmt.Errorf("can't fail terminal deployment")
 	}
@@ -119,13 +126,6 @@ func (d *Deployment) Pause(args *structs.DeploymentPauseRequest, reply *structs.
 	}
 	defer metrics.MeasureSince([]string{"nomad", "deployment", "pause"}, time.Now())
 
-	// Check namespace submit-job permissions
-	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilitySubmitJob) {
-		return structs.ErrPermissionDenied
-	}
-
 	// Validate the arguments
 	if args.DeploymentID == "" {
 		return fmt.Errorf("missing deployment ID")
@@ -146,6 +146,13 @@ func (d *Deployment) Pause(args *structs.DeploymentPauseRequest, reply *structs.
 		return fmt.Errorf("deployment not found")
 	}
 
+	// Check namespace submit-job permissions
+	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(deploy.Namespace, acl.NamespaceCapabilitySubmitJob) {
+		return structs.ErrPermissionDenied
+	}
+
 	if !deploy.Active() {
 		if args.Pause {
 			return fmt.Errorf("can't pause terminal deployment")
@@ -165,13 +172,6 @@ func (d *Deployment) Promote(args *structs.DeploymentPromoteRequest, reply *stru
 	}
 	defer metrics.MeasureSince([]string{"nomad", "deployment", "promote"}, time.Now())
 
-	// Check namespace submit-job permissions
-	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilitySubmitJob) {
-		return structs.ErrPermissionDenied
-	}
-
 	// Validate the arguments
 	if args.DeploymentID == "" {
 		return fmt.Errorf("missing deployment ID")
@@ -192,6 +192,13 @@ func (d *Deployment) Promote(args *structs.DeploymentPromoteRequest, reply *stru
 		return fmt.Errorf("deployment not found")
 	}
 
+	// Check namespace submit-job permissions
+	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(deploy.Namespace, acl.NamespaceCapabilitySubmitJob) {
+		return structs.ErrPermissionDenied
+	}
+
 	if !deploy.Active() {
 		return fmt.Errorf("can't promote terminal deployment")
 	}
@@ -208,13 +215,6 @@ func (d *Deployment) SetAllocHealth(args *structs.DeploymentAllocHealthRequest,
 	}
 	defer metrics.MeasureSince([]string{"nomad", "deployment", "set_alloc_health"}, time.Now())
 
-	// Check namespace submit-job permissions
-	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
-		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilitySubmitJob) {
-		return structs.ErrPermissionDenied
-	}
-
 	// Validate the arguments
 	if args.DeploymentID == "" {
 		return fmt.Errorf("missing deployment ID")
@@ -239,6 +239,13 @@ func (d *Deployment) SetAllocHealth(args *structs.DeploymentAllocHealthRequest,
 		return fmt.Errorf("deployment not found")
 	}
 
+	// Check namespace submit-job permissions
+	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if aclObj != nil && !aclObj.AllowNsOp(deploy.Namespace, acl.NamespaceCapabilitySubmitJob) {
+		return structs.ErrPermissionDenied
+	}
+
 	if !deploy.Active() {
 		return fmt.Errorf("can't set health of allocations for a terminal deployment")
 	}
@@ -254,7 +261,8 @@ func (d *Deployment) List(args *structs.DeploymentListRequest, reply *structs.De
 	}
 	defer metrics.MeasureSince([]string{"nomad", "deployment", "list"}, time.Now())
 
-	// Check namespace read-job permissions
+	// Check namespace read-job permissions against request namespace since
+	// results are filtered by request namespace.
 	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
 		return err
 	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilityReadJob) {
@@ -310,10 +318,14 @@ func (d *Deployment) Allocations(args *structs.DeploymentSpecificRequest, reply
 	}
 	defer metrics.MeasureSince([]string{"nomad", "deployment", "allocations"}, time.Now())
 
-	// Check namespace read-job permissions
-	if aclObj, err := d.srv.ResolveToken(args.AuthToken); err != nil {
+	// Check namespace read-job permissions against the request namespace.
+	// Must re-check against the alloc namespace when they return to ensure
+	// there's no namespace mismatch.
+	allowNsOp := acl.NamespaceValidator(acl.NamespaceCapabilityReadJob)
+	aclObj, err := d.srv.ResolveToken(args.AuthToken)
+	if err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilityReadJob) {
+	} else if !allowNsOp(aclObj, args.RequestNamespace()) {
 		return structs.ErrPermissionDenied
 	}
 
@@ -328,6 +340,15 @@ func (d *Deployment) Allocations(args *structs.DeploymentSpecificRequest, reply
 				return err
 			}
 
+			// Deployments do not span namespaces so just check the
+			// first allocs namespace.
+			if len(allocs) > 0 {
+				ns := allocs[0].Namespace
+				if ns != args.RequestNamespace() && !allowNsOp(aclObj, ns) {
+					return structs.ErrPermissionDenied
+				}
+			}
+
 			stubs := make([]*structs.AllocListStub, 0, len(allocs))
 			for _, alloc := range allocs {
 				stubs = append(stubs, alloc.Stub())

nomad/eval_endpoint.go

@@ -34,10 +34,12 @@ func (e *Eval) GetEval(args *structs.EvalSpecificRequest,
 	}
 	defer metrics.MeasureSince([]string{"nomad", "eval", "get_eval"}, time.Now())
 
-	// Check for read-job permissions
-	if aclObj, err := e.srv.ResolveToken(args.AuthToken); err != nil {
+	// Check for read-job permissions before performing blocking query.
+	allowNsOp := acl.NamespaceValidator(acl.NamespaceCapabilityReadJob)
+	aclObj, err := e.srv.ResolveToken(args.AuthToken)
+	if err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilityReadJob) {
+	} else if !allowNsOp(aclObj, args.RequestNamespace()) {
 		return structs.ErrPermissionDenied
 	}
 
@@ -55,6 +57,11 @@ func (e *Eval) GetEval(args *structs.EvalSpecificRequest,
 			// Setup the output
 			reply.Eval = out
 			if out != nil {
+				// Re-check namespace in case it differs from request.
+				if !allowNsOp(aclObj, out.Namespace) {
+					return structs.ErrPermissionDenied
+				}
+
 				reply.Index = out.ModifyIndex
 			} else {
 				// Use the last index that affected the nodes table
@@ -389,9 +396,11 @@ func (e *Eval) Allocations(args *structs.EvalSpecificRequest,
 	defer metrics.MeasureSince([]string{"nomad", "eval", "allocations"}, time.Now())
 
 	// Check for read-job permissions
-	if aclObj, err := e.srv.ResolveToken(args.AuthToken); err != nil {
+	allowNsOp := acl.NamespaceValidator(acl.NamespaceCapabilityReadJob)
+	aclObj, err := e.srv.ResolveToken(args.AuthToken)
+	if err != nil {
 		return err
-	} else if aclObj != nil && !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilityReadJob) {
+	} else if !allowNsOp(aclObj, args.RequestNamespace()) {
 		return structs.ErrPermissionDenied
 	}
 
@@ -408,6 +417,13 @@ func (e *Eval) Allocations(args *structs.EvalSpecificRequest,
 
 			// Convert to a stub
 			if len(allocs) > 0 {
+				// Evaluations do not span namespaces so just check the
+				// first allocs namespace.
+				ns := allocs[0].Namespace
+				if ns != args.RequestNamespace() && !allowNsOp(aclObj, ns) {
+					return structs.ErrPermissionDenied
+				}
+
 				reply.Allocations = make([]*structs.AllocListStub, 0, len(allocs))
 				for _, alloc := range allocs {
 					reply.Allocations = append(reply.Allocations, alloc.Stub())

nomad/structs/errors.go

@@ -16,6 +16,7 @@ const (
 	errUnknownMethod              = "Unknown rpc method"
 	errUnknownNomadVersion        = "Unable to determine Nomad version"
 	errNodeLacksRpc               = "Node does not support RPC; requires 0.8 or later"
+	errMissingAllocID             = "Missing allocation ID"
 
 	// Prefix based errors that are used to check if the error is of a given
 	// type. These errors should be created with the associated constructor.
@@ -36,6 +37,7 @@ var (
 	ErrUnknownMethod              = errors.New(errUnknownMethod)
 	ErrUnknownNomadVersion        = errors.New(errUnknownNomadVersion)
 	ErrNodeLacksRpc               = errors.New(errNodeLacksRpc)
+	ErrMissingAllocID             = errors.New(errMissingAllocID)
 )
 
 // IsErrNoLeader returns whether the error is due to there being no leader.

nomad/structs/structs.go

@@ -207,6 +207,13 @@ type QueryOptions struct {
 	Region string
 
 	// Namespace is the target namespace for the query.
+	//
+	// Since handlers do not have a default value set they should access
+	// the Namespace via the RequestNamespace method.
+	//
+	// Requests accessing specific namespaced objects must check ACLs
+	// against the namespace of the object, not the namespace in the
+	// request.
 	Namespace string
 
 	// If set, wait until query exceeds given index. Must be provided
@@ -233,6 +240,11 @@ func (q QueryOptions) RequestRegion() string {
 	return q.Region
 }
 
+// RequestNamespace returns the request's namespace or the default namespace if
+// no explicit namespace was sent.
+//
+// Requests accessing specific namespaced objects must check ACLs against the
+// namespace of the object, not the namespace in the request.
 func (q QueryOptions) RequestNamespace() string {
 	if q.Namespace == "" {
 		return DefaultNamespace
@@ -254,6 +266,13 @@ type WriteRequest struct {
 	Region string
 
 	// Namespace is the target namespace for the write.
+	//
+	// Since RPC handlers do not have a default value set they should
+	// access the Namespace via the RequestNamespace method.
+	//
+	// Requests accessing specific namespaced objects must check ACLs
+	// against the namespace of the object, not the namespace in the
+	// request.
 	Namespace string
 
 	// AuthToken is secret portion of the ACL token used for the request
@@ -267,6 +286,11 @@ func (w WriteRequest) RequestRegion() string {
 	return w.Region
 }
 
+// RequestNamespace returns the request's namespace or the default namespace if
+// no explicit namespace was sent.
+//
+// Requests accessing specific namespaced objects must check ACLs against the
+// namespace of the object, not the namespace in the request.
 func (w WriteRequest) RequestNamespace() string {
 	if w.Namespace == "" {
 		return DefaultNamespace

nomad/util.go

@@ -8,6 +8,7 @@ import (
 	"path/filepath"
 	"strconv"
 
+	memdb "github.com/hashicorp/go-memdb"
 	version "github.com/hashicorp/go-version"
 	"github.com/hashicorp/nomad/nomad/state"
 	"github.com/hashicorp/nomad/nomad/structs"
@@ -275,3 +276,28 @@ func nodeSupportsRpc(node *structs.Node) error {
 
 	return nil
 }
+
+// AllocGetter is an interface for retrieving allocations by ID. It is
+// satisfied by *state.StateStore and *state.StateSnapshot.
+type AllocGetter interface {
+	AllocByID(ws memdb.WatchSet, id string) (*structs.Allocation, error)
+}
+
+// getAlloc retrieves an allocation by ID and namespace. If the allocation is
+// nil, an error is returned.
+func getAlloc(state AllocGetter, allocID string) (*structs.Allocation, error) {
+	if allocID == "" {
+		return nil, structs.ErrMissingAllocID
+	}
+
+	alloc, err := state.AllocByID(nil, allocID)
+	if err != nil {
+		return nil, err
+	}
+
+	if alloc == nil {
+		return nil, structs.NewErrUnknownAllocation(allocID)
+	}
+
+	return alloc, nil
+}

testutil/wait.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/kr/pretty"
 	testing "github.com/mitchellh/go-testing-interface"
 	"github.com/stretchr/testify/require"
 )
@@ -125,13 +126,14 @@ func WaitForVotingMembers(t testing.T, rpc rpcFn, nPeers int) {
 	})
 }
 
+// RegisterJobWithToken registers a job and uses the job's Region and Namespace.
 func RegisterJobWithToken(t testing.T, rpc rpcFn, job *structs.Job, token string) {
 	WaitForResult(func() (bool, error) {
 		args := &structs.JobRegisterRequest{}
 		args.Job = job
-		args.WriteRequest.Region = "global"
+		args.WriteRequest.Region = job.Region
 		args.AuthToken = token
-		args.Namespace = structs.DefaultNamespace
+		args.Namespace = job.Namespace
 		var jobResp structs.JobRegisterResponse
 		err := rpc("Job.Register", args, &jobResp)
 		return err == nil, fmt.Errorf("Job.Register error: %v", err)
@@ -154,16 +156,18 @@ func WaitForRunningWithToken(t testing.T, rpc rpcFn, job *structs.Job, token str
 	WaitForResult(func() (bool, error) {
 		args := &structs.JobSpecificRequest{}
 		args.JobID = job.ID
-		args.QueryOptions.Region = "global"
+		args.QueryOptions.Region = job.Region
 		args.AuthToken = token
-		args.Namespace = structs.DefaultNamespace
+		args.Namespace = job.Namespace
 		err := rpc("Job.Allocations", args, &resp)
 		if err != nil {
 			return false, fmt.Errorf("Job.Allocations error: %v", err)
 		}
 
 		if len(resp.Allocations) == 0 {
-			return false, fmt.Errorf("0 allocations")
+			evals := structs.JobEvaluationsResponse{}
+			require.NoError(t, rpc("Job.Evaluations", args, &evals), "error looking up evals")
+			return false, fmt.Errorf("0 allocations; evals: %s", pretty.Sprint(evals.Evaluations))
 		}
 
 		for _, alloc := range resp.Allocations {