Spouts check the scheme of stored URLs before emitting

core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java

@@ -22,11 +22,14 @@
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 import org.apache.storm.spout.SpoutOutputCollector;
 import org.apache.storm.task.TopologyContext;
 import org.apache.storm.topology.OutputFieldsDeclarer;
@@ -88,6 +91,9 @@
 
     protected ScopedCounter eventCounter;
 
+    /** Schemes which may be emitted from the store, from the {@code protocols} config key. */
+    protected Set<String> allowedSchemes;
+
     protected URLBuffer buffer;
 
     protected SpoutOutputCollector collector;
@@ -116,6 +122,21 @@
 
         buffer = URLBuffer.createInstance(stormConf);
 
+        /*
+         * The store is the crawl instruction set: whatever ends up in it is
+         * fetched. Schemes which are not configured for the crawl must not
+         * re-enter the topology from there, so rows are checked before they
+         * are emitted - URL filtering only runs on the discovery path.
+         */
+        allowedSchemes =
+                ConfUtils.loadListFromConf("protocols", stormConf).stream()
+                        .map(String::trim)
+                        .map(String::toLowerCase)
+                        .collect(Collectors.toSet());
+        if (allowedSchemes.isEmpty()) {
+            allowedSchemes = Set.of("http", "https");
+        }
+
         CrawlerMetrics.registerGauge(context, stormConf, "buffer_size", buffer::size, 10);
         CrawlerMetrics.registerGauge(context, stormConf, "numQueues", buffer::numQueues, 10);
 
@@ -191,7 +212,7 @@
             timeLastQuerySent = System.currentTimeMillis();
         }
 
-        if (buffer.hasNext()) {
+        while (buffer.hasNext()) {
             // track how long the buffer had been empty for
             if (timestampEmptyBuffer != -1) {
                 eventCounter
@@ -201,11 +222,21 @@
             }
             List<Object> fields = buffer.next();
             String url = fields.get(0).toString();
+            if (!schemeAllowed(url)) {
+                LOG.warn(
+                        "Stored URL {} not fetched: its scheme is not in the configured list",
+                        url);
+                eventCounter.scope("skipped.scheme").incrBy(1);
+                // try the next entry the buffer holds; a rejected row stays in
+                // the store and is skipped again on every query
+                continue;
+            }
             this.collector.emit(fields, url);
             beingProcessed.put(url, null);
             eventCounter.scope("emitted").incrBy(1);
             return;
-        } else if (timestampEmptyBuffer == -1) {
+        }
+        if (timestampEmptyBuffer == -1 && !buffer.hasNext()) {
             timestampEmptyBuffer = System.currentTimeMillis();
         }
 
@@ -223,11 +254,19 @@
         timeLastQuerySent = System.currentTimeMillis();
     }
 
+    /** Checks the scheme of a stored URL against the configured {@code protocols} list. */
+    protected boolean schemeAllowed(String url) {
+        int colon = url.indexOf(':');
+        if (colon <= 0) {
+            return false;
+        }
+        return allowedSchemes.contains(url.substring(0, colon).toLowerCase(Locale.ROOT));
+    }
+
     /**
      * Returns the amount of time to wait if the backend was queried too recently and needs
      * throttling or -1 if the backend can be queried straight away.
-     */
-    private long throttleQueries() {
+     */    private long throttleQueries() {
         if (timeLastQuerySent != 0) {
             // check that we allowed some time between queries
             long difference = System.currentTimeMillis() - timeLastQuerySent;

core/src/test/java/org/apache/stormcrawler/persistence/AbstractQueryingSpoutSchemeTest.java

@@ -0,0 +1,105 @@
+/*
+ * 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.stormcrawler.persistence;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.TestUtil;
+import org.apache.stormcrawler.spout.mocks.FileSpoutOutputCollectorMock;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * A row whose URL uses a scheme outside the configured {@code protocols} list must not be emitted:
+ * URL filtering only runs on the discovery path, so the spout is the last place where the schemes
+ * re-entering the topology from the store can be constrained.
+ */
+class AbstractQueryingSpoutSchemeTest {
+
+    /** Minimal spout which returns whatever a backend row would contain. */
+    private static class StoredRowSpout extends AbstractQueryingSpout {
+
+        private final String url;
+
+        StoredRowSpout(String url) {
+            this.url = url;
+        }
+
+        @Override
+        protected void populateBuffer() {
+            buffer.add(url, new Metadata());
+            markQueryReceivedNow();
+        }
+
+        @Override
+        public void declareOutputFields(OutputFieldsDeclarer declarer) {
+            declarer.declare(new Fields("url", "metadata"));
+        }
+    }
+
+    private static Map<String, Object> conf() {
+        Map<String, Object> conf = new HashMap<>();
+        conf.put(
+                "urlbuffer.class",
+                "org.apache.stormcrawler.persistence.urlbuffer.SimpleURLBuffer");
+        return conf;
+    }
+
+    @Test
+    void httpUrlsFromTheBackendAreEmitted() {
+        StoredRowSpout spout = new StoredRowSpout("https://example.com/page.html");
+        FileSpoutOutputCollectorMock collector = new FileSpoutOutputCollectorMock();
+        spout.open(conf(), TestUtil.getMockedTopologyContext(), collector);
+        spout.activate();
+        // first call fills the buffer, second one emits from it
+        spout.nextTuple();
+        spout.nextTuple();
+        Assertions.assertNotNull(collector.getTuple());
+        Assertions.assertEquals("https://example.com/page.html", collector.getTuple().get(0));
+    }
+
+    @Test
+    void urlsWithAnUnexpectedSchemeAreNotEmitted() {
+        StoredRowSpout spout = new StoredRowSpout("file:///etc/hosts");
+        FileSpoutOutputCollectorMock collector = new FileSpoutOutputCollectorMock();
+        spout.open(conf(), TestUtil.getMockedTopologyContext(), collector);
+        spout.activate();
+        // first call fills the buffer, second one emits from it
+        spout.nextTuple();
+        spout.nextTuple();
+        Assertions.assertNull(
+                collector.getTuple(),
+                "the spout emitted a stored URL whose scheme is not in the configured list: "
+                        + collector.getTuple());
+    }
+
+    @Test
+    void uppercaseSchemeIsAllowed() {
+        StoredRowSpout spout = new StoredRowSpout("HTTPS://example.com/page.html");
+        FileSpoutOutputCollectorMock collector = new FileSpoutOutputCollectorMock();
+        spout.open(conf(), TestUtil.getMockedTopologyContext(), collector);
+        spout.activate();
+        spout.nextTuple();
+        spout.nextTuple();
+        Assertions.assertNotNull(collector.getTuple());
+        Assertions.assertEquals("HTTPS://example.com/page.html", collector.getTuple().get(0));
+    }
+}