Add configurations for allowed protocols for HTTP and HDFS inputSourc…

…es/firehoses (#10830)

* Allow only HTTP and HTTPS protocols for the HTTP inputSource

* rename

* Update core/src/main/java/org/apache/druid/data/input/impl/HttpInputSource.java

Co-authored-by: Abhishek Agarwal <1477457+abhishekagarwal87@users.noreply.github.com>

* fix http firehose and update doc

* HDFS inputSource

* add configs for allowed protocols

* fix checkstyle and doc

* more checkstyle

* remove stale doc

* remove more doc

* Apply doc suggestions from code review

Co-authored-by: Charles Smith <38529548+techdocsmith@users.noreply.github.com>

* update hdfs address in docs

* fix test

Co-authored-by: Abhishek Agarwal <1477457+abhishekagarwal87@users.noreply.github.com>
Co-authored-by: Charles Smith <38529548+techdocsmith@users.noreply.github.com>

core/src/main/java/org/apache/druid/data/input/impl/HttpInputSource.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.data.input.impl;
 
+import com.fasterxml.jackson.annotation.JacksonInject;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.Preconditions;
@@ -28,6 +29,8 @@
 import org.apache.druid.data.input.InputSourceReader;
 import org.apache.druid.data.input.InputSplit;
 import org.apache.druid.data.input.SplitHintSpec;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.metadata.PasswordProvider;
 
 import javax.annotation.Nullable;
@@ -45,18 +48,31 @@ public class HttpInputSource extends AbstractInputSource implements SplittableIn
   private final String httpAuthenticationUsername;
   @Nullable
   private final PasswordProvider httpAuthenticationPasswordProvider;
+  private final HttpInputSourceConfig config;
 
   @JsonCreator
   public HttpInputSource(
       @JsonProperty("uris") List<URI> uris,
       @JsonProperty("httpAuthenticationUsername") @Nullable String httpAuthenticationUsername,
-      @JsonProperty("httpAuthenticationPassword") @Nullable PasswordProvider httpAuthenticationPasswordProvider
+      @JsonProperty("httpAuthenticationPassword") @Nullable PasswordProvider httpAuthenticationPasswordProvider,
+      @JacksonInject HttpInputSourceConfig config
   )
   {
     Preconditions.checkArgument(uris != null && !uris.isEmpty(), "Empty URIs");
+    throwIfInvalidProtocols(config, uris);
     this.uris = uris;
     this.httpAuthenticationUsername = httpAuthenticationUsername;
     this.httpAuthenticationPasswordProvider = httpAuthenticationPasswordProvider;
+    this.config = config;
+  }
+
+  public static void throwIfInvalidProtocols(HttpInputSourceConfig config, List<URI> uris)
+  {
+    for (URI uri : uris) {
+      if (!config.getAllowedProtocols().contains(StringUtils.toLowerCase(uri.getScheme()))) {
+        throw new IAE("Only %s protocols are allowed", config.getAllowedProtocols());
+      }
+    }
   }
 
   @JsonProperty
@@ -97,7 +113,8 @@ public SplittableInputSource<URI> withSplit(InputSplit<URI> split)
     return new HttpInputSource(
         Collections.singletonList(split.get()),
         httpAuthenticationUsername,
-        httpAuthenticationPasswordProvider
+        httpAuthenticationPasswordProvider,
+        config
     );
   }
 
@@ -129,16 +146,17 @@ public boolean equals(Object o)
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-    HttpInputSource source = (HttpInputSource) o;
-    return Objects.equals(uris, source.uris) &&
-           Objects.equals(httpAuthenticationUsername, source.httpAuthenticationUsername) &&
-           Objects.equals(httpAuthenticationPasswordProvider, source.httpAuthenticationPasswordProvider);
+    HttpInputSource that = (HttpInputSource) o;
+    return Objects.equals(uris, that.uris) &&
+           Objects.equals(httpAuthenticationUsername, that.httpAuthenticationUsername) &&
+           Objects.equals(httpAuthenticationPasswordProvider, that.httpAuthenticationPasswordProvider) &&
+           Objects.equals(config, that.config);
   }
 
   @Override
   public int hashCode()
   {
-    return Objects.hash(uris, httpAuthenticationUsername, httpAuthenticationPasswordProvider);
+    return Objects.hash(uris, httpAuthenticationUsername, httpAuthenticationPasswordProvider, config);
   }
 
   @Override

core/src/main/java/org/apache/druid/data/input/impl/HttpInputSourceConfig.java

@@ -0,0 +1,83 @@
+/*
+ * 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.druid.data.input.impl;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableSet;
+import org.apache.druid.java.util.common.StringUtils;
+
+import javax.annotation.Nullable;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class HttpInputSourceConfig
+{
+  @VisibleForTesting
+  public static final Set<String> DEFAULT_ALLOWED_PROTOCOLS = ImmutableSet.of("http", "https");
+
+  @JsonProperty
+  private final Set<String> allowedProtocols;
+
+  @JsonCreator
+  public HttpInputSourceConfig(
+      @JsonProperty("allowedProtocols") @Nullable Set<String> allowedProtocols
+  )
+  {
+    this.allowedProtocols = allowedProtocols == null || allowedProtocols.isEmpty()
+                            ? DEFAULT_ALLOWED_PROTOCOLS
+                            : allowedProtocols.stream().map(StringUtils::toLowerCase).collect(Collectors.toSet());
+  }
+
+  public Set<String> getAllowedProtocols()
+  {
+    return allowedProtocols;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    HttpInputSourceConfig that = (HttpInputSourceConfig) o;
+    return Objects.equals(allowedProtocols, that.allowedProtocols);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(allowedProtocols);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "HttpInputSourceConfig{" +
+           ", allowedProtocols=" + allowedProtocols +
+           '}';
+  }
+}
+

core/src/test/java/org/apache/druid/data/input/impl/HttpInputSourceConfigTest.java

@@ -0,0 +1,56 @@
+/*
+ * 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.druid.data.input.impl;
+
+import com.google.common.collect.ImmutableSet;
+import nl.jqno.equalsverifier.EqualsVerifier;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class HttpInputSourceConfigTest
+{
+
+  @Test
+  public void testEquals()
+  {
+    EqualsVerifier.forClass(HttpInputSourceConfig.class).usingGetClass().verify();
+  }
+
+  @Test
+  public void testNullAllowedProtocolsUseDefault()
+  {
+    HttpInputSourceConfig config = new HttpInputSourceConfig(null);
+    Assert.assertEquals(HttpInputSourceConfig.DEFAULT_ALLOWED_PROTOCOLS, config.getAllowedProtocols());
+  }
+
+  @Test
+  public void testEmptyAllowedProtocolsUseDefault()
+  {
+    HttpInputSourceConfig config = new HttpInputSourceConfig(ImmutableSet.of());
+    Assert.assertEquals(HttpInputSourceConfig.DEFAULT_ALLOWED_PROTOCOLS, config.getAllowedProtocols());
+  }
+
+  @Test
+  public void testCustomAllowedProtocols()
+  {
+    HttpInputSourceConfig config = new HttpInputSourceConfig(ImmutableSet.of("druid"));
+    Assert.assertEquals(ImmutableSet.of("druid"), config.getAllowedProtocols());
+  }
+}

core/src/test/java/org/apache/druid/data/input/impl/HttpInputSourceTest.java

@@ -19,29 +19,87 @@
 
 package org.apache.druid.data.input.impl;
 
+import com.fasterxml.jackson.databind.InjectableValues.Std;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import org.apache.druid.data.input.InputSource;
 import org.apache.druid.metadata.DefaultPasswordProvider;
 import org.junit.Assert;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import java.io.IOException;
 import java.net.URI;
 
 public class HttpInputSourceTest
 {
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
   @Test
   public void testSerde() throws IOException
   {
+    HttpInputSourceConfig httpInputSourceConfig = new HttpInputSourceConfig(null);
     final ObjectMapper mapper = new ObjectMapper();
+    mapper.setInjectableValues(new Std().addValue(HttpInputSourceConfig.class, httpInputSourceConfig));
     final HttpInputSource source = new HttpInputSource(
         ImmutableList.of(URI.create("http://test.com/http-test")),
         "myName",
-        new DefaultPasswordProvider("myPassword")
+        new DefaultPasswordProvider("myPassword"),
+        httpInputSourceConfig
     );
     final byte[] json = mapper.writeValueAsBytes(source);
     final HttpInputSource fromJson = (HttpInputSource) mapper.readValue(json, InputSource.class);
     Assert.assertEquals(source, fromJson);
   }
+
+  @Test
+  public void testConstructorAllowsOnlyDefaultProtocols()
+  {
+    new HttpInputSource(
+        ImmutableList.of(URI.create("http:///")),
+        "myName",
+        new DefaultPasswordProvider("myPassword"),
+        new HttpInputSourceConfig(null)
+    );
+
+    new HttpInputSource(
+        ImmutableList.of(URI.create("https:///")),
+        "myName",
+        new DefaultPasswordProvider("myPassword"),
+        new HttpInputSourceConfig(null)
+    );
+
+    expectedException.expect(IllegalArgumentException.class);
+    expectedException.expectMessage("Only [http, https] protocols are allowed");
+    new HttpInputSource(
+        ImmutableList.of(URI.create("my-protocol:///")),
+        "myName",
+        new DefaultPasswordProvider("myPassword"),
+        new HttpInputSourceConfig(null)
+    );
+  }
+
+  @Test
+  public void testConstructorAllowsOnlyCustomProtocols()
+  {
+    final HttpInputSourceConfig customConfig = new HttpInputSourceConfig(ImmutableSet.of("druid"));
+    new HttpInputSource(
+        ImmutableList.of(URI.create("druid:///")),
+        "myName",
+        new DefaultPasswordProvider("myPassword"),
+        customConfig
+    );
+
+    expectedException.expect(IllegalArgumentException.class);
+    expectedException.expectMessage("Only [druid] protocols are allowed");
+    new HttpInputSource(
+        ImmutableList.of(URI.create("https:///")),
+        "myName",
+        new DefaultPasswordProvider("myPassword"),
+        customConfig
+    );
+  }
 }

docs/configuration/index.md

@@ -515,6 +515,28 @@ This deep storage is used to interface with Cassandra.  Note that the `druid-cas
 |`druid.storage.keyspace`|Cassandra key space.|none|
 
 
+### Ingestion Security Configuration
+
+#### HDFS input source
+
+You can set the following property to specify permissible protocols for
+the [HDFS input source](../ingestion/native-batch.md#hdfs-input-source) and the [HDFS firehose](../ingestion/native-batch.md#hdfsfirehose).
+
+|Property|Possible Values|Description|Default|
+|--------|---------------|-----------|-------|
+|`druid.ingestion.hdfs.allowedProtocols`|List of protocols|Allowed protocols for the HDFS input source and HDFS firehose.|["hdfs"]|
+
+
+#### HTTP input source
+
+You can set the following property to specify permissible protocols for
+the [HTTP input source](../ingestion/native-batch.md#http-input-source) and the [HTTP firehose](../ingestion/native-batch.md#httpfirehose).
+
+|Property|Possible Values|Description|Default|
+|--------|---------------|-----------|-------|
+|`druid.ingestion.http.allowedProtocols`|List of protocols|Allowed protocols for the HTTP input source and HTTP firehose.|["http", "https"]|
+
+
 ### Task Logging
 
 If you are running the indexing service in remote mode, the task logs must be stored in S3, Azure Blob Store, Google Cloud Storage or HDFS.
@@ -1355,6 +1377,7 @@ The amount of direct memory needed by Druid is at least
 ensure at least this amount of direct memory is available by providing `-XX:MaxDirectMemorySize=<VALUE>` at the command
 line.
 
+
 #### Query Configurations
 
 See [general query configuration](#general-query-configuration).