From 44bb12e320c4e1e095af620578dd8ef685f35261 Mon Sep 17 00:00:00 2001 From: abhinav-phi Date: Sun, 6 Sep 2026 04:16:41 +0530 Subject: [PATCH] Spouts check the scheme of stored URLs before emitting (#2085) The frontier is the crawl instruction set: whatever ends up in the store is fetched, with no scheme check and no filtering, since URL filtering only runs on the discovery path. A row whose URL uses a scheme the operator never intended to crawl was emitted as long as a protocol implementation was registered for it. AbstractQueryingSpout.nextTuple now checks the scheme of each buffered URL against the configured protocols list before emitting, skips rows with other schemes at WARN with a skipped.scheme counter, and keeps draining the buffer so one bad row does not block the ones behind it. --- .../persistence/AbstractQueryingSpout.java | 47 +++++++- .../AbstractQueryingSpoutSchemeTest.java | 105 ++++++++++++++++++ 2 files changed, 148 insertions(+), 4 deletions(-) create mode 100644 core/src/test/java/org/apache/stormcrawler/persistence/AbstractQueryingSpoutSchemeTest.java diff --git a/core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java b/core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java index a1a164a29..08e09b2d9 100644 --- a/core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java +++ b/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 @@ public abstract class AbstractQueryingSpout extends BaseRichSpout { protected ScopedCounter eventCounter; + /** Schemes which may be emitted from the store, from the {@code protocols} config key. */ + protected Set allowedSchemes; + protected URLBuffer buffer; protected SpoutOutputCollector collector; @@ -116,6 +122,21 @@ public void open( 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 @@ public void nextTuple() { 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 @@ public void nextTuple() { } List 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 @@ public void nextTuple() { 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; diff --git a/core/src/test/java/org/apache/stormcrawler/persistence/AbstractQueryingSpoutSchemeTest.java b/core/src/test/java/org/apache/stormcrawler/persistence/AbstractQueryingSpoutSchemeTest.java new file mode 100644 index 000000000..30f93f92f --- /dev/null +++ b/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 conf() { + Map 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)); + } +}