Persistence spouts emit stored URLs without checking the scheme

[rzo1]

What happens

AbstractQueryingSpout.nextTuple() takes the next entry out of the URL buffer and emits it unchanged. The per-backend spouts fill that buffer straight from the query results. No scheme check and no filtering happens between the store and the fetcher: shipped topologies wire the spout to URLPartitionerBolt and then to FetcherBolt, and URL filtering only runs on the discovery path, in StatusEmitterBolt.filterOutlink and URLFilterBolt. A row whose URL uses a scheme the operator never intended to crawl is fetched all the same, as long as a protocol implementation is registered for it.

Where

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

            List<Object> fields = buffer.next();
            String url = fields.get(0).toString();
            this.collector.emit(fields, url);

The same applies to the backend spouts that feed the buffer, for example SQLSpout.processRow, the OpenSearch AbstractSpout, SolrSpout.handleSuccess and the urlfrontier Spout.onNext.

Why it matters

The frontier is the crawl instruction set, so write access to it already means a lot of control, and this is not a way in by itself. It does mean there is no second line of defence at the point where URLs re-enter the topology: whatever ends up in the store is fetched, including schemes the operator's filter chain would reject on the discovery path. Combined with the file scheme being registered by default, a single bad row is a local file read rather than a wasted fetch. A scheme check in the spout is cheap and would keep the two paths consistent.

Reproduction

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

/*
 * 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;

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());
    }
}

Run it:

mvn -pl core test -Dtest=AbstractQueryingSpoutSchemeTest

It drives a minimal spout with a mocked context and the existing FileSpoutOutputCollectorMock. The control case with an https URL passes; the case with a file:// row asserts the wanted behaviour and fails on main.

AbstractQueryingSpoutSchemeTest.urlsWithAnUnexpectedSchemeAreNotEmitted -- Time elapsed: 0.010 s <<< FAILURE!
org.opentest4j.AssertionFailedError: the spout emitted a stored URL whose scheme is not in the configured list: [file:///etc/hosts, ] ==> expected: <null> but was: <[file:///etc/hosts, ]>

Tests run: 2, Failures: 1, Errors: 0, Skipped: 0

Suggested fix

In AbstractQueryingSpout.nextTuple(), check the scheme of the URL against the configured protocols list (or a dedicated key) before emitting, and count the rejects on the existing event counter so operators can see them. Do not run the full URL filter chain there: that is hot path cost on every tuple and would silently drop rows an operator deliberately put in the store. Log rejected rows at WARN with the URL, since a rejected row otherwise stays in the store and is retried on every query.

[abhinav-phi]

Hi! I'd like to take this issue.
Plan: I'll make persistence spouts validate the scheme of stored URLs before emitting (scheme allowlist / filter check), and add tests for stored URLs with unexpected schemes.
/assign take