You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by xi...@apache.org on 2026/08/31 06:31:15 UTC
(tinkerpop) branch 3.7-dev updated: Add Remote Restrictions in Hadoop/Spark OLAP
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a commit to branch 3.7-dev
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/3.7-dev by this push:
new ae53653365 Add Remote Restrictions in Hadoop/Spark OLAP
ae53653365 is described below
commit ae53653365363cdcbdee8d6ffa20ca430dab3478
Author: Yang Xia <55...@users.noreply.github.com>
AuthorDate: Sun Aug 30 20:33:09 2026 -0700
Add Remote Restrictions in Hadoop/Spark OLAP
Remotely submitted traversal may name only the built-in `gryo` and `graphson` formats, and its
`io().with(key, value)` and `GraphComputer.configure(key, value)` options are no longer transferred automatically.
Naming another reader or writer class, selecting `ScriptInputFormat` or `ScriptOutputFormat`, or setting a key that
has not been approved now fails with an error rather than loading the class or applying the key.
To keep using a custom format, a script format, or additional `with()` or `configure()` keys from a remote traversal,
they must be listed in the graph configuration.
Assisted-by: Claude Code:claude-opus-4-8
---
.gitignore | 3 +
CHANGELOG.asciidoc | 2 +
.../implementations-hadoop-start.asciidoc | 31 ++
docs/src/upgrade/release-3.7.x.asciidoc | 46 +++
hadoop-gremlin/conf/hadoop-script.properties | 2 +
.../computer/AbstractHadoopGraphComputer.java | 31 ++
.../traversal/step/sideEffect/HadoopIoStep.java | 103 +++++--
.../structure/io/script/ScriptRecordReader.java | 17 +-
.../structure/io/script/ScriptRecordWriter.java | 13 +
.../structure/io/util/ConfigValueParser.java | 61 ++++
.../structure/io/util/OlapClassLoadingPolicy.java | 190 +++++++++++++
.../structure/io/util/OlapConfigKeyPolicy.java | 114 ++++++++
.../step/sideEffect/HadoopIoStepTest.java | 234 +++++++++++++++
.../io/script/ScriptRecordReaderTest.java | 83 ++++++
.../io/script/ScriptRecordReaderWriterTest.java | 3 +
.../io/script/ScriptRecordWriterTest.java | 86 ++++++
.../structure/io/util/ConfigValueParserTest.java | 62 ++++
.../io/util/OlapClassLoadingPolicyTest.java | 202 +++++++++++++
.../structure/io/util/OlapConfigKeyPolicyTest.java | 137 +++++++++
.../spark/process/computer/SparkGraphComputer.java | 91 +++++-
.../SparkGraphComputerConfigSecurityTest.java | 281 ++++++++++++++++++
.../SparkHadoopIoSecurityIntegrateTest.java | 145 ++++++++++
.../SparkLocalClusterIoSecurityIntegrateTest.java | 315 +++++++++++++++++++++
.../computer/SystemPropertyProbeVertexProgram.java | 122 ++++++++
24 files changed, 2334 insertions(+), 40 deletions(-)
diff --git a/.gitignore b/.gitignore
index aeef7709f9..079782e2f3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,5 +1,8 @@
*.log
target/
+# Spark local-cluster OLAP ITs leave an untracked "file:"-scheme output dir under the module
+file:/
+**/file:/
*.iml
**/.idea/
*.iws
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 21474400d4..cfb2b6bf03 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -30,6 +30,8 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Added verification to `io()` ensuring a class named by `IO.reader`, `IO.writer` or `IO.registry` implements the expected interface before loading and initializing it.
* Fixed `io().write()` to resolve its `GraphWriter` before opening the output file, so that a writer which cannot be constructed no longer truncates the target.
* Fixed `TinkerGraph` GraphSON deserialization to fail fast with `JsonParseException` on malformed `tinker:graph` input.
+* Restricted Hadoop/Spark OLAP `io()` from a remotely submitted traversal to operator-approved reader/writer classes and `with()` keys, adding the `gremlin.io.trusted`, `gremlin.io.approvedClasses` and `gremlin.io.approvedGraphConfigKeys` options.
+* Restricted OLAP `GraphComputer.configure()` from a remotely submitted traversal to built-in and operator-approved keys, adding the `gremlin.io.approvedComputerConfigKeys` option.
* Fixed `subgraph()` to throw a descriptive error identifying the required `Edge` input instead of an internal `ClassCastException` when the traversal produces a non-edge value.
* Fixed `where(P)` to throw a descriptive error identifying the required String scope key (and suggesting `is(P)` for value comparisons) instead of an internal `ClassCastException` when given a non-String predicate value.
* Fixed `PeerPressure.property_name` in `gremlin-python` incorrectly mapping to the `pageRank` property name token.
diff --git a/docs/src/reference/implementations-hadoop-start.asciidoc b/docs/src/reference/implementations-hadoop-start.asciidoc
index 9a6a6f4f68..469130ce63 100644
--- a/docs/src/reference/implementations-hadoop-start.asciidoc
+++ b/docs/src/reference/implementations-hadoop-start.asciidoc
@@ -128,6 +128,10 @@ engines (<<sparkgraphcomputer,`SparkGraphComputer`>> refer to their respective d
|gremlin.hadoop.graphWriter |The class that the graph output file(s) are written with (e.g. an `OutputFormat`).
|gremlin.hadoop.jarsInDistributedCache |Whether to upload the Hadoop-Gremlin jars to a distributed cache (necessary if jars are not on the machines' classpaths).
|gremlin.hadoop.defaultGraphComputer |The default `GraphComputer` to use when `graph.compute()` is called. This is optional.
+|gremlin.io.trusted |Enables the full OLAP IO behavior, permitting any reader/writer class, `GraphComputer.configure()` key, and the `ScriptInputFormat`/`ScriptOutputFormat` formats. Suits an in-process deployment that does not accept remote traversals. Defaults to `false`. This is optional.
+|gremlin.io.approvedClasses |A comma-separated allow-list of fully-qualified class names a remotely submitted traversal may load (e.g. a custom reader/writer, the Spark vertex-program interceptor, or `ScriptInputFormat`/`ScriptOutputFormat` to permit script IO). Ignored when `gremlin.io.trusted` is `true`. This is optional.
+|gremlin.io.approvedComputerConfigKeys |A comma-separated allow-list of OLAP computer-configuration keys a remotely submitted traversal may set via `GraphComputer.configure()`. Ignored when `gremlin.io.trusted` is `true`. This is optional.
+|gremlin.io.approvedGraphConfigKeys |A comma-separated allow-list of graph/IO configuration keys a remotely submitted traversal may set via `io().with(key, value)`. Separate from `gremlin.io.approvedComputerConfigKeys` (the two govern different surfaces). Ignored when `gremlin.io.trusted` is `true`. This is optional.
|=========================================================
Along with the properties above, the numerous link:http://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/core-default.xml[Hadoop specific properties]
@@ -138,6 +142,33 @@ underlying OLAP engine (e.g. Spark, etc.) works and understand the numerous para
these systems. Such knowledge can help alleviate out of memory exceptions, slow load times, slow processing times,
garbage collection issues, etc.
+[[remote-olap-configuration]]
+==== Remote OLAP Configuration
+
+When an OLAP traversal is submitted remotely, the `io()` step and `GraphComputer.configure(key, value)` let that
+traversal introduce classes and configuration keys into the graph configuration. By default a remotely submitted
+traversal may use the built-in `gryo` and `graphson` formats together with the configuration keys Hadoop-Gremlin needs
+for normal execution. Any other reader or writer class, the `ScriptInputFormat` and `ScriptOutputFormat` formats, and
+any additional `io().with(key, value)` or `GraphComputer.configure(key, value)` key are applied only when they have
+been approved in the graph configuration. A class or key that has not been approved is rejected rather than loaded or
+applied.
+
+Approvals are declared through the properties listed above. `gremlin.io.approvedClasses` names the fully-qualified
+classes a remote traversal may load, such as a custom `InputFormat` or `OutputFormat`, a Spark vertex-program
+interceptor, or the script formats. `gremlin.io.approvedGraphConfigKeys` names the `io().with(key, value)` keys a
+remote traversal may set, and `gremlin.io.approvedComputerConfigKeys` names the `GraphComputer.configure(key, value)`
+keys it may set. The two key lists are kept separate because `io().with()` and `configure()` govern different
+configuration surfaces, so approving a key on one does not approve it on the other. Setting `gremlin.io.trusted` to
+`true` enables all of this configuration at once and suits an in-process deployment that does not accept remote
+traversals.
+
+These lists are meant for genuine tuning keys. The keys Hadoop-Gremlin sets for itself from the traversal or the graph
+properties file (`gremlin.hadoop.graphReader`, `gremlin.hadoop.graphWriter`, `gremlin.hadoop.graphFilter`,
+`gremlin.hadoop.defaultGraphComputer`, `gremlin.vertexProgram`, and `gremlin.mapReduce`) are managed by the framework
+and do not need to be added to them. The typed fluent setters on `SparkGraphComputer` such as `master(...)`,
+`serializer(...)`, `sparkKryoRegistrator(...)`, and `persistContext(...)` are called from embedded code and are
+unaffected by these lists.
+
=== OLTP Hadoop-Gremlin
image:hadoop-pipes.png[width=180,float=left] It is possible to execute OLTP operations over a `HadoopGraph`.
diff --git a/docs/src/upgrade/release-3.7.x.asciidoc b/docs/src/upgrade/release-3.7.x.asciidoc
index f59e366613..05524f9299 100644
--- a/docs/src/upgrade/release-3.7.x.asciidoc
+++ b/docs/src/upgrade/release-3.7.x.asciidoc
@@ -88,6 +88,35 @@ GryoIo.build().graph(graph).onMapper(m -> ((GryoMapper.Builder) m)
See: link:https://issues.apache.org/jira/browse/TINKERPOP-3278[TINKERPOP-3278]
+==== OLAP Restrictions
+
+The OLAP `io()` step on `HadoopGraph` (`g.withComputer().io(path)...`) historically transferred every
+`with(key, value)` option into the graph configuration and accepted any class name as the reader or writer. As of
+3.7.7 a remotely submitted traversal may name only the built-in `gryo` and `graphson` formats, and its
+`io().with(key, value)` and `GraphComputer.configure(key, value)` options are no longer transferred automatically.
+Naming another reader or writer class, selecting `ScriptInputFormat` or `ScriptOutputFormat`, or setting a key that
+has not been approved now fails with an error rather than loading the class or applying the key.
+
+To keep using a custom format, a script format, or additional `with()` or `configure()` keys from a remote traversal,
+they must be listed in the graph configuration. Setting `gremlin.io.trusted=true` instead restores the previous
+behavior in full for an in-process deployment:
+
+[source,properties]
+----
+# approve specific reader/writer, interceptor, or script format classes
+gremlin.io.approvedClasses=org.example.MyInputFormat,org.example.MyOutputFormat
+# approve specific io().with() and GraphComputer.configure() keys
+gremlin.io.approvedGraphConfigKeys=some.graph.option
+gremlin.io.approvedComputerConfigKeys=spark.executor.memory
+# or restore the previous behavior in full
+gremlin.io.trusted=true
+----
+
+Deployments that use only the built-in formats and configure OLAP through the graph properties file, rather than
+through `with()` on a remote traversal, are unaffected.
+
+See: link:https://tinkerpop.apache.org/docs/3.7.7/reference/#remote-olap-configuration[Remote OLAP Configuration]
+
==== conjoin() Step Null Handling
The `conjoin()` step previously returned `null` when elements in the incoming list are `null`. This behavior has
@@ -241,6 +270,23 @@ Registration is only required for in Java for deserialization, which typically o
registration, read attempts fail with a `Class not recognized` message naming the class, raised as an
`IOException` for GraphBinary, a `MismatchedInputException` for GraphSON and a `RuntimeException` for Gryo.
+===== Hadoop/Spark OLAP IO Class Loading
+
+As of 3.7.7, a remotely submitted OLAP traversal loads a reader, writer, or Spark vertex-program interceptor class only
+after that class has been approved in the graph configuration. Providers that ship custom Hadoop input or output
+formats, or a custom interceptor, must make those classes approvable by declaring them in the graph configuration or by
+listing them in `gremlin.io.approvedClasses`. The Spark vertex-program interceptor
+(`gremlin.hadoop.vertexProgramInterceptor`) is validated against the approved classes at its load site, regardless of
+how its value entered the configuration.
+
+The typed fluent setters on `SparkGraphComputer` (`master(...)`, `serializer(...)`, `sparkKryoRegistrator(...)`,
+`persistContext(...)`) are reachable only from embedded Java code and never from a remote traversal, so they continue
+to work unchanged. Only a `configure(key, value)` call with a key that has not been approved is affected. Embedded code
+that needs arbitrary `configure()` keys can set `gremlin.io.trusted=true`, or set the configuration in the graph
+properties file.
+
+See: link:https://tinkerpop.apache.org/docs/3.7.7/reference/#remote-olap-configuration[Remote OLAP Configuration]
+
== TinkerPop 3.7.6
*Release Date: April 1, 2026*
diff --git a/hadoop-gremlin/conf/hadoop-script.properties b/hadoop-gremlin/conf/hadoop-script.properties
index e2cfc32064..96b281495d 100644
--- a/hadoop-gremlin/conf/hadoop-script.properties
+++ b/hadoop-gremlin/conf/hadoop-script.properties
@@ -16,6 +16,8 @@
# under the License.
gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph
+# ScriptInputFormat compiles and runs a script, so it requires trusted IO to be enabled (this is a trusted local job).
+gremlin.io.trusted=true
gremlin.hadoop.graphReader=org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptInputFormat
gremlin.hadoop.graphWriter=org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONOutputFormat
gremlin.hadoop.jarsInDistributedCache=true
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
index c07aad9e8c..1d55e56015 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
@@ -26,6 +26,7 @@ import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapConfigKeyPolicy;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.GraphFilter;
@@ -45,6 +46,7 @@ import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
@@ -71,11 +73,40 @@ public abstract class AbstractHadoopGraphComputer implements GraphComputer {
protected GraphFilter graphFilter = new GraphFilter();
+ /**
+ * Configuration key holding a comma-separated allow-list of OLAP computer-configuration keys an operator permits
+ * to be set from an untrusted traversal via {@link #configure(String, Object)}. Canonical definition lives on
+ * {@link OlapConfigKeyPolicy}; re-exported here for callers/tests that reference it on the computer.
+ */
+ public static final String APPROVED_COMPUTER_CONFIG_KEYS = OlapConfigKeyPolicy.APPROVED_COMPUTER_CONFIG_KEYS;
+
public AbstractHadoopGraphComputer(final HadoopGraph hadoopGraph) {
this.hadoopGraph = hadoopGraph;
this.logger = LoggerFactory.getLogger(this.getClass());
}
+ /**
+ * The framework configuration keys a concrete computer's own strategies set via {@link #configure(String, Object)}
+ * during normal execution; always permitted so ordinary OLAP is not broken. Empty by default; a computer overrides
+ * this to contribute its keys, keeping that knowledge in the computer's own module rather than this base class.
+ */
+ protected Set<String> builtinApprovedConfigKeys() {
+ return Collections.emptySet();
+ }
+
+ /**
+ * Guards a computer configuration key supplied via {@link #configure(String, Object)} against untrusted OLAP
+ * input, delegating to {@link OlapConfigKeyPolicy} with this computer's built-in framework keys
+ * ({@link #builtinApprovedConfigKeys()}) and the {@link OlapConfigKeyPolicy#APPROVED_COMPUTER_CONFIG_KEYS} operator
+ * allow-list. Trust-boundary keys are never settable (self-elevation); trusted deployments permit all keys; untrusted
+ * deployments permit only built-in or operator-approved keys, so a remotely injected key (for example
+ * {@code spark.executor.extraJavaOptions} or {@code fs.*.impl}) is rejected before it can reach a reflective sink.
+ */
+ protected void checkConfigurationKeyPermitted(final String key) {
+ OlapConfigKeyPolicy.checkConfigKeyPermitted(this.hadoopGraph.configuration(), key,
+ builtinApprovedConfigKeys(), OlapConfigKeyPolicy.APPROVED_COMPUTER_CONFIG_KEYS);
+ }
+
@Override
public GraphComputer vertices(final Traversal<Vertex, Vertex> vertexFilter) {
this.graphFilter.setVertexFilter(vertexFilter);
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStep.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStep.java
index 6cffa12a21..78f01c69e3 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStep.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStep.java
@@ -34,13 +34,20 @@ import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSo
import org.apache.tinkerpop.gremlin.process.traversal.step.ReadWriting;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.Parameters;
import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapConfigKeyPolicy;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
+import java.util.Collections;
+
/**
* An OLAP oriented step for doing IO operations with {@link GraphTraversalSource#io(String)} which uses the
* {@link CloneVertexProgram} for its implementation. Standard Hadoop OLAP configurations can be passed using the
- * {@link GraphTraversal#with(String, Object)} step modulator as all options aside from those in {@link IO} will be
- * transferred.
+ * {@link GraphTraversal#with(String, Object)} step modulator: in a trusted deployment (see
+ * {@link OlapClassLoadingPolicy#TRUSTED}) all options aside from those in {@link IO} are transferred. By default a
+ * traversal is treated as untrusted, in which case reader/writer selection is restricted to approved formats and other
+ * {@code with(k,v)} options are not transferred into the graph configuration, so a remote caller cannot drive
+ * reflective class loading, script execution, or native deserialization through OLAP IO.
*
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
@@ -110,16 +117,9 @@ public class HadoopIoStep extends VertexProgramStep implements ReadWriting {
}
private void configureForRead(final Graph graph) {
- final String inputFormatClassNameOrKeyword = parameters.get(IO.reader, this::detectReader).get(0);
- String inputFormatClassName;
- if (inputFormatClassNameOrKeyword.equals(IO.graphson))
- inputFormatClassName = GraphSONInputFormat.class.getName();
- else if (inputFormatClassNameOrKeyword.equals(IO.gryo))
- inputFormatClassName = GryoInputFormat.class.getName();
- else if (inputFormatClassNameOrKeyword.equals(IO.graphml))
- throw new IllegalStateException("GraphML is not a supported file format for OLAP");
- else
- inputFormatClassName = inputFormatClassNameOrKeyword;
+ final boolean trusted = OlapClassLoadingPolicy.isTrusted(graph.configuration());
+ final String reader = parameters.get(IO.reader, this::detectReader).get(0);
+ final String inputFormatClassName = resolveFormat(graph, reader, trusted, GryoInputFormat.class, GraphSONInputFormat.class);
graph.configuration().setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, inputFormatClassName);
graph.configuration().setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, file);
@@ -128,17 +128,10 @@ public class HadoopIoStep extends VertexProgramStep implements ReadWriting {
}
private void configureForWrite(final Graph graph) {
- final String outputFormatClassNameOrKeyword = parameters.get(IO.writer, this::detectWriter).get(0);
- String outputFormatClassName;
- if (outputFormatClassNameOrKeyword.equals(IO.graphson))
- outputFormatClassName = GraphSONOutputFormat.class.getName();
- else if (outputFormatClassNameOrKeyword.equals(IO.gryo))
- outputFormatClassName = GryoOutputFormat.class.getName();
- else if (outputFormatClassNameOrKeyword.equals(IO.graphml))
- throw new IllegalStateException("GraphML is not a supported file format for OLAP");
- else
- outputFormatClassName = outputFormatClassNameOrKeyword;
-
+ final boolean trusted = OlapClassLoadingPolicy.isTrusted(graph.configuration());
+ final String writer = parameters.get(IO.writer, this::detectWriter).get(0);
+ final String outputFormatClassName = resolveFormat(graph, writer, trusted, GryoOutputFormat.class, GraphSONOutputFormat.class);
+
graph.configuration().setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, outputFormatClassName);
graph.configuration().setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, file);
@@ -146,24 +139,78 @@ public class HadoopIoStep extends VertexProgramStep implements ReadWriting {
}
/**
- * Overwrites all configurations from values passed using {@link GraphTraversal#with(String, Object)}.
+ * Resolves a reader/writer token to a concrete format class name. A built-in keyword maps to the corresponding
+ * built-in format; GraphML is unsupported for OLAP; any other value is an explicit class name that, in untrusted
+ * mode, must resolve to an approved format. Centralizing this keeps the untrusted approval check in a single place
+ * for both the read and write paths.
+ */
+ private String resolveFormat(final Graph graph, final String nameOrKeyword, final boolean trusted,
+ final Class<?> gryoFormat, final Class<?> graphsonFormat) {
+ if (nameOrKeyword.equals(IO.graphson))
+ return graphsonFormat.getName();
+ else if (nameOrKeyword.equals(IO.gryo))
+ return gryoFormat.getName();
+ else if (nameOrKeyword.equals(IO.graphml))
+ throw new IllegalStateException("GraphML is not a supported file format for OLAP");
+ else {
+ // an explicit format class name -- in untrusted mode it must resolve to an approved format
+ if (!trusted && !approvedFormats(graph).isApproved(nameOrKeyword))
+ throw new IllegalArgumentException(formatRejection(nameOrKeyword));
+ return nameOrKeyword;
+ }
+ }
+
+ /**
+ * Copies configurations from values passed using {@link GraphTraversal#with(String, Object)} into the graph
+ * configuration. Each key is gated by {@link OlapConfigKeyPolicy}: in untrusted (default) mode a remote traversal
+ * may set only keys the operator approved via {@link OlapConfigKeyPolicy#APPROVED_GRAPH_CONFIG_KEYS} (trust-boundary
+ * keys are never settable); trusted deployments (see {@link OlapClassLoadingPolicy#TRUSTED}) copy all options. This
+ * keeps {@code io().with()} consistent with a graph computer's {@code configure()} while each surface keeps its own
+ * operator allow-list. The open reflective config surface (formats, computers, serializers, codecs, filesystem
+ * implementations, scripts) therefore stays closed to untrusted input unless an operator explicitly opens a key.
*/
private void addParametersToConfiguration(final Graph graph) {
- parameters.getRaw(IO.writer, IO.writer, IO.registry).entrySet().forEach(kv -> {
+ parameters.getRaw(IO.reader, IO.writer, IO.registry).entrySet().forEach(kv -> {
+ final String key = kv.getKey().toString();
+ OlapConfigKeyPolicy.checkConfigKeyPermitted(graph.configuration(), key,
+ Collections.emptySet(), OlapConfigKeyPolicy.APPROVED_GRAPH_CONFIG_KEYS);
if (kv.getValue().size() == 1)
- graph.configuration().setProperty(kv.getKey().toString(), kv.getValue().get(0));
+ graph.configuration().setProperty(key, kv.getValue().get(0));
else {
// reset the default configuration with the first option then add to that for List options
for (int ix = 0; ix < kv.getValue().size(); ix++) {
if (ix == 0)
- graph.configuration().setProperty(kv.getKey().toString(), kv.getValue().get(ix));
+ graph.configuration().setProperty(key, kv.getValue().get(ix));
else
- graph.configuration().addProperty(kv.getKey().toString(), kv.getValue().get(ix));
+ graph.configuration().addProperty(key, kv.getValue().get(ix));
}
}
});
}
+ /**
+ * The formats a remote (untrusted) traversal may name explicitly: the built-in Gryo/GraphSON formats, any format
+ * class the operator already declared in trusted graph configuration, and any class the operator approved via
+ * {@link OlapClassLoadingPolicy#APPROVED_CLASSES}. It is seeded only from the trusted base configuration and must be
+ * built before any {@code with(k,v)} values are merged into the configuration, so an attacker-supplied reader/writer
+ * value can never approve itself.
+ */
+ private static OlapClassLoadingPolicy approvedFormats(final Graph graph) {
+ return OlapClassLoadingPolicy.build()
+ .approve(GryoInputFormat.class, GraphSONInputFormat.class, GryoOutputFormat.class, GraphSONOutputFormat.class)
+ .approveFromConfigValues(graph.configuration(),
+ Constants.GREMLIN_HADOOP_GRAPH_READER, Constants.GREMLIN_HADOOP_GRAPH_WRITER)
+ .approveFrom(graph.configuration())
+ .create();
+ }
+
+ private static String formatRejection(final String name) {
+ return String.format(
+ "The format '%s' is not an approved OLAP format for a remote traversal. Use a built-in format ('%s'/'%s'), " +
+ "declare it in trusted graph configuration, or add it to '%s'.",
+ name, IO.gryo, IO.graphson, OlapClassLoadingPolicy.APPROVED_CLASSES);
+ }
+
private String detectReader() {
if (file.endsWith(".kryo"))
return GryoInputFormat.class.getName();
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java
index bce982063f..5aa73eb47f 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReader.java
@@ -37,6 +37,7 @@ import org.apache.tinkerpop.gremlin.process.computer.util.VertexProgramHelper;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.T;
import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph;
import javax.script.Bindings;
@@ -76,10 +77,22 @@ public final class ScriptRecordReader extends RecordReader<NullWritable, VertexW
@Override
public void initialize(final InputSplit genericSplit, final TaskAttemptContext context) throws IOException {
- this.lineRecordReader.initialize(genericSplit, context);
final Configuration configuration = context.getConfiguration();
+ // ScriptInputFormat compiles and executes a script, so it is restricted regardless of how it was selected. It
+ // is permitted when the deployment is trusted, or when the operator has explicitly approved ScriptInputFormat
+ // via gremlin.io.approvedClasses in the graph configuration -- a narrower opt-in than trusting the whole
+ // deployment. Note this honors only the explicit approved-class list, not the configured graphReader (which
+ // would be circular here). This backstops the HadoopIoStep boundary against any other path that reaches this
+ // reader.
+ final org.apache.commons.configuration2.Configuration graphConfiguration = ConfUtil.makeApacheConfiguration(configuration);
+ if (!OlapClassLoadingPolicy.isTrusted(graphConfiguration) &&
+ !OlapClassLoadingPolicy.build().approveFrom(graphConfiguration).create().isApproved(ScriptInputFormat.class.getName()))
+ throw new IllegalStateException(String.format(
+ "ScriptInputFormat compiles and runs a script and is only permitted for trusted IO; set '%s' to true, or add '%s' to '%s', in trusted graph configuration to enable it.",
+ OlapClassLoadingPolicy.TRUSTED, ScriptInputFormat.class.getName(), OlapClassLoadingPolicy.APPROVED_CLASSES));
+ this.lineRecordReader.initialize(genericSplit, context);
if (configuration.get(Constants.GREMLIN_HADOOP_GRAPH_FILTER, null) != null)
- this.graphFilter = VertexProgramHelper.deserialize(ConfUtil.makeApacheConfiguration(configuration), Constants.GREMLIN_HADOOP_GRAPH_FILTER);
+ this.graphFilter = VertexProgramHelper.deserialize(graphConfiguration, Constants.GREMLIN_HADOOP_GRAPH_FILTER);
this.engine = manager.getEngineByName(configuration.get(SCRIPT_ENGINE, "gremlin-groovy"));
final FileSystem fs = FileSystem.get(configuration);
try (final InputStream stream = fs.open(new Path(configuration.get(SCRIPT_FILE)));
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordWriter.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordWriter.java
index 0641a50da5..0131d6250b 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordWriter.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordWriter.java
@@ -27,6 +27,8 @@ import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.jsr223.CachedGremlinScriptEngineManager;
import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptEngineManager;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
+import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import javax.script.Bindings;
import javax.script.ScriptEngine;
@@ -62,6 +64,17 @@ public final class ScriptRecordWriter extends RecordWriter<NullWritable, VertexW
public ScriptRecordWriter(final DataOutputStream out, final TaskAttemptContext context) throws IOException {
this.out = out;
final Configuration configuration = context.getConfiguration();
+ // ScriptOutputFormat compiles and runs a script, so it is restricted regardless of how it was selected. It is
+ // permitted when the deployment is trusted, or when the operator has explicitly approved ScriptOutputFormat
+ // via gremlin.io.approvedClasses in the graph configuration -- a narrower opt-in than trusting the whole
+ // deployment. Note this honors only the explicit approved-class list, not the configured graphWriter (which
+ // would be circular here). Mirrors the ScriptRecordReader guard and backstops the HadoopIoStep boundary.
+ final org.apache.commons.configuration2.Configuration graphConfiguration = ConfUtil.makeApacheConfiguration(configuration);
+ if (!OlapClassLoadingPolicy.isTrusted(graphConfiguration) &&
+ !OlapClassLoadingPolicy.build().approveFrom(graphConfiguration).create().isApproved(ScriptOutputFormat.class.getName()))
+ throw new IllegalStateException(String.format(
+ "ScriptOutputFormat compiles and runs a script and is only permitted for trusted IO; set '%s' to true, or add '%s' to '%s', in trusted graph configuration to enable it.",
+ OlapClassLoadingPolicy.TRUSTED, ScriptOutputFormat.class.getName(), OlapClassLoadingPolicy.APPROVED_CLASSES));
this.engine = manager.getEngineByName(configuration.get(SCRIPT_ENGINE, "gremlin-groovy"));
final FileSystem fs = FileSystem.get(configuration);
try {
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/ConfigValueParser.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/ConfigValueParser.java
new file mode 100644
index 0000000000..ead26cb5c2
--- /dev/null
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/ConfigValueParser.java
@@ -0,0 +1,61 @@
+/*
+ * 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.tinkerpop.gremlin.hadoop.structure.io.util;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+/**
+ * Parses a configuration value that holds a comma-separated string, a multi-valued property, or a single token into an
+ * ordered set of trimmed, non-empty entries. Shared by the allow-lists that read such values so they interpret an
+ * operator's lists identically: class names in {@link OlapClassLoadingPolicy} and configuration keys in
+ * {@link OlapConfigKeyPolicy}. It is deliberately value-type-neutral -- it knows nothing about classes or keys -- so
+ * neither of those depends on the other.
+ * <p/>
+ * Package-private: only the same-package allow-lists use it, so it is kept off the module's public API surface.
+ */
+final class ConfigValueParser {
+
+ private ConfigValueParser() {
+ }
+
+ /**
+ * Splits {@code value} into trimmed, non-empty entries, preserving order and de-duplicating. Accepts a
+ * {@link Collection} (each element split on commas) or a single object whose {@code toString()} is split on commas;
+ * {@code null} (and {@code null} elements) yield no entries.
+ */
+ static Set<String> parse(final Object value) {
+ final Set<String> parsed = new LinkedHashSet<>();
+ if (value instanceof Collection) {
+ for (final Object entry : (Collection<?>) value)
+ if (null != entry) addSplit(parsed, entry.toString());
+ } else if (null != value) {
+ addSplit(parsed, value.toString());
+ }
+ return parsed;
+ }
+
+ private static void addSplit(final Set<String> parsed, final String value) {
+ for (final String part : value.split(",")) {
+ final String trimmed = part.trim();
+ if (!trimmed.isEmpty()) parsed.add(trimmed);
+ }
+ }
+}
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapClassLoadingPolicy.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapClassLoadingPolicy.java
new file mode 100644
index 0000000000..f8bf10d5b6
--- /dev/null
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapClassLoadingPolicy.java
@@ -0,0 +1,190 @@
+/*
+ * 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.tinkerpop.gremlin.hadoop.structure.io.util;
+
+import org.apache.commons.configuration2.Configuration;
+
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+/**
+ * The class-loading policy for untrusted OLAP IO: an allow-list of fully-qualified class names that IO code is
+ * permitted to resolve and load when the class name originates from an untrusted source (for example, a value supplied
+ * on a remote traversal), together with the trust flag ({@link #TRUSTED}) that governs whether the allow-list applies.
+ * The allow-list is a plain membership check paired with a {@link #resolve(String, Class)} helper that never consults a
+ * class loader for a name that has not been approved first. It is the class-loading counterpart to
+ * {@link OlapConfigKeyPolicy}, which governs configuration keys.
+ * <p/>
+ * The policy itself carries no knowledge of which classes are "built in" or where names come from; callers assemble
+ * one from whatever trusted sources apply to them — product built-ins, the operator-supplied
+ * {@code gremlin.io.approvedClasses} configuration (see {@link Builder#approveFrom(Configuration)}), and any classes
+ * already declared in a trusted graph configuration. This keeps the type module-neutral: modules such as
+ * {@code hadoop-gremlin} and {@code spark-gremlin} contribute their own built-in class names rather than having them
+ * hard-coded here.
+ */
+public final class OlapClassLoadingPolicy {
+
+ /**
+ * Configuration key holding a comma-separated list (or multi-valued property) of additional fully-qualified class
+ * names an operator approves for resolution from untrusted IO.
+ */
+ public static final String APPROVED_CLASSES = "gremlin.io.approvedClasses";
+
+ /**
+ * Configuration key that opts a deployment into <em>trusted</em> IO, restoring full class loading (no allow-list
+ * enforcement). It defaults to {@code false} (untrusted), so the allow-list applies unless an operator explicitly
+ * turns it on. This flag must only ever be read from trusted, operator-controlled configuration — never from
+ * per-request/traversal options — and callers that copy untrusted options into configuration must refuse to
+ * copy this key, so an untrusted request cannot elevate itself.
+ */
+ public static final String TRUSTED = "gremlin.io.trusted";
+
+ /**
+ * Returns whether the given (trusted, operator-controlled) configuration opts into trusted IO. Absent or non-boolean
+ * values are treated as {@code false} so IO is untrusted by default.
+ */
+ public static boolean isTrusted(final Configuration configuration) {
+ return null != configuration && configuration.getBoolean(TRUSTED, false);
+ }
+
+ private final Set<String> approved;
+
+ private OlapClassLoadingPolicy(final Set<String> approved) {
+ this.approved = approved;
+ }
+
+ public static Builder build() {
+ return new Builder();
+ }
+
+ /**
+ * Returns {@code true} if the given fully-qualified class name has been approved.
+ */
+ public boolean isApproved(final String className) {
+ return className != null && approved.contains(className);
+ }
+
+ /**
+ * The unmodifiable set of approved fully-qualified class names.
+ */
+ public Set<String> approvedClasses() {
+ return Collections.unmodifiableSet(approved);
+ }
+
+ /**
+ * Resolves an approved class name to a {@link Class} assignable to {@code expectedType}. The approval check happens
+ * <em>before</em> any class loader is consulted, so a name that is not on the allow-list is rejected without being
+ * loaded (and therefore without running its static initializer).
+ *
+ * @throws IllegalArgumentException if the name is not approved, cannot be loaded, or is not assignable to
+ * {@code expectedType}
+ */
+ @SuppressWarnings("unchecked")
+ public <T> Class<? extends T> resolve(final String className, final Class<T> expectedType) {
+ if (!isApproved(className))
+ throw new IllegalArgumentException(String.format(
+ "The class '%s' is not approved for resolution from untrusted IO. If it is trusted, add it to the '%s' configuration.",
+ className, APPROVED_CLASSES));
+
+ final Class<?> clazz;
+ try {
+ // load without initializing: the assignability check below runs before any static initializer, so a class
+ // that turns out not to be assignable is rejected without its initializer ever running
+ clazz = Class.forName(className, false, OlapClassLoadingPolicy.class.getClassLoader());
+ } catch (final ClassNotFoundException e) {
+ throw new IllegalArgumentException(String.format("The approved class '%s' could not be loaded", className), e);
+ }
+
+ if (!expectedType.isAssignableFrom(clazz))
+ throw new IllegalArgumentException(String.format(
+ "The approved class '%s' is not a '%s'", className, expectedType.getName()));
+
+ return (Class<? extends T>) clazz;
+ }
+
+ public static final class Builder {
+
+ private final Set<String> approved = new LinkedHashSet<>();
+
+ private Builder() {
+ }
+
+ /**
+ * Approves the given fully-qualified class names. {@code null} entries are ignored.
+ */
+ public Builder approve(final String... classNames) {
+ if (null != classNames)
+ for (final String className : classNames)
+ addTrimmed(className);
+ return this;
+ }
+
+ /**
+ * Approves the given classes by their fully-qualified names. {@code null} entries are ignored.
+ */
+ public Builder approve(final Class<?>... classes) {
+ if (null != classes)
+ for (final Class<?> clazz : classes)
+ if (null != clazz) addTrimmed(clazz.getName());
+ return this;
+ }
+
+ /**
+ * Approves the names listed under {@link #APPROVED_CLASSES} in the given configuration, if present. The value
+ * may be a comma-separated {@link String} or a multi-valued property that resolves to a {@code Collection}.
+ */
+ public Builder approveFrom(final Configuration configuration) {
+ if (null != configuration && configuration.containsKey(APPROVED_CLASSES))
+ approveValue(configuration.getProperty(APPROVED_CLASSES));
+ return this;
+ }
+
+ /**
+ * Auto-seeds the registry from a trusted configuration by approving the class name(s) held under each of the
+ * given keys, if present. This lets a deployment's already-declared classes (graph reader/writer, graph
+ * computer, serializer, registrator, ...) be trusted by construction, so they resolve without an operator also
+ * listing them under {@link #APPROVED_CLASSES}. Values may be single strings, comma-separated strings, or
+ * multi-valued properties. The keys to read are supplied by the caller because they are module-specific
+ * (gremlin-core does not know the Hadoop/Spark configuration keys).
+ */
+ public Builder approveFromConfigValues(final Configuration configuration, final String... keys) {
+ if (null != configuration && null != keys)
+ for (final String key : keys)
+ if (null != key && configuration.containsKey(key))
+ approveValue(configuration.getProperty(key));
+ return this;
+ }
+
+ private void approveValue(final Object value) {
+ approved.addAll(ConfigValueParser.parse(value));
+ }
+
+ private void addTrimmed(final String className) {
+ if (null != className) {
+ final String trimmed = className.trim();
+ if (!trimmed.isEmpty()) approved.add(trimmed);
+ }
+ }
+
+ public OlapClassLoadingPolicy create() {
+ return new OlapClassLoadingPolicy(new LinkedHashSet<>(approved));
+ }
+ }
+}
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapConfigKeyPolicy.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapConfigKeyPolicy.java
new file mode 100644
index 0000000000..cd1dff3b60
--- /dev/null
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapConfigKeyPolicy.java
@@ -0,0 +1,114 @@
+/*
+ * 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.tinkerpop.gremlin.hadoop.structure.io.util;
+
+import org.apache.commons.configuration2.Configuration;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * Decides whether an OLAP configuration key/value may be set from an untrusted traversal. It is shared by the two
+ * distinct surfaces that copy caller-supplied options into graph/computer configuration — a graph computer's
+ * {@code configure(String, Object)} and an {@code io().with(String, Object)} step — so both behave the same way,
+ * while each keeps its <em>own</em> operator allow-list (a computer-config key set is a different operator decision than
+ * a graph/IO key set, so approving a key on one surface must not open it on the other).
+ * <p/>
+ * The policy is deny-by-default and, for an untrusted deployment, permits only a caller's built-in framework keys or the
+ * keys the operator approved via the supplied allow-list configuration key. Trust and the allow-lists are read only from
+ * the operator-controlled graph configuration, never from the key/value being set. The keys that govern the trust
+ * boundary itself ({@link OlapClassLoadingPolicy#TRUSTED}, {@link OlapClassLoadingPolicy#APPROVED_CLASSES}, and the two
+ * approved-config-key lists) are never settable through either surface — even in a trusted deployment — so a
+ * traversal cannot elevate its own trust.
+ */
+public final class OlapConfigKeyPolicy {
+
+ /**
+ * Configuration key holding a comma-separated allow-list of computer-configuration keys an operator permits to be
+ * set from an untrusted traversal via a graph computer's {@code configure(String, Object)}.
+ */
+ public static final String APPROVED_COMPUTER_CONFIG_KEYS = "gremlin.io.approvedComputerConfigKeys";
+
+ /**
+ * Configuration key holding a comma-separated allow-list of graph/IO configuration keys an operator permits to be
+ * set from an untrusted traversal via {@code io().with(String, Object)}. Separate from
+ * {@link #APPROVED_COMPUTER_CONFIG_KEYS} because {@code io().with()} and {@code configure()} govern different
+ * surfaces.
+ */
+ public static final String APPROVED_GRAPH_CONFIG_KEYS = "gremlin.io.approvedGraphConfigKeys";
+
+ private OlapConfigKeyPolicy() {
+ }
+
+ /**
+ * Returns whether the given key governs the IO trust boundary and therefore may never be set through
+ * {@code configure()} or {@code io().with()} (regardless of trust or allow-list), so a traversal cannot grant
+ * itself trust.
+ */
+ public static boolean isMetaKey(final String key) {
+ return OlapClassLoadingPolicy.TRUSTED.equals(key)
+ || OlapClassLoadingPolicy.APPROVED_CLASSES.equals(key)
+ || APPROVED_COMPUTER_CONFIG_KEYS.equals(key)
+ || APPROVED_GRAPH_CONFIG_KEYS.equals(key);
+ }
+
+ /**
+ * Returns whether {@code key} may be set from an untrusted traversal: never for a {@link #isMetaKey(String) meta
+ * key}; always for a trusted deployment; otherwise only if it is one of {@code builtinKeys} or is listed under
+ * {@code approvedKeysConfigKey} in the (operator-controlled) graph configuration.
+ */
+ public static boolean isConfigKeyPermitted(final Configuration graphConfiguration, final String key,
+ final Set<String> builtinKeys, final String approvedKeysConfigKey) {
+ if (isMetaKey(key))
+ return false;
+ if (OlapClassLoadingPolicy.isTrusted(graphConfiguration))
+ return true;
+ return (null != builtinKeys && builtinKeys.contains(key))
+ || approvedKeys(graphConfiguration, approvedKeysConfigKey).contains(key);
+ }
+
+ /**
+ * Enforces {@link #isConfigKeyPermitted}, throwing {@link IllegalArgumentException} with a clear message when the
+ * key is a meta key (self-elevation) or is not permitted for an untrusted traversal.
+ */
+ public static void checkConfigKeyPermitted(final Configuration graphConfiguration, final String key,
+ final Set<String> builtinKeys, final String approvedKeysConfigKey) {
+ if (isMetaKey(key))
+ throw new IllegalArgumentException(String.format(
+ "The configuration key '%s' governs the IO trust boundary and cannot be set via configure() or io().with().",
+ key));
+ if (!isConfigKeyPermitted(graphConfiguration, key, builtinKeys, approvedKeysConfigKey))
+ throw new IllegalArgumentException(String.format(
+ "The OLAP configuration key '%s' is not permitted for an untrusted traversal. Approve it via '%s' in " +
+ "trusted graph configuration, or set '%s' to true for a trusted deployment.",
+ key, approvedKeysConfigKey, OlapClassLoadingPolicy.TRUSTED));
+ }
+
+ /**
+ * Reads the operator-approved key names under {@code approvedKeysConfigKey} via the shared
+ * {@link ConfigValueParser} (so an operator's lists are parsed identically to class-name lists), while staying
+ * independent of {@link OlapClassLoadingPolicy}'s policy: these are configuration <em>keys</em>, not class names.
+ */
+ private static Set<String> approvedKeys(final Configuration graphConfiguration, final String approvedKeysConfigKey) {
+ if (null != graphConfiguration && null != approvedKeysConfigKey
+ && graphConfiguration.containsKey(approvedKeysConfigKey))
+ return ConfigValueParser.parse(graphConfiguration.getProperty(approvedKeysConfigKey));
+ return Collections.emptySet();
+ }
+}
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStepTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStepTest.java
new file mode 100644
index 0000000000..09c6043d27
--- /dev/null
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStepTest.java
@@ -0,0 +1,234 @@
+/*
+ * 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.tinkerpop.gremlin.hadoop.process.computer.traversal.step.sideEffect;
+
+import org.apache.commons.configuration2.BaseConfiguration;
+import org.apache.commons.configuration2.Configuration;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONInputFormat;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONOutputFormat;
+import org.apache.tinkerpop.gremlin.process.traversal.IO;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
+import org.apache.tinkerpop.gremlin.process.traversal.step.ReadWriting;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapConfigKeyPolicy;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class HadoopIoStepTest {
+
+ private static HadoopIoStep readStepWith(final String key, final Object value) {
+ final HadoopIoStep step = new HadoopIoStep(__.start().asAdmin(), "graph.kryo");
+ step.setMode(ReadWriting.Mode.READING);
+ step.configure(key, value);
+ return step;
+ }
+
+ private static HadoopIoStep writeStepWith(final String key, final Object value) {
+ final HadoopIoStep step = new HadoopIoStep(__.start().asAdmin(), "output");
+ step.setMode(ReadWriting.Mode.WRITING);
+ step.configure(key, value);
+ return step;
+ }
+
+ // A raw graphReader key set via io().with() is denied by the default config-key policy: it is neither a built-in
+ // nor an operator-approved key. There is no class-value-specific gate here and the value never reaches class
+ // loading -- this simply pins that the internal graphReader key stays out of the default allow-list. (The reader
+ // FQCN supplied via the IO.reader token is gated separately; see shouldRejectReaderClassNameFromUntrustedTraversal.)
+ @Test
+ public void shouldDenyGraphReaderKeyByDefaultWhenUntrusted() {
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = readStepWith(Constants.GREMLIN_HADOOP_GRAPH_READER, "evil.Reader");
+ try {
+ step.generateProgram(graph, null);
+ fail("a raw graphReader key from a remote traversal must be denied by default in untrusted mode");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ // graphFilter is denied by the same default config-key policy because it is not a built-in/approved key. This is
+ // NOT a graphFilter-deserialization guard: the value is never deserialized here, it is rejected at the key gate
+ // before use (deserialization hardening of graphFilter is a separate concern). The test pins that graphFilter
+ // stays out of the default allow-list, so an untrusted traversal cannot inject one.
+ @Test
+ public void shouldDenyGraphFilterKeyByDefaultWhenUntrusted() {
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = readStepWith(Constants.GREMLIN_HADOOP_GRAPH_FILTER, "rO0ABXNy"); // never deserialized -- rejected at the key gate
+ try {
+ step.generateProgram(graph, null);
+ fail("graphFilter is not an approved io() key, so it must be denied for an untrusted traversal");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldRejectSelfElevationFromUntrustedTraversal() {
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = readStepWith(OlapClassLoadingPolicy.TRUSTED, true);
+ try {
+ step.generateProgram(graph, null);
+ fail("a remote traversal must not be able to set the IO trust flag on itself");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldAllowClassValuedOptionInTrustedConfiguration() {
+ final Configuration config = new BaseConfiguration();
+ config.setProperty(OlapClassLoadingPolicy.TRUSTED, true);
+ final HadoopGraph graph = HadoopGraph.open(config);
+ final HadoopIoStep step = readStepWith(Constants.GREMLIN_HADOOP_GRAPH_READER, "my.custom.Reader");
+ step.generateProgram(graph, null); // must not throw in trusted mode
+ assertEquals("my.custom.Reader", graph.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_READER));
+ }
+
+ @Test
+ public void shouldAcceptAnyReaderClassNameWhenTrusted() {
+ // a trusted deployment restores the previous behavior: an arbitrary reader FQCN via IO.reader is accepted
+ final Configuration config = new BaseConfiguration();
+ config.setProperty(OlapClassLoadingPolicy.TRUSTED, true);
+ final HadoopGraph graph = HadoopGraph.open(config);
+ final HadoopIoStep step = readStepWith(IO.reader, "my.custom.Reader");
+ step.generateProgram(graph, null); // must not throw in trusted mode
+ assertEquals("my.custom.Reader", graph.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_READER));
+ }
+
+ @Test
+ public void shouldAcceptAnyWriterClassNameWhenTrusted() {
+ // a trusted deployment restores the previous behavior: an arbitrary writer FQCN via IO.writer is accepted
+ final Configuration config = new BaseConfiguration();
+ config.setProperty(OlapClassLoadingPolicy.TRUSTED, true);
+ final HadoopGraph graph = HadoopGraph.open(config);
+ final HadoopIoStep step = writeStepWith(IO.writer, "my.custom.Writer");
+ step.generateProgram(graph, null); // must not throw in trusted mode
+ assertEquals("my.custom.Writer", graph.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_WRITER));
+ }
+
+ @Test
+ public void shouldRejectUnapprovedOptionFromUntrustedTraversal() {
+ // untrusted io().with() permits only operator-approved keys; an unapproved key is rejected
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = readStepWith("spark.executor.memory", "2g");
+ try {
+ step.generateProgram(graph, null);
+ fail("a remote traversal must not set an unapproved OLAP configuration option in untrusted mode");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldAcceptApprovedGraphConfigKeyFromUntrustedTraversal() {
+ // io().with() permits keys the operator approved via its own list (separate from the computer list)
+ final Configuration config = new BaseConfiguration();
+ config.setProperty(OlapConfigKeyPolicy.APPROVED_GRAPH_CONFIG_KEYS, "my.graph.option");
+ final HadoopGraph graph = HadoopGraph.open(config);
+ final HadoopIoStep step = readStepWith("my.graph.option", "v");
+ step.generateProgram(graph, null); // approved for the io() surface -> copied
+ assertEquals("v", graph.configuration().getString("my.graph.option"));
+ }
+
+ @Test
+ public void shouldRejectApprovedGraphConfigKeysMetaKeyFromUntrustedTraversal() {
+ // the approved-key list itself is a trust-boundary meta-key and is never settable via io().with()
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = readStepWith(OlapConfigKeyPolicy.APPROVED_GRAPH_CONFIG_KEYS, "attacker.key");
+ try {
+ step.generateProgram(graph, null);
+ fail("the approved-graph-config-keys meta-key must not be settable via io().with()");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapConfigKeyPolicy.APPROVED_GRAPH_CONFIG_KEYS));
+ }
+ }
+
+ @Test
+ public void shouldRejectReaderClassNameFromUntrustedTraversal() {
+ // an FQCN supplied via the IO.reader token must not bypass the approved-format guard
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = readStepWith(IO.reader, "org.attacker.Rce");
+ try {
+ step.generateProgram(graph, null);
+ fail("a reader FQCN from a remote traversal must be rejected in untrusted mode");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.APPROVED_CLASSES));
+ }
+ }
+
+ @Test
+ public void shouldAcceptReaderKeywordFromUntrustedTraversal() {
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = readStepWith(IO.reader, IO.graphson);
+ step.generateProgram(graph, null); // a built-in keyword is always allowed
+ assertEquals(GraphSONInputFormat.class.getName(),
+ graph.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_READER));
+ }
+
+ @Test
+ public void shouldAcceptApprovedReaderClassNameFromUntrustedTraversal() {
+ final Configuration config = new BaseConfiguration();
+ config.setProperty(OlapClassLoadingPolicy.APPROVED_CLASSES, "org.provider.CustomInputFormat");
+ final HadoopGraph graph = HadoopGraph.open(config);
+ final HadoopIoStep step = readStepWith(IO.reader, "org.provider.CustomInputFormat");
+ step.generateProgram(graph, null); // approved via the operator's approvedClasses list
+ assertEquals("org.provider.CustomInputFormat",
+ graph.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_READER));
+ }
+
+ // The write path (configureForWrite) resolves the writer format symmetrically to the read path; the following
+ // mirror the reader FQCN/keyword/approved cases for the writer side.
+
+ @Test
+ public void shouldRejectWriterClassNameFromUntrustedTraversal() {
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = writeStepWith(IO.writer, "org.attacker.Rce");
+ try {
+ step.generateProgram(graph, null);
+ fail("a writer FQCN from a remote traversal must be rejected in untrusted mode");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.APPROVED_CLASSES));
+ }
+ }
+
+ @Test
+ public void shouldAcceptWriterKeywordFromUntrustedTraversal() {
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration());
+ final HadoopIoStep step = writeStepWith(IO.writer, IO.graphson);
+ step.generateProgram(graph, null); // a built-in keyword is always allowed
+ assertEquals(GraphSONOutputFormat.class.getName(),
+ graph.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_WRITER));
+ }
+
+ @Test
+ public void shouldAcceptApprovedWriterClassNameFromUntrustedTraversal() {
+ final Configuration config = new BaseConfiguration();
+ config.setProperty(OlapClassLoadingPolicy.APPROVED_CLASSES, "org.provider.CustomOutputFormat");
+ final HadoopGraph graph = HadoopGraph.open(config);
+ final HadoopIoStep step = writeStepWith(IO.writer, "org.provider.CustomOutputFormat");
+ step.generateProgram(graph, null); // approved via the operator's approvedClasses list
+ assertEquals("org.provider.CustomOutputFormat",
+ graph.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_WRITER));
+ }
+}
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReaderTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReaderTest.java
new file mode 100644
index 0000000000..c07c3e853f
--- /dev/null
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReaderTest.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.tinkerpop.gremlin.hadoop.structure.io.script;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class ScriptRecordReaderTest {
+
+ @Test
+ public void shouldRefuseToRunScriptInUntrustedMode() throws Exception {
+ final Configuration configuration = new Configuration(false);
+ configuration.set(ScriptRecordReader.SCRIPT_FILE, "irrelevant.groovy");
+ // default (untrusted) -- the trust check runs before the split/filesystem is touched, so a null split is fine
+ final TaskAttemptContext context = new TaskAttemptContextImpl(configuration, new TaskAttemptID());
+ try {
+ new ScriptRecordReader().initialize(null, context);
+ fail("ScriptRecordReader must refuse to compile/run a script in untrusted mode");
+ } catch (final IllegalStateException ise) {
+ assertTrue(ise.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldRunScriptWhenFormatApprovedInUntrustedMode() throws Exception {
+ final Configuration configuration = new Configuration(false);
+ configuration.set(ScriptRecordReader.SCRIPT_FILE, "irrelevant.groovy");
+ // untrusted, but the operator explicitly approved ScriptInputFormat -> the trust gate must pass (init then
+ // fails later on the null split, not on the gate)
+ configuration.set(OlapClassLoadingPolicy.APPROVED_CLASSES, ScriptInputFormat.class.getName());
+ final TaskAttemptContext context = new TaskAttemptContextImpl(configuration, new TaskAttemptID());
+ try {
+ new ScriptRecordReader().initialize(null, context);
+ } catch (final Exception e) {
+ // a downstream failure on the null split is fine; only the trust-gate rejection (which names the trust
+ // flag) means the approved-class opt-in did not take effect
+ assertFalse("an approved ScriptInputFormat must pass the trust gate, but was rejected: " + e.getMessage(),
+ String.valueOf(e.getMessage()).contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldRejectScriptConfiguredAsGraphReaderButNotApprovedInUntrustedMode() throws Exception {
+ final Configuration configuration = new Configuration(false);
+ configuration.set(ScriptRecordReader.SCRIPT_FILE, "irrelevant.groovy");
+ // the gate seeds ONLY from gremlin.io.approvedClasses, never from the configured graphReader -- otherwise
+ // naming ScriptInputFormat as the reader would self-approve it. Configure it as the reader but do NOT approve
+ // it; the gate must still reject.
+ configuration.set(Constants.GREMLIN_HADOOP_GRAPH_READER, ScriptInputFormat.class.getName());
+ final TaskAttemptContext context = new TaskAttemptContextImpl(configuration, new TaskAttemptID());
+ try {
+ new ScriptRecordReader().initialize(null, context);
+ fail("ScriptInputFormat configured as the graphReader must not self-approve; the gate must still reject");
+ } catch (final IllegalStateException ise) {
+ assertTrue(ise.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+}
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReaderWriterTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReaderWriterTest.java
index 50ade2e39a..86f0e7baea 100644
--- a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReaderWriterTest.java
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordReaderWriterTest.java
@@ -26,6 +26,7 @@ import org.apache.tinkerpop.gremlin.features.TestFiles;
import org.apache.tinkerpop.gremlin.hadoop.HadoopGraphProvider;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.RecordReaderWriterTest;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
import java.io.File;
@@ -37,6 +38,8 @@ public class ScriptRecordReaderWriterTest extends RecordReaderWriterTest {
@Override
protected Configuration configure(final File outputDirectory) {
final Configuration configuration = super.configure(outputDirectory);
+ // ScriptInputFormat requires trusted IO; this is a trusted local test that legitimately runs the script
+ configuration.set(OlapClassLoadingPolicy.TRUSTED, "true");
configuration.set(ScriptRecordReader.SCRIPT_FILE, TestFiles.PATHS.get("script-input-grateful-dead.groovy"));
configuration.set(ScriptRecordWriter.SCRIPT_FILE, TestFiles.PATHS.get("script-output-grateful-dead.groovy"));
return configuration;
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordWriterTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordWriterTest.java
new file mode 100644
index 0000000000..14f5bcb7a2
--- /dev/null
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/script/ScriptRecordWriterTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.tinkerpop.gremlin.hadoop.structure.io.script;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class ScriptRecordWriterTest {
+
+ @Test
+ public void shouldRefuseToRunScriptInUntrustedMode() throws Exception {
+ final Configuration configuration = new Configuration(false);
+ configuration.set(ScriptRecordWriter.SCRIPT_FILE, "irrelevant.groovy");
+ // default (untrusted) -- the trust check runs before the script is opened/compiled
+ final TaskAttemptContext context = new TaskAttemptContextImpl(configuration, new TaskAttemptID());
+ try {
+ new ScriptRecordWriter(new DataOutputStream(new ByteArrayOutputStream()), context);
+ fail("ScriptRecordWriter must refuse to compile/run a script in untrusted mode");
+ } catch (final IllegalStateException ise) {
+ assertTrue(ise.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldRunScriptWhenFormatApprovedInUntrustedMode() throws Exception {
+ final Configuration configuration = new Configuration(false);
+ configuration.set(ScriptRecordWriter.SCRIPT_FILE, "irrelevant.groovy");
+ // untrusted, but the operator explicitly approved ScriptOutputFormat -> the trust gate must pass (construction
+ // then fails later opening the missing script file, not on the gate)
+ configuration.set(OlapClassLoadingPolicy.APPROVED_CLASSES, ScriptOutputFormat.class.getName());
+ final TaskAttemptContext context = new TaskAttemptContextImpl(configuration, new TaskAttemptID());
+ try {
+ new ScriptRecordWriter(new DataOutputStream(new ByteArrayOutputStream()), context);
+ } catch (final Exception e) {
+ // a downstream failure opening the missing script file is fine; only the trust-gate rejection (which names
+ // the trust flag) means the approved-class opt-in did not take effect
+ assertFalse("an approved ScriptOutputFormat must pass the trust gate, but was rejected: " + e.getMessage(),
+ String.valueOf(e.getMessage()).contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldRejectScriptConfiguredAsGraphWriterButNotApprovedInUntrustedMode() throws Exception {
+ final Configuration configuration = new Configuration(false);
+ configuration.set(ScriptRecordWriter.SCRIPT_FILE, "irrelevant.groovy");
+ // the gate seeds ONLY from gremlin.io.approvedClasses, never from the configured graphWriter -- otherwise
+ // naming ScriptOutputFormat as the writer would self-approve it. Configure it as the writer but do NOT approve
+ // it; the gate must still reject.
+ configuration.set(Constants.GREMLIN_HADOOP_GRAPH_WRITER, ScriptOutputFormat.class.getName());
+ final TaskAttemptContext context = new TaskAttemptContextImpl(configuration, new TaskAttemptID());
+ try {
+ new ScriptRecordWriter(new DataOutputStream(new ByteArrayOutputStream()), context);
+ fail("ScriptOutputFormat configured as the graphWriter must not self-approve; the gate must still reject");
+ } catch (final IllegalStateException ise) {
+ assertTrue(ise.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+}
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/ConfigValueParserTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/ConfigValueParserTest.java
new file mode 100644
index 0000000000..373f6d666d
--- /dev/null
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/ConfigValueParserTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.tinkerpop.gremlin.hadoop.structure.io.util;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Direct tests for the canonical operator-list parser shared by {@link OlapClassLoadingPolicy} and
+ * {@link OlapConfigKeyPolicy}. Pins the parsing contract here so a regression points at the parser rather than a
+ * consumer.
+ */
+public class ConfigValueParserTest {
+
+ @Test
+ public void shouldParseCommaSeparatedString() {
+ assertEquals(Arrays.asList("a.B", "c.D", "e.F"), new ArrayList<>(ConfigValueParser.parse("a.B,c.D,e.F")));
+ }
+
+ @Test
+ public void shouldParseMultiValuedCollectionSplittingEachElementAndSkippingNulls() {
+ assertEquals(Arrays.asList("a.B", "c.D", "e.F"),
+ new ArrayList<>(ConfigValueParser.parse(Arrays.asList("a.B", null, "c.D,e.F"))));
+ }
+
+ @Test
+ public void shouldTrimWhitespaceAndSkipBlankEntries() {
+ assertEquals(Arrays.asList("a.B", "c.D"), new ArrayList<>(ConfigValueParser.parse(" a.B , , c.D ,")));
+ }
+
+ @Test
+ public void shouldYieldNothingForNull() {
+ assertTrue(ConfigValueParser.parse(null).isEmpty());
+ }
+
+ @Test
+ public void shouldPreserveOrderAndDeduplicate() {
+ assertEquals(Arrays.asList("a.B", "c.D", "e.F"),
+ new ArrayList<>(ConfigValueParser.parse("a.B, c.D, a.B, e.F, c.D")));
+ }
+}
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapClassLoadingPolicyTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapClassLoadingPolicyTest.java
new file mode 100644
index 0000000000..4d835013a6
--- /dev/null
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapClassLoadingPolicyTest.java
@@ -0,0 +1,202 @@
+/*
+ * 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.tinkerpop.gremlin.hadoop.structure.io.util;
+
+import org.apache.commons.configuration2.BaseConfiguration;
+import org.apache.commons.configuration2.Configuration;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class OlapClassLoadingPolicyTest {
+
+ @Test
+ public void shouldReportMembership() {
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approve("a.B", "c.D").create();
+ assertTrue(policy.isApproved("a.B"));
+ assertTrue(policy.isApproved("c.D"));
+ assertFalse(policy.isApproved("x.Y"));
+ assertFalse(policy.isApproved(null));
+ }
+
+ @Test
+ public void shouldResolveApprovedAssignableClass() {
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approve(ArrayList.class).create();
+ final Class<? extends List> resolved = policy.resolve("java.util.ArrayList", List.class);
+ assertEquals(ArrayList.class, resolved);
+ }
+
+ @Test
+ public void shouldRejectUnapprovedClass() {
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approve("java.util.ArrayList").create();
+ try {
+ policy.resolve("java.util.LinkedList", List.class);
+ fail("resolve() must reject a class that is not on the allow-list");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.APPROVED_CLASSES));
+ }
+ }
+
+ /**
+ * Pins two security invariants: (a) an unapproved name is rejected <em>before</em> a class loader is consulted, so
+ * its static initializer never runs; and (b) resolving an approved name loads it <em>without</em> initializing it
+ * (non-initializing {@code Class.forName}), so a class that would fail the assignability check never runs its
+ * initializer either. Uses a probe class referenced only by name (never as a class literal, so it is not loaded by
+ * this test) whose static initializer records a side effect. Part (c) is a positive control proving the side effect
+ * is genuinely observable, so (a) and (b) are not vacuous.
+ */
+ @Test
+ public void shouldRejectUnapprovedBeforeLoadingAndResolveApprovedWithoutInitializing() throws Exception {
+ final String probeName = OlapClassLoadingPolicyTest.class.getName() + "$InitProbe";
+ assertFalse("probe must start uninitialized", PROBE_INITIALIZED);
+
+ // (a) unapproved -> rejected before Class.forName -> initializer must not run
+ final OlapClassLoadingPolicy empty = OlapClassLoadingPolicy.build().create();
+ try {
+ empty.resolve(probeName, Object.class);
+ fail("resolve() must reject the unapproved probe");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.APPROVED_CLASSES));
+ }
+ assertFalse("unapproved probe must be rejected WITHOUT loading/initializing it", PROBE_INITIALIZED);
+
+ // (b) approved -> resolve() loads it but does NOT initialize it (non-initializing load)
+ final OlapClassLoadingPolicy approved = OlapClassLoadingPolicy.build().approve(probeName).create();
+ final Class<?> resolved = approved.resolve(probeName, Object.class);
+ assertEquals(probeName, resolved.getName());
+ assertFalse("resolve() must load the approved class WITHOUT running its static initializer", PROBE_INITIALIZED);
+
+ // (c) positive control: actually using the class initializes it -- proves the probe is observable
+ Class.forName(probeName); // initializing load
+ assertTrue("using the resolved class must run its initializer", PROBE_INITIALIZED);
+ }
+
+ @Test
+ public void shouldRejectApprovedButNotAssignableClass() {
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approve("java.lang.String").create();
+ try {
+ policy.resolve("java.lang.String", List.class);
+ fail("resolve() must reject a class that is not assignable to the expected type");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains("is not a"));
+ }
+ }
+
+ @Test
+ public void shouldRejectApprovedButUnloadableClass() {
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approve("does.not.Exist").create();
+ try {
+ policy.resolve("does.not.Exist", Object.class);
+ fail("resolve() must fail when an approved class cannot be loaded");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getCause() instanceof ClassNotFoundException);
+ }
+ }
+
+ @Test
+ public void shouldApproveFromCommaSeparatedConfiguration() {
+ final Configuration configuration = new BaseConfiguration();
+ configuration.setProperty(OlapClassLoadingPolicy.APPROVED_CLASSES, "a.B, c.D ,e.F");
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approveFrom(configuration).create();
+ assertTrue(policy.isApproved("a.B"));
+ assertTrue(policy.isApproved("c.D"));
+ assertTrue(policy.isApproved("e.F"));
+ }
+
+ @Test
+ public void shouldIgnoreBlankEntriesInCommaSeparatedConfiguration() {
+ final Configuration configuration = new BaseConfiguration();
+ configuration.setProperty(OlapClassLoadingPolicy.APPROVED_CLASSES, "a.B,,c.D, ,");
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approveFrom(configuration).create();
+ assertTrue(policy.isApproved("a.B"));
+ assertTrue(policy.isApproved("c.D"));
+ assertFalse(policy.isApproved(""));
+ assertEquals(2, policy.approvedClasses().size());
+ }
+
+ @Test
+ public void shouldIgnoreNullClassEntries() {
+ assertTrue(OlapClassLoadingPolicy.build().approve((Class<?>) null).create().approvedClasses().isEmpty());
+ }
+
+ @Test
+ public void shouldApproveFromMultiValuedConfiguration() {
+ final Configuration configuration = new BaseConfiguration();
+ configuration.addProperty(OlapClassLoadingPolicy.APPROVED_CLASSES, Arrays.asList("a.B", "c.D"));
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approveFrom(configuration).create();
+ assertTrue(policy.isApproved("a.B"));
+ assertTrue(policy.isApproved("c.D"));
+ }
+
+ @Test
+ public void shouldIgnoreMissingConfigurationKey() {
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build().approveFrom(new BaseConfiguration()).create();
+ assertTrue(policy.approvedClasses().isEmpty());
+ }
+
+ @Test
+ public void shouldAutoSeedFromConfiguredClassValuedKeys() {
+ final Configuration configuration = new BaseConfiguration();
+ configuration.setProperty("gremlin.hadoop.graphReader", "com.provider.Reader");
+ configuration.setProperty("gremlin.hadoop.graphWriter", "com.provider.Writer");
+ final OlapClassLoadingPolicy policy = OlapClassLoadingPolicy.build()
+ .approveFromConfigValues(configuration, "gremlin.hadoop.graphReader", "gremlin.hadoop.graphWriter", "absent.key")
+ .create();
+ assertTrue(policy.isApproved("com.provider.Reader"));
+ assertTrue(policy.isApproved("com.provider.Writer"));
+ assertFalse(policy.isApproved("absent.key"));
+ }
+
+ @Test
+ public void shouldDefaultToUntrusted() {
+ assertFalse(OlapClassLoadingPolicy.isTrusted(new BaseConfiguration()));
+ assertFalse(OlapClassLoadingPolicy.isTrusted(null));
+ }
+
+ @Test
+ public void shouldReadTrustedFlagFromConfiguration() {
+ final Configuration configuration = new BaseConfiguration();
+ configuration.setProperty(OlapClassLoadingPolicy.TRUSTED, true);
+ assertTrue(OlapClassLoadingPolicy.isTrusted(configuration));
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void shouldExposeApprovedNamesAsUnmodifiable() {
+ OlapClassLoadingPolicy.build().approve("a.B").create().approvedClasses().add("c.D");
+ }
+
+ /** Set by {@link InitProbe}'s static initializer. Kept on the (already-loaded) test class so reading it does not
+ * initialize the probe. */
+ public static volatile boolean PROBE_INITIALIZED = false;
+
+ /** A class that is never referenced as a class literal, so it is loaded only if something resolves it by name. Its
+ * static initializer flips {@link #PROBE_INITIALIZED} so a test can observe whether it was loaded. */
+ public static class InitProbe {
+ static {
+ PROBE_INITIALIZED = true;
+ }
+ }
+}
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapConfigKeyPolicyTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapConfigKeyPolicyTest.java
new file mode 100644
index 0000000000..a5e54c8b96
--- /dev/null
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/util/OlapConfigKeyPolicyTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.tinkerpop.gremlin.hadoop.structure.io.util;
+
+import org.apache.commons.configuration2.BaseConfiguration;
+import org.apache.commons.configuration2.Configuration;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapConfigKeyPolicy.APPROVED_COMPUTER_CONFIG_KEYS;
+import static org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapConfigKeyPolicy.APPROVED_GRAPH_CONFIG_KEYS;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Unit tests for the shared OLAP config-key policy used by both {@code configure()} and {@code io().with()}. The two
+ * surfaces reuse the policy <em>mechanism</em> but keep <em>separate</em> operator lists.
+ */
+public class OlapConfigKeyPolicyTest {
+
+ private static Configuration untrusted() {
+ return new BaseConfiguration();
+ }
+
+ private static Configuration trusted() {
+ final Configuration c = new BaseConfiguration();
+ c.setProperty(OlapClassLoadingPolicy.TRUSTED, true);
+ return c;
+ }
+
+ @Test
+ public void shouldDenyUnapprovedKeyWhenUntrusted() {
+ assertFalse(OlapConfigKeyPolicy.isConfigKeyPermitted(untrusted(), "spark.executor.extraJavaOptions",
+ Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS));
+ }
+
+ @Test
+ public void shouldPermitBuiltinKeyWhenUntrusted() {
+ assertTrue(OlapConfigKeyPolicy.isConfigKeyPermitted(untrusted(), "framework.key",
+ Collections.singleton("framework.key"), APPROVED_COMPUTER_CONFIG_KEYS));
+ }
+
+ @Test
+ public void shouldPermitOperatorApprovedKeyWhenUntrusted() {
+ final Configuration c = untrusted();
+ c.setProperty(APPROVED_GRAPH_CONFIG_KEYS, "my.graph.key");
+ assertTrue(OlapConfigKeyPolicy.isConfigKeyPermitted(c, "my.graph.key",
+ Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS));
+ }
+
+ // The point of separate lists: approving a key for one surface must NOT open it on the other surface.
+ @Test
+ public void shouldKeepComputerAndGraphListsSeparate() {
+ final Configuration computerApproved = untrusted();
+ computerApproved.setProperty(APPROVED_COMPUTER_CONFIG_KEYS, "shared.key");
+ assertTrue("approved for the computer surface",
+ OlapConfigKeyPolicy.isConfigKeyPermitted(computerApproved, "shared.key", Collections.emptySet(), APPROVED_COMPUTER_CONFIG_KEYS));
+ assertFalse("must NOT be permitted on the graph/io surface via the computer list",
+ OlapConfigKeyPolicy.isConfigKeyPermitted(computerApproved, "shared.key", Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS));
+
+ final Configuration graphApproved = untrusted();
+ graphApproved.setProperty(APPROVED_GRAPH_CONFIG_KEYS, "shared.key");
+ assertTrue("approved for the graph/io surface",
+ OlapConfigKeyPolicy.isConfigKeyPermitted(graphApproved, "shared.key", Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS));
+ assertFalse("must NOT be permitted on the computer surface via the graph list",
+ OlapConfigKeyPolicy.isConfigKeyPermitted(graphApproved, "shared.key", Collections.emptySet(), APPROVED_COMPUTER_CONFIG_KEYS));
+ }
+
+ @Test
+ public void shouldPermitAnyNonMetaKeyWhenTrusted() {
+ assertTrue(OlapConfigKeyPolicy.isConfigKeyPermitted(trusted(), "spark.executor.extraJavaOptions",
+ Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS));
+ }
+
+ // meta-keys govern the trust boundary and are never settable via either surface, even when trusted (self-elevation).
+ @Test
+ public void shouldRejectMetaKeysEvenWhenTrusted() {
+ for (final String metaKey : new String[]{OlapClassLoadingPolicy.TRUSTED, OlapClassLoadingPolicy.APPROVED_CLASSES,
+ APPROVED_COMPUTER_CONFIG_KEYS, APPROVED_GRAPH_CONFIG_KEYS}) {
+ assertFalse("meta-key '" + metaKey + "' must never be permitted",
+ OlapConfigKeyPolicy.isConfigKeyPermitted(trusted(), metaKey, Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS));
+ assertTrue("isMetaKey must classify '" + metaKey + "'", OlapConfigKeyPolicy.isMetaKey(metaKey));
+ }
+ }
+
+ @Test
+ public void shouldParseCommaSeparatedApprovedKeysAndIgnoreBlanks() {
+ final Configuration c = untrusted();
+ c.setProperty(APPROVED_GRAPH_CONFIG_KEYS, "a.key, b.key ,, c.key");
+ for (final String key : new String[]{"a.key", "b.key", "c.key"})
+ assertTrue(key, OlapConfigKeyPolicy.isConfigKeyPermitted(c, key, Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS));
+ assertFalse("blank entry must not become an approved key",
+ OlapConfigKeyPolicy.isConfigKeyPermitted(c, "", Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS));
+ }
+
+ @Test
+ public void shouldThrowNamingTheSurfaceListForUnapprovedUntrustedKey() {
+ try {
+ OlapConfigKeyPolicy.checkConfigKeyPermitted(untrusted(), "spark.executor.extraJavaOptions",
+ Collections.emptySet(), APPROVED_GRAPH_CONFIG_KEYS);
+ fail("an unapproved key must be rejected for an untrusted traversal");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue("message must name the surface's approved-key list",
+ iae.getMessage().contains(APPROVED_GRAPH_CONFIG_KEYS));
+ assertTrue("message must name the trust flag", iae.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldThrowMetaKeyMessageForTrustBoundaryKey() {
+ try {
+ OlapConfigKeyPolicy.checkConfigKeyPermitted(trusted(), OlapClassLoadingPolicy.TRUSTED,
+ Collections.emptySet(), APPROVED_COMPUTER_CONFIG_KEYS);
+ fail("a trust-boundary meta-key must never be settable, even when trusted");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+}
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
index 52450d9c0b..b6393cc3b6 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
@@ -45,6 +45,7 @@ import org.apache.tinkerpop.gremlin.hadoop.structure.io.FileSystemStorage;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.GraphFilterAware;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.HadoopPoolShimService;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
@@ -61,6 +62,7 @@ import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.Sp
import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.SparkInterceptorStrategy;
import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.SparkSingleIterationStrategy;
import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.interceptor.SparkCloneVertexProgramInterceptor;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.interceptor.SparkStarBarrierInterceptor;
import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.apache.tinkerpop.gremlin.spark.structure.io.InputFormatRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.InputOutputHelper;
@@ -81,6 +83,7 @@ import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimServiceLo
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executor;
@@ -144,25 +147,91 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
return this;
}
+ /**
+ * The framework keys SparkGraphComputer's own optimization strategies set via {@link #configure(String, Object)}
+ * during normal execution; always permitted so ordinary OLAP is not broken. The class-valued
+ * {@link Constants#GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR} value is additionally validated at its load site.
+ */
+ private static final Set<String> BUILTIN_APPROVED_CONFIG_KEYS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
+ Constants.GREMLIN_SPARK_SKIP_PARTITIONER,
+ Constants.GREMLIN_SPARK_SKIP_GRAPH_CACHE,
+ Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR)));
+
+ @Override
+ protected Set<String> builtinApprovedConfigKeys() {
+ return BUILTIN_APPROVED_CONFIG_KEYS;
+ }
+
@Override
public SparkGraphComputer configure(final String key, final Object value) {
+ checkConfigurationKeyPermitted(key);
+ return set(key, value);
+ }
+
+ /**
+ * Sets a configuration value directly, bypassing the untrusted {@link #configure(String, Object)} key policy. Used
+ * by the typed fluent setters below, which are reachable only from embedded Java code (never from a remote
+ * traversal, whose config always arrives through {@code configure(String, Object)}), so they are not subject to the
+ * untrusted allow-list.
+ * <p/>
+ * These keys are intentionally <em>not</em> auto-trusted in {@link #builtinApprovedConfigKeys()}. Most fluent
+ * setters are individually benign ({@link #persistContext(boolean)}, {@link #graphStorageLevel(StorageLevel)}, and
+ * similar), but the setter key-set as a whole also includes class-load sinks -- {@link #serializer(Class)} and
+ * {@link #sparkKryoRegistrator(Class)} name classes Spark loads reflectively, and {@link #master(String)} redirects
+ * the driver's cluster manager -- so the set is gated uniformly rather than auto-trusting individual keys (which
+ * would reopen those sinks to remote {@code configure()}). Keeping them embedded-only via {@code set()} preserves
+ * the typed Java API; an operator that wants a specific benign key settable remotely can list it in
+ * {@code gremlin.io.approvedComputerConfigKeys} instead.
+ */
+ private SparkGraphComputer set(final String key, final Object value) {
this.sparkConfiguration.setProperty(key, value);
return this;
}
+ /**
+ * Resolves the configured {@link SparkVertexProgramInterceptor} class. Trust is read from the pristine operator
+ * graph configuration ({@code hadoopGraph.configuration()}), never from a post-merge configuration that could
+ * carry injected keys. Untrusted deployments restrict the interceptor to approved classes: the built-ins, an
+ * interceptor the operator declared in the pristine graph configuration ({@code gremlin.hadoop.vertexProgramInterceptor},
+ * auto-seeded as {@link org.apache.tinkerpop.gremlin.hadoop.process.computer.traversal.step.sideEffect.HadoopIoStep}
+ * seeds its reader/writer), and operator {@code gremlin.io.approvedClasses} -- so an injected value cannot drive
+ * arbitrary class loading. Trusted deployments load it directly, with an assignability check for a clear failure.
+ * <p/>
+ * Package-private (rather than {@code private}) so the trust/approval branches can be unit-tested directly without
+ * standing up a full Spark job.
+ */
+ @SuppressWarnings("unchecked")
+ Class<? extends SparkVertexProgramInterceptor> resolveInterceptorClass(final String interceptorClassName) throws ClassNotFoundException {
+ final org.apache.commons.configuration2.Configuration trustedConfiguration = this.hadoopGraph.configuration();
+ if (OlapClassLoadingPolicy.isTrusted(trustedConfiguration)) {
+ // load without initializing so a class that fails the assignability check never runs its static initializer
+ final Class<?> clazz = Class.forName(interceptorClassName, false, SparkGraphComputer.class.getClassLoader());
+ if (!SparkVertexProgramInterceptor.class.isAssignableFrom(clazz))
+ throw new IllegalArgumentException(String.format("The class '%s' is not a '%s'",
+ interceptorClassName, SparkVertexProgramInterceptor.class.getName()));
+ return (Class<? extends SparkVertexProgramInterceptor>) clazz;
+ }
+ return OlapClassLoadingPolicy.build()
+ .approve(SparkCloneVertexProgramInterceptor.class, SparkStarBarrierInterceptor.class)
+ .approveFromConfigValues(trustedConfiguration, Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR)
+ .approveFrom(trustedConfiguration)
+ .create()
+ .resolve(interceptorClassName, SparkVertexProgramInterceptor.class);
+ }
+
/**
* Sets the configuration option for {@code spark.master} which is the cluster manager to connect to which may be
* one of the <a href="https://spark.apache.org/docs/latest/submitting-applications.html#master-urls">allowed master URLs</a>.
*/
public SparkGraphComputer master(final String clusterManager) {
- return configure(SparkLauncher.SPARK_MASTER, clusterManager);
+ return set(SparkLauncher.SPARK_MASTER, clusterManager);
}
/**
* Determines if the Spark context should be left open preventing Spark from garbage collecting unreferenced RDDs.
*/
public SparkGraphComputer persistContext(final boolean persist) {
- return configure(GREMLIN_SPARK_PERSIST_CONTEXT, persist);
+ return set(GREMLIN_SPARK_PERSIST_CONTEXT, persist);
}
/**
@@ -170,18 +239,18 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
* to use {@code StorageLevel#MEMORY_ONLY()}
*/
public SparkGraphComputer graphStorageLevel(final StorageLevel storageLevel) {
- return configure(GREMLIN_SPARK_GRAPH_STORAGE_LEVEL, storageLevel.description());
+ return set(GREMLIN_SPARK_GRAPH_STORAGE_LEVEL, storageLevel.description());
}
public SparkGraphComputer persistStorageLevel(final StorageLevel storageLevel) {
- return configure(GREMLIN_SPARK_PERSIST_STORAGE_LEVEL, storageLevel.description());
+ return set(GREMLIN_SPARK_PERSIST_STORAGE_LEVEL, storageLevel.description());
}
/**
* Determines if the graph RDD should be partitioned or not. By default, this value is {@code false}.
*/
public SparkGraphComputer skipPartitioner(final boolean skip) {
- return configure(GREMLIN_SPARK_SKIP_PARTITIONER, skip);
+ return set(GREMLIN_SPARK_SKIP_PARTITIONER, skip);
}
/**
@@ -189,7 +258,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
* {@link #graphStorageLevel(StorageLevel)} is ignored. By default, this value is {@code false}.
*/
public SparkGraphComputer skipGraphCache(final boolean skip) {
- return configure(GREMLIN_SPARK_SKIP_GRAPH_CACHE, skip);
+ return set(GREMLIN_SPARK_SKIP_GRAPH_CACHE, skip);
}
/**
@@ -197,7 +266,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
* set to {@code org.apache.spark.serializer.KryoSerializer}.
*/
public SparkGraphComputer serializer(final Class<? extends Serializer> serializer) {
- return configure(SPARK_SERIALIZER, serializer.getCanonicalName());
+ return set(SPARK_SERIALIZER, serializer.getCanonicalName());
}
/**
@@ -205,7 +274,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
* default this value is set to TinkerPop's {@link GryoRegistrator}.
*/
public SparkGraphComputer sparkKryoRegistrator(final Class<? extends KryoRegistrator> registrator) {
- return configure(Constants.SPARK_KRYO_REGISTRATOR, registrator.getCanonicalName());
+ return set(Constants.SPARK_KRYO_REGISTRATOR, registrator.getCanonicalName());
}
/**
@@ -213,7 +282,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
* will result in an error. By default this value is {@code false}.
*/
public SparkGraphComputer kryoRegistrationRequired(final boolean required) {
- return configure(SPARK_KRYO_REGISTRATION_REQUIRED, required);
+ return set(SPARK_KRYO_REGISTRATION_REQUIRED, required);
}
@Override
@@ -393,8 +462,10 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
// if there is a registered VertexProgramInterceptor, use it to bypass the GraphComputer semantics
if (graphComputerConfiguration.containsKey(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR)) {
try {
+ final Class<? extends SparkVertexProgramInterceptor> interceptorClass = this.resolveInterceptorClass(
+ graphComputerConfiguration.getString(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR));
final SparkVertexProgramInterceptor<VertexProgram> interceptor =
- (SparkVertexProgramInterceptor) Class.forName(graphComputerConfiguration.getString(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR)).newInstance();
+ (SparkVertexProgramInterceptor) interceptorClass.newInstance();
computedGraphRDD = interceptor.apply(this.vertexProgram, loadedGraphRDD, memory);
} catch (final ClassNotFoundException | IllegalAccessException | InstantiationException e) {
throw new IllegalStateException(e.getMessage());
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerConfigSecurityTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerConfigSecurityTest.java
new file mode 100644
index 0000000000..e0be662cf8
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerConfigSecurityTest.java
@@ -0,0 +1,281 @@
+/*
+ * 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.tinkerpop.gremlin.spark.process.computer;
+
+import org.apache.commons.configuration2.BaseConfiguration;
+import org.apache.commons.configuration2.Configuration;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.launcher.SparkLauncher;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.process.computer.AbstractHadoopGraphComputer;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapConfigKeyPolicy;
+import org.apache.tinkerpop.gremlin.process.computer.Computer;
+import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.SparkVertexProgramInterceptor;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.interceptor.SparkCloneVertexProgramInterceptor;
+import org.apache.tinkerpop.gremlin.spark.process.computer.traversal.strategy.optimization.interceptor.SparkStarBarrierInterceptor;
+import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Unit tests for the OLAP computer-config key allow-list enforced at {@link SparkGraphComputer#configure(String, Object)}
+ * (delegating to {@link AbstractHadoopGraphComputer#checkConfigurationKeyPermitted(String)}) and the interceptor
+ * class validation at {@link SparkGraphComputer#resolveInterceptorClass(String)}. No Spark context is created.
+ */
+public class SparkGraphComputerConfigSecurityTest {
+
+ private static SparkGraphComputer computer(final boolean trusted, final String approvedKeysCsv) {
+ final Configuration c = new BaseConfiguration();
+ if (trusted)
+ c.setProperty(OlapClassLoadingPolicy.TRUSTED, true);
+ if (null != approvedKeysCsv)
+ c.setProperty(AbstractHadoopGraphComputer.APPROVED_COMPUTER_CONFIG_KEYS, approvedKeysCsv);
+ return new SparkGraphComputer(HadoopGraph.open(c));
+ }
+
+ @Test
+ public void shouldRejectUnknownConfigKeyWhenUntrusted() {
+ try {
+ computer(false, null).configure("spark.executor.extraJavaOptions", "-Devil");
+ fail("an unapproved computer-config key must be rejected for an untrusted traversal");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(AbstractHadoopGraphComputer.APPROVED_COMPUTER_CONFIG_KEYS));
+ }
+ }
+
+ @Test
+ public void shouldAllowFrameworkKeyWhenUntrusted() {
+ // TinkerPop's own Spark strategies set these via configure() during normal execution
+ computer(false, null).configure(Constants.GREMLIN_SPARK_SKIP_PARTITIONER, true);
+ computer(false, null).configure(Constants.GREMLIN_SPARK_SKIP_GRAPH_CACHE, true);
+ computer(false, null).configure(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR, "x");
+ }
+
+ @Test
+ public void shouldAllowOperatorApprovedKeyWhenUntrusted() {
+ computer(false, "my.custom.key,another.key").configure("my.custom.key", "v"); // must not throw
+ computer(false, "my.custom.key,another.key").configure("another.key", "v"); // must not throw
+ }
+
+ @Test
+ public void shouldRejectSelfElevationWhenUntrusted() {
+ // the trust meta-key is categorically refused by configure(), independent of trust/allow-list
+ try {
+ computer(false, null).configure(OlapClassLoadingPolicy.TRUSTED, true);
+ fail("a traversal must not grant itself trust through configure()");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ @Test
+ public void shouldRejectMetaKeysEvenIfOperatorApprovesThem() {
+ // even if an operator mistakenly lists the meta-keys, they remain non-settable via configure()
+ for (final String metaKey : new String[]{OlapClassLoadingPolicy.TRUSTED, OlapClassLoadingPolicy.APPROVED_CLASSES,
+ OlapConfigKeyPolicy.APPROVED_COMPUTER_CONFIG_KEYS, OlapConfigKeyPolicy.APPROVED_GRAPH_CONFIG_KEYS}) {
+ try {
+ computer(false, metaKey).configure(metaKey, "whatever");
+ fail("meta-key '" + metaKey + "' must never be settable via configure()");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(metaKey));
+ }
+ }
+ }
+
+ @Test
+ public void shouldNotBreakFluentSettersWhenUntrusted() throws Exception {
+ // the typed fluent setters are embedded-only and must keep working on a default (untrusted) graph
+ final SparkGraphComputer computer = computer(false, null)
+ .serializer(GryoSerializer.class)
+ .master("local[4]")
+ .kryoRegistrationRequired(true)
+ .persistContext(false);
+ // assert the values actually landed (not merely that the setters didn't throw): a setter silently routed
+ // through a no-op would still "not throw", so read back the resulting spark configuration.
+ final Configuration spark = sparkConfigurationOf(computer);
+ assertEquals("local[4]", spark.getString(SparkLauncher.SPARK_MASTER));
+ assertEquals(GryoSerializer.class.getCanonicalName(), spark.getString(Constants.SPARK_SERIALIZER));
+ assertTrue(spark.getBoolean(Constants.SPARK_KRYO_REGISTRATION_REQUIRED, false));
+ assertFalse(spark.getBoolean(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true));
+ }
+
+ // Reads SparkGraphComputer's private spark configuration for value assertions. Test-local reflection is preferable
+ // to widening production API with a getter that would expose internal mutable configuration.
+ private static Configuration sparkConfigurationOf(final SparkGraphComputer computer) throws Exception {
+ final Field field = SparkGraphComputer.class.getDeclaredField("sparkConfiguration");
+ field.setAccessible(true);
+ return (Configuration) field.get(computer);
+ }
+
+ @Test
+ public void shouldLandAllowedConfigKeyValueWhenUntrusted() throws Exception {
+ // an allowed key must not only pass the gate but actually store its value on the spark configuration
+ final SparkGraphComputer framework = computer(false, null);
+ framework.configure(Constants.GREMLIN_SPARK_SKIP_PARTITIONER, true);
+ assertTrue(sparkConfigurationOf(framework).getBoolean(Constants.GREMLIN_SPARK_SKIP_PARTITIONER, false));
+
+ final SparkGraphComputer approved = computer(false, "my.custom.key");
+ approved.configure("my.custom.key", "v");
+ assertEquals("v", sparkConfigurationOf(approved).getString("my.custom.key"));
+ }
+
+ @Test
+ public void shouldAllowAnyConfigKeyWhenTrusted() {
+ // trusted deployment retains full pass-through
+ computer(true, null).configure("spark.executor.extraJavaOptions", "-Dok");
+ computer(true, null).configure("fs.file.impl", "some.FileSystem");
+ }
+
+ // the gate must also fire through the real OLAP surface: Computer.apply() loops configure(). This throws before
+ // any SparkContext is created, so it needs no cluster.
+ @Test
+ public void shouldBlockInjectedConfigKeyThroughComputerApplyWhenUntrusted() {
+ final HadoopGraph graph = HadoopGraph.open(new BaseConfiguration()); // untrusted
+ try {
+ Computer.compute(SparkGraphComputer.class)
+ .configure("spark.executor.extraJavaOptions", "-Devil")
+ .apply(graph);
+ fail("Computer.apply must reject an injected computer-config key on an untrusted graph");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(AbstractHadoopGraphComputer.APPROVED_COMPUTER_CONFIG_KEYS));
+ }
+ }
+
+ // The interceptor KEY is a permitted framework key, so its VALUE (a class name) must be validated at the load site:
+ // untrusted deployments may resolve only approved interceptors, so an injected value cannot drive arbitrary class
+ // loading, while the built-ins that normal execution relies on stay resolvable.
+ @Test
+ public void shouldRestrictInterceptorClassAtLoadSiteWhenUntrusted() throws Exception {
+ assertEquals(SparkCloneVertexProgramInterceptor.class,
+ computer(false, null).resolveInterceptorClass(SparkCloneVertexProgramInterceptor.class.getName()));
+ assertEquals(SparkStarBarrierInterceptor.class,
+ computer(false, null).resolveInterceptorClass(SparkStarBarrierInterceptor.class.getName()));
+ try {
+ computer(false, null).resolveInterceptorClass("java.lang.Runtime");
+ fail("an unapproved interceptor class must be rejected before loading on an untrusted graph");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapClassLoadingPolicy.APPROVED_CLASSES));
+ }
+ }
+
+ // Trusted deployments load any interceptor directly, but a non-interceptor class still fails fast with a clear
+ // assignability message (rather than a later ClassCastException deep in the job).
+ @Test
+ public void shouldLoadAnyAssignableInterceptorButRejectNonInterceptorWhenTrusted() throws Exception {
+ assertEquals(SparkStarBarrierInterceptor.class,
+ computer(true, null).resolveInterceptorClass(SparkStarBarrierInterceptor.class.getName()));
+ try {
+ computer(true, null).resolveInterceptorClass("java.lang.Runtime");
+ fail("a class that is not a SparkVertexProgramInterceptor must be rejected even when trusted");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(SparkVertexProgramInterceptor.class.getName()));
+ }
+ }
+
+ // An interceptor the operator declared in the graph configuration (gremlin.hadoop.vertexProgramInterceptor) is
+ // auto-approved for untrusted resolution -- mirroring how HadoopIoStep auto-seeds its reader/writer -- so a
+ // provider's configured interceptor resolves without also being listed in gremlin.io.approvedClasses.
+ @Test
+ public void shouldApproveInterceptorDeclaredInGraphConfigurationWhenUntrusted() throws Exception {
+ final Configuration c = new BaseConfiguration(); // untrusted
+ c.setProperty(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR, ConfiguredInterceptor.class.getName());
+ final SparkGraphComputer computer = new SparkGraphComputer(HadoopGraph.open(c));
+ assertEquals(ConfiguredInterceptor.class,
+ computer.resolveInterceptorClass(ConfiguredInterceptor.class.getName()));
+ }
+
+ // The auto-seed must approve ONLY the operator-declared interceptor, not an arbitrary other class: the seed reads
+ // the pristine graph config, never the post-merge value being resolved. Declaring one interceptor must not
+ // approve a different injected class -- a regression that seeded from the resolved (post-merge) value would be
+ // circular and this test would catch it.
+ @Test
+ public void shouldNotApproveArbitraryInterceptorWhenAnotherIsDeclaredWhenUntrusted() throws Exception {
+ final Configuration c = new BaseConfiguration(); // untrusted
+ c.setProperty(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR, ConfiguredInterceptor.class.getName());
+ final SparkGraphComputer computer = new SparkGraphComputer(HadoopGraph.open(c));
+ try {
+ computer.resolveInterceptorClass("java.lang.Runtime");
+ fail("declaring one interceptor must not approve an arbitrary other class");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage(), iae.getMessage().contains(OlapClassLoadingPolicy.APPROVED_CLASSES));
+ }
+ }
+
+ // separate lists: a key approved on the io() (graph) list must NOT open it for computer.configure(), which reads
+ // only the computer list. Proves the two surfaces keep independent operator allow-lists.
+ @Test
+ public void shouldNotHonorGraphConfigKeysListForComputerConfigureWhenUntrusted() {
+ final Configuration c = new BaseConfiguration();
+ c.setProperty(OlapConfigKeyPolicy.APPROVED_GRAPH_CONFIG_KEYS, "spark.executor.extraJavaOptions");
+ final SparkGraphComputer computer = new SparkGraphComputer(HadoopGraph.open(c));
+ try {
+ computer.configure("spark.executor.extraJavaOptions", "-Devil");
+ fail("configure() must not honor the io() (graph) approved-key list");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(OlapConfigKeyPolicy.APPROVED_COMPUTER_CONFIG_KEYS));
+ }
+ }
+
+ // a class rejected by resolveInterceptorClass's trusted branch must not run its static initializer (it is loaded
+ // non-initializing before the assignability check) -- proven with a probe referenced only by name.
+ @Test
+ public void shouldNotInitializeRejectedInterceptorClassWhenTrusted() throws Exception {
+ assertFalse("probe must start uninitialized", NON_INTERCEPTOR_PROBE_INITIALIZED);
+ try {
+ computer(true, null).resolveInterceptorClass(NonInterceptorProbe.class.getName());
+ fail("a non-interceptor class must be rejected even when trusted");
+ } catch (final IllegalArgumentException iae) {
+ assertTrue(iae.getMessage().contains(SparkVertexProgramInterceptor.class.getName()));
+ }
+ assertFalse("a rejected interceptor class must not have run its static initializer (non-init load)",
+ NON_INTERCEPTOR_PROBE_INITIALIZED);
+ }
+
+ /** Set by {@link NonInterceptorProbe}'s static initializer; kept on the (already-loaded) test class so reading it
+ * does not initialize the probe. */
+ public static volatile boolean NON_INTERCEPTOR_PROBE_INITIALIZED = false;
+
+ /** A non-interceptor class initialized only if something resolves it and runs its initializer. */
+ public static final class NonInterceptorProbe {
+ static {
+ NON_INTERCEPTOR_PROBE_INITIALIZED = true;
+ }
+ }
+
+ // A non-built-in interceptor used only to exercise operator-declared interceptor resolution; apply() is never
+ // invoked by resolveInterceptorClass (which only resolves the class), so it is a no-op.
+ public static final class ConfiguredInterceptor implements SparkVertexProgramInterceptor<VertexProgram> {
+ @Override
+ public JavaPairRDD<Object, VertexWritable> apply(final VertexProgram vertexProgram,
+ final JavaPairRDD<Object, VertexWritable> graph,
+ final SparkMemory memory) {
+ return graph;
+ }
+ }
+}
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopIoSecurityIntegrateTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopIoSecurityIntegrateTest.java
new file mode 100644
index 0000000000..3868ee59c2
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopIoSecurityIntegrateTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.tinkerpop.gremlin.spark.process.computer;
+
+import org.apache.commons.configuration2.Configuration;
+import org.apache.tinkerpop.gremlin.TestHelper;
+import org.apache.tinkerpop.gremlin.features.TestFiles;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoOutputFormat;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptInputFormat;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
+import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
+import org.apache.tinkerpop.gremlin.process.computer.clone.CloneVertexProgram;
+import org.apache.tinkerpop.gremlin.spark.AbstractSparkTest;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.io.IoCore;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
+import org.junit.Test;
+
+import java.util.UUID;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Validates the OLAP IO trust boundary on a real (local) SparkGraphComputer job, using the config-driven
+ * reader/writer + {@code graph.compute().program()} execution shape (not the {@code io()} step, which is guard-blocked
+ * in this build):
+ * <ul>
+ * <li>standard Gryo config-driven OLAP is unaffected;</li>
+ * <li>a {@code ScriptInputFormat} job is blocked unless {@code gremlin.io.trusted} is set;</li>
+ * <li>the trust flag propagates through the job configuration to the record reader, so a trusted script job runs.</li>
+ * </ul>
+ * Local mode runs executors in-JVM; cross-JVM configuration serialization is covered by
+ * {@link SparkLocalClusterIoSecurityIntegrateTest}.
+ */
+public class SparkHadoopIoSecurityIntegrateTest extends AbstractSparkTest {
+
+ private static final String SCRIPT_FILE = "gremlin.hadoop.scriptInputFormat.script";
+
+ private String writeModernInputGraph() throws Exception {
+ final String input = TestHelper.makeTestDataFile(SparkHadoopIoSecurityIntegrateTest.class,
+ UUID.randomUUID().toString(), "input.kryo");
+ final TinkerGraph modern = TinkerFactory.createModern();
+ modern.io(IoCore.gryo()).writeGraph(input);
+ modern.close();
+ return input;
+ }
+
+ private static long countCauseChainFor(final Throwable t, final String marker) {
+ for (Throwable c = t; c != null; c = c.getCause())
+ if (c.getMessage() != null && c.getMessage().contains(marker)) return 1;
+ return 0;
+ }
+
+ // baseline Gryo config-driven OLAP traversal -- must be unaffected by the restriction.
+ @Test
+ public void shouldRunBuiltInGryoOlapUnaffected() throws Exception {
+ final String input = writeModernInputGraph();
+ final Configuration configuration = getBaseConfiguration();
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, input);
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ final Graph graph = GraphFactory.open(configuration);
+ assertEquals(6L, graph.traversal().withComputer(SparkGraphComputer.class).V().count().next().longValue());
+ }
+
+ // config-driven CloneVertexProgram Gryo->Gryo IO -- the config path HadoopIoStep never touches; unaffected.
+ @Test
+ public void shouldCloneWithBuiltInFormatsUnaffected() throws Exception {
+ final String input = writeModernInputGraph();
+ final String output = TestHelper.makeTestDataDirectory(SparkHadoopIoSecurityIntegrateTest.class, UUID.randomUUID().toString());
+ final Configuration configuration = getBaseConfiguration();
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, input);
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, GryoOutputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, output);
+ final Graph graph = GraphFactory.open(configuration);
+ graph.compute(SparkGraphComputer.class).program(CloneVertexProgram.build().create()).submit().get();
+ // read the output back to prove the round-trip completed
+ final Configuration readBack = getBaseConfiguration();
+ readBack.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, output);
+ readBack.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ assertEquals(6L, GraphFactory.open(readBack).traversal().withComputer(SparkGraphComputer.class).V().count().next().longValue());
+ }
+
+ // ScriptInputFormat job WITHOUT the trust flag -- must be blocked at the (in-JVM) executor record reader.
+ @Test
+ public void shouldBlockScriptInputFormatWhenUntrusted() {
+ final Configuration configuration = scriptInputConfiguration(false);
+ try {
+ GraphFactory.open(configuration).traversal().withComputer(SparkGraphComputer.class).V().count().next();
+ fail("a ScriptInputFormat OLAP job must be blocked when gremlin.io.trusted is not set");
+ } catch (final Throwable t) {
+ assertTrue("rejection must name the trust flag; was: " + t,
+ countCauseChainFor(t, OlapClassLoadingPolicy.TRUSTED) > 0);
+ }
+ }
+
+ // ScriptInputFormat job WITH gremlin.io.trusted=true -- must run, proving the flag reaches the reader.
+ @Test
+ public void shouldAllowScriptInputFormatWhenTrusted() {
+ final Configuration configuration = scriptInputConfiguration(true);
+ final long count = GraphFactory.open(configuration).traversal().withComputer(SparkGraphComputer.class).V().count().next();
+ assertTrue("a trusted ScriptInputFormat job should load vertices; got " + count, count > 0);
+ }
+
+ // Propagation micro-check: the trust flag survives the graph-config -> Hadoop-config transform SparkGraphComputer uses.
+ @Test
+ public void shouldPropagateTrustFlagThroughConfUtil() {
+ final Configuration configuration = scriptInputConfiguration(true);
+ final org.apache.hadoop.conf.Configuration hadoop = ConfUtil.makeHadoopConfiguration(configuration);
+ assertTrue("gremlin.io.trusted must survive ConfUtil into the Hadoop Configuration shipped to executors",
+ hadoop.getBoolean(OlapClassLoadingPolicy.TRUSTED, false));
+ }
+
+ private Configuration scriptInputConfiguration(final boolean trusted) {
+ final Configuration configuration = getBaseConfiguration();
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, ScriptInputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, TestFiles.PATHS.get("grateful-dead.txt"));
+ configuration.setProperty(SCRIPT_FILE, TestFiles.PATHS.get("script-input-grateful-dead.groovy"));
+ if (trusted)
+ configuration.setProperty(OlapClassLoadingPolicy.TRUSTED, true);
+ return configuration;
+ }
+}
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkLocalClusterIoSecurityIntegrateTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkLocalClusterIoSecurityIntegrateTest.java
new file mode 100644
index 0000000000..b4b556b050
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkLocalClusterIoSecurityIntegrateTest.java
@@ -0,0 +1,315 @@
+/*
+ * 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.tinkerpop.gremlin.spark.process.computer;
+
+import org.apache.commons.configuration2.BaseConfiguration;
+import org.apache.commons.configuration2.Configuration;
+import org.apache.spark.launcher.SparkLauncher;
+import org.apache.tinkerpop.gremlin.TestHelper;
+import org.apache.tinkerpop.gremlin.features.TestFiles;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.process.computer.AbstractHadoopGraphComputer;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.HadoopPools;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptInputFormat;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptOutputFormat;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy;
+import org.apache.tinkerpop.gremlin.process.computer.Computer;
+import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
+import org.apache.tinkerpop.gremlin.process.computer.clone.CloneVertexProgram;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
+import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimServiceLoader;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.UUID;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.junit.Assume.assumeTrue;
+
+/**
+ * Validates the OLAP trust boundary on <b>real, forked executor JVMs</b> via Spark {@code local-cluster} mode. Unlike
+ * {@link SparkHadoopIoSecurityIntegrateTest} (local[N], in-JVM), the gated code here runs in a separate executor
+ * process, so cross-JVM behavior is exercised for real. Two boundaries are covered:
+ * <ul>
+ * <li><b>script reader/writer gates</b> — a {@code ScriptInputFormat} job (blocked untrusted / allowed trusted) at
+ * the executor's record reader and, symmetrically, a {@code ScriptOutputFormat} job at the executor's record
+ * writer, proving {@code gremlin.io.trusted} is serialized across the JVM boundary into the executor task
+ * configuration on both the read and write sides.</li>
+ * <li><b>computer-config gate</b> — an untrusted traversal that injects {@code spark.executor.extraJavaOptions}
+ * through {@code withComputer(Computer.compute(...).configure(...))} is refused at the driver {@code configure()}
+ * boundary before any executor launches; the trusted counterpart proves that same option genuinely crosses into
+ * a forked executor JVM (a probe vertex program observes the injected {@code -D}), so the guarded sink is a real
+ * executor-JVM code-injection vector. The interceptor class name is likewise validated at its load site.</li>
+ * </ul>
+ * <p/>
+ * Requires a real Spark distribution: set {@code SPARK_HOME} (local-cluster uses it to launch executors). When that
+ * distribution is a <i>binary</i> release, also export {@code SPARK_SCALA_VERSION} matching the distro (e.g.
+ * {@code 2.12} for {@code spark-*-bin-hadoop3}); Spark's {@code AbstractCommandBuilder#getScalaVersion()} otherwise
+ * probes for a source-tree layout ({@code $SPARK_HOME/launcher/target/scala-*}) that a binary release lacks and the
+ * executor fork dies with "Cannot find any build directories." That variable is read via {@code System.getenv} at JVM
+ * start, so it must be exported into the (fork of the) test JVM — it cannot be supplied through Spark configuration.
+ * The test skips when {@code SPARK_HOME} is absent, so it is inert in ordinary CI.
+ * <p/>
+ * Run: {@code SPARK_HOME=<dist> SPARK_SCALA_VERSION=2.12 mvn integration-test failsafe:verify -pl spark-gremlin
+ * -DskipTests -DskipIntegrationTests=false -Dit.test=SparkLocalClusterIoSecurityIntegrateTest}.
+ */
+public class SparkLocalClusterIoSecurityIntegrateTest {
+
+ private static final String SCRIPT_FILE = "gremlin.hadoop.scriptInputFormat.script";
+ private static final String SCRIPT_OUTPUT_FILE = "gremlin.hadoop.scriptOutputFormat.script";
+
+ @After
+ @Before
+ public void cleanup() {
+ Spark.close();
+ HadoopPools.close();
+ KryoShimServiceLoader.close();
+ }
+
+ private static Configuration localClusterConfiguration(final boolean trusted) {
+ assumeTrue("SPARK_HOME must point to a real Spark distribution to launch local-cluster executors",
+ System.getenv("SPARK_HOME") != null && !System.getenv("SPARK_HOME").isEmpty());
+ final Configuration c = new BaseConfiguration();
+ // two forked executor JVMs, 1 core / 1g each -> genuine cross-JVM config serialization
+ c.setProperty(SparkLauncher.SPARK_MASTER, "local-cluster[2,1,1024]");
+ c.setProperty(Constants.SPARK_SERIALIZER, GryoSerializer.class.getCanonicalName());
+ c.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
+ c.setProperty(Graph.GRAPH, HadoopGraph.class.getName());
+ // ship the driver's classpath to the forked executors so our patched jars load there
+ final String cp = System.getProperty("java.class.path");
+ c.setProperty("spark.executor.extraClassPath", cp);
+ c.setProperty("spark.driver.extraClassPath", cp);
+ c.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, ScriptInputFormat.class.getCanonicalName());
+ c.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, TestFiles.PATHS.get("grateful-dead.txt"));
+ c.setProperty(SCRIPT_FILE, TestFiles.PATHS.get("script-input-grateful-dead.groovy"));
+ if (trusted)
+ c.setProperty(OlapClassLoadingPolicy.TRUSTED, true);
+ return c;
+ }
+
+ /**
+ * Switches the given config to a built-in Gryo read of the modern graph (no {@code ScriptInputFormat}), so a
+ * test's only possible failure is the behavior under test rather than the (separately covered) script-reader gate.
+ */
+ private static Configuration withBuiltInGryoInput(final Configuration c) {
+ c.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ c.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, TestFiles.PATHS.get("tinkerpop-modern-v3.kryo"));
+ c.clearProperty(SCRIPT_FILE);
+ return c;
+ }
+
+ /**
+ * Reads the grateful-dead graph via built-in Gryo (no {@code ScriptInputFormat}) and writes it via
+ * {@code ScriptOutputFormat}, so the only script-gated component is the write side. The output script requires
+ * grateful-dead labels ({@code song}/{@code artist}), so this reads that graph rather than the modern one.
+ */
+ private static Configuration withScriptOutputOfGratefulDead(final Configuration c) {
+ c.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
+ c.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, TestFiles.PATHS.get("grateful-dead-v3.kryo"));
+ c.clearProperty(SCRIPT_FILE);
+ c.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, ScriptOutputFormat.class.getCanonicalName());
+ c.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION,
+ TestHelper.makeTestDataDirectory(SparkLocalClusterIoSecurityIntegrateTest.class, UUID.randomUUID().toString()));
+ c.setProperty(SCRIPT_OUTPUT_FILE, TestFiles.PATHS.get("script-output-grateful-dead.groovy"));
+ return c;
+ }
+
+ private static boolean stackMentions(final Throwable t, final String marker) {
+ final java.io.StringWriter sw = new java.io.StringWriter();
+ t.printStackTrace(new java.io.PrintWriter(sw));
+ return sw.toString().contains(marker);
+ }
+
+ /** Recursively reports whether {@code dir} holds any non-empty regular file (an OutputFormat wrote part files). */
+ private static boolean containsNonEmptyFile(final File dir) {
+ final File[] children = dir.listFiles();
+ if (null == children)
+ return dir.isFile() && dir.length() > 0;
+ for (final File child : children) {
+ if (child.isDirectory()) {
+ if (containsNonEmptyFile(child)) return true;
+ } else if (child.length() > 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ // Baseline sanity on a real cluster: a built-in Gryo read runs across forked executors.
+ @Test
+ public void shouldReadBuiltInGryoOnForkedExecutors() {
+ final Configuration c = withBuiltInGryoInput(localClusterConfiguration(true));
+ final long count = GraphFactory.open(c).traversal().withComputer(SparkGraphComputer.class).V().count().next();
+ assertTrue("built-in Gryo OLAP must run on forked executors; got " + count, count > 0);
+ }
+
+ // Cross-JVM block: ScriptInputFormat, untrusted -> executor task fails with our trust-flag message.
+ @Test
+ public void shouldBlockScriptInputFormatOnForkedExecutorsWhenUntrusted() {
+ final Configuration c = localClusterConfiguration(false);
+ try {
+ GraphFactory.open(c).traversal().withComputer(SparkGraphComputer.class).V().count().next();
+ fail("ScriptInputFormat must be blocked on the executor when gremlin.io.trusted is not set");
+ } catch (final Throwable t) {
+ assertTrue("executor-side rejection must name the trust flag; stack was: " + t,
+ stackMentions(t, OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ // Cross-JVM allow (the validity gate): ScriptInputFormat, trusted -> flag reaches the executor, job runs.
+ @Test
+ public void shouldAllowScriptInputFormatOnForkedExecutorsWhenTrusted() {
+ final Configuration c = localClusterConfiguration(true);
+ final long count = GraphFactory.open(c).traversal().withComputer(SparkGraphComputer.class).V().count().next();
+ assertTrue("a trusted ScriptInputFormat job must run on forked executors; got " + count, count > 0);
+ }
+
+ // Cross-JVM allow via the narrower opt-in: untrusted, but the operator approved ScriptInputFormat in
+ // gremlin.io.approvedClasses. Proves the approved-class list reaches the executor's ScriptRecordReader gate and
+ // enables the job cross-JVM, not just the global trust flag.
+ @Test
+ public void shouldAllowScriptInputFormatOnForkedExecutorsWhenFormatApproved() {
+ final Configuration c = localClusterConfiguration(false);
+ c.setProperty(OlapClassLoadingPolicy.APPROVED_CLASSES, ScriptInputFormat.class.getName());
+ final long count = GraphFactory.open(c).traversal().withComputer(SparkGraphComputer.class).V().count().next();
+ assertTrue("an approved ScriptInputFormat job must run on forked executors (untrusted + approvedClasses); got " + count, count > 0);
+ }
+
+ // Cross-JVM block, WRITE side: ScriptOutputFormat, untrusted -> the executor's ScriptRecordWriter refuses to
+ // construct, so the write task fails with our trust-flag message. The read side is built-in Gryo, so the only
+ // script-gated component is the writer.
+ @Test
+ public void shouldBlockScriptOutputFormatOnForkedExecutorsWhenUntrusted() {
+ final Configuration c = withScriptOutputOfGratefulDead(localClusterConfiguration(false));
+ try {
+ GraphFactory.open(c).compute(SparkGraphComputer.class)
+ .program(CloneVertexProgram.build().create()).submit().get();
+ fail("ScriptOutputFormat must be blocked on the executor when gremlin.io.trusted is not set");
+ } catch (final Throwable t) {
+ assertTrue("executor-side write rejection must name the trust flag; stack was: " + t,
+ stackMentions(t, OlapClassLoadingPolicy.TRUSTED));
+ }
+ }
+
+ // Cross-JVM allow, WRITE side: ScriptOutputFormat, trusted -> the writer runs on the executor (compiling and
+ // executing the output script per vertex) and produces output, proving the flag reaches the write task too.
+ @Test
+ public void shouldAllowScriptOutputFormatOnForkedExecutorsWhenTrusted() throws Exception {
+ final Configuration c = withScriptOutputOfGratefulDead(localClusterConfiguration(true));
+ final File outputLocation = new File(c.getString(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION));
+ GraphFactory.open(c).compute(SparkGraphComputer.class)
+ .program(CloneVertexProgram.build().create()).submit().get();
+ assertTrue("a trusted ScriptOutputFormat job must write script output on forked executors",
+ containsNonEmptyFile(outputLocation));
+ }
+
+ // Cross-JVM allow, WRITE side via the narrower opt-in: untrusted, but the operator approved ScriptOutputFormat in
+ // gremlin.io.approvedClasses -> the executor's ScriptRecordWriter gate passes and produces output. Symmetric to the
+ // read-side approved-format scenario.
+ @Test
+ public void shouldAllowScriptOutputFormatOnForkedExecutorsWhenFormatApproved() throws Exception {
+ final Configuration c = withScriptOutputOfGratefulDead(localClusterConfiguration(false));
+ c.setProperty(OlapClassLoadingPolicy.APPROVED_CLASSES, ScriptOutputFormat.class.getName());
+ final File outputLocation = new File(c.getString(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION));
+ GraphFactory.open(c).compute(SparkGraphComputer.class)
+ .program(CloneVertexProgram.build().create()).submit().get();
+ assertTrue("an approved ScriptOutputFormat job must write output on forked executors (untrusted + approvedClasses)",
+ containsNonEmptyFile(outputLocation));
+ }
+
+ // an untrusted OLAP traversal that injects spark.executor.extraJavaOptions through the real
+ // withComputer(Computer.compute(...).configure(...)) surface -> refused at the driver configure() boundary, so the
+ // executor-JVM option never reaches any forked process (no context is even created). Uses built-in Gryo input so
+ // the only thing that can reject the traversal is the config-key gate (not the ScriptInputFormat reader).
+ @Test
+ public void shouldBlockInjectedExecutorJvmOptionOnUntrustedGraph() {
+ final Configuration c = withBuiltInGryoInput(localClusterConfiguration(false));
+ try {
+ GraphFactory.open(c).traversal()
+ .withComputer(Computer.compute(SparkGraphComputer.class)
+ .configure(SparkLauncher.EXECUTOR_EXTRA_JAVA_OPTIONS, "-Dtinkerpop.tierb.rce=owned"))
+ .V().count().next();
+ fail("an untrusted OLAP traversal must not inject " + SparkLauncher.EXECUTOR_EXTRA_JAVA_OPTIONS
+ + " via configure()");
+ } catch (final Throwable t) {
+ assertTrue("refusal must name the operator approval key; stack was: " + t,
+ stackMentions(t, AbstractHadoopGraphComputer.APPROVED_COMPUTER_CONFIG_KEYS));
+ }
+ }
+
+ // the same injected spark.executor.extraJavaOptions is permitted on a trusted graph and provably crosses into a
+ // forked executor JVM: a probe vertex program runs on the executor and reports (via Memory) that it observed the
+ // injected -D. This proves the guarded sink is real and executor-affecting, not theoretical.
+ @Test
+ public void shouldReachForkedExecutorJvmWithInjectedOptionWhenTrusted() throws Exception {
+ final Configuration c = withBuiltInGryoInput(localClusterConfiguration(true));
+ final String property = "tinkerpop.tierb.canary";
+ final String value = "REACHED";
+ final ComputerResult result = GraphFactory.open(c)
+ .compute(SparkGraphComputer.class)
+ .configure(SparkLauncher.EXECUTOR_EXTRA_JAVA_OPTIONS, "-D" + property + "=" + value)
+ .program(SystemPropertyProbeVertexProgram.forProperty(property, value))
+ .submit().get();
+ assertTrue("the injected -D" + property + " must be observed inside a forked executor JVM",
+ result.memory().<Boolean>get(SystemPropertyProbeVertexProgram.MEMORY_OBSERVED));
+ }
+
+ // probe integrity (paired with the test above): the same probe, run WITHOUT injecting the option, must not observe
+ // the canary on any executor -- proving the probe genuinely detects the injected -D (so the "reach" test above is
+ // meaningful) and that an ordinary trusted job carries no such property.
+ @Test
+ public void shouldNotObserveCanaryOnForkedExecutorWithoutInjection() throws Exception {
+ final Configuration c = withBuiltInGryoInput(localClusterConfiguration(true));
+ final ComputerResult result = GraphFactory.open(c)
+ .compute(SparkGraphComputer.class)
+ .program(SystemPropertyProbeVertexProgram.forProperty("tinkerpop.tierb.canary", "REACHED"))
+ .submit().get();
+ assertFalse("no executor JVM should see the canary when the option is not injected",
+ result.memory().<Boolean>get(SystemPropertyProbeVertexProgram.MEMORY_OBSERVED));
+ }
+
+ // the interceptor KEY is a framework-approved config key, so an untrusted OLAP job may set it via configure() --
+ // but its VALUE (a class name) is validated at the load site by resolveInterceptorClass during submit(): an
+ // unapproved interceptor class is refused (message names gremlin.io.approvedClasses), so it cannot drive arbitrary
+ // reflective class loading. A CloneVertexProgram job deterministically reaches the interceptor branch -- when the
+ // interceptor key is already present, the computer skips its built-in auto-selection and resolves the set value.
+ @Test
+ public void shouldBlockInjectedUnapprovedInterceptorClassWhenUntrusted() {
+ final Configuration c = withBuiltInGryoInput(localClusterConfiguration(false));
+ try {
+ GraphFactory.open(c).compute(SparkGraphComputer.class)
+ .configure(Constants.GREMLIN_HADOOP_VERTEX_PROGRAM_INTERCEPTOR, "java.lang.Runtime")
+ .program(CloneVertexProgram.build().create())
+ .submit().get();
+ fail("an untrusted OLAP job must not resolve an unapproved vertexProgramInterceptor class");
+ } catch (final Throwable t) {
+ assertTrue("interceptor rejection must name the approved-classes key; stack was: " + t,
+ stackMentions(t, OlapClassLoadingPolicy.APPROVED_CLASSES));
+ }
+ }
+}
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SystemPropertyProbeVertexProgram.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SystemPropertyProbeVertexProgram.java
new file mode 100644
index 0000000000..a68c38c786
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SystemPropertyProbeVertexProgram.java
@@ -0,0 +1,122 @@
+/*
+ * 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.tinkerpop.gremlin.spark.process.computer;
+
+import org.apache.commons.configuration2.Configuration;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.computer.Memory;
+import org.apache.tinkerpop.gremlin.process.computer.MemoryComputeKey;
+import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
+import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
+import org.apache.tinkerpop.gremlin.process.traversal.Operator;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * A tiny test-only {@link VertexProgram} whose {@link #execute(Vertex, Messenger, Memory)} runs <b>on the executor</b>
+ * and checks whether a given JVM system property is visible there with an expected value, reducing the observation to
+ * the driver via a single boolean {@link Memory} key ({@link Operator#or}: {@code true} iff <i>any</i> executor saw
+ * it). It lets a {@code local-cluster} test prove that a value injected into
+ * {@code spark.executor.extraJavaOptions} genuinely crosses into a forked executor JVM — the sink the
+ * {@code configure()} allow-list guards against untrusted OLAP input.
+ * <p/>
+ * The program is reconstructed on each executor by {@link VertexProgram#createVertexProgram(Graph, Configuration)}
+ * (reflective no-arg constructor + {@link #loadState(Graph, Configuration)}), so its only state — the property name
+ * and expected value — is carried in {@link #storeState(Configuration)}, not via Java serialization.
+ */
+public final class SystemPropertyProbeVertexProgram implements VertexProgram<Object> {
+
+ private static final String CONFIG_PROPERTY_NAME = "gremlin.tierb.probe.propertyName";
+ private static final String CONFIG_EXPECTED_VALUE = "gremlin.tierb.probe.expectedValue";
+
+ /**
+ * Boolean {@link Memory} key readable on the driver after the job: {@code true} iff at least one executor JVM
+ * observed the probed system property with the expected value.
+ */
+ public static final String MEMORY_OBSERVED = "gremlin.tierb.probe.observedOnExecutor";
+
+ private String propertyName;
+ private String expectedValue;
+
+ public static SystemPropertyProbeVertexProgram forProperty(final String propertyName, final String expectedValue) {
+ final SystemPropertyProbeVertexProgram vp = new SystemPropertyProbeVertexProgram();
+ vp.propertyName = propertyName;
+ vp.expectedValue = expectedValue;
+ return vp;
+ }
+
+ @Override
+ public void storeState(final Configuration configuration) {
+ VertexProgram.super.storeState(configuration);
+ configuration.setProperty(CONFIG_PROPERTY_NAME, this.propertyName);
+ configuration.setProperty(CONFIG_EXPECTED_VALUE, this.expectedValue);
+ }
+
+ @Override
+ public void loadState(final Graph graph, final Configuration configuration) {
+ this.propertyName = configuration.getString(CONFIG_PROPERTY_NAME);
+ this.expectedValue = configuration.getString(CONFIG_EXPECTED_VALUE);
+ }
+
+ @Override
+ public void setup(final Memory memory) {
+ memory.set(MEMORY_OBSERVED, false);
+ }
+
+ @Override
+ public void execute(final Vertex vertex, final Messenger<Object> messenger, final Memory memory) {
+ // runs inside the (forked) executor JVM: report whether the injected -D reached this process
+ memory.add(MEMORY_OBSERVED, this.expectedValue.equals(System.getProperty(this.propertyName)));
+ }
+
+ @Override
+ public boolean terminate(final Memory memory) {
+ return true; // a single iteration is enough to sample the executor JVM
+ }
+
+ @Override
+ public Set<MemoryComputeKey> getMemoryComputeKeys() {
+ return Collections.singleton(MemoryComputeKey.of(MEMORY_OBSERVED, Operator.or, true, false));
+ }
+
+ @Override
+ public Set<MessageScope> getMessageScopes(final Memory memory) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ @SuppressWarnings("CloneDoesntCallSuperClone")
+ public SystemPropertyProbeVertexProgram clone() {
+ return forProperty(this.propertyName, this.expectedValue);
+ }
+
+ @Override
+ public GraphComputer.ResultGraph getPreferredResultGraph() {
+ return GraphComputer.ResultGraph.ORIGINAL;
+ }
+
+ @Override
+ public GraphComputer.Persist getPreferredPersist() {
+ return GraphComputer.Persist.NOTHING;
+ }
+}