Role changes to support enforcing workflow restrictions (#96744)

This PR implements necessary changes to `Role` classes in order to enforce 
workflow restrictions for API keys. Main change is around resolving roles from 
role descriptors, where a role can either be effectively empty (or not) 
depending on its workflow restriction. 

Currently, the only workflow supported is `search_application_query` which 
allows restricting access only to Search Application Search API.

For simplicity, when creating an API key, it's possible to specify only a single 
role descriptor with workflows restriction. For example:

```
POST /_security/api_key
{
  "name": "my_restricted_api_key",
  "role_descriptors": {
    "my_restricted_role": {
      "indices": [
        {
          "names": ["books"],
          "privileges": ["read"]
        }
      ],
      "restriction": {
        "workflows": ["search_application_query"]
      }
    }
  }
}
```

Relates to #96215

Author: slobodanadamovic

docs/changelog/96744.yaml

@@ -0,0 +1,5 @@
+pr: 96744
+summary: Support restricting access of API keys to only certain workflows
+area: Authorization
+type: feature
+issues: []

server/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -1838,6 +1838,12 @@ private enum ElasticsearchExceptionHandle {
             org.elasticsearch.http.HttpHeadersValidationException::new,
             169,
             TransportVersion.V_8_9_0
+        ),
+        ROLE_RESTRICTION_EXCEPTION(
+            ElasticsearchRoleRestrictionException.class,
+            ElasticsearchRoleRestrictionException::new,
+            170,
+            TransportVersion.V_8_500_016
         );
 
         final Class<? extends ElasticsearchException> exceptionClass;

server/src/main/java/org/elasticsearch/ElasticsearchRoleRestrictionException.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.rest.RestStatus;
+
+import java.io.IOException;
+
+/**
+ * This exception is thrown to indicate that the access has been denied because of role restrictions that
+ * an authenticated subject might have (e.g. not allowed to access certain APIs).
+ * This differs from other 403 error in sense that it's additional access control that is enforced after role
+ * is resolved and before permissions are checked.
+ */
+public class ElasticsearchRoleRestrictionException extends ElasticsearchSecurityException {
+
+    public ElasticsearchRoleRestrictionException(String msg, Throwable cause, Object... args) {
+        super(msg, RestStatus.FORBIDDEN, cause, args);
+    }
+
+    public ElasticsearchRoleRestrictionException(String msg, Object... args) {
+        this(msg, null, args);
+    }
+
+    public ElasticsearchRoleRestrictionException(StreamInput in) throws IOException {
+        super(in);
+    }
+}

server/src/main/java/org/elasticsearch/TransportVersion.java

@@ -140,9 +140,10 @@ private static TransportVersion registerTransportVersion(int id, String uniqueId
     public static final TransportVersion V_8_500_013 = registerTransportVersion(8_500_013, "f65b85ac-db5e-4558-a487-a1dde4f6a33a");
     public static final TransportVersion V_8_500_014 = registerTransportVersion(8_500_014, "D115A2E1-1739-4A02-AB7B-64F6EA157EFB");
     public static final TransportVersion V_8_500_015 = registerTransportVersion(8_500_015, "651216c9-d54f-4189-9fe1-48d82d276863");
+    public static final TransportVersion V_8_500_016 = registerTransportVersion(8_500_016, "492C94FB-AAEA-4C9E-8375-BDB67A398584");
 
     private static class CurrentHolder {
-        private static final TransportVersion CURRENT = findCurrent(V_8_500_015);
+        private static final TransportVersion CURRENT = findCurrent(V_8_500_016);
 
         // finds the pluggable current version, or uses the given fallback
         private static TransportVersion findCurrent(TransportVersion fallback) {

server/src/test/java/org/elasticsearch/ExceptionSerializationTests.java

@@ -829,6 +829,7 @@ public void testIds() {
         ids.put(167, UnsupportedAggregationOnDownsampledIndex.class);
         ids.put(168, DocumentParsingException.class);
         ids.put(169, HttpHeadersValidationException.class);
+        ids.put(170, ElasticsearchRoleRestrictionException.class);
 
         Map<Class<? extends ElasticsearchException>, Integer> reverse = new HashMap<>();
         for (Map.Entry<Integer, Class<? extends ElasticsearchException>> entry : ids.entrySet()) {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/LimitedRole.java

@@ -51,6 +51,7 @@ public final class LimitedRole implements Role {
     public LimitedRole(Role baseRole, Role limitedByRole) {
         this.baseRole = Objects.requireNonNull(baseRole);
         this.limitedByRole = Objects.requireNonNull(limitedByRole, "limited by role is required to create limited role");
+        assert false == limitedByRole.hasWorkflowsRestriction() : "limited-by role must not have workflows restriction";
     }
 
     @Override
@@ -74,6 +75,28 @@ public RemoteIndicesPermission remoteIndices() {
         throw new UnsupportedOperationException("cannot retrieve remote indices permission on limited role");
     }
 
+    @Override
+    public boolean hasWorkflowsRestriction() {
+        return baseRole.hasWorkflowsRestriction() || limitedByRole.hasWorkflowsRestriction();
+    }
+
+    @Override
+    public Role forWorkflow(String workflow) {
+        Role baseRestricted = baseRole.forWorkflow(workflow);
+        if (baseRestricted == EMPTY_RESTRICTED_BY_WORKFLOW) {
+            return EMPTY_RESTRICTED_BY_WORKFLOW;
+        }
+        Role limitedByRestricted = limitedByRole.forWorkflow(workflow);
+        if (limitedByRestricted == EMPTY_RESTRICTED_BY_WORKFLOW) {
+            return EMPTY_RESTRICTED_BY_WORKFLOW;
+        }
+        if (baseRestricted == baseRole && limitedByRestricted == limitedByRole) {
+            return this;
+        } else {
+            return baseRestricted.limitedBy(limitedByRestricted);
+        }
+    }
+
     @Override
     public ApplicationPermission application() {
         throw new UnsupportedOperationException("cannot retrieve application permission on limited role");

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/Role.java

@@ -27,6 +27,8 @@
 import org.elasticsearch.xpack.core.security.authz.privilege.ConfigurableClusterPrivilege;
 import org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege;
 import org.elasticsearch.xpack.core.security.authz.privilege.Privilege;
+import org.elasticsearch.xpack.core.security.authz.restriction.WorkflowResolver;
+import org.elasticsearch.xpack.core.security.authz.restriction.WorkflowsRestriction;
 import org.elasticsearch.xpack.core.security.support.Automatons;
 
 import java.util.ArrayList;
@@ -46,6 +48,8 @@ public interface Role {
 
     Role EMPTY = builder(new RestrictedIndices(Automatons.EMPTY)).build();
 
+    Role EMPTY_RESTRICTED_BY_WORKFLOW = builder(new RestrictedIndices(Automatons.EMPTY)).workflows(Set.of()).build();
+
     String[] names();
 
     ClusterPermission cluster();
@@ -58,6 +62,19 @@ public interface Role {
 
     RemoteIndicesPermission remoteIndices();
 
+    boolean hasWorkflowsRestriction();
+
+    /**
+     * This method returns an effective role for the given workflow if role has workflows restriction
+     * (i.e. {@link #hasWorkflowsRestriction} is true). Otherwise, this method returns an unchanged role.
+     *
+     * The returned effective role can be an {@link #EMPTY_RESTRICTED_BY_WORKFLOW} when the given workflow is
+     * not one of the workflows to which this role is restricted.
+     *
+     * The workflows to which a role can be restricted are static and defined in {@link WorkflowResolver}.
+     */
+    Role forWorkflow(@Nullable String workflow);
+
     /**
      * Whether the Role has any field or document level security enabled index privileges
      * @return
@@ -176,7 +193,7 @@ IndicesAccessControl authorize(
     /***
      * Creates a {@link LimitedRole} that uses this Role as base and the given role as limited-by.
      */
-    default LimitedRole limitedBy(Role role) {
+    default Role limitedBy(Role role) {
         return new LimitedRole(this, role);
     }
 
@@ -200,6 +217,7 @@ class Builder {
         private final Map<Set<String>, List<IndicesPermissionGroupDefinition>> remoteGroups = new HashMap<>();
         private final List<Tuple<ApplicationPrivilege, Set<String>>> applicationPrivs = new ArrayList<>();
         private final RestrictedIndices restrictedIndices;
+        private WorkflowsRestriction workflowsRestriction = WorkflowsRestriction.NONE;
 
         private Builder(RestrictedIndices restrictedIndices, String[] names) {
             this.restrictedIndices = restrictedIndices;
@@ -259,6 +277,15 @@ public Builder addApplicationPrivilege(ApplicationPrivilege privilege, Set<Strin
             return this;
         }
 
+        public Builder workflows(Set<String> workflowNames) {
+            if (workflowNames == null) {
+                this.workflowsRestriction = WorkflowsRestriction.NONE;
+            } else {
+                this.workflowsRestriction = new WorkflowsRestriction(workflowNames);
+            }
+            return this;
+        }
+
         public SimpleRole build() {
             final IndicesPermission indices;
             if (groups.isEmpty()) {
@@ -301,7 +328,7 @@ public SimpleRole build() {
             final ApplicationPermission applicationPermission = applicationPrivs.isEmpty()
                 ? ApplicationPermission.NONE
                 : new ApplicationPermission(applicationPrivs);
-            return new SimpleRole(names, cluster, indices, applicationPermission, runAs, remoteIndices);
+            return new SimpleRole(names, cluster, indices, applicationPermission, runAs, remoteIndices, workflowsRestriction);
         }
 
         private static class IndicesPermissionGroupDefinition {
@@ -392,6 +419,10 @@ static SimpleRole buildFromRoleDescriptor(
             builder.runAs(new Privilege(Sets.newHashSet(rdRunAs), rdRunAs));
         }
 
+        if (roleDescriptor.hasWorkflowsRestriction()) {
+            builder.workflows(Sets.newHashSet(roleDescriptor.getRestriction().getWorkflows()));
+        }
+
         return builder.build();
     }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/SimpleRole.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.xpack.core.security.authz.permission.IndicesPermission.IsResourceAuthorizedPredicate;
 import org.elasticsearch.xpack.core.security.authz.privilege.ApplicationPrivilegeDescriptor;
 import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilege;
+import org.elasticsearch.xpack.core.security.authz.restriction.WorkflowsRestriction;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -51,21 +52,24 @@ public class SimpleRole implements Role {
     private final ApplicationPermission application;
     private final RunAsPermission runAs;
     private final RemoteIndicesPermission remoteIndices;
+    private final WorkflowsRestriction workflowsRestriction;
 
     SimpleRole(
         String[] names,
         ClusterPermission cluster,
         IndicesPermission indices,
         ApplicationPermission application,
         RunAsPermission runAs,
-        RemoteIndicesPermission remoteIndices
+        RemoteIndicesPermission remoteIndices,
+        WorkflowsRestriction workflowsRestriction
     ) {
         this.names = names;
         this.cluster = Objects.requireNonNull(cluster);
         this.indices = Objects.requireNonNull(indices);
         this.application = Objects.requireNonNull(application);
         this.runAs = Objects.requireNonNull(runAs);
         this.remoteIndices = Objects.requireNonNull(remoteIndices);
+        this.workflowsRestriction = Objects.requireNonNull(workflowsRestriction);
     }
 
     @Override
@@ -98,6 +102,20 @@ public RemoteIndicesPermission remoteIndices() {
         return remoteIndices;
     }
 
+    @Override
+    public boolean hasWorkflowsRestriction() {
+        return workflowsRestriction.hasWorkflows();
+    }
+
+    @Override
+    public Role forWorkflow(String workflow) {
+        if (workflowsRestriction.isWorkflowAllowed(workflow)) {
+            return this;
+        } else {
+            return EMPTY_RESTRICTED_BY_WORKFLOW;
+        }
+    }
+
     @Override
     public boolean hasFieldOrDocumentLevelSecurity() {
         return indices.hasFieldOrDocumentLevelSecurity();

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/restriction/WorkflowsRestriction.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.security.authz.restriction;
+
+import org.elasticsearch.core.Nullable;
+
+import java.util.Set;
+import java.util.function.Predicate;
+
+public final class WorkflowsRestriction {
+
+    /**
+     * Default behaviour is no restriction which allows all workflows.
+     */
+    public static final WorkflowsRestriction NONE = new WorkflowsRestriction(null);
+
+    private final Set<String> names;
+    private final Predicate<String> predicate;
+
+    public WorkflowsRestriction(Set<String> names) {
+        this.names = names;
+        if (names == null) {
+            // No restriction, all workflows are allowed
+            this.predicate = name -> true;
+        } else if (names.isEmpty()) {
+            // Empty restriction, no workflow is allowed
+            this.predicate = name -> false;
+        } else {
+            this.predicate = name -> {
+                if (name == null) {
+                    return false;
+                } else {
+                    return names.contains(name);
+                }
+            };
+        }
+    }
+
+    public boolean hasWorkflows() {
+        return this.names != null;
+    }
+
+    public boolean isWorkflowAllowed(@Nullable String workflow) {
+        return predicate.test(workflow);
+    }
+
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/permission/LimitedRoleTests.java

@@ -38,6 +38,8 @@
 import org.elasticsearch.xpack.core.security.authz.privilege.ApplicationPrivilegeTests;
 import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilegeResolver;
 import org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege;
+import org.elasticsearch.xpack.core.security.authz.restriction.Workflow;
+import org.elasticsearch.xpack.core.security.authz.restriction.WorkflowResolver;
 import org.elasticsearch.xpack.core.security.support.Automatons;
 import org.junit.Before;
 
@@ -54,6 +56,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -767,6 +770,41 @@ public void testHasPrivilegesForIndexPatterns() {
         }
     }
 
+    public void testForWorkflowRestriction() {
+        // Test when role is restricted to the same workflow as originating workflow
+        {
+            Workflow workflow = WorkflowResolver.SEARCH_APPLICATION_QUERY_WORKFLOW;
+            Role baseRole = Role.builder(EMPTY_RESTRICTED_INDICES, "role-a")
+                .add(IndexPrivilege.READ, "index-a")
+                .workflows(Set.of(workflow.name()))
+                .build();
+            Role limitedBy = Role.builder(EMPTY_RESTRICTED_INDICES, "role-b").add(IndexPrivilege.READ, "index-a").build();
+            Role role = baseRole.limitedBy(limitedBy);
+            assertThat(role.hasWorkflowsRestriction(), equalTo(true));
+            assertThat(role.forWorkflow(workflow.name()), sameInstance(role));
+        }
+        // Test restriction when role is not restricted regardless of originating workflow
+        {
+            String originatingWorkflow = randomBoolean() ? null : WorkflowResolver.SEARCH_APPLICATION_QUERY_WORKFLOW.name();
+            Role baseRole = Role.builder(EMPTY_RESTRICTED_INDICES, "role-a").add(IndexPrivilege.READ, "index-a").build();
+            Role limitedBy = Role.builder(EMPTY_RESTRICTED_INDICES, "role-b").add(IndexPrivilege.READ, "index-a").build();
+            Role role = baseRole.limitedBy(limitedBy);
+            assertThat(role.forWorkflow(originatingWorkflow), sameInstance(role));
+            assertThat(role.hasWorkflowsRestriction(), equalTo(false));
+        }
+        // Test when role is restricted but originating workflow is not allowed
+        {
+            Role baseRole = Role.builder(EMPTY_RESTRICTED_INDICES, "role-a")
+                .add(IndexPrivilege.READ, "index-a")
+                .workflows(Set.of(WorkflowResolver.SEARCH_APPLICATION_QUERY_WORKFLOW.name()))
+                .build();
+            Role limitedBy = Role.builder(EMPTY_RESTRICTED_INDICES, "role-b").add(IndexPrivilege.READ, "index-a").build();
+            Role role = baseRole.limitedBy(limitedBy);
+            assertThat(role.forWorkflow(randomFrom(randomAlphaOfLength(9), null, "")), sameInstance(Role.EMPTY_RESTRICTED_BY_WORKFLOW));
+            assertThat(role.hasWorkflowsRestriction(), equalTo(true));
+        }
+    }
+
     public void testGetApplicationPrivilegesByResource() {
         final ApplicationPrivilege app1Read = defineApplicationPrivilege("app1", "read", "data:read/*");
         final ApplicationPrivilege app1All = defineApplicationPrivilege("app1", "all", "*");