RBAC integration (#168)

PR includes:
- RBAC integretion in 'rest' module
- updated tests
- added dependencies in build.gradle files
- V3 schema in archive module for storing roles and groups from json_data into rbac_labels
- javadoc

Added new package 'rbac' to rest sub-module with sub-package 'annotations' for spring AOP.

Added new classes:

RbacHttpFilter
First step of validation.
Handles:
- no validation healthcheck
- stores all groups and roles values from request
- creates object of type UserType for further processing
- prevents triggering endpoints when request is not validated

RbacAccessAspect
Aspect class for handling annotated rest controller methods.
Handles:
- triggering methods only accessible by admin
- triggering BulkResponse methods and modifing returned object
- triggering methods that require certain parameters in path or in body of request
- when user has not access, 403 is thrown

RbacDbHandler
Validation of certain parameteres before reaching endpoints and returning of certain object (omitting direct endpoints).
Handles:
- checks if certain parameters are present in db and if user has access due to provided values in groups and roles of header
- returns all Task/Workflow definitions and SeachResult object with param Workflow/WorkflowSummary directly

RbacProperties
Properties class to return admin roles and groups specified in properties file.

UserType
POJO for creating user object.

TYPE: Improvement
JIRA: DEP-686

Signed-off-by: jmasar <[email protected]>
Co-authored-by: jmasar <[email protected]>

client/src/main/java/com/netflix/conductor/client/http/ClientRequestHandler.java

@@ -33,6 +33,9 @@
 public class ClientRequestHandler {
     private final Client client;
 
+    private final String FROM_HEADER = "from";
+    private final String HEADER_VALUE = "test";
+
     public ClientRequestHandler(
             ClientConfig config, ClientHandler handler, ClientFilter... filters) {
         ObjectMapper objectMapper = new ObjectMapperProvider().getObjectMapper();
@@ -60,22 +63,25 @@ public BulkResponse delete(URI uri, Object body) {
         if (body != null) {
             return client.resource(uri)
                     .type(MediaType.APPLICATION_JSON_TYPE)
+                    .header(FROM_HEADER, HEADER_VALUE)
                     .delete(BulkResponse.class, body);
         } else {
-            client.resource(uri).delete();
+            client.resource(uri).header(FROM_HEADER, HEADER_VALUE).delete();
         }
         return null;
     }
 
     public ClientResponse get(URI uri) {
         return client.resource(uri)
                 .accept(MediaType.APPLICATION_JSON, MediaType.TEXT_PLAIN)
+                .header(FROM_HEADER, HEADER_VALUE)
                 .get(ClientResponse.class);
     }
 
     public WebResource.Builder getWebResourceBuilder(URI URI, Object entity) {
         return client.resource(URI)
                 .type(MediaType.APPLICATION_JSON)
+                .header(FROM_HEADER, HEADER_VALUE)
                 .entity(entity)
                 .accept(MediaType.TEXT_PLAIN, MediaType.APPLICATION_JSON);
     }

core/src/main/java/com/netflix/conductor/core/dal/ExecutionDAOFacade.java

@@ -759,6 +759,40 @@ public void populateTaskData(TaskModel taskModel) {
         }
     }
 
+    public boolean hasAccess(Object[] args, List<String> labels) {
+        return executionDAO.hasAccess(args, labels);
+    }
+
+    public boolean exists(Object[] args) {
+        return executionDAO.exists(args);
+    }
+
+    public List<String> getUserWorkflowIds(List<String> labels) {
+        return executionDAO.getUserWorkflowIds(labels);
+    }
+
+    public List<String> getPresentIds(List<String> ids) {
+        return executionDAO.getPresentIds(ids);
+    }
+
+    public SearchResult<String> getSearchResultIds(List<String> roles) {
+        return executionDAO.getSearchResultIds(roles);
+    }
+
+    public SearchResult<WorkflowSummary> getSummaries(SearchResult<String> searchResultIds) {
+        return indexDAO.getSummaries(searchResultIds);
+    }
+
+    public SearchResult<Workflow> getUserWorkflows(SearchResult<String> searchResultIds) {
+        final List<Workflow> workflows =
+                searchResultIds.getResults().stream()
+                        .map(id -> executionDAO.getWorkflow(id, false))
+                        .filter(Objects::nonNull)
+                        .map(WorkflowModel::toWorkflow)
+                        .toList();
+        return new SearchResult<>(workflows.size(), workflows);
+    }
+
     class DelayWorkflowUpdate implements Runnable {
 
         private final String workflowId;

core/src/main/java/com/netflix/conductor/core/index/NoopIndexDAO.java

@@ -160,4 +160,9 @@ public List<String> searchArchivableWorkflows(String indexName, long archiveTtlD
     public long getWorkflowCount(String query, String freeText) {
         return 0;
     }
+
+    @Override
+    public SearchResult<WorkflowSummary> getSummaries(SearchResult<String> searchResultIds) {
+        return new SearchResult<>(0, Collections.emptyList());
+    }
 }

core/src/main/java/com/netflix/conductor/dao/ExecutionDAO.java

@@ -16,6 +16,7 @@
 
 import com.netflix.conductor.common.metadata.events.EventExecution;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
+import com.netflix.conductor.common.run.SearchResult;
 import com.netflix.conductor.model.TaskModel;
 import com.netflix.conductor.model.WorkflowModel;
 
@@ -222,4 +223,14 @@ List<WorkflowModel> getWorkflowsByCorrelationId(
     default List<WorkflowModel> getWorkflowFamily(String workflowId, boolean summaryOnly) {
         throw new UnsupportedOperationException();
     }
+
+    boolean hasAccess(Object[] args, List<String> labels);
+
+    boolean exists(Object[] args);
+
+    List<String> getUserWorkflowIds(List<String> labels);
+
+    List<String> getPresentIds(List<String> ids);
+
+    SearchResult<String> getSearchResultIds(List<String> roles);
 }

core/src/main/java/com/netflix/conductor/dao/IndexDAO.java

@@ -247,4 +247,6 @@ CompletableFuture<Void> asyncUpdateTask(
      * @return Number of matches for the query
      */
     long getWorkflowCount(String query, String freeText);
+
+    SearchResult<WorkflowSummary> getSummaries(SearchResult<String> searchResultIds);
 }

core/src/main/java/com/netflix/conductor/dao/MetadataDAO.java

@@ -15,6 +15,7 @@
 import java.util.List;
 import java.util.Optional;
 
+import com.netflix.conductor.common.metadata.BaseDef;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
 
@@ -86,4 +87,12 @@ public interface MetadataDAO {
      * @return List the latest versions of the workflow definitions
      */
     List<WorkflowDef> getAllWorkflowDefsLatestVersions();
+
+    boolean hasAccess(Object[] args, List<String> labels, String uri);
+
+    boolean exists(Object[] args, String uri);
+
+    List<? extends BaseDef> getUserTaskDefs(List<String> roles);
+
+    List<? extends BaseDef> getUserWorkflowDefs(List<String> roles);
 }

core/src/main/java/com/netflix/conductor/service/ExecutionService.java

@@ -611,4 +611,32 @@ public ExternalStorageLocation getExternalStorageLocation(
     public List<String> getWorkflowPath(String workflowId) {
         return executionDAOFacade.getWorkflowPath(workflowId);
     }
+
+    public boolean hasAccess(Object[] args, List<String> labels) {
+        return executionDAOFacade.hasAccess(args, labels);
+    }
+
+    public boolean exists(Object[] args) {
+        return executionDAOFacade.exists(args);
+    }
+
+    public List<String> getUserWorkflowIds(List<String> labels) {
+        return executionDAOFacade.getUserWorkflowIds(labels);
+    }
+
+    public List<String> getPresentIds(List<String> ids) {
+        return executionDAOFacade.getPresentIds(ids);
+    }
+
+    public SearchResult<String> getSearchResultIds(List<String> roles) {
+        return executionDAOFacade.getSearchResultIds(roles);
+    }
+
+    public SearchResult<WorkflowSummary> getSummaries(SearchResult<String> searchResultIds) {
+        return executionDAOFacade.getSummaries(searchResultIds);
+    }
+
+    public SearchResult<Workflow> getUserWorkflows(SearchResult<String> searchResultIds) {
+        return executionDAOFacade.getUserWorkflows(searchResultIds);
+    }
 }

core/src/main/java/com/netflix/conductor/service/MetadataService.java

@@ -23,6 +23,7 @@
 
 import org.springframework.validation.annotation.Validated;
 
+import com.netflix.conductor.common.metadata.BaseDef;
 import com.netflix.conductor.common.metadata.events.EventHandler;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
@@ -156,4 +157,12 @@ List<EventHandler> getEventHandlersForEvent(
             boolean activeOnly);
 
     List<WorkflowDef> getWorkflowDefsLatestVersions();
+
+    boolean hasAccess(Object[] args, List<String> labels, String uri);
+
+    boolean exists(Object[] args, String uri);
+
+    List<? extends BaseDef> getUserTaskDefs(List<String> roles);
+
+    List<? extends BaseDef> getUserWorkflowDefs(List<String> roles);
 }

core/src/main/java/com/netflix/conductor/service/MetadataServiceImpl.java

@@ -23,6 +23,7 @@
 import org.springframework.stereotype.Service;
 
 import com.netflix.conductor.common.constraints.OwnerEmailMandatoryConstraint;
+import com.netflix.conductor.common.metadata.BaseDef;
 import com.netflix.conductor.common.metadata.events.EventHandler;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
@@ -223,6 +224,26 @@ public List<WorkflowDef> getWorkflowDefsLatestVersions() {
         return metadataDAO.getAllWorkflowDefsLatestVersions();
     }
 
+    @Override
+    public boolean hasAccess(Object[] args, List<String> labels, String uri) {
+        return metadataDAO.hasAccess(args, labels, uri);
+    }
+
+    @Override
+    public boolean exists(Object[] args, String uri) {
+        return metadataDAO.exists(args, uri);
+    }
+
+    @Override
+    public List<? extends BaseDef> getUserTaskDefs(List<String> roles) {
+        return metadataDAO.getUserTaskDefs(roles);
+    }
+
+    @Override
+    public List<? extends BaseDef> getUserWorkflowDefs(List<String> roles) {
+        return metadataDAO.getUserWorkflowDefs(roles);
+    }
+
     public Map<String, ? extends Iterable<WorkflowDefSummary>> getWorkflowNamesAndVersions() {
         List<WorkflowDef> workflowDefs = metadataDAO.getAllWorkflowDefs();
 

core/src/main/java/com/netflix/conductor/service/WorkflowService.java

@@ -401,4 +401,18 @@ ExternalStorageLocation getExternalStorageLocation(
     List<String> getWorkflowPath(String workflowId);
 
     List<Workflow> getWorkflowFamily(String workflowId, boolean summaryOnly);
+
+    boolean hasAccess(Object[] args, List<String> labels);
+
+    boolean exists(Object[] args);
+
+    List<String> getUserWorkflowIds(List<String> labels);
+
+    List<String> getPresentIds(List<String> ids);
+
+    SearchResult<String> getSearchResultIds(List<String> roles);
+
+    SearchResult<WorkflowSummary> getSummaries(SearchResult<String> searchResultIds);
+
+    SearchResult<Workflow> getUserWorkflows(SearchResult<String> searchResultIds);
 }

core/src/main/java/com/netflix/conductor/service/WorkflowServiceImpl.java

@@ -195,6 +195,41 @@ public List<Workflow> getWorkflowFamily(String workflowId, boolean summaryOnly)
         return workflows;
     }
 
+    @Override
+    public boolean hasAccess(Object[] args, List<String> labels) {
+        return executionService.hasAccess(args, labels);
+    }
+
+    @Override
+    public boolean exists(Object[] args) {
+        return executionService.exists(args);
+    }
+
+    @Override
+    public List<String> getUserWorkflowIds(List<String> labels) {
+        return executionService.getUserWorkflowIds(labels);
+    }
+
+    @Override
+    public List<String> getPresentIds(List<String> ids) {
+        return executionService.getPresentIds(ids);
+    }
+
+    @Override
+    public SearchResult<String> getSearchResultIds(List<String> roles) {
+        return executionService.getSearchResultIds(roles);
+    }
+
+    @Override
+    public SearchResult<WorkflowSummary> getSummaries(SearchResult<String> searchResultIds) {
+        return executionService.getSummaries(searchResultIds);
+    }
+
+    @Override
+    public SearchResult<Workflow> getUserWorkflows(SearchResult<String> searchResultIds) {
+        return executionService.getUserWorkflows(searchResultIds);
+    }
+
     /**
      * Removes the workflow from the system.
      *

es6-persistence/src/main/java/com/netflix/conductor/es6/dao/index/ElasticSearchDAOV6.java

@@ -680,6 +680,12 @@ public long getWorkflowCount(String query, String freeText) {
         return count(query, freeText, WORKFLOW_DOC_TYPE);
     }
 
+    @Override
+    public SearchResult<WorkflowSummary> getSummaries(SearchResult<String> searchResultIds) {
+        throw new UnsupportedOperationException(
+                "getSummaries is not supported in ElasticSearchDAOV6");
+    }
+
     @Override
     public SearchResult<String> searchTasks(
             String query, String freeText, int start, int count, List<String> sort) {

es6-persistence/src/main/java/com/netflix/conductor/es6/dao/index/ElasticSearchRestDAOV6.java

@@ -1117,6 +1117,12 @@ public long getWorkflowCount(String query, String freeText) {
         }
     }
 
+    @Override
+    public SearchResult<WorkflowSummary> getSummaries(SearchResult<String> searchResultIds) {
+        throw new UnsupportedOperationException(
+                "getSummaries is not supported in ElasticSearchRestDAOV6");
+    }
+
     private long getObjectCounts(String structuredQuery, String freeTextQuery, String docType)
             throws ParserException, IOException {
         QueryBuilder queryBuilder = boolQueryBuilder(structuredQuery, freeTextQuery);

redis-persistence/src/main/java/com/netflix/conductor/redis/dao/RedisExecutionDAO.java

@@ -23,6 +23,7 @@
 
 import com.netflix.conductor.common.metadata.events.EventExecution;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
+import com.netflix.conductor.common.run.SearchResult;
 import com.netflix.conductor.core.config.ConductorProperties;
 import com.netflix.conductor.core.exception.TransientException;
 import com.netflix.conductor.dao.ConcurrentExecutionLimitDAO;
@@ -726,6 +727,31 @@ public void removeEventExecution(EventExecution eventExecution) {
         }
     }
 
+    @Override
+    public boolean hasAccess(Object[] args, List<String> labels) {
+        return false;
+    }
+
+    @Override
+    public boolean exists(Object[] args) {
+        return false;
+    }
+
+    @Override
+    public List<String> getUserWorkflowIds(List<String> labels) {
+        return List.of();
+    }
+
+    @Override
+    public List<String> getPresentIds(List<String> ids) {
+        return List.of();
+    }
+
+    @Override
+    public SearchResult<String> getSearchResultIds(List<String> roles) {
+        return new SearchResult<>(0, Collections.emptyList());
+    }
+
     public List<EventExecution> getEventExecutions(
             String eventHandlerName, String eventName, String messageId, int max) {
         try {

redis-persistence/src/main/java/com/netflix/conductor/redis/dao/RedisMetadataDAO.java

@@ -29,6 +29,7 @@
 import org.springframework.context.annotation.Conditional;
 import org.springframework.stereotype.Component;
 
+import com.netflix.conductor.common.metadata.BaseDef;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
 import com.netflix.conductor.core.config.ConductorProperties;
@@ -320,6 +321,28 @@ public List<WorkflowDef> getAllWorkflowDefsLatestVersions() {
         return workflows;
     }
 
+    @Override
+    public boolean hasAccess(Object[] args, List<String> labels, String uri) {
+        throw new UnsupportedOperationException("hasAccess is not supported in RedisMetadataDAO");
+    }
+
+    @Override
+    public boolean exists(Object[] args, String uri) {
+        throw new UnsupportedOperationException("exists is not supported in RedisMetadataDAO");
+    }
+
+    @Override
+    public List<? extends BaseDef> getUserTaskDefs(List<String> roles) {
+        throw new UnsupportedOperationException(
+                "getUserTaskDefs is not supported in RedisMetadataDAO");
+    }
+
+    @Override
+    public List<? extends BaseDef> getUserWorkflowDefs(List<String> roles) {
+        throw new UnsupportedOperationException(
+                "getUserWorkflowDefs is not supported in RedisMetadataDAO");
+    }
+
     private void _createOrUpdate(WorkflowDef workflowDef) {
         // First set the workflow def
         jedisProxy.hset(