[GOBBLIN-1304] Adds group ownership service

Closes apache#3142 from Will-Lo/add-group-ownership-
flows

Author: Will-Lo

gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java

@@ -140,6 +140,7 @@ public class ConfigurationKeys {
   public static final String FLOW_ALLOW_CONCURRENT_EXECUTION = "flow.allowConcurrentExecution";
   public static final String FLOW_EXPLAIN_KEY = "flow.explain";
   public static final String FLOW_UNSCHEDULE_KEY = "flow.unschedule";
+  public static final String FLOW_OWNING_GROUP_KEY = "flow.owningGroup";
 
   /**
    * Common topology configuration properties.

gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java

@@ -129,4 +129,7 @@ public class ServiceConfigKeys {
 
   public static final String FORCE_LEADER = GOBBLIN_SERVICE_PREFIX + "forceLeader";
   public static final boolean DEFAULT_FORCE_LEADER = false;
+  // Group Membership authentication service
+  public static final String GROUP_OWNERSHIP_SERVICE_CLASS = GOBBLIN_SERVICE_PREFIX + "groupOwnershipService.class";
+  public static final String DEFAULT_GROUP_OWNERSHIP_SERVICE = "org.apache.gobblin.service.NoopGroupOwnershipService";
 }

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/FlowConfig.pdl

@@ -26,6 +26,11 @@ record FlowConfig {
    */
   explain: boolean = false
 
+  /**
+   * Optional string name of group that the requester belongs to for group ownership of flows.
+   */
+  owningGroup: optional string
+
   /**
    * Properties for the flow. These properties are passed to the compiled Gobblin jobs.
    */

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowconfigs.snapshot.json

@@ -76,6 +76,11 @@
       "type" : "boolean",
       "doc" : "Return the compiled flow as a string. If enabled, the flow is not added.",
       "default" : false
+    }, {
+      "name" : "owningGroup",
+      "type" : "string",
+      "doc" : "Optional string name of group that the requester belongs to for group ownership of flows.",
+      "optional" : true
     }, {
       "name" : "properties",
       "type" : {

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowconfigsV2.snapshot.json

@@ -67,6 +67,11 @@
       "type" : "boolean",
       "doc" : "Return the compiled flow as a string. If enabled, the flow is not added.",
       "default" : false
+    }, {
+      "name" : "owningGroup",
+      "type" : "string",
+      "doc" : "Optional string name of group that the requester belongs to for group ownership of flows.",
+      "optional" : true
     }, {
       "name" : "properties",
       "type" : {

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-client/src/test/java/org/apache/gobblin/service/FlowConfigV2Test.java

@@ -18,11 +18,13 @@
 package org.apache.gobblin.service;
 
 import java.io.File;
+import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
+import org.mortbay.jetty.HttpStatus;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -62,12 +64,18 @@ public class FlowConfigV2Test {
   private EmbeddedRestliServer _server;
   private File _testDirectory;
   private TestRequesterService _requesterService;
+  private GroupOwnershipService groupOwnershipService;
+  private File groupConfigFile;
 
   private static final String TEST_SPEC_STORE_DIR = "/tmp/flowConfigV2Test/";
   private static final String TEST_GROUP_NAME = "testGroup1";
   private static final String TEST_FLOW_NAME = "testFlow1";
   private static final String TEST_FLOW_NAME_2 = "testFlow2";
   private static final String TEST_FLOW_NAME_3 = "testFlow3";
+  private static final String TEST_FLOW_NAME_4 = "testFlow4";
+  private static final String TEST_FLOW_NAME_5 = "testFlow5";
+  private static final String TEST_FLOW_NAME_6 = "testFlow6";
+  private static final String TEST_FLOW_NAME_7 = "testFlow7";
   private static final String TEST_SCHEDULE = "0 1/0 * ? * *";
   private static final String TEST_TEMPLATE_URI = "FS:///templates/test.template";
 
@@ -90,6 +98,15 @@ public void setUp() throws Exception {
 
     _requesterService = new TestRequesterService(ConfigFactory.empty());
 
+    this.groupConfigFile = new File(_testDirectory + "/TestGroups.json");
+    String groups ="{\"testGroup\": \"testName,testName2\"}";
+    Files.write(groups.getBytes(), this.groupConfigFile);
+    Config groupServiceConfig = ConfigBuilder.create()
+        .addPrimitive(LocalGroupOwnershipService.GROUP_MEMBER_LIST, this.groupConfigFile.getAbsolutePath())
+        .build();
+
+    groupOwnershipService = new LocalGroupOwnershipService(groupServiceConfig);
+
     Injector injector = Guice.createInjector(new Module() {
       @Override
       public void configure(Binder binder) {
@@ -98,6 +115,7 @@ public void configure(Binder binder) {
         // been made
         binder.bindConstant().annotatedWith(Names.named(FlowConfigsV2Resource.INJECT_READY_TO_USE)).to(Boolean.TRUE);
         binder.bind(RequesterService.class).annotatedWith(Names.named(FlowConfigsV2Resource.INJECT_REQUESTER_SERVICE)).toInstance(_requesterService);
+        binder.bind(GroupOwnershipService.class).annotatedWith(Names.named(FlowConfigsV2Resource.INJECT_GROUP_OWNERSHIP_SERVICE)).toInstance(groupOwnershipService);
       }
     });
 
@@ -185,20 +203,102 @@ public void testBadPartialUpdate() throws Exception {
     _client.partialUpdateFlowConfig(flowId, flowConfigPatch);
   }
 
-  @Test (expectedExceptions = RestLiResponseException.class)
+  @Test
   public void testDisallowedRequester() throws Exception {
-    ServiceRequester testRequester = new ServiceRequester("testName", "testType", "testFrom");
-    _requesterService.setRequester(testRequester);
+    try {
+      ServiceRequester testRequester = new ServiceRequester("testName", "testType", "testFrom");
+      _requesterService.setRequester(testRequester);
+
+      Map<String, String> flowProperties = Maps.newHashMap();
+      flowProperties.put("param1", "value1");
+
+      FlowConfig flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_4))
+          .setTemplateUris(TEST_TEMPLATE_URI)
+          .setProperties(new StringMap(flowProperties));
+      _client.createFlowConfig(flowConfig);
+
+      testRequester.setName("testName2");
+      _client.deleteFlowConfig(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_4));
+    } catch (RestLiResponseException e) {
+      Assert.assertEquals(e.getStatus(), HttpStatus.ORDINAL_401_Unauthorized);
+    }
+  }
 
+  @Test
+  public void testGroupRequesterAllowed() throws Exception {
+    ServiceRequester testRequester = new ServiceRequester("testName", "USER_PRINCIPAL", "testFrom");
+    _requesterService.setRequester(testRequester);
     Map<String, String> flowProperties = Maps.newHashMap();
-    flowProperties.put("param1", "value1");
 
-    FlowConfig flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME))
-        .setTemplateUris(TEST_TEMPLATE_URI).setProperties(new StringMap(flowProperties));
-    _client.createFlowConfig(flowConfig);
+    FlowConfig flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_5))
+        .setTemplateUris(TEST_TEMPLATE_URI)
+        .setProperties(new StringMap(flowProperties))
+        .setOwningGroup("testGroup");
+
+     _client.createFlowConfig(flowConfig);
 
     testRequester.setName("testName2");
-    _client.deleteFlowConfig(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME));
+    _client.deleteFlowConfig(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_5));
+  }
+
+  @Test
+  public void testGroupRequesterRejected() throws Exception {
+    try {
+      ServiceRequester testRequester = new ServiceRequester("testName", "USER_PRINCIPAL", "testFrom");
+      _requesterService.setRequester(testRequester);
+      Map<String, String> flowProperties = Maps.newHashMap();
+
+      FlowConfig flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_6))
+          .setTemplateUris(TEST_TEMPLATE_URI)
+          .setProperties(new StringMap(flowProperties))
+          .setOwningGroup("testGroup");
+
+      _client.createFlowConfig(flowConfig);
+
+      testRequester.setName("testName3");
+      _client.deleteFlowConfig(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_6));
+    } catch (RestLiResponseException e) {
+      Assert.assertEquals(e.getStatus(), HttpStatus.ORDINAL_401_Unauthorized);
+    }
+  }
+
+  @Test
+  public void testLocalGroupOwnershipUpdates() throws Exception {
+    try {
+      ServiceRequester testRequester = new ServiceRequester("testName", "USER_PRINCIPAL", "testFrom");
+      _requesterService.setRequester(testRequester);
+      Map<String, String> flowProperties = Maps.newHashMap();
+
+      FlowConfig flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_7))
+          .setTemplateUris(TEST_TEMPLATE_URI)
+          .setProperties(new StringMap(flowProperties))
+          .setOwningGroup("testGroup2");
+
+      _client.createFlowConfig(flowConfig);
+
+    } catch (RestLiResponseException e) {
+      Assert.assertEquals(e.getStatus(), HttpStatus.ORDINAL_401_Unauthorized);
+    }
+
+    String filePath = this.groupConfigFile.getAbsolutePath();
+    this.groupConfigFile.delete();
+    this.groupConfigFile = new File(filePath);
+    String groups ="{\"testGroup2\": \"testName,testName3\"}";
+    Files.write(groups.getBytes(), this.groupConfigFile);
+
+    ServiceRequester testRequester = new ServiceRequester("testName", "USER_PRINCIPAL", "testFrom");
+    _requesterService.setRequester(testRequester);
+    Map<String, String> flowProperties = Maps.newHashMap();
+
+    FlowConfig flowConfig = new FlowConfig().setId(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_7))
+        .setTemplateUris(TEST_TEMPLATE_URI)
+        .setProperties(new StringMap(flowProperties))
+        .setOwningGroup("testGroup2");
+
+    // this should no longer fail as the localGroupOwnership service should have updated as the file changed
+    _client.createFlowConfig(flowConfig);
+    testRequester.setName("testName3");
+    _client.deleteFlowConfig(new FlowId().setFlowGroup(TEST_GROUP_NAME).setFlowName(TEST_FLOW_NAME_7));
   }
 
   @AfterClass(alwaysRun = true)
@@ -226,5 +326,16 @@ public TestRequesterService(Config config) {
     public List<ServiceRequester> findRequesters(BaseResource resource) {
       return requester == null ? Lists.newArrayList() : Lists.newArrayList(requester);
     }
+
+    @Override
+    public boolean isRequesterAllowed(
+        List<ServiceRequester> originalRequesterList, List<ServiceRequester> currentRequesterList) {
+      for (ServiceRequester s: currentRequesterList) {
+        if (originalRequesterList.contains(s)) {
+          return true;
+        }
+      }
+      return false;
+    }
   }
 }

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-client/src/test/resources/TestGroups.json

@@ -0,0 +1,3 @@
+{
+  "testGroup": "testName,testName2"
+}

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowConfigResourceLocalHandler.java

@@ -248,6 +248,10 @@ public static FlowSpec createFlowSpecForConfig(FlowConfig flowConfig) {
       configBuilder.addPrimitive(ConfigurationKeys.FLOW_EXPLAIN_KEY, flowConfig.isExplain());
     }
 
+    if (flowConfig.hasOwningGroup()) {
+      configBuilder.addPrimitive(ConfigurationKeys.FLOW_OWNING_GROUP_KEY, flowConfig.getOwningGroup());
+    }
+
     Config config = configBuilder.build();
 
     Config configWithFallback;

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowConfigsV2Resource.java

@@ -56,7 +56,7 @@ public class FlowConfigsV2Resource extends ComplexKeyResourceTemplate<FlowId, Fl
   public static final String FLOW_CONFIG_GENERATOR_INJECT_NAME = "flowConfigsV2ResourceHandler";
   public static final String INJECT_REQUESTER_SERVICE = "v2RequesterService";
   public static final String INJECT_READY_TO_USE = "v2ReadyToUse";
-
+  public static final String INJECT_GROUP_OWNERSHIP_SERVICE = "v2GroupOwnershipService";
   private static final Set<String> ALLOWED_METADATA = ImmutableSet.of("delete.state.store");
 
 
@@ -77,6 +77,10 @@ public class FlowConfigsV2Resource extends ComplexKeyResourceTemplate<FlowId, Fl
   @Named(INJECT_READY_TO_USE)
   private Boolean readyToUse;
 
+  @Inject
+  @Named(INJECT_GROUP_OWNERSHIP_SERVICE)
+  private GroupOwnershipService groupOwnershipService;
+
   public FlowConfigsV2Resource() {
   }
 
@@ -128,11 +132,14 @@ public List<FlowConfig> getFilteredFlows(@Context PagingContext context,
   @ReturnEntity
   @Override
   public CreateKVResponse create(FlowConfig flowConfig) {
-    List<ServiceRequester> requestorList = this.requesterService.findRequesters(this);
+    List<ServiceRequester> requesterList = this.requesterService.findRequesters(this);
     try {
-      String serialized = RequesterService.serialize(requestorList);
+      String serialized = RequesterService.serialize(requesterList);
       flowConfig.getProperties().put(RequesterService.REQUESTER_LIST, serialized);
       LOG.info("Rest requester list is " + serialized);
+      if (flowConfig.hasOwningGroup() && !this.groupOwnershipService.isMemberOfGroup(requesterList, flowConfig.getOwningGroup())) {
+        throw new FlowConfigLoggedException(HttpStatus.S_401_UNAUTHORIZED, "Requester not part of owning group specified");
+      }
     } catch (IOException e) {
       throw new FlowConfigLoggedException(HttpStatus.S_401_UNAUTHORIZED, "cannot get who is the requester", e);
     }
@@ -148,7 +155,7 @@ public CreateKVResponse create(FlowConfig flowConfig) {
    */
   @Override
   public UpdateResponse update(ComplexResourceKey<FlowId, FlowStatusId> key, FlowConfig flowConfig) {
-    FlowConfigsResource.checkRequester(this.requesterService, get(key), this.requesterService.findRequesters(this));
+    checkRequester(this.requesterService, get(key), this.requesterService.findRequesters(this));
     String flowGroup = key.getKey().getFlowGroup();
     String flowName = key.getKey().getFlowName();
     FlowId flowId = new FlowId().setFlowGroup(flowGroup).setFlowName(flowName);
@@ -163,7 +170,7 @@ public UpdateResponse update(ComplexResourceKey<FlowId, FlowStatusId> key, FlowC
    */
   @Override
   public UpdateResponse update(ComplexResourceKey<FlowId, FlowStatusId> key, PatchRequest<FlowConfig> flowConfigPatch) {
-    FlowConfigsResource.checkRequester(this.requesterService, get(key), this.requesterService.findRequesters(this));
+    checkRequester(this.requesterService, get(key), this.requesterService.findRequesters(this));
     String flowGroup = key.getKey().getFlowGroup();
     String flowName = key.getKey().getFlowName();
     FlowId flowId = new FlowId().setFlowGroup(flowGroup).setFlowName(flowName);
@@ -177,7 +184,7 @@ public UpdateResponse update(ComplexResourceKey<FlowId, FlowStatusId> key, Patch
    */
   @Override
   public UpdateResponse delete(ComplexResourceKey<FlowId, FlowStatusId> key) {
-    FlowConfigsResource.checkRequester(this.requesterService, get(key), this.requesterService.findRequesters(this));
+    checkRequester(this.requesterService, get(key), this.requesterService.findRequesters(this));
     String flowGroup = key.getKey().getFlowGroup();
     String flowName = key.getKey().getFlowName();
     FlowId flowId = new FlowId().setFlowGroup(flowGroup).setFlowName(flowName);
@@ -200,4 +207,37 @@ private Properties getHeaders() {
     }
     return headerProperties;
   }
+
+  /**
+   * Check that all {@link ServiceRequester}s in this request are contained within the original service requester list
+   * or is part of the original requester's owning group when the flow was submitted. If they are not, throw a {@link FlowConfigLoggedException} with {@link HttpStatus#S_401_UNAUTHORIZED}.
+   * If there is a failure when deserializing the original requester list, throw a {@link FlowConfigLoggedException} with
+   * {@link HttpStatus#S_400_BAD_REQUEST}.
+   * @param requesterService the {@link RequesterService} used to verify the requester
+   * @param originalFlowConfig original flow config to find original requester
+   * @param requesterList list of requesters for this request
+   */
+  public void checkRequester(
+      RequesterService requesterService, FlowConfig originalFlowConfig, List<ServiceRequester> requesterList) {
+    if (requesterList == null) {
+      return;
+    }
+
+    try {
+      String serializedOriginalRequesterList = originalFlowConfig.getProperties().get(RequesterService.REQUESTER_LIST);
+      if (serializedOriginalRequesterList != null) {
+        List<ServiceRequester> originalRequesterList = RequesterService.deserialize(serializedOriginalRequesterList);
+        if (!requesterService.isRequesterAllowed(originalRequesterList, requesterList)) {
+          // if the requester is not whitelisted or the original requester, reject the requester if it is not part of the owning group
+          // of the original requester
+          if (!(originalFlowConfig.hasOwningGroup() && this.groupOwnershipService.isMemberOfGroup(
+              requesterList, originalFlowConfig.getOwningGroup()))) {
+            throw new FlowConfigLoggedException(HttpStatus.S_401_UNAUTHORIZED, "Requester not allowed to make this request");
+          }
+        }
+      }
+    } catch (IOException e) {
+      throw new FlowConfigLoggedException(HttpStatus.S_400_BAD_REQUEST, "Failed to get original requester list", e);
+    }
+  }
 }

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/GroupOwnershipService.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.service;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Service for handling group ownership of flows
+ */
+public abstract class GroupOwnershipService {
+
+   /**
+    * @return true if any of the serviceRequesters belong in the group
+    */
+   public abstract boolean isMemberOfGroup(List<ServiceRequester> serviceRequesters, String group);
+
+   /**
+    * Extracts ServiceRequester names
+    * @param requesterList
+    * @return a list of service requester names
+    */
+   protected static List<String> extractRequesterNames(List<ServiceRequester> requesterList) {
+      return requesterList.stream()
+          .map(requester -> requester.getName())
+          .collect(Collectors.toList());
+   }
+}