This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 05c3f2bfef Add TLS and authentication options to the Arrow Flight
server, fixes #8394 (#8401)
05c3f2bfef is described below
commit 05c3f2bfef3b2d040ef9927397763ebd7653f1cb
Author: Bart Maertens <[email protected]>
AuthorDate: Wed Sep 16 10:41:57 2026 +0200
Add TLS and authentication options to the Arrow Flight server, fixes #8394
(#8401)
---
THREAT_MODEL.md | 9 +-
docs/hop-user-manual/modules/ROOT/nav.adoc | 4 +
.../ROOT/pages/hop-tools/hop-arrow-command.adoc | 77 ++++++++-
.../metadata-types/data-stream/arrow-flight.adoc | 33 ++++
.../org/apache/hop/arrow/command/ArrowCommand.java | 42 ++++-
.../datastream/flight/ArrowFlightDataStream.java | 136 +++++++++++++++-
.../hop/arrow/flight/ArrowFlightSecurity.java | 163 +++++++++++++++++++
.../apache/hop/arrow/flight/ArrowFlightServer.java | 73 ++++++++-
.../apache/hop/arrow/flight/HopFlightProducer.java | 7 +-
.../flight/messages/messages_en_US.properties | 14 ++
.../ArrowFlightDataStreamAuthenticationTest.java | 171 ++++++++++++++++++++
.../ArrowFlightDataStreamSerializationTest.java | 178 +++++++++++++++++++++
.../hop/arrow/flight/ArrowFlightSecurityTest.java | 114 +++++++++++++
.../ArrowFlightServerAuthenticationTest.java | 161 +++++++++++++++++++
14 files changed, 1169 insertions(+), 13 deletions(-)
diff --git a/THREAT_MODEL.md b/THREAT_MODEL.md
index 0a97593869..4696a98d37 100644
--- a/THREAT_MODEL.md
+++ b/THREAT_MODEL.md
@@ -54,6 +54,7 @@ the `hop-run` CLI, or submitted to a **Hop Server** for
local/remote execution.
| Execution engine | `engine/`, `engine-beam/` | Runs pipelines/workflows the
operator authored — incl. transforms that touch files, DBs, network, and
**scripting/exec** steps. |
| Hop Server (servlet / HTTP) | `engine/` — package `org.apache.hop.www` +
`HopServerMeta` in `org.apache.hop.server`; launched by the `hop-server`
command (`org.apache.hop.www.HopServer`). *(The `org.apache.hop.server`
connection helpers `HttpUtil`/`ServerConnectionManager` — see §9 — live in
`core/`.)* | The **network trust boundary**: an embedded Jetty server whose
servlets accept pipeline/workflow run requests over HTTP (`/hop/execPipeline`,
`/hop/addPipeline`+`/hop/startExec`, `/hop [...]
| Hop JSON API | `engine/src/main/java/org/apache/hop/www/api/` | **Part of
Hop Server**, not a separate deployable. Mounted on `/hop/api/v1/`
([`WebServer.java`](engine/src/main/java/org/apache/hop/www/WebServer.java))
and exposes `execute/sync` (run a web service), `metadata` CRUD, `plugins` and
`location` execution info. It sits inside the same `ConstraintSecurityHandler`
as every servlet, so it inherits Hop Server's Basic/JAAS auth (§8). |
+| Arrow Flight server (gRPC) | `plugins/tech/arrow/` — package
`org.apache.hop.arrow.flight`; launched by the `hop arrow` command
(`org.apache.hop.arrow.command.ArrowCommand`) | A **second network listener**,
separate from Hop Server: a gRPC/Arrow Flight endpoint that hands rows to and
from Data Stream metadata elements. Binds `0.0.0.0:33333` by default. TLS
(incl. mutual TLS) and username/password authentication are configurable but
**off unless the operator passes the options** (§8) — [...]
| GUI (desktop / web) | `ui/` (`hop-ui`, SWT core), `rcp/` (desktop fragment),
`rap/` (`hop-ui-rap`, RAP/RWT **web** GUI) | Authoring surface used by the
trusted operator. `rap/` is the **web GUI**, not a server. |
| Plugins | `plugins/` | The large transform/action set, incl. scripting
(GraalJS/Rhino/Groovy), shell/exec, SQL, and per-DB/per-cloud connectors
(`plugins/tech/*`). |
| Connection / driver layer | `lib-jdbc/` (bundled JDBC drivers), `core/`
(`org.apache.hop.core.database`, `org.apache.hop.metadata`) | DB/file/cloud
credentials + JDBC drivers the operator configures. (There is no standalone
`metadata/` module — connection/metadata code is in `core/`.) |
@@ -205,8 +206,12 @@ where strict outbound TLS verification is required (see
§9).
the `<sslConfig>` block of `hop-server.xml`, but **TLS is off by default**
(the
shipped config starts a plain-HTTP listener; only auth is on). Outbound JDBC
TLS is fully driver-delegated; HTTP/REST/cloud transforms use the JVM trust
- store unless given a per-transform truststore. Enabling TLS is an operator
- responsibility (§10).
+ store unless given a per-transform truststore. The **Arrow Flight server**
+ (`hop arrow`) is likewise plaintext and unauthenticated unless started with
+ `--arrow-flight-tls-certificate`/`--arrow-flight-tls-key` (optionally
+ `--arrow-flight-tls-client-ca` for mutual TLS) and
+ `--arrow-flight-username`/`--arrow-flight-password`; it logs a warning when
it
+ starts without them. Enabling TLS is an operator responsibility (§10).
## §9 Security properties the project does *not* provide
diff --git a/docs/hop-user-manual/modules/ROOT/nav.adoc
b/docs/hop-user-manual/modules/ROOT/nav.adoc
index 09dbb22042..47c2894cbf 100644
--- a/docs/hop-user-manual/modules/ROOT/nav.adoc
+++ b/docs/hop-user-manual/modules/ROOT/nav.adoc
@@ -496,6 +496,10 @@ under the License.
** xref:metadata-types/databricks-connection.adoc[Databricks Connection]
** xref:metadata-types/databricks-vfs-connection.adoc[Databricks VFS
Connection]
** xref:metadata-types/data-set.adoc[Data Set]
+** xref:metadata-types/data-stream/data-stream.adoc[Data Stream]
+*** xref:metadata-types/data-stream/arrow-file-stream.adoc[Apache Arrow File
Stream]
+*** xref:metadata-types/data-stream/arrow-flight.adoc[Apache Arrow Flight
Stream]
+*** xref:metadata-types/data-stream/arrow-random-access-file.adoc[Apache Arrow
Random Access File]
** xref:metadata-types/dbt-project.adoc[dbt Project]
** xref:metadata-types/execution-data-profile.adoc[]
** xref:metadata-types/execution-information-location.adoc[]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/hop-tools/hop-arrow-command.adoc
b/docs/hop-user-manual/modules/ROOT/pages/hop-tools/hop-arrow-command.adoc
index b0d968775f..5bbe33f044 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/hop-tools/hop-arrow-command.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/hop-tools/hop-arrow-command.adoc
@@ -33,15 +33,37 @@ The `hop arrow` sub-command is used to start Arrow-related
services.
----
sh hop arrow --help
Usage: hop arrow [-hV] [--arrow-flight-host=<hostname>]
- [--arrow-flight-port=<port>] [-e=<environmentOption>]
+ [--arrow-flight-password=<password>]
+ [--arrow-flight-port=<port>]
+ [--arrow-flight-tls-certificate=<tlsCertificateFile>]
+
[--arrow-flight-tls-client-ca=<tlsClientCertificateAuthorityFile>]
+ [--arrow-flight-tls-key=<tlsPrivateKeyFile>]
+ [--arrow-flight-username=<username>] [-e=<environmentOption>]
[-j=<projectOption>]
Run the hop arrow command to start a flight or socket server
--arrow-flight-host=<hostname>
The hostname on which the Apache Arrow Flight server will
listen, defaults to 0.0.0.0
+ --arrow-flight-password=<password>
+ The password belonging to --arrow-flight-username. Can be an
+ encrypted password or a variable expression.
--arrow-flight-port=<port>
The port on which the Apache Arrow Flight server will listen,
defaults to 33333
+ --arrow-flight-tls-certificate=<tlsCertificateFile>
+ The PEM file with the certificate chain of the Apache Arrow
+ Flight server. Specify this together with
+ --arrow-flight-tls-key to serve over TLS.
+ --arrow-flight-tls-client-ca=<tlsClientCertificateAuthorityFile>
+ The PEM file with the certificate authority used to verify
+ client certificates (mutual TLS). Needs TLS to be enabled.
+ --arrow-flight-tls-key=<tlsPrivateKeyFile>
+ The PEM file with the PKCS#8 private key of the Apache Arrow
+ Flight server certificate
+ --arrow-flight-username=<username>
+ The user name clients need to present to the Apache Arrow
+ Flight server. Without it the server accepts
+ unauthenticated clients.
-e, --environment=<environmentOption>
The name of the lifecycle environment to use
-h, --help Show this help message and exit.
@@ -72,10 +94,63 @@ The rows are stored in memory so that they can be picked up.
Reading data from the Flight server with Hop is as simple as referencing the
same Data Stream with the xref
+== Transport security and authentication
+
+By default the Flight server speaks plain gRPC and accepts every client that
can reach the port.
+
+WARNING: Without the options below, anything that can open a connection to the
port can read the data the server streams, and that data travels in the clear.
+Until you enable TLS and authentication, treat the Flight server as a
trusted-network-only service and keep it off any interface it does not need to
be on (use `--arrow-flight-host` to bind it to a single address).
+
+=== TLS
+
+Pass a certificate chain and its PKCS#8 private key to serve over TLS:
+
+[source,bash]
+----
+hop arrow --project my-project \
+ --arrow-flight-tls-certificate /etc/hop/flight/server.crt \
+ --arrow-flight-tls-key /etc/hop/flight/server.key
+----
+
+Both files are read through xref:vfs.adoc[Apache VFS], so they can live
anywhere Hop can reach.
+The clients then connect with `grpc+tls://` instead of `grpc://`.
+
+To also require clients to present a certificate of their own (mutual TLS),
add the certificate authority to verify them with:
+
+[source,bash]
+----
+ --arrow-flight-tls-client-ca /etc/hop/flight/client-ca.crt
+----
+
+Client certificate verification needs TLS to be enabled.
+
+=== Authentication
+
+Pass a user name and password to make the server reject anonymous clients:
+
+[source,bash]
+----
+hop arrow --project my-project \
+ --arrow-flight-username hop \
+ --arrow-flight-password '${FLIGHT_PASSWORD}'
+----
+
+The password accepts a variable expression or a password encrypted with the
xref:hop-tools/hop-encrypt.adoc[hop-encrypt] tool, so it does not have to
appear in plain text in a startup script.
+
+Clients authenticate with the user name and password, and the server hands
them a bearer token which is used for the calls that follow.
+The xref:metadata-types/data-stream/arrow-flight.adoc[Apache Arrow Flight Data
Stream] has matching **Username** and **Password** fields; other clients use
the standard Flight `Authorization` handshake, for example
`pyarrow.flight.FlightClient.authenticate_basic_token()`.
+
+NOTE: Authentication without TLS means the credentials are sent in the clear.
The server logs a warning when you do that. Enable both.
+
=== Supported options:
* `--arrow-flight-host` : Bind address (default: 0.0.0.0)
* `--arrow-flight-port` : Listening port (default: 33333)
+* `--arrow-flight-tls-certificate` : PEM file with the server certificate
chain, enables TLS
+* `--arrow-flight-tls-key` : PEM file with the PKCS#8 private key for that
certificate
+* `--arrow-flight-tls-client-ca` : PEM file with the CA used to verify client
certificates (mutual TLS)
+* `--arrow-flight-username` : The user name clients need to present
+* `--arrow-flight-password` : The password for that user, optionally encrypted
or a variable
* `--project` : Used to find the Data Stream metadata to reference
* `--environment`: Used to find the project and metadata but also set
variables.
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/metadata-types/data-stream/arrow-flight.adoc
b/docs/hop-user-manual/modules/ROOT/pages/metadata-types/data-stream/arrow-flight.adoc
index daf25ad093..69c8d81494 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/metadata-types/data-stream/arrow-flight.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/metadata-types/data-stream/arrow-flight.adoc
@@ -50,8 +50,31 @@ The number of rows that Apache Arrow will use per batch.
(default: 500)
| *Maximum Buffer Size* |
The maximum number of rows that will be kept in the buffer on the Hop Flight
server. Set it high enough to avoid losing rows. As a ballpark figure, take the
throughput in rows/s and multiply that by 10 for the buffer size to avoid
issues. (default: 10M)
+| *Hop Flight hostname* |
+The hostname on which the Hop Flight server can be reached. (default:
localhost)
+
+| *Hop Flight port* |
+The port on which the Hop Flight server can be reached. (default: 33333)
+
+| *Use TLS* |
+Connect over TLS instead of plain gRPC. The server needs to be started with a
certificate and a private key, see the
xref:hop-tools/hop-arrow-command.adoc[`hop arrow` command].
+
+| *Verify the server certificate* |
+Verify the certificate the Flight server presents. Only switch this off for
testing with a self-signed certificate. (default: enabled)
+
+| *Trusted certificates file* |
+Optional PEM file with the certificate authority to trust when verifying the
server certificate. Leave it empty to use the JVM trust store.
+
+| *Client certificate file* / *Client private key file* |
+Optional PEM files with the client certificate and its private key, for a
server started with `--arrow-flight-tls-client-ca` (mutual TLS). Specify both
or neither.
+
+| *Username* / *Password* |
+The credentials to authenticate with, for a server started with
`--arrow-flight-username`. Leave the username empty when the server does not
require authentication.
+
|===
+TIP: The TLS and credential fields are only used by Hop itself when it
connects to the server. The hostname, port and *Use TLS* settings are also what
the server advertises to clients in the endpoint it returns, so they need to
describe how clients reach the server, not how it binds.
+
IMPORTANT: This is an IPC system, not a safe data queue. The purpose is to
hand data over to the receiving party as soon as possible. That is why there
is no blocking happening when we write data to the Flight server. Rows are
kept in memory to avoid stalling the gRPC back-end system as this would cause
data to get lost. The only time we wait is when we read data from the Flight
server. That is why it's recommended to start reading with one process before
you write data with another [...]
== Important Behavior
@@ -59,6 +82,7 @@ IMPORTANT: This is an IPC system, not a safe data queue. The
purpose is to hand
* The **stream name** is simply the name of the Data Stream metadata element.
* Schema validation is strict: the client must send data using exactly the
same schema defined in the **Static Schema** field.
* The Flight server must be started separately using the `hop arrow`
xref:hop-tools/hop-arrow-command.adoc[command].
+* Transport security and authentication are off unless the server is started
with the matching options. Without them the data travels in the clear and
anything that can reach the port can read it.
== Related
@@ -76,6 +100,15 @@ import pyarrow as pa
# Connect to your Flight server
client = flight.FlightClient("grpc://localhost:33333") # or
"grpc://0.0.0.0:33333"
+# On a server started with TLS and credentials, connect like this instead:
+#
+# with open("server.crt", "rb") as f:
+# client = flight.FlightClient("grpc+tls://localhost:33333",
tls_root_certs=f.read())
+# token_pair = client.authenticate_basic_token(b"hop", b"s3cr3t")
+# options = flight.FlightCallOptions(headers=[token_pair])
+#
+# and pass `options` to get_flight_info() and do_get().
+
# Define the stream name (must match what your server expects: the name of the
Data Stream metadata element)
stream_name = "FlightStream"
diff --git
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/command/ArrowCommand.java
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/command/ArrowCommand.java
index bce73b81c2..9b3d2e127e 100644
---
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/command/ArrowCommand.java
+++
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/command/ArrowCommand.java
@@ -22,10 +22,12 @@ import java.util.Map;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
import lombok.Setter;
+import org.apache.hop.arrow.flight.ArrowFlightSecurity;
import org.apache.hop.arrow.flight.ArrowFlightServer;
import org.apache.hop.core.Const;
import org.apache.hop.core.config.plugin.ConfigPlugin;
import org.apache.hop.core.config.plugin.IConfigOptions;
+import org.apache.hop.core.encryption.Encr;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.logging.ILogChannel;
import org.apache.hop.core.logging.LogChannel;
@@ -64,6 +66,36 @@ public class ArrowCommand implements Runnable, IHopCommand,
IHasHopMetadataProvi
"The port on which the Apache Arrow Flight server will listen,
defaults to 33333")
private String port = "33333";
+ @CommandLine.Option(
+ names = {"--arrow-flight-tls-certificate"},
+ description =
+ "The PEM file with the certificate chain of the Apache Arrow Flight
server. Specify this together with --arrow-flight-tls-key to serve over TLS.")
+ private String tlsCertificateFile;
+
+ @CommandLine.Option(
+ names = {"--arrow-flight-tls-key"},
+ description =
+ "The PEM file with the PKCS#8 private key of the Apache Arrow Flight
server certificate")
+ private String tlsPrivateKeyFile;
+
+ @CommandLine.Option(
+ names = {"--arrow-flight-tls-client-ca"},
+ description =
+ "The PEM file with the certificate authority used to verify client
certificates (mutual TLS). Needs TLS to be enabled.")
+ private String tlsClientCertificateAuthorityFile;
+
+ @CommandLine.Option(
+ names = {"--arrow-flight-username"},
+ description =
+ "The user name clients need to present to the Apache Arrow Flight
server. Without it the server accepts unauthenticated clients.")
+ private String username;
+
+ @CommandLine.Option(
+ names = {"--arrow-flight-password"},
+ description =
+ "The password belonging to --arrow-flight-username. Can be an
encrypted password or a variable expression.")
+ private String password;
+
public ArrowCommand() {
// Nothing specific to set
}
@@ -104,9 +136,17 @@ public class ArrowCommand implements Runnable,
IHopCommand, IHasHopMetadataProvi
String realHostname = Const.NVL(variables.resolve(hostname), "0.0.0.0");
int realPort = Const.toInt(variables.resolve(port), 33333);
+ ArrowFlightSecurity security =
+ new ArrowFlightSecurity(
+ variables.resolve(tlsCertificateFile),
+ variables.resolve(tlsPrivateKeyFile),
+ variables.resolve(tlsClientCertificateAuthorityFile),
+ variables.resolve(username),
+
Encr.decryptPasswordOptionallyEncrypted(variables.resolve(password)));
+
// Start the flight server
ArrowFlightServer server =
- new ArrowFlightServer(realHostname, realPort, variables,
metadataProvider, log);
+ new ArrowFlightServer(realHostname, realPort, security, variables,
metadataProvider, log);
server.start();
// For now, we'll wait indefinitely
diff --git
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStream.java
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStream.java
index 5d0747d473..de2d5478e6 100644
---
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStream.java
+++
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStream.java
@@ -18,11 +18,13 @@
package org.apache.hop.arrow.datastream.flight;
+import java.io.ByteArrayInputStream;
import java.util.ArrayList;
import java.util.List;
import lombok.Getter;
import lombok.Setter;
import org.apache.arrow.flight.AsyncPutListener;
+import org.apache.arrow.flight.CallOption;
import org.apache.arrow.flight.FlightClient;
import org.apache.arrow.flight.FlightClient.ClientStreamListener;
import org.apache.arrow.flight.FlightDescriptor;
@@ -33,12 +35,15 @@ import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.hop.arrow.datastream.shared.ArrowBaseDataStream;
+import org.apache.hop.arrow.flight.ArrowFlightSecurity;
import org.apache.hop.core.Const;
+import org.apache.hop.core.encryption.Encr;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.gui.plugin.GuiElementType;
import org.apache.hop.core.gui.plugin.GuiPlugin;
import org.apache.hop.core.gui.plugin.GuiWidgetElement;
import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.util.Utils;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.datastream.metadata.DataStreamMeta;
import org.apache.hop.datastream.plugin.DataStreamPlugin;
@@ -105,12 +110,77 @@ public class ArrowFlightDataStream extends
ArrowBaseDataStream {
@HopMetadataProperty(key = "port")
protected String port;
+ @GuiWidgetElement(
+ order = "20500-arrow-flight-data-stream-tls",
+ parentId = DataStreamMeta.GUI_WIDGETS_PARENT_ID,
+ type = GuiElementType.CHECKBOX,
+ label = "i18n::ArrowFlightDataStream.Tls.Label",
+ toolTip = "i18n::ArrowFlightDataStream.Tls.Tooltip")
+ @HopMetadataProperty(key = "tls")
+ protected boolean tls;
+
+ @GuiWidgetElement(
+ order = "20600-arrow-flight-data-stream-verify-server",
+ parentId = DataStreamMeta.GUI_WIDGETS_PARENT_ID,
+ type = GuiElementType.CHECKBOX,
+ label = "i18n::ArrowFlightDataStream.VerifyServer.Label",
+ toolTip = "i18n::ArrowFlightDataStream.VerifyServer.Tooltip")
+ @HopMetadataProperty(key = "verifyServer", defaultBoolean = true)
+ protected boolean verifyServer;
+
+ @GuiWidgetElement(
+ order = "20700-arrow-flight-data-stream-trusted-certificates",
+ parentId = DataStreamMeta.GUI_WIDGETS_PARENT_ID,
+ type = GuiElementType.FILENAME,
+ label = "i18n::ArrowFlightDataStream.TrustedCertificates.Label",
+ toolTip = "i18n::ArrowFlightDataStream.TrustedCertificates.Tooltip")
+ @HopMetadataProperty(key = "trustedCertificatesFile")
+ protected String trustedCertificatesFile;
+
+ @GuiWidgetElement(
+ order = "20800-arrow-flight-data-stream-client-certificate",
+ parentId = DataStreamMeta.GUI_WIDGETS_PARENT_ID,
+ type = GuiElementType.FILENAME,
+ label = "i18n::ArrowFlightDataStream.ClientCertificate.Label",
+ toolTip = "i18n::ArrowFlightDataStream.ClientCertificate.Tooltip")
+ @HopMetadataProperty(key = "clientCertificateFile")
+ protected String clientCertificateFile;
+
+ @GuiWidgetElement(
+ order = "20900-arrow-flight-data-stream-client-key",
+ parentId = DataStreamMeta.GUI_WIDGETS_PARENT_ID,
+ type = GuiElementType.FILENAME,
+ label = "i18n::ArrowFlightDataStream.ClientKey.Label",
+ toolTip = "i18n::ArrowFlightDataStream.ClientKey.Tooltip")
+ @HopMetadataProperty(key = "clientKeyFile")
+ protected String clientKeyFile;
+
+ @GuiWidgetElement(
+ order = "21000-arrow-flight-data-stream-username",
+ parentId = DataStreamMeta.GUI_WIDGETS_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label = "i18n::ArrowFlightDataStream.Username.Label",
+ toolTip = "i18n::ArrowFlightDataStream.Username.Tooltip")
+ @HopMetadataProperty(key = "username")
+ protected String username;
+
+ @GuiWidgetElement(
+ order = "21100-arrow-flight-data-stream-password",
+ parentId = DataStreamMeta.GUI_WIDGETS_PARENT_ID,
+ type = GuiElementType.TEXT,
+ password = true,
+ label = "i18n::ArrowFlightDataStream.Password.Label",
+ toolTip = "i18n::ArrowFlightDataStream.Password.Tooltip")
+ @HopMetadataProperty(key = "password", password = true)
+ protected String password;
+
private int realBufferSize;
private int realBatchSize;
private String realHostname;
private int realPort;
private SchemaDefinition schemaDefinition;
private FlightClient flightClient;
+ private CallOption[] callOptions;
private ClientStreamListener clientStreamListener;
private FlightInfo readFlightInfo;
private FlightStream readFlightStream;
@@ -129,6 +199,8 @@ public class ArrowFlightDataStream extends
ArrowBaseDataStream {
batchSize = "10000";
hostname = "localhost";
port = "33333";
+ verifyServer = true;
+ callOptions = new CallOption[0];
}
@SuppressWarnings("CopyConstructorMissesField")
@@ -137,6 +209,15 @@ public class ArrowFlightDataStream extends
ArrowBaseDataStream {
this.bufferSize = s.bufferSize;
this.batchSize = s.batchSize;
this.schemaDefinitionName = s.schemaDefinitionName;
+ this.hostname = s.hostname;
+ this.port = s.port;
+ this.tls = s.tls;
+ this.verifyServer = s.verifyServer;
+ this.trustedCertificatesFile = s.trustedCertificatesFile;
+ this.clientCertificateFile = s.clientCertificateFile;
+ this.clientKeyFile = s.clientKeyFile;
+ this.username = s.username;
+ this.password = s.password;
}
@Override
@@ -192,15 +273,58 @@ public class ArrowFlightDataStream extends
ArrowBaseDataStream {
flightClient.startPut(
FlightDescriptor.path(dataStreamMeta.getName()),
vectorSchemaRoot,
- new AsyncPutListener());
+ new AsyncPutListener(),
+ callOptions);
}
private void buildFlightClient() throws HopException {
try {
// Get a flight client going.
//
- Location location = Location.forGrpcInsecure(realHostname, realPort);
- flightClient = FlightClient.builder(rootAllocator, location).build();
+ Location location =
+ tls
+ ? Location.forGrpcTls(realHostname, realPort)
+ : Location.forGrpcInsecure(realHostname, realPort);
+ FlightClient.Builder builder = FlightClient.builder(rootAllocator,
location);
+
+ if (tls) {
+ builder.verifyServer(verifyServer);
+
+ String realTrustedCertificates =
variables.resolve(trustedCertificatesFile);
+ if (!Utils.isEmpty(realTrustedCertificates)) {
+ builder.trustedCertificates(
+ new
ByteArrayInputStream(ArrowFlightSecurity.readPemFile(realTrustedCertificates)));
+ }
+
+ String realClientCertificate =
variables.resolve(clientCertificateFile);
+ String realClientKey = variables.resolve(clientKeyFile);
+ if (!Utils.isEmpty(realClientCertificate) ||
!Utils.isEmpty(realClientKey)) {
+ if (Utils.isEmpty(realClientCertificate) ||
Utils.isEmpty(realClientKey)) {
+ throw new HopException(
+ "Please specify both a client certificate and a client key to
connect to the Flight server with mutual TLS.");
+ }
+ builder.clientCertificate(
+ new
ByteArrayInputStream(ArrowFlightSecurity.readPemFile(realClientCertificate)),
+ new
ByteArrayInputStream(ArrowFlightSecurity.readPemFile(realClientKey)));
+ }
+ }
+
+ flightClient = builder.build();
+
+ // Authenticate if the server asks us to. We get a bearer token back
which we then pass
+ // along with every call we make.
+ //
+ String realUsername = variables.resolve(username);
+ if (!Utils.isEmpty(realUsername)) {
+ String realPassword =
Encr.decryptPasswordOptionallyEncrypted(variables.resolve(password));
+ callOptions =
+ flightClient
+ .authenticateBasicToken(realUsername, Const.NVL(realPassword,
""))
+ .map(option -> new CallOption[] {option})
+ .orElseGet(() -> new CallOption[0]);
+ } else {
+ callOptions = new CallOption[0];
+ }
} catch (Exception e) {
throw new HopException(
"Error connecting to Flight server " + realHostname + ":" +
realPort, e);
@@ -296,12 +420,14 @@ public class ArrowFlightDataStream extends
ArrowBaseDataStream {
private void initializeStreamReading() throws HopException {
buildFlightClient();
- readFlightInfo =
flightClient.getInfo(FlightDescriptor.path(dataStreamMeta.getName()));
+ readFlightInfo =
+ flightClient.getInfo(FlightDescriptor.path(dataStreamMeta.getName()),
callOptions);
if (readFlightInfo.getEndpoints().isEmpty()) {
throw new HopException(
"No endpoint tickets found in flight server matching " +
dataStreamMeta.getName());
}
- readFlightStream =
flightClient.getStream(readFlightInfo.getEndpoints().get(0).getTicket());
+ readFlightStream =
+
flightClient.getStream(readFlightInfo.getEndpoints().get(0).getTicket(),
callOptions);
readVectorSchemaRoot = readFlightStream.getRoot();
readSchema = readVectorSchemaRoot.getSchema();
this.rowMeta = buildRowMeta(readSchema);
diff --git
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/ArrowFlightSecurity.java
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/ArrowFlightSecurity.java
new file mode 100644
index 0000000000..b526a55e74
--- /dev/null
+++
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/ArrowFlightSecurity.java
@@ -0,0 +1,163 @@
+/*
+ * 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.hop.arrow.flight;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.apache.arrow.flight.CallStatus;
+import org.apache.arrow.flight.auth2.BasicCallHeaderAuthenticator;
+import org.apache.arrow.flight.auth2.CallHeaderAuthenticator;
+import org.apache.arrow.flight.auth2.GeneratedBearerTokenAuthenticator;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.util.Utils;
+import org.apache.hop.core.vfs.HopVfs;
+
+/**
+ * The transport security and authentication settings of the Hop Apache Arrow
Flight server. Both
+ * are optional and disabled when nothing is configured, which keeps the plain
gRPC behavior of
+ * earlier versions.
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class ArrowFlightSecurity {
+
+ /** Neither TLS nor authentication: plain gRPC on an open port. */
+ public static final ArrowFlightSecurity NONE = new ArrowFlightSecurity();
+
+ /** The PEM file with the server certificate chain. Enables TLS together
with the private key. */
+ private String certificateFile;
+
+ /** The PEM file with the (PKCS#8) private key belonging to the server
certificate. */
+ private String privateKeyFile;
+
+ /**
+ * The PEM file with the certificate authority used to verify client
certificates. Enables mutual
+ * TLS.
+ */
+ private String clientCertificateAuthorityFile;
+
+ /** The user name clients need to present, or empty to accept
unauthenticated clients. */
+ private String username;
+
+ /** The password belonging to {@link #username}. */
+ private String password;
+
+ public boolean isTlsEnabled() {
+ return !Utils.isEmpty(certificateFile) || !Utils.isEmpty(privateKeyFile);
+ }
+
+ public boolean isMutualTlsEnabled() {
+ return !Utils.isEmpty(clientCertificateAuthorityFile);
+ }
+
+ public boolean isAuthenticationEnabled() {
+ return !Utils.isEmpty(username);
+ }
+
+ /**
+ * Verify that the combination of options makes sense and warn about the
ones that are legal but
+ * unwise.
+ *
+ * @param log the channel to log warnings on
+ * @throws HopException in case the options can't be used to start a server
+ */
+ public void validate(ILogChannel log) throws HopException {
+ if (!Utils.isEmpty(certificateFile) && Utils.isEmpty(privateKeyFile)) {
+ throw new HopException(
+ "Please also specify the TLS private key file for the Arrow Flight
server certificate.");
+ }
+ if (Utils.isEmpty(certificateFile) && !Utils.isEmpty(privateKeyFile)) {
+ throw new HopException(
+ "Please also specify the TLS certificate file for the Arrow Flight
server private key.");
+ }
+ if (isMutualTlsEnabled() && !isTlsEnabled()) {
+ throw new HopException(
+ "Client certificate verification needs TLS: please also specify a
certificate and private key for the Arrow Flight server.");
+ }
+ if (isAuthenticationEnabled() && Utils.isEmpty(password)) {
+ throw new HopException(
+ "Please also specify a password for Arrow Flight server user '" +
username + "'.");
+ }
+
+ if (!isTlsEnabled()) {
+ if (isAuthenticationEnabled()) {
+ log.logMinimal(
+ "WARNING: the Arrow Flight server is not using TLS: credentials
and data are sent in the clear.");
+ } else {
+ log.logMinimal(
+ "WARNING: the Arrow Flight server has no TLS and no authentication
configured: anything that can reach the port can read the data it streams.");
+ }
+ }
+ }
+
+ /**
+ * Builds the authenticator which validates the credentials clients present
and hands them a
+ * bearer token for the calls that follow.
+ *
+ * @return the authenticator, never null when {@link
#isAuthenticationEnabled()}
+ */
+ public CallHeaderAuthenticator createAuthenticator() {
+ String expectedUsername = username;
+ String expectedPassword = password;
+ return new GeneratedBearerTokenAuthenticator(
+ new BasicCallHeaderAuthenticator(
+ (user, pass) -> {
+ if (equalsConstantTime(expectedUsername, user)
+ && equalsConstantTime(expectedPassword, pass)) {
+ return () -> user;
+ }
+ throw CallStatus.UNAUTHENTICATED
+ .withDescription("Invalid credentials for the Hop Arrow
Flight server")
+ .toRuntimeException();
+ }));
+ }
+
+ /**
+ * Reads a certificate or key file through Apache VFS. The content is kept
in memory: these files
+ * are small and the Flight builders only consume the streams when the
server or client is built.
+ *
+ * @param filename the VFS filename to read
+ * @return the file content
+ * @throws HopException in case the file could not be read
+ */
+ public static byte[] readPemFile(String filename) throws HopException {
+ try (InputStream inputStream = HopVfs.getInputStream(filename)) {
+ return inputStream.readAllBytes();
+ } catch (IOException e) {
+ throw new HopException("Unable to read certificate or key file '" +
filename + "'", e);
+ }
+ }
+
+ private static boolean equalsConstantTime(String expected, String actual) {
+ if (expected == null || actual == null) {
+ return false;
+ }
+ return MessageDigest.isEqual(
+ expected.getBytes(StandardCharsets.UTF_8),
actual.getBytes(StandardCharsets.UTF_8));
+ }
+}
diff --git
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/ArrowFlightServer.java
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/ArrowFlightServer.java
index c45a5398de..981605b8c9 100644
---
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/ArrowFlightServer.java
+++
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/ArrowFlightServer.java
@@ -18,6 +18,9 @@
package org.apache.hop.arrow.flight;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
import lombok.Getter;
import lombok.Setter;
import org.apache.arrow.flight.FlightProducer;
@@ -36,6 +39,7 @@ public class ArrowFlightServer {
private int port;
private final IVariables variables;
private final IHopMetadataProvider metadataProvider;
+ private final ArrowFlightSecurity security;
private final RootAllocator allocator;
private final Location location;
@@ -50,24 +54,87 @@ public class ArrowFlightServer {
IHopMetadataProvider metadataProvider,
ILogChannel log)
throws HopException {
+ this(hostname, port, ArrowFlightSecurity.NONE, variables,
metadataProvider, log);
+ }
+
+ public ArrowFlightServer(
+ String hostname,
+ int port,
+ ArrowFlightSecurity security,
+ IVariables variables,
+ IHopMetadataProvider metadataProvider,
+ ILogChannel log)
+ throws HopException {
this.hostname = hostname;
this.port = port;
+ this.security = security == null ? ArrowFlightSecurity.NONE : security;
this.variables = variables;
this.metadataProvider = metadataProvider;
this.log = log;
+ this.security.validate(log);
+
this.allocator = new RootAllocator();
- this.location = Location.forGrpcInsecure(hostname, port);
+ this.location =
+ this.security.isTlsEnabled()
+ ? Location.forGrpcTls(hostname, port)
+ : Location.forGrpcInsecure(hostname, port);
FlightProducer producer = new HopFlightProducer(variables,
metadataProvider, allocator);
- flightServer = FlightServer.builder(allocator, location, producer).build();
+ flightServer = buildFlightServer(producer);
+ }
+
+ private FlightServer buildFlightServer(FlightProducer producer) throws
HopException {
+ FlightServer.Builder builder = FlightServer.builder(allocator, location,
producer);
+
+ if (security.isTlsEnabled()) {
+ byte[] certificate =
ArrowFlightSecurity.readPemFile(security.getCertificateFile());
+ byte[] privateKey =
ArrowFlightSecurity.readPemFile(security.getPrivateKeyFile());
+ try (InputStream certificateStream = new
ByteArrayInputStream(certificate);
+ InputStream privateKeyStream = new ByteArrayInputStream(privateKey))
{
+ builder.useTls(certificateStream, privateKeyStream);
+ } catch (IOException e) {
+ throw new HopException(
+ "Unable to configure TLS for the Flight server on " + hostname +
":" + port, e);
+ }
+
+ if (security.isMutualTlsEnabled()) {
+ byte[] clientCa =
+
ArrowFlightSecurity.readPemFile(security.getClientCertificateAuthorityFile());
+ try (InputStream clientCaStream = new ByteArrayInputStream(clientCa)) {
+ builder.useMTlsClientVerification(clientCaStream);
+ } catch (IOException e) {
+ throw new HopException(
+ "Unable to configure client certificate verification for the
Flight server on "
+ + hostname
+ + ":"
+ + port,
+ e);
+ }
+ }
+ }
+
+ if (security.isAuthenticationEnabled()) {
+ builder.headerAuthenticator(security.createAuthenticator());
+ }
+
+ return builder.build();
}
public void start() throws HopException {
try {
flightServer.start();
- log.logBasic("Apache Arrow Flight server listening on " + hostname + ":"
+ port);
+ log.logBasic(
+ "Apache Arrow Flight server listening on "
+ + hostname
+ + ":"
+ + port
+ + " (TLS: "
+ + (security.isTlsEnabled() ? "enabled" : "disabled")
+ + ", authentication: "
+ + (security.isAuthenticationEnabled() ? "enabled" : "disabled")
+ + ")");
} catch (Exception e) {
throw new HopException("Unable to start Flight server on " + hostname +
":" + port, e);
}
diff --git
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/HopFlightProducer.java
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/HopFlightProducer.java
index e72f7df2ad..30cfef7a48 100644
---
a/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/HopFlightProducer.java
+++
b/plugins/tech/arrow/src/main/java/org/apache/hop/arrow/flight/HopFlightProducer.java
@@ -170,7 +170,12 @@ public class HopFlightProducer extends NoOpFlightProducer {
String hostname =
Const.NVL(variables.resolve(flightDataStream.getHostname()), "0.0.0.0");
int port = Const.toInt(variables.resolve(flightDataStream.getPort()),
33333);
- Location location = Location.forGrpcInsecure(hostname, port);
+ // The endpoint we hand back needs the same scheme clients use to reach
this server.
+ //
+ Location location =
+ flightDataStream.isTls()
+ ? Location.forGrpcTls(hostname, port)
+ : Location.forGrpcInsecure(hostname, port);
buffer =
new FlightStreamBuffer(expectedSchema, rowMeta, rowSet, bufferSize,
batchSize, location);
streamMap.put(streamName, buffer);
diff --git
a/plugins/tech/arrow/src/main/resources/org/apache/hop/arrow/datastream/flight/messages/messages_en_US.properties
b/plugins/tech/arrow/src/main/resources/org/apache/hop/arrow/datastream/flight/messages/messages_en_US.properties
index cf733ffa00..a82095d4c6 100644
---
a/plugins/tech/arrow/src/main/resources/org/apache/hop/arrow/datastream/flight/messages/messages_en_US.properties
+++
b/plugins/tech/arrow/src/main/resources/org/apache/hop/arrow/datastream/flight/messages/messages_en_US.properties
@@ -25,3 +25,17 @@ ArrowFlightDataStream.Hostname.Label=Hop Flight hostname
ArrowFlightDataStream.Hostname.Tooltip=The hostname on which we can reach the
Hop Apache Arrow Flight server
ArrowFlightDataStream.Port.Label=Hop Flight port
ArrowFlightDataStream.Port.Tooltip=The port on which we can reach the Hop
Apache Arrow Flight server
+ArrowFlightDataStream.Tls.Label=Use TLS
+ArrowFlightDataStream.Tls.Tooltip=Connect to the Hop Apache Arrow Flight
server over TLS. The server needs to be started with a certificate and a
private key.
+ArrowFlightDataStream.VerifyServer.Label=Verify the server certificate
+ArrowFlightDataStream.VerifyServer.Tooltip=Verify the certificate the Flight
server presents. Only switch this off for testing with a self-signed
certificate.
+ArrowFlightDataStream.TrustedCertificates.Label=Trusted certificates file
+ArrowFlightDataStream.TrustedCertificates.Tooltip=Optional PEM file with the
certificate authority to trust when verifying the Flight server certificate.
Leave empty to use the JVM trust store.
+ArrowFlightDataStream.ClientCertificate.Label=Client certificate file
+ArrowFlightDataStream.ClientCertificate.Tooltip=Optional PEM file with the
client certificate to present to the Flight server (mutual TLS).
+ArrowFlightDataStream.ClientKey.Label=Client private key file
+ArrowFlightDataStream.ClientKey.Tooltip=The PEM file with the private key
belonging to the client certificate (mutual TLS).
+ArrowFlightDataStream.Username.Label=Username
+ArrowFlightDataStream.Username.Tooltip=The user name to authenticate with on
the Flight server. Leave empty when the server does not require authentication.
+ArrowFlightDataStream.Password.Label=Password
+ArrowFlightDataStream.Password.Tooltip=The password to authenticate with on
the Flight server.
diff --git
a/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStreamAuthenticationTest.java
b/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStreamAuthenticationTest.java
new file mode 100644
index 0000000000..71c1544b1b
--- /dev/null
+++
b/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStreamAuthenticationTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.hop.arrow.datastream.flight;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+
+import java.util.List;
+import org.apache.hop.arrow.flight.ArrowFlightSecurity;
+import org.apache.hop.arrow.flight.ArrowFlightServer;
+import org.apache.hop.core.HopClientEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.datastream.metadata.DataStreamMeta;
+import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
+import org.apache.hop.staticschema.metadata.SchemaDefinition;
+import org.apache.hop.staticschema.metadata.SchemaFieldDefinition;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Round trips rows through an authenticated Flight server using the data
stream itself, so the
+ * credentials on the Hop client side are exercised on all three calls it
makes: getFlightInfo,
+ * doPut and doGet.
+ */
+class ArrowFlightDataStreamAuthenticationTest {
+
+ private static final String STREAM_NAME = "secured-stream";
+ private static final String SCHEMA_NAME = "secured-schema";
+ private static final String USERNAME = "hop";
+ private static final String PASSWORD = "s3cr3t";
+
+ private final IVariables variables = new Variables();
+ private final ILogChannel log = mock(ILogChannel.class);
+
+ private MemoryMetadataProvider metadataProvider;
+ private ArrowFlightServer server;
+ private ArrowFlightDataStream writer;
+ private ArrowFlightDataStream reader;
+
+ @BeforeAll
+ static void initHop() throws Exception {
+ HopClientEnvironment.init();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (writer != null) {
+ writer.close();
+ }
+ if (reader != null) {
+ reader.close();
+ }
+ if (server != null) {
+ server.shutdown();
+ server.getFlightServer().awaitTermination();
+ }
+ }
+
+ /**
+ * Starts a secured server and registers the metadata the producer and the
client both read. The
+ * port is only known after the server is up, so it's set on the data stream
afterwards: both
+ * sides read it lazily.
+ */
+ private ArrowFlightDataStream startServerAndRegisterMetadata(String
password) throws Exception {
+ metadataProvider = new MemoryMetadataProvider();
+
+ SchemaDefinition schemaDefinition = new SchemaDefinition();
+ schemaDefinition.setName(SCHEMA_NAME);
+ schemaDefinition.setFieldDefinitions(
+ List.of(
+ new SchemaFieldDefinition("id", "Integer"),
+ new SchemaFieldDefinition("name", "String")));
+
metadataProvider.getSerializer(SchemaDefinition.class).save(schemaDefinition);
+
+ ArrowFlightDataStream dataStream = new ArrowFlightDataStream();
+ dataStream.setSchemaDefinitionName(SCHEMA_NAME);
+ dataStream.setHostname("localhost");
+ dataStream.setUsername(USERNAME);
+ dataStream.setPassword(password);
+
+ DataStreamMeta dataStreamMeta = new DataStreamMeta();
+ dataStreamMeta.setName(STREAM_NAME);
+ dataStreamMeta.setDataStream(dataStream);
+ metadataProvider.getSerializer(DataStreamMeta.class).save(dataStreamMeta);
+
+ ArrowFlightSecurity security = new ArrowFlightSecurity();
+ security.setUsername(USERNAME);
+ security.setPassword(PASSWORD);
+
+ server = new ArrowFlightServer("localhost", 0, security, variables,
metadataProvider, log);
+ server.start();
+ dataStream.setPort(Integer.toString(server.getFlightServer().getPort()));
+
+ return dataStream;
+ }
+
+ private DataStreamMeta loadStreamMeta() throws HopException {
+ return
metadataProvider.getSerializer(DataStreamMeta.class).load(STREAM_NAME);
+ }
+
+ @Test
+ void rowsRoundTripThroughAnAuthenticatedServer() throws Exception {
+ startServerAndRegisterMetadata(PASSWORD);
+ DataStreamMeta dataStreamMeta = loadStreamMeta();
+ IRowMeta rowMeta =
+
metadataProvider.getSerializer(SchemaDefinition.class).load(SCHEMA_NAME).getRowMeta();
+
+ // Write a handful of rows. This authenticates and then does a doPut.
+ //
+ writer = (ArrowFlightDataStream) dataStreamMeta.getDataStream();
+ writer.initialize(variables, metadataProvider, true, dataStreamMeta);
+ writer.setRowMeta(rowMeta);
+ for (int i = 0; i < 5; i++) {
+ writer.writeRow(new Object[] {(long) i, "row-" + i});
+ }
+ writer.setOutputDone();
+
+ // Read them back. This authenticates again and then does a getFlightInfo
and a doGet.
+ //
+ DataStreamMeta readMeta = loadStreamMeta();
+ reader = (ArrowFlightDataStream) readMeta.getDataStream();
+ reader.setPort(Integer.toString(server.getFlightServer().getPort()));
+ reader.initialize(variables, metadataProvider, false, readMeta);
+
+ IRowMeta readRowMeta = reader.getRowMeta();
+ assertEquals(2, readRowMeta.size());
+
+ for (int i = 0; i < 5; i++) {
+ Object[] row = reader.readRow();
+ assertEquals((long) i, row[0]);
+ assertEquals("row-" + i, row[1]);
+ }
+ assertNull(reader.readRow());
+ }
+
+ @Test
+ void theWrongPasswordIsRefusedByTheServer() throws Exception {
+ startServerAndRegisterMetadata("not-the-password");
+ DataStreamMeta dataStreamMeta = loadStreamMeta();
+ IRowMeta rowMeta =
+
metadataProvider.getSerializer(SchemaDefinition.class).load(SCHEMA_NAME).getRowMeta();
+
+ writer = (ArrowFlightDataStream) dataStreamMeta.getDataStream();
+ writer.initialize(variables, metadataProvider, true, dataStreamMeta);
+
+ assertThrows(HopException.class, () -> writer.setRowMeta(rowMeta));
+ }
+}
diff --git
a/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStreamSerializationTest.java
b/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStreamSerializationTest.java
new file mode 100644
index 0000000000..ac31abf761
--- /dev/null
+++
b/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/datastream/flight/ArrowFlightDataStreamSerializationTest.java
@@ -0,0 +1,178 @@
+/*
+ * 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.hop.arrow.datastream.flight;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.stream.Stream;
+import org.apache.hop.core.HopClientEnvironment;
+import org.apache.hop.core.encryption.Encr;
+import org.apache.hop.core.encryption.HopTwoWayPasswordEncoder;
+import org.apache.hop.core.plugins.PluginRegistry;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.datastream.metadata.DataStreamMeta;
+import org.apache.hop.datastream.plugin.DataStreamPlugin;
+import org.apache.hop.datastream.plugin.DataStreamPluginType;
+import org.apache.hop.metadata.api.IHopMetadataSerializer;
+import org.apache.hop.metadata.serializer.json.JsonMetadataProvider;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * The TLS and credential settings are only useful if they survive a save and
a load, and the
+ * password must not sit in the metadata in plain text.
+ */
+class ArrowFlightDataStreamSerializationTest {
+
+ private static final String STREAM_NAME = "secured-stream";
+
+ @TempDir private Path metadataFolder;
+
+ private JsonMetadataProvider metadataProvider;
+
+ @BeforeAll
+ static void initHop() throws Exception {
+ HopClientEnvironment.init();
+ // The data stream plugin isn't picked up by annotation scanning in a
plain unit test.
+ PluginRegistry.getInstance()
+ .registerPluginClass(
+ ArrowFlightDataStream.class.getName(),
+ DataStreamPluginType.class,
+ DataStreamPlugin.class);
+ }
+
+ @BeforeEach
+ void setUp() {
+ metadataProvider =
+ new JsonMetadataProvider(
+ new HopTwoWayPasswordEncoder(),
+ metadataFolder.toString(),
+ Variables.getADefaultVariableSpace());
+ }
+
+ private DataStreamMeta saveStream(ArrowFlightDataStream dataStream) throws
Exception {
+ DataStreamMeta meta = new DataStreamMeta();
+ meta.setName(STREAM_NAME);
+ meta.setDataStream(dataStream);
+ IHopMetadataSerializer<DataStreamMeta> serializer =
+ metadataProvider.getSerializer(DataStreamMeta.class);
+ serializer.save(meta);
+ return serializer.load(STREAM_NAME);
+ }
+
+ private String savedJson() throws Exception {
+ try (Stream<Path> files = Files.walk(metadataFolder)) {
+ Path file =
+ files
+ .filter(p -> p.toString().endsWith(".json"))
+ .max(Comparator.comparing(Path::toString))
+ .orElseThrow();
+ return Files.readString(file, StandardCharsets.UTF_8);
+ }
+ }
+
+ @Test
+ void everySecuritySettingSurvivesASaveAndLoad() throws Exception {
+ ArrowFlightDataStream dataStream = new ArrowFlightDataStream();
+ dataStream.setHostname("flight.example.com");
+ dataStream.setPort("44444");
+ dataStream.setTls(true);
+ dataStream.setVerifyServer(false);
+ dataStream.setTrustedCertificatesFile("/etc/hop/ca.crt");
+ dataStream.setClientCertificateFile("/etc/hop/client.crt");
+ dataStream.setClientKeyFile("/etc/hop/client.key");
+ dataStream.setUsername("hop");
+ dataStream.setPassword("s3cr3t");
+
+ ArrowFlightDataStream loaded = (ArrowFlightDataStream)
saveStream(dataStream).getDataStream();
+
+ assertEquals("flight.example.com", loaded.getHostname());
+ assertEquals("44444", loaded.getPort());
+ assertTrue(loaded.isTls());
+ assertFalse(loaded.isVerifyServer());
+ assertEquals("/etc/hop/ca.crt", loaded.getTrustedCertificatesFile());
+ assertEquals("/etc/hop/client.crt", loaded.getClientCertificateFile());
+ assertEquals("/etc/hop/client.key", loaded.getClientKeyFile());
+ assertEquals("hop", loaded.getUsername());
+ assertEquals("s3cr3t", loaded.getPassword());
+ }
+
+ @Test
+ void thePasswordIsNotStoredInPlainText() throws Exception {
+ ArrowFlightDataStream dataStream = new ArrowFlightDataStream();
+ dataStream.setUsername("hop");
+ dataStream.setPassword("s3cr3t");
+ saveStream(dataStream);
+
+ String json = savedJson();
+ assertFalse(json.contains("s3cr3t"), "The password was written to the
metadata in plain text");
+
assertTrue(json.contains(Encr.encryptPasswordIfNotUsingVariables("s3cr3t")));
+ }
+
+ @Test
+ void aVariableIsStoredAsIsSoItStaysResolvable() throws Exception {
+ ArrowFlightDataStream dataStream = new ArrowFlightDataStream();
+ dataStream.setUsername("hop");
+ dataStream.setPassword("${FLIGHT_PASSWORD}");
+ saveStream(dataStream);
+
+ assertTrue(savedJson().contains("${FLIGHT_PASSWORD}"));
+ }
+
+ /**
+ * A stream saved before this change has none of the new keys. It has to
keep working, which means
+ * plain gRPC, no credentials, and server verification left on rather than
silently off.
+ */
+ @Test
+ void aStreamSavedBeforeTheSecurityOptionsExistedStillLoads() throws
Exception {
+ Path folder = metadataFolder.resolve("data-stream");
+ Files.createDirectories(folder);
+ Files.writeString(
+ folder.resolve(STREAM_NAME + ".json"),
+ """
+ {
+ "dataStream": {
+ "ArrowFlightStream": {
+ "batchSize": "10000",
+ "bufferSize": "10000000",
+ "hostname": "localhost",
+ "port": "33333",
+ "schemaDefinition": "some-schema"
+ }
+ }
+ }
+ """);
+
+ DataStreamMeta meta =
metadataProvider.getSerializer(DataStreamMeta.class).load(STREAM_NAME);
+ ArrowFlightDataStream loaded = (ArrowFlightDataStream)
meta.getDataStream();
+
+ assertFalse(loaded.isTls(), "An existing stream must not suddenly expect
TLS");
+ assertTrue(loaded.isVerifyServer(), "Server verification must default to
on, not off");
+ assertEquals("localhost", loaded.getHostname());
+ assertEquals("33333", loaded.getPort());
+ }
+}
diff --git
a/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/flight/ArrowFlightSecurityTest.java
b/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/flight/ArrowFlightSecurityTest.java
new file mode 100644
index 0000000000..c900a054ca
--- /dev/null
+++
b/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/flight/ArrowFlightSecurityTest.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.hop.arrow.flight;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.vfs.HopVfs;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class ArrowFlightSecurityTest {
+
+ @TempDir private java.nio.file.Path tempDir;
+
+ private final ILogChannel log = mock(ILogChannel.class);
+
+ @Test
+ void noOptionsMeansNoTlsAndNoAuthentication() throws HopException {
+ ArrowFlightSecurity security = new ArrowFlightSecurity();
+
+ assertFalse(security.isTlsEnabled());
+ assertFalse(security.isMutualTlsEnabled());
+ assertFalse(security.isAuthenticationEnabled());
+ assertDoesNotThrow(() -> security.validate(log));
+ }
+
+ @Test
+ void certificateAndKeyTogetherEnableTls() throws HopException {
+ ArrowFlightSecurity security = new ArrowFlightSecurity();
+ security.setCertificateFile("/tmp/server.crt");
+ security.setPrivateKeyFile("/tmp/server.key");
+
+ assertTrue(security.isTlsEnabled());
+ security.validate(log);
+ }
+
+ @Test
+ void aCertificateWithoutAKeyIsRejected() {
+ ArrowFlightSecurity security = new ArrowFlightSecurity();
+ security.setCertificateFile("/tmp/server.crt");
+
+ assertThrows(HopException.class, () -> security.validate(log));
+ }
+
+ @Test
+ void aKeyWithoutACertificateIsRejected() {
+ ArrowFlightSecurity security = new ArrowFlightSecurity();
+ security.setPrivateKeyFile("/tmp/server.key");
+
+ assertThrows(HopException.class, () -> security.validate(log));
+ }
+
+ @Test
+ void clientCertificateVerificationWithoutTlsIsRejected() {
+ ArrowFlightSecurity security = new ArrowFlightSecurity();
+ security.setClientCertificateAuthorityFile("/tmp/ca.crt");
+
+ assertTrue(security.isMutualTlsEnabled());
+ assertThrows(HopException.class, () -> security.validate(log));
+ }
+
+ @Test
+ void aUsernameWithoutAPasswordIsRejected() {
+ ArrowFlightSecurity security = new ArrowFlightSecurity();
+ security.setUsername("hop");
+
+ assertTrue(security.isAuthenticationEnabled());
+ assertThrows(HopException.class, () -> security.validate(log));
+ }
+
+ @Test
+ void certificateFilesAreReadThroughVfs() throws Exception {
+ String filename = tempDir.resolve("certificate.pem").toString();
+ byte[] content =
+ "-----BEGIN CERTIFICATE-----\nnot a real
one\n".getBytes(StandardCharsets.UTF_8);
+ try (OutputStream outputStream = HopVfs.getOutputStream(filename, false)) {
+ outputStream.write(content);
+ }
+
+ assertArrayEquals(content, ArrowFlightSecurity.readPemFile(filename));
+ }
+
+ @Test
+ void readingAMissingCertificateFileThrows() {
+ assertThrows(
+ HopException.class,
+ () ->
ArrowFlightSecurity.readPemFile(tempDir.resolve("does-not-exist.pem").toString()));
+ }
+}
diff --git
a/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/flight/ArrowFlightServerAuthenticationTest.java
b/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/flight/ArrowFlightServerAuthenticationTest.java
new file mode 100644
index 0000000000..d9ca935811
--- /dev/null
+++
b/plugins/tech/arrow/src/test/java/org/apache/hop/arrow/flight/ArrowFlightServerAuthenticationTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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.hop.arrow.flight;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.util.Optional;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.Criteria;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightRuntimeException;
+import org.apache.arrow.flight.FlightStatusCode;
+import org.apache.arrow.flight.Location;
+import org.apache.arrow.flight.grpc.CredentialCallOption;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Starts a real Flight server and checks that the credentials configured on
it are actually
+ * enforced. These tests use plain gRPC: they cover the authentication half of
the security options,
+ * which is independent of the transport.
+ */
+class ArrowFlightServerAuthenticationTest {
+
+ private static final String USERNAME = "hop";
+ private static final String PASSWORD = "s3cr3t";
+
+ private final ILogChannel log = mock(ILogChannel.class);
+
+ private ArrowFlightServer server;
+ private RootAllocator clientAllocator;
+ private FlightClient client;
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (client != null) {
+ client.close();
+ }
+ if (clientAllocator != null) {
+ clientAllocator.close();
+ }
+ if (server != null) {
+ server.shutdown();
+ server.getFlightServer().awaitTermination();
+ }
+ }
+
+ private FlightClient startServerAndConnect(ArrowFlightSecurity security)
throws Exception {
+ // Port 0 lets the OS pick a free port for us.
+ //
+ server =
+ new ArrowFlightServer(
+ "localhost", 0, security, new Variables(), new
MemoryMetadataProvider(), log);
+ server.start();
+
+ clientAllocator = new RootAllocator();
+ client =
+ FlightClient.builder(
+ clientAllocator,
+ Location.forGrpcInsecure("localhost",
server.getFlightServer().getPort()))
+ .build();
+ return client;
+ }
+
+ private static ArrowFlightSecurity withCredentials() {
+ ArrowFlightSecurity security = new ArrowFlightSecurity();
+ security.setUsername(USERNAME);
+ security.setPassword(PASSWORD);
+ return security;
+ }
+
+ /**
+ * Any call made without credentials. {@code listFlights} is not implemented
by the Hop producer,
+ * so reaching the producer at all shows up as UNIMPLEMENTED rather than
UNAUTHENTICATED.
+ */
+ private static FlightStatusCode callStatusCode(FlightClient client,
CallOption... callOptions) {
+ FlightRuntimeException exception =
+ assertThrows(
+ FlightRuntimeException.class,
+ () -> client.listFlights(Criteria.ALL, callOptions).forEach(info
-> {}));
+ return exception.status().code();
+ }
+
+ @Test
+ void aServerWithoutCredentialsAcceptsAnyClient() throws Exception {
+ FlightClient flightClient =
startServerAndConnect(ArrowFlightSecurity.NONE);
+
+ // The call reaches the producer, which doesn't implement listFlights.
+ //
+ assertEquals(FlightStatusCode.UNIMPLEMENTED, callStatusCode(flightClient));
+ }
+
+ @Test
+ void aServerWithCredentialsRejectsAnonymousClients() throws Exception {
+ FlightClient flightClient = startServerAndConnect(withCredentials());
+
+ assertEquals(FlightStatusCode.UNAUTHENTICATED,
callStatusCode(flightClient));
+ }
+
+ @Test
+ void aServerWithCredentialsRejectsTheWrongPassword() throws Exception {
+ FlightClient flightClient = startServerAndConnect(withCredentials());
+
+ FlightRuntimeException exception =
+ assertThrows(
+ FlightRuntimeException.class,
+ () -> flightClient.authenticateBasicToken(USERNAME, "wrong"));
+ assertEquals(FlightStatusCode.UNAUTHENTICATED, exception.status().code());
+ }
+
+ @Test
+ void aServerWithCredentialsRejectsTheWrongUsername() throws Exception {
+ FlightClient flightClient = startServerAndConnect(withCredentials());
+
+ FlightRuntimeException exception =
+ assertThrows(
+ FlightRuntimeException.class,
+ () -> flightClient.authenticateBasicToken("somebody-else",
PASSWORD));
+ assertEquals(FlightStatusCode.UNAUTHENTICATED, exception.status().code());
+ }
+
+ @Test
+ void aServerWithCredentialsAcceptsTheRightOnes() throws Exception {
+ FlightClient flightClient = startServerAndConnect(withCredentials());
+
+ Optional<CredentialCallOption> bearerToken =
+ flightClient.authenticateBasicToken(USERNAME, PASSWORD);
+ assertTrue(bearerToken.isPresent(), "Expected a bearer token back from the
Flight server");
+
+ CredentialCallOption callOption = bearerToken.get();
+ assertNotNull(callOption);
+
+ // With the token the call gets through to the producer.
+ //
+ assertEquals(FlightStatusCode.UNIMPLEMENTED, callStatusCode(flightClient,
callOption));
+ }
+}
