You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by cd...@apache.org on 2026/06/23 17:14:17 UTC
(plc4x-extras) branch develop updated (96719d6 -> afc56ed)
This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x-extras.git
from 96719d6 chore(deps-dev): bump org.cyclonedx:cyclonedx-maven-plugin (#716)
new 2514ed7 chore: Disabled the s7event module as it's completely out of sync with the refactored driver.
new 0859a6c chore: Made the integration modules build again.
new afc56ed feat: Updated the plc4x server and driver to operate via TLS and to support authentication.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
RELEASE_NOTES | 19 ++
plc4j/examples/pom.xml | 2 +-
plc4j/integrations/apache-calcite/pom.xml | 6 +
plc4j/integrations/apache-nifi/README.md | 54 ++--
plc4j/tools/plc4x-server/pom.xml | 22 +-
.../plc4x/java/tools/plc4xserver/Plc4xServer.java | 285 ++++++++++++++-----
.../java/tools/plc4xserver/ServerTlsContext.java | 153 +++++++++++
.../plc4xserver/protocol/Plc4xServerAdapter.java | 302 ++++++++++++---------
.../protocol/SocketTransportInstance.java | 121 +++++++++
.../java/tools/plc4xserver/Plc4xServerTest.java | 58 ++--
10 files changed, 767 insertions(+), 255 deletions(-)
create mode 100644 plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/ServerTlsContext.java
create mode 100644 plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/protocol/SocketTransportInstance.java
(plc4x-extras) 01/03: chore: Disabled the s7event module as it's completely out of sync with the refactored driver.
Posted by cd...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x-extras.git
commit 2514ed7a6719679bb5ebf885298a352d00c671b1
Author: Christofer Dutz <ch...@c-ware.de>
AuthorDate: Tue Jun 23 19:12:14 2026 +0200
chore: Disabled the s7event module as it's completely out of sync with the refactored driver.
---
plc4j/examples/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/plc4j/examples/pom.xml b/plc4j/examples/pom.xml
index a66dc74..125a81a 100644
--- a/plc4j/examples/pom.xml
+++ b/plc4j/examples/pom.xml
@@ -60,7 +60,7 @@
<module>hello-world-plc4x-subscribe</module>
<module>hello-world-plc4x-write</module>
<module>poll-loop</module>
- <module>plc4j-s7event</module>
+ <!--module>plc4j-s7event</module-->
</modules>
<build>
(plc4x-extras) 03/03: feat: Updated the plc4x server and driver to operate via TLS and to support authentication.
Posted by cd...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x-extras.git
commit afc56ed6b0c457110a32d98251f44bebb6731264
Author: Christofer Dutz <ch...@c-ware.de>
AuthorDate: Tue Jun 23 19:14:09 2026 +0200
feat: Updated the plc4x server and driver to operate via TLS and to support authentication.
---
RELEASE_NOTES | 19 ++
plc4j/tools/plc4x-server/pom.xml | 22 +-
.../plc4x/java/tools/plc4xserver/Plc4xServer.java | 285 ++++++++++++++-----
.../java/tools/plc4xserver/ServerTlsContext.java | 153 +++++++++++
.../plc4xserver/protocol/Plc4xServerAdapter.java | 302 ++++++++++++---------
.../protocol/SocketTransportInstance.java | 121 +++++++++
.../java/tools/plc4xserver/Plc4xServerTest.java | 58 ++--
7 files changed, 733 insertions(+), 227 deletions(-)
diff --git a/RELEASE_NOTES b/RELEASE_NOTES
index 1fc99b6..9dc6606 100644
--- a/RELEASE_NOTES
+++ b/RELEASE_NOTES
@@ -4,9 +4,28 @@
New Features
------------
+- PLC4X Server: Now supports TLS as a transport (the default).
+ When no keystore is configured, the server generates an
+ ephemeral self-signed certificate on startup and logs its
+ SHA-256 fingerprint so clients can pin/trust it. Plaintext TCP
+ remains available as an explicit opt-in
+ (plc4x.server.plaintext=true).
+- PLC4X Server: Now enforces mandatory username/password
+ authentication; no operation is permitted before a successful
+ auth handshake. If no credentials are configured, the server
+ starts with the default user "toddy" and a generated secure
+ password that is printed to the console once at startup.
+ Configured credentials are never logged.
Incompatible changes
--------------------
+- PLC4X Server: Was migrated from the SPI2 (Netty) core to SPI3.
+ It now listens via a plain ServerSocket/SSLServerSocket and no
+ longer depends on Netty.
+- PLC4X Server: Authentication is now mandatory and TLS is the
+ default transport. Existing clients must provide credentials
+ and connect via TLS (or explicitly opt into plaintext). See the
+ matching note in the core 'plc4x' driver release notes.
Bug Fixes
---------
diff --git a/plc4j/tools/plc4x-server/pom.xml b/plc4j/tools/plc4x-server/pom.xml
index 9366bea..42daebe 100644
--- a/plc4j/tools/plc4x-server/pom.xml
+++ b/plc4j/tools/plc4x-server/pom.xml
@@ -120,15 +120,25 @@
<scope>runtime</scope>
</dependency>
+ <!-- SPI3 transport primitives (RingBuffer / TransportInstance) used to adapt accepted
+ sockets to the PLC4X message codec. -->
<dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty-buffer</artifactId>
- <version>4.1.123.Final</version>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-transports-api</artifactId>
+ <version>${plc4x.version}</version>
+ </dependency>
+
+ <!-- Bouncy Castle is used to generate an ephemeral self-signed server certificate when
+ no keystore is configured. -->
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcprov-jdk18on</artifactId>
+ <version>1.84</version>
</dependency>
<dependency>
- <groupId>io.netty</groupId>
- <artifactId>netty-transport</artifactId>
- <version>4.1.123.Final</version>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcpkix-jdk18on</artifactId>
+ <version>1.84</version>
</dependency>
<dependency>
diff --git a/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/Plc4xServer.java b/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/Plc4xServer.java
index 2227ec8..40657f6 100644
--- a/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/Plc4xServer.java
+++ b/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/Plc4xServer.java
@@ -21,54 +21,84 @@ package org.apache.plc4x.java.tools.plc4xserver;
import static java.lang.Runtime.getRuntime;
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
import java.io.IOException;
import java.net.ServerSocket;
-import java.util.Arrays;
+import java.net.Socket;
+import java.security.SecureRandom;
+import java.util.Base64;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import java.util.function.ToIntFunction;
+import javax.net.ssl.SSLServerSocket;
+import javax.net.ssl.SSLServerSocketFactory;
+import org.apache.plc4x.java.api.PlcConnectionManager;
import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
+import org.apache.plc4x.java.plc4x.Plc4xMessageCodec;
import org.apache.plc4x.java.plc4x.readwrite.Constants;
-import org.apache.plc4x.java.plc4x.readwrite.Plc4xMessage;
-import org.apache.plc4x.java.spi.connection.GeneratedProtocolMessageCodec;
-import org.apache.plc4x.java.spi.generation.ByteOrder;
+import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException;
import org.apache.plc4x.java.tools.plc4xserver.protocol.Plc4xServerAdapter;
+import org.apache.plc4x.java.tools.plc4xserver.protocol.SocketTransportInstance;
+import org.apache.plc4x.java.utils.cache.CachedPlcConnectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * TCP/TLS server that relays {@code plc4x} proxy requests to PLCs reachable from this machine.
+ *
+ * <p>Security model:</p>
+ * <ul>
+ * <li><b>Authentication is mandatory.</b> Every connection must present a username and
+ * password before any operation. If none are configured the server uses the default user
+ * {@code toddy} with a freshly generated secure password that is printed to the console once
+ * at startup. Explicitly configured credentials are never logged.</li>
+ * <li><b>TLS is the default transport.</b> Without a configured keystore the server generates
+ * an ephemeral self-signed certificate and prints its SHA-256 fingerprint so clients can pin
+ * it. Plaintext TCP is available as an explicit opt-in for trusted networks/testing.</li>
+ * </ul>
+ */
public class Plc4xServer {
public static final String SERVER_PORT_PROPERTY = "plc4x.server.port";
public static final String SERVER_PORT_ENVIRONMENT_VARIABLE = "PLC4X_SERVER_PORT";
+ public static final String SERVER_USERNAME_PROPERTY = "plc4x.server.username";
+ public static final String SERVER_USERNAME_ENVIRONMENT_VARIABLE = "PLC4X_SERVER_USERNAME";
+ public static final String SERVER_PASSWORD_PROPERTY = "plc4x.server.password";
+ public static final String SERVER_PASSWORD_ENVIRONMENT_VARIABLE = "PLC4X_SERVER_PASSWORD";
+ public static final String SERVER_PLAINTEXT_PROPERTY = "plc4x.server.plaintext";
+ public static final String SERVER_PLAINTEXT_ENVIRONMENT_VARIABLE = "PLC4X_SERVER_PLAINTEXT";
+ public static final String SERVER_KEYSTORE_PROPERTY = "plc4x.server.keystore";
+ public static final String SERVER_KEYSTORE_PASSWORD_PROPERTY = "plc4x.server.keystore-password";
+
+ public static final String DEFAULT_USERNAME = "toddy";
public static int DEFAULT_PORT = Constants.PLC4XTCPDEFAULTPORT;
- private static final Logger LOG = LoggerFactory.getLogger(Plc4xServerAdapter.class);
+ private static final Logger LOG = LoggerFactory.getLogger(Plc4xServer.class);
+
+ private final PlcConnectionManager connectionManager = CachedPlcConnectionManager.getBuilder().build();
- private EventLoopGroup loopGroup;
- private EventLoopGroup workerGroup;
- private ChannelFuture channelFuture;
private Integer port;
+ private String username;
+ private String password;
+ private boolean plaintext = false;
+ private String keystorePath;
+ private String keystorePassword;
+
+ private ServerSocket serverSocket;
+ private Thread acceptThread;
+ private ExecutorService connectionExecutor;
+ private volatile boolean running = false;
public static void main(String[] args) throws Exception {
final Plc4xServer server = new Plc4xServer();
Future<Void> serverFuture = server.start(
- Arrays.stream(args).findFirst() // port number given as first command line argument
- .or(() -> Optional.ofNullable(System.getProperty(SERVER_PORT_PROPERTY)))
- .or(() -> Optional.ofNullable(System.getenv(SERVER_PORT_ENVIRONMENT_VARIABLE)))
- .map(Integer::parseInt)
- .orElse(DEFAULT_PORT)
+ Arrays_findFirst(args) // port number given as first command line argument
+ .or(() -> Optional.ofNullable(System.getProperty(SERVER_PORT_PROPERTY)))
+ .or(() -> Optional.ofNullable(System.getenv(SERVER_PORT_ENVIRONMENT_VARIABLE)))
+ .map(Integer::parseInt)
+ .orElse(DEFAULT_PORT)
);
CompletableFuture<Void> serverRunning = new CompletableFuture<>();
getRuntime().addShutdownHook(new Thread(() -> serverRunning.complete(null)));
@@ -86,89 +116,196 @@ public class Plc4xServer {
}
}
+ private static Optional<String> Arrays_findFirst(String[] args) {
+ return args.length > 0 ? Optional.of(args[0]) : Optional.empty();
+ }
+
public Integer getPort() {
return port;
}
+ /**
+ * The effective username clients must authenticate with.
+ */
+ public String getUsername() {
+ return username;
+ }
+
+ /**
+ * The effective password clients must authenticate with (generated if none was configured).
+ */
+ public String getPassword() {
+ return password;
+ }
+
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public void setPlaintext(boolean plaintext) {
+ this.plaintext = plaintext;
+ }
+
+ public void setKeystore(String keystorePath, String keystorePassword) {
+ this.keystorePath = keystorePath;
+ this.keystorePassword = keystorePassword;
+ }
+
public Future<Void> start() {
return start(0);
}
public Future<Void> start(int port) {
- if (port == 0) {
- this.port = findRandomFreePort();
- } else {
- this.port = port;
- }
-
- if (loopGroup != null) {
+ if (running) {
return CompletableFuture.completedFuture(null);
}
+ this.port = (port == 0) ? findRandomFreePort() : port;
+
+ resolveCredentials();
+ resolvePlaintext();
- loopGroup = new NioEventLoopGroup();
- workerGroup = new NioEventLoopGroup();
+ try {
+ serverSocket = plaintext ? new ServerSocket(this.port) : createTlsServerSocket(this.port);
+ } catch (Exception e) {
+ return CompletableFuture.failedFuture(
+ new PlcRuntimeException("Failed to start PLC4X server", e));
+ }
- channelFuture = new ServerBootstrap()
- .group(loopGroup, workerGroup)
- .channel(NioServerSocketChannel.class)
- .childHandler(new SocketChannelChannelInitializer())
- .option(ChannelOption.SO_BACKLOG, 128)
- .childOption(ChannelOption.SO_KEEPALIVE, true)
- .bind(this.port);
+ running = true;
+ connectionExecutor = Executors.newVirtualThreadPerTaskExecutor();
+ acceptThread = new Thread(this::acceptLoop, "Plc4xServer-Accept");
+ acceptThread.setDaemon(true);
+ acceptThread.start();
- return channelFuture;
+ LOG.info("PLC4X server listening on port {} ({})", this.port, plaintext ? "plaintext TCP" : "TLS");
+ return CompletableFuture.completedFuture(null);
}
public void stop() {
- if (workerGroup == null) {
- return;
+ running = false;
+ if (serverSocket != null) {
+ try {
+ serverSocket.close();
+ } catch (IOException e) {
+ LOG.debug("Error closing server socket", e);
+ }
+ serverSocket = null;
}
+ if (connectionExecutor != null) {
+ connectionExecutor.shutdownNow();
+ connectionExecutor = null;
+ }
+ }
- channelFuture.cancel(true);
-
- workerGroup.shutdownGracefully();
- loopGroup.shutdownGracefully();
-
- workerGroup = null;
- loopGroup = null;
+ private void acceptLoop() {
+ while (running) {
+ final Socket socket;
+ try {
+ socket = serverSocket.accept();
+ } catch (IOException e) {
+ if (running) {
+ LOG.debug("Accept failed", e);
+ }
+ return;
+ }
+ connectionExecutor.submit(() -> handleConnection(socket));
+ }
}
- private static class SocketChannelChannelInitializer extends ChannelInitializer<SocketChannel> {
+ private void handleConnection(Socket socket) {
+ try (socket) {
+ SocketTransportInstance transport = new SocketTransportInstance(socket);
+ // The codec needs a message handler and the adapter needs the codec to send replies,
+ // so wire them through a one-slot holder to break the construction cycle.
+ final Plc4xServerAdapter[] holder = new Plc4xServerAdapter[1];
+ Plc4xMessageCodec codec = new Plc4xMessageCodec(transport, msg -> holder[0].handle(msg));
+ holder[0] = new Plc4xServerAdapter(connectionManager, codec, username, password);
- @Override
- public void initChannel(SocketChannel channel) {
- ChannelPipeline pipeline = channel.pipeline();
- pipeline.addLast(
- new GeneratedProtocolMessageCodec<>(
- Plc4xMessage.class,
- Plc4xMessage::staticParse,
- ByteOrder.BIG_ENDIAN,
- new ByteLengthEstimator(),
- null
- )
- );
- pipeline.addLast(new Plc4xServerAdapter());
+ while (running && transport.fill()) {
+ codec.processIncomingData();
+ }
+ } catch (MessageCodecException e) {
+ LOG.debug("Protocol error - dropping connection", e);
+ } catch (IOException e) {
+ LOG.debug("Connection I/O error", e);
}
}
- private static class ByteLengthEstimator implements ToIntFunction<ByteBuf> {
+ private SSLServerSocket createTlsServerSocket(int port) throws Exception {
+ ServerTlsContext tlsContext;
+ String keystore = keystorePath != null ? keystorePath : System.getProperty(SERVER_KEYSTORE_PROPERTY);
+ if (keystore != null) {
+ String pwd = keystorePassword != null
+ ? keystorePassword : System.getProperty(SERVER_KEYSTORE_PASSWORD_PROPERTY);
+ tlsContext = ServerTlsContext.fromKeystore(keystore, pwd, null);
+ } else {
+ tlsContext = ServerTlsContext.selfSigned();
+ LOG.info("No keystore configured - generated an ephemeral self-signed certificate.");
+ LOG.info("Server certificate SHA-256 fingerprint: {}", tlsContext.getCertificateFingerprint());
+ }
+ SSLServerSocketFactory factory = tlsContext.getSslContext().getServerSocketFactory();
+ return (SSLServerSocket) factory.createServerSocket(port);
+ }
- @Override
- public int applyAsInt(ByteBuf byteBuf) {
- if (byteBuf.readableBytes() >= 3) {
- return byteBuf.getUnsignedShort(byteBuf.readerIndex() + 1);
+ /**
+ * Resolves the effective credentials from explicit config, system properties or environment,
+ * falling back to the default user with a generated password. Generated passwords are printed
+ * once; configured passwords are never logged.
+ */
+ private void resolveCredentials() {
+ if (username == null) {
+ username = Optional.ofNullable(System.getProperty(SERVER_USERNAME_PROPERTY))
+ .or(() -> Optional.ofNullable(System.getenv(SERVER_USERNAME_ENVIRONMENT_VARIABLE)))
+ .orElse(DEFAULT_USERNAME);
+ }
+ boolean generated = false;
+ if (password == null) {
+ String configured = Optional.ofNullable(System.getProperty(SERVER_PASSWORD_PROPERTY))
+ .or(() -> Optional.ofNullable(System.getenv(SERVER_PASSWORD_ENVIRONMENT_VARIABLE)))
+ .orElse(null);
+ if (configured != null) {
+ password = configured;
+ } else {
+ password = generateSecurePassword();
+ generated = true;
}
- return -1;
+ }
+ if (generated) {
+ // Intentionally printed to stdout (not just the log) so it is visible on first start.
+ System.out.println("============================================================");
+ System.out.println(" No PLC4X server credentials configured - generated defaults:");
+ System.out.println(" username: " + username);
+ System.out.println(" password: " + password);
+ System.out.println(" Provide plc4x.server.username/password to set your own.");
+ System.out.println("============================================================");
+ } else {
+ LOG.info("Using configured credentials for user '{}'", username);
}
}
+ private void resolvePlaintext() {
+ if (!plaintext) {
+ plaintext = Boolean.parseBoolean(System.getProperty(SERVER_PLAINTEXT_PROPERTY))
+ || Boolean.parseBoolean(System.getenv(SERVER_PLAINTEXT_ENVIRONMENT_VARIABLE) == null
+ ? "false" : System.getenv(SERVER_PLAINTEXT_ENVIRONMENT_VARIABLE));
+ }
+ }
+
+ private static String generateSecurePassword() {
+ byte[] bytes = new byte[24];
+ new SecureRandom().nextBytes(bytes);
+ return Base64.getUrlEncoder().withoutPadding().encodeToString(bytes);
+ }
+
private static int findRandomFreePort() {
- final int port;
try (ServerSocket socket = new ServerSocket(0)) {
- port = socket.getLocalPort();
+ return socket.getLocalPort();
} catch (IOException e) {
- throw new RuntimeException("Couldn't find any free port.", e);
+ throw new PlcRuntimeException("Couldn't find any free port.", e);
}
- return port;
}
}
diff --git a/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/ServerTlsContext.java b/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/ServerTlsContext.java
new file mode 100644
index 0000000..3a0b8a2
--- /dev/null
+++ b/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/ServerTlsContext.java
@@ -0,0 +1,153 @@
+/*
+ * 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
+ *
+ * https://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.plc4x.java.tools.plc4xserver;
+
+import org.bouncycastle.asn1.x500.X500Name;
+import org.bouncycastle.asn1.x509.GeneralName;
+import org.bouncycastle.asn1.x509.GeneralNames;
+import org.bouncycastle.asn1.x509.Extension;
+import org.bouncycastle.cert.X509v3CertificateBuilder;
+import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter;
+import org.bouncycastle.cert.jcajce.JcaX509v3CertificateBuilder;
+import org.bouncycastle.operator.ContentSigner;
+import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import java.math.BigInteger;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.KeyStore;
+import java.security.MessageDigest;
+import java.security.SecureRandom;
+import java.security.cert.X509Certificate;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Date;
+import java.util.HexFormat;
+
+/**
+ * Holds the {@link SSLContext} the server listens with, plus the SHA-256 fingerprint of the
+ * presented certificate. Either loads an operator-provided keystore or, when none is configured,
+ * generates an ephemeral self-signed certificate so TLS works with zero configuration. The
+ * fingerprint lets a client pin/trust the auto-generated identity.
+ */
+public final class ServerTlsContext {
+
+ private final SSLContext sslContext;
+ private final String certificateFingerprint;
+ private final boolean selfSigned;
+
+ private ServerTlsContext(SSLContext sslContext, String certificateFingerprint, boolean selfSigned) {
+ this.sslContext = sslContext;
+ this.certificateFingerprint = certificateFingerprint;
+ this.selfSigned = selfSigned;
+ }
+
+ public SSLContext getSslContext() {
+ return sslContext;
+ }
+
+ public String getCertificateFingerprint() {
+ return certificateFingerprint;
+ }
+
+ public boolean isSelfSigned() {
+ return selfSigned;
+ }
+
+ /**
+ * Builds a TLS context from an existing PKCS12/JKS keystore.
+ */
+ public static ServerTlsContext fromKeystore(String keystorePath, String keystorePassword,
+ String keystoreType) throws Exception {
+ char[] password = keystorePassword == null ? new char[0] : keystorePassword.toCharArray();
+ KeyStore keyStore = KeyStore.getInstance(keystoreType == null ? "PKCS12" : keystoreType);
+ try (var in = new java.io.FileInputStream(keystorePath)) {
+ keyStore.load(in, password);
+ }
+ SSLContext sslContext = buildContext(keyStore, password);
+ return new ServerTlsContext(sslContext, fingerprintOfFirst(keyStore), false);
+ }
+
+ /**
+ * Generates an ephemeral self-signed certificate and wraps it in a fresh TLS context.
+ */
+ public static ServerTlsContext selfSigned() throws Exception {
+ KeyPairGenerator keyPairGenerator = KeyPairGenerator.getInstance("RSA");
+ keyPairGenerator.initialize(2048, new SecureRandom());
+ KeyPair keyPair = keyPairGenerator.generateKeyPair();
+
+ X500Name subject = new X500Name("CN=PLC4X-Server");
+ Instant now = Instant.now();
+ Date notBefore = Date.from(now.minus(1, ChronoUnit.HOURS));
+ Date notAfter = Date.from(now.plus(3650, ChronoUnit.DAYS));
+ BigInteger serial = BigInteger.valueOf(System.nanoTime());
+
+ X509v3CertificateBuilder certBuilder = new JcaX509v3CertificateBuilder(
+ subject, serial, notBefore, notAfter, subject, keyPair.getPublic());
+ // Add localhost / loopback SANs so common-name verification can succeed for local use.
+ GeneralNames sans = new GeneralNames(new GeneralName[]{
+ new GeneralName(GeneralName.dNSName, "localhost"),
+ new GeneralName(GeneralName.iPAddress, "127.0.0.1")
+ });
+ certBuilder.addExtension(Extension.subjectAlternativeName, false, sans);
+
+ ContentSigner signer = new JcaContentSignerBuilder("SHA256withRSA").build(keyPair.getPrivate());
+ X509Certificate certificate = new JcaX509CertificateConverter()
+ .getCertificate(certBuilder.build(signer));
+
+ char[] password = new char[0];
+ KeyStore keyStore = KeyStore.getInstance("PKCS12");
+ keyStore.load(null, password);
+ keyStore.setKeyEntry("plc4x-server", keyPair.getPrivate(), password,
+ new X509Certificate[]{certificate});
+
+ SSLContext sslContext = buildContext(keyStore, password);
+ return new ServerTlsContext(sslContext, fingerprint(certificate), true);
+ }
+
+ private static SSLContext buildContext(KeyStore keyStore, char[] password) throws Exception {
+ KeyManagerFactory keyManagerFactory =
+ KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ keyManagerFactory.init(keyStore, password);
+ SSLContext sslContext = SSLContext.getInstance("TLS");
+ sslContext.init(keyManagerFactory.getKeyManagers(), null, new SecureRandom());
+ return sslContext;
+ }
+
+ private static String fingerprintOfFirst(KeyStore keyStore) throws Exception {
+ var aliases = keyStore.aliases();
+ while (aliases.hasMoreElements()) {
+ String alias = aliases.nextElement();
+ var cert = keyStore.getCertificate(alias);
+ if (cert instanceof X509Certificate x509) {
+ return fingerprint(x509);
+ }
+ }
+ return "unknown";
+ }
+
+ private static String fingerprint(X509Certificate certificate) throws Exception {
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ byte[] hash = digest.digest(certificate.getEncoded());
+ return HexFormat.ofDelimiter(":").withUpperCase().formatHex(hash);
+ }
+
+}
diff --git a/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/protocol/Plc4xServerAdapter.java b/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/protocol/Plc4xServerAdapter.java
index 7194b24..3086fc1 100644
--- a/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/protocol/Plc4xServerAdapter.java
+++ b/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/protocol/Plc4xServerAdapter.java
@@ -18,8 +18,6 @@
*/
package org.apache.plc4x.java.tools.plc4xserver.protocol;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.plc4x.java.api.PlcConnection;
import org.apache.plc4x.java.api.PlcConnectionManager;
import org.apache.plc4x.java.api.messages.PlcReadRequest;
@@ -28,149 +26,209 @@ import org.apache.plc4x.java.api.messages.PlcWriteRequest;
import org.apache.plc4x.java.api.messages.PlcWriteResponse;
import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.api.value.PlcValue;
+import org.apache.plc4x.java.plc4x.Plc4xMessageCodec;
import org.apache.plc4x.java.plc4x.readwrite.*;
-import org.apache.plc4x.java.utils.cache.CachedPlcConnectionManager;
+import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
-public class Plc4xServerAdapter extends ChannelInboundHandlerAdapter {
+/**
+ * Per-connection handler for the PLC4X proxy protocol. One instance lives for the lifetime of a
+ * single accepted socket; the {@link SocketTransportInstance} read loop feeds the codec, which
+ * invokes {@link #handle(Plc4xMessage)} for every decoded message.
+ *
+ * <p>Authentication is mandatory: a connection must complete a successful {@code AUTH_REQUEST}
+ * before any {@code CONNECT}/{@code READ}/{@code WRITE} is honoured. Anything else is answered
+ * with {@code ACCESS_DENIED}.</p>
+ */
+public class Plc4xServerAdapter implements Consumer<Plc4xMessage> {
private final Logger logger = LoggerFactory.getLogger(Plc4xServerAdapter.class);
private final PlcConnectionManager connectionManager;
- private final AtomicInteger connectionIdGenerator;
- private final ConcurrentHashMap<Integer, String> connectionUrls;
+ private final Plc4xMessageCodec codec;
+ private final byte[] expectedUsername;
+ private final byte[] expectedPassword;
+
+ private final AtomicInteger connectionIdGenerator = new AtomicInteger(1);
+ private final ConcurrentHashMap<Integer, String> connectionUrls = new ConcurrentHashMap<>();
- public Plc4xServerAdapter() {
- connectionManager = CachedPlcConnectionManager.getBuilder().build();
- connectionIdGenerator = new AtomicInteger(1);
- connectionUrls = new ConcurrentHashMap<>();
+ private volatile boolean authenticated = false;
+
+ public Plc4xServerAdapter(PlcConnectionManager connectionManager, Plc4xMessageCodec codec,
+ String expectedUsername, String expectedPassword) {
+ this.connectionManager = connectionManager;
+ this.codec = codec;
+ this.expectedUsername = expectedUsername.getBytes(StandardCharsets.UTF_8);
+ this.expectedPassword = expectedPassword.getBytes(StandardCharsets.UTF_8);
}
@Override
- public void channelRead(ChannelHandlerContext ctx, Object msg) {
- if (msg instanceof Plc4xMessage) {
- final Plc4xMessage plc4xMessage = (Plc4xMessage) msg;
- switch (plc4xMessage.getRequestType()) {
- case CONNECT_REQUEST: {
- Plc4xConnectRequest request = (Plc4xConnectRequest) plc4xMessage;
- try (final PlcConnection ignored = connectionManager.getConnection(request.getConnectionString())) {
- //connection.ping().get();
- final int connectionId = connectionIdGenerator.getAndIncrement();
- connectionUrls.put(connectionId, request.getConnectionString());
- Plc4xConnectResponse response = new Plc4xConnectResponse(
- request.getRequestId(), connectionId, Plc4xResponseCode.OK);
- ctx.writeAndFlush(response);
- } catch (Exception e) {
- Plc4xConnectResponse response = new Plc4xConnectResponse(
- request.getRequestId(), 0, Plc4xResponseCode.INVALID_ADDRESS);
- ctx.writeAndFlush(response);
- }
- break;
+ public void accept(Plc4xMessage message) {
+ handle(message);
+ }
+
+ public void handle(Plc4xMessage plc4xMessage) {
+ switch (plc4xMessage.getRequestType()) {
+ case AUTH_REQUEST:
+ handleAuth((Plc4xAuthRequest) plc4xMessage);
+ break;
+ case CONNECT_REQUEST:
+ if (requireAuth(plc4xMessage)) {
+ handleConnect((Plc4xConnectRequest) plc4xMessage);
+ }
+ break;
+ case READ_REQUEST:
+ if (requireAuth(plc4xMessage)) {
+ handleRead((Plc4xReadRequest) plc4xMessage);
}
- case READ_REQUEST: {
- final Plc4xReadRequest request = (Plc4xReadRequest) plc4xMessage;
- String connectionUrl = connectionUrls.get(request.getConnectionId());
- try (final PlcConnection connection = connectionManager.getConnection(connectionUrl)) {
- // Build a read request for all tags in the request.
- final PlcReadRequest.Builder builder = connection.readRequestBuilder();
- for (Plc4xTagRequest requestTag : request.getTags()) {
- builder.addTagAddress(requestTag.getTag().getName(), requestTag.getTag().getTagQuery());
- }
- final PlcReadRequest rr = builder.build();
-
- // Execute the query.
- // (It has to be synchronously when working with the connection cache)
- final PlcReadResponse apiReadResponse = rr.execute().get();
-
- // Create the response.
- List<Plc4xTagValueResponse> tags = new ArrayList<>(apiReadResponse.getTagNames().size());
- for (Plc4xTagRequest plc4xRequestTag : request.getTags()) {
- final PlcResponseCode responseCode = apiReadResponse.getResponseCode(plc4xRequestTag.getTag().getName());
- Plc4xResponseCode resCode;
- Plc4xValueType valueType;
- PlcValue value;
- if(responseCode == PlcResponseCode.OK) {
- resCode = Plc4xResponseCode.OK;
- value = apiReadResponse.getPlcValue(plc4xRequestTag.getTag().getName());
- final String valueTypeName = value.getClass().getSimpleName();
- // Cut off the "Plc" prefix to get the name of the PlcValueType.
- valueType = Plc4xValueType.valueOf(valueTypeName.substring(3));
- } else {
- resCode = Plc4xResponseCode.INVALID_ADDRESS;
- value = null;
- valueType = Plc4xValueType.NULL;
- }
- tags.add(new Plc4xTagValueResponse(
- plc4xRequestTag.getTag(), resCode, valueType, value));
- }
- Plc4xReadResponse response = new Plc4xReadResponse(
- request.getRequestId(), request.getConnectionId(), Plc4xResponseCode.OK, tags);
-
- // Send the response.
- ctx.writeAndFlush(response);
- } catch (Exception e) {
- logger.error("Error executing request", e);
- Plc4xReadResponse response = new Plc4xReadResponse(
- request.getRequestId(), request.getConnectionId(),
- Plc4xResponseCode.INVALID_ADDRESS, Collections.emptyList());
- ctx.writeAndFlush(response);
- }
- break;
+ break;
+ case WRITE_REQUEST:
+ if (requireAuth(plc4xMessage)) {
+ handleWrite((Plc4xWriteRequest) plc4xMessage);
+ }
+ break;
+ default:
+ logger.debug("Ignoring unsupported request type {}", plc4xMessage.getRequestType());
+ }
+ }
+
+ private void handleAuth(Plc4xAuthRequest request) {
+ boolean ok = constantTimeEquals(expectedUsername, request.getUsername().getBytes(StandardCharsets.UTF_8))
+ & constantTimeEquals(expectedPassword, request.getPassword().getBytes(StandardCharsets.UTF_8));
+ authenticated = ok;
+ // Never log the supplied credentials, just the outcome.
+ logger.info("Authentication {}", ok ? "succeeded" : "failed");
+ send(new Plc4xAuthResponse(request.getRequestId(),
+ ok ? Plc4xResponseCode.OK : Plc4xResponseCode.ACCESS_DENIED));
+ }
+
+ /**
+ * Returns {@code true} if the connection is authenticated. Otherwise emits a type-appropriate
+ * {@code ACCESS_DENIED} response and returns {@code false}.
+ */
+ private boolean requireAuth(Plc4xMessage message) {
+ if (authenticated) {
+ return true;
+ }
+ logger.warn("Rejecting {} - connection is not authenticated", message.getRequestType());
+ switch (message.getRequestType()) {
+ case CONNECT_REQUEST -> send(new Plc4xConnectResponse(
+ message.getRequestId(), 0, Plc4xResponseCode.ACCESS_DENIED));
+ case READ_REQUEST -> send(new Plc4xReadResponse(message.getRequestId(),
+ ((Plc4xReadRequest) message).getConnectionId(), Plc4xResponseCode.ACCESS_DENIED,
+ Collections.emptyList()));
+ case WRITE_REQUEST -> send(new Plc4xWriteResponse(message.getRequestId(),
+ ((Plc4xWriteRequest) message).getConnectionId(), Plc4xResponseCode.ACCESS_DENIED,
+ Collections.emptyList()));
+ default -> { /* nothing to answer */ }
+ }
+ return false;
+ }
+
+ private void handleConnect(Plc4xConnectRequest request) {
+ try (final PlcConnection ignored = connectionManager.getConnection(request.getConnectionString())) {
+ final int connectionId = connectionIdGenerator.getAndIncrement();
+ connectionUrls.put(connectionId, request.getConnectionString());
+ send(new Plc4xConnectResponse(request.getRequestId(), connectionId, Plc4xResponseCode.OK));
+ } catch (Exception e) {
+ send(new Plc4xConnectResponse(request.getRequestId(), 0, Plc4xResponseCode.INVALID_ADDRESS));
+ }
+ }
+
+ private void handleRead(Plc4xReadRequest request) {
+ String connectionUrl = connectionUrls.get(request.getConnectionId());
+ try (final PlcConnection connection = connectionManager.getConnection(connectionUrl)) {
+ final PlcReadRequest.Builder builder = connection.readRequestBuilder();
+ for (Plc4xTagRequest requestTag : request.getTags()) {
+ builder.addTagAddress(requestTag.getTag().getName(), requestTag.getTag().getTagQuery());
+ }
+ final PlcReadRequest rr = builder.build();
+
+ // Execute synchronously (required when working with the connection cache).
+ final PlcReadResponse apiReadResponse = rr.execute().get();
+
+ List<Plc4xTagValueResponse> tags = new ArrayList<>(apiReadResponse.getTagNames().size());
+ for (Plc4xTagRequest plc4xRequestTag : request.getTags()) {
+ final PlcResponseCode responseCode =
+ apiReadResponse.getResponseCode(plc4xRequestTag.getTag().getName());
+ Plc4xResponseCode resCode;
+ Plc4xValueType valueType;
+ PlcValue value;
+ if (responseCode == PlcResponseCode.OK) {
+ resCode = Plc4xResponseCode.OK;
+ value = apiReadResponse.getPlcValue(plc4xRequestTag.getTag().getName());
+ final String valueTypeName = value.getClass().getSimpleName();
+ // Cut off the "Plc" prefix to get the name of the PlcValueType.
+ valueType = Plc4xValueType.valueOf(valueTypeName.substring(3));
+ } else {
+ resCode = Plc4xResponseCode.INVALID_ADDRESS;
+ value = null;
+ valueType = Plc4xValueType.NULL;
}
+ tags.add(new Plc4xTagValueResponse(plc4xRequestTag.getTag(), resCode, valueType, value));
+ }
+ send(new Plc4xReadResponse(request.getRequestId(), request.getConnectionId(),
+ Plc4xResponseCode.OK, tags));
+ } catch (Exception e) {
+ logger.error("Error executing read request", e);
+ send(new Plc4xReadResponse(request.getRequestId(), request.getConnectionId(),
+ Plc4xResponseCode.INVALID_ADDRESS, Collections.emptyList()));
+ }
+ }
- case WRITE_REQUEST:
- final Plc4xWriteRequest plc4xWriteRequest = (Plc4xWriteRequest) plc4xMessage;
- String connectionUrl = connectionUrls.get(plc4xWriteRequest.getConnectionId());
- try (final PlcConnection connection = connectionManager.getConnection(connectionUrl)) {
- // Build a write request for all tags in the request.
- final PlcWriteRequest.Builder builder = connection.writeRequestBuilder();
- for (Plc4xTagValueRequest plc4xRequestTag : plc4xWriteRequest.getTags()) {
- builder.addTagAddress(plc4xRequestTag.getTag().getName(),
- plc4xRequestTag.getTag().getTagQuery(), plc4xRequestTag.getValue().getObject());
- }
- final PlcWriteRequest apiWriteRequest = builder.build();
-
- // Execute the query
- // (It has to be synchronously when working with the connection cache)
- final PlcWriteResponse apiWriteResponse = apiWriteRequest.execute().get();
-
- // Create the response.
- List<Plc4xTagResponse> plc4xTags =
- new ArrayList<>(apiWriteResponse.getTagNames().size());
- for (Plc4xTagValueRequest plc4xRequestTag : plc4xWriteRequest.getTags()) {
- final PlcResponseCode apiResponseCode =
- apiWriteResponse.getResponseCode(plc4xRequestTag.getTag().getName());
- Plc4xResponseCode resCode;
- if(apiResponseCode == PlcResponseCode.OK) {
- resCode = Plc4xResponseCode.OK;
- } else {
- resCode = Plc4xResponseCode.INVALID_ADDRESS;
- }
- plc4xTags.add(new Plc4xTagResponse(plc4xRequestTag.getTag(), resCode));
- }
- Plc4xWriteResponse plc4xWriteResponse = new Plc4xWriteResponse(
- plc4xWriteRequest.getRequestId(), plc4xWriteRequest.getConnectionId(),
- Plc4xResponseCode.OK, plc4xTags);
-
- // Send the response.
- ctx.writeAndFlush(plc4xWriteResponse);
- } catch (Exception e) {
- logger.error("Error executing request", e);
- Plc4xWriteResponse response = new Plc4xWriteResponse(
- plc4xWriteRequest.getRequestId(), plc4xWriteRequest.getConnectionId(),
- Plc4xResponseCode.INVALID_ADDRESS, Collections.emptyList());
- ctx.writeAndFlush(response);
- }
- break;
+ private void handleWrite(Plc4xWriteRequest request) {
+ String connectionUrl = connectionUrls.get(request.getConnectionId());
+ try (final PlcConnection connection = connectionManager.getConnection(connectionUrl)) {
+ final PlcWriteRequest.Builder builder = connection.writeRequestBuilder();
+ for (Plc4xTagValueRequest plc4xRequestTag : request.getTags()) {
+ builder.addTagAddress(plc4xRequestTag.getTag().getName(),
+ plc4xRequestTag.getTag().getTagQuery(), plc4xRequestTag.getValue().getObject());
}
+ final PlcWriteRequest apiWriteRequest = builder.build();
+
+ // Execute synchronously (required when working with the connection cache).
+ final PlcWriteResponse apiWriteResponse = apiWriteRequest.execute().get();
+
+ List<Plc4xTagResponse> plc4xTags = new ArrayList<>(apiWriteResponse.getTagNames().size());
+ for (Plc4xTagValueRequest plc4xRequestTag : request.getTags()) {
+ final PlcResponseCode apiResponseCode =
+ apiWriteResponse.getResponseCode(plc4xRequestTag.getTag().getName());
+ Plc4xResponseCode resCode = apiResponseCode == PlcResponseCode.OK
+ ? Plc4xResponseCode.OK : Plc4xResponseCode.INVALID_ADDRESS;
+ plc4xTags.add(new Plc4xTagResponse(plc4xRequestTag.getTag(), resCode));
+ }
+ send(new Plc4xWriteResponse(request.getRequestId(), request.getConnectionId(),
+ Plc4xResponseCode.OK, plc4xTags));
+ } catch (Exception e) {
+ logger.error("Error executing write request", e);
+ send(new Plc4xWriteResponse(request.getRequestId(), request.getConnectionId(),
+ Plc4xResponseCode.INVALID_ADDRESS, Collections.emptyList()));
+ }
+ }
+
+ private void send(Plc4xMessage message) {
+ try {
+ codec.send(message);
+ } catch (MessageCodecException e) {
+ logger.error("Failed to send response", e);
}
}
+ /**
+ * Length-aware constant-time comparison to avoid leaking credential length/content via timing.
+ */
+ private static boolean constantTimeEquals(byte[] expected, byte[] actual) {
+ return MessageDigest.isEqual(expected, actual);
+ }
+
}
diff --git a/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/protocol/SocketTransportInstance.java b/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/protocol/SocketTransportInstance.java
new file mode 100644
index 0000000..cc05687
--- /dev/null
+++ b/plc4j/tools/plc4x-server/src/main/java/org/apache/plc4x/java/tools/plc4xserver/protocol/SocketTransportInstance.java
@@ -0,0 +1,121 @@
+/*
+ * 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
+ *
+ * https://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.plc4x.java.tools.plc4xserver.protocol;
+
+import org.apache.plc4x.java.spi.transports.api.RingBuffer;
+import org.apache.plc4x.java.spi.transports.api.TransportInstance;
+import org.apache.plc4x.java.spi.transports.api.config.TransportConfiguration;
+import org.apache.plc4x.java.spi.transports.api.exceptions.TransportException;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+
+/**
+ * Adapts an accepted server-side {@link Socket} to the SPI3 {@link TransportInstance} contract
+ * so the regular PLC4X message codec can be reused on the server side.
+ *
+ * <p>SPI3 has no server transport (it is purely client/outbound), but a {@code MessageCodecBase}
+ * only needs the byte-level peek/read/write surface. Inbound bytes are pulled from the socket by
+ * {@link #fill()} into a {@link RingBuffer}; the codec then drains complete messages from that
+ * buffer via {@link #getNumBytesAvailable()} / {@link #peekReadableBytes(int)} / {@link #read(int)}.
+ * The maximum PLC4X proxy message is bounded by a {@code uint16} length, so a buffer slightly
+ * larger than 64 KiB always holds a full frame.</p>
+ */
+public class SocketTransportInstance implements TransportInstance<TransportConfiguration> {
+
+ private static final int BUFFER_CAPACITY = 256 * 1024;
+ private static final int READ_CHUNK = 8 * 1024;
+
+ private final Socket socket;
+ private final InputStream in;
+ private final OutputStream out;
+ private final RingBuffer ringBuffer = new RingBuffer(BUFFER_CAPACITY);
+ private final byte[] readChunk = new byte[READ_CHUNK];
+
+ public SocketTransportInstance(Socket socket) throws IOException {
+ this.socket = socket;
+ this.in = socket.getInputStream();
+ this.out = socket.getOutputStream();
+ }
+
+ /**
+ * Blocks until at least one byte arrives from the socket and appends it to the ring buffer.
+ *
+ * @return {@code false} on end-of-stream (peer closed), {@code true} otherwise.
+ */
+ public boolean fill() throws IOException {
+ int read = in.read(readChunk);
+ if (read == -1) {
+ return false;
+ }
+ if (read > 0) {
+ ringBuffer.write(readChunk, 0, read);
+ }
+ return true;
+ }
+
+ @Override
+ public TransportConfiguration getConfiguration() {
+ return null;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return !socket.isClosed() && socket.isConnected();
+ }
+
+ @Override
+ public int getNumBytesAvailable() {
+ return ringBuffer.availableForReading();
+ }
+
+ @Override
+ public byte[] peekReadableBytes(int numBytes) {
+ return ringBuffer.peek(numBytes);
+ }
+
+ @Override
+ public byte[] read(int numBytes) {
+ return ringBuffer.read(numBytes);
+ }
+
+ @Override
+ public void write(byte[] bytes) throws TransportException {
+ try {
+ synchronized (out) {
+ out.write(bytes);
+ out.flush();
+ }
+ } catch (IOException e) {
+ throw new TransportException("Failed to write to socket", e);
+ }
+ }
+
+ @Override
+ public void close() throws TransportException {
+ try {
+ socket.close();
+ } catch (IOException e) {
+ throw new TransportException("Failed to close socket", e);
+ }
+ }
+
+}
diff --git a/plc4j/tools/plc4x-server/src/test/java/org/apache/plc4x/java/tools/plc4xserver/Plc4xServerTest.java b/plc4j/tools/plc4x-server/src/test/java/org/apache/plc4x/java/tools/plc4xserver/Plc4xServerTest.java
index 0e35957..326bed3 100644
--- a/plc4j/tools/plc4x-server/src/test/java/org/apache/plc4x/java/tools/plc4xserver/Plc4xServerTest.java
+++ b/plc4j/tools/plc4x-server/src/test/java/org/apache/plc4x/java/tools/plc4xserver/Plc4xServerTest.java
@@ -22,6 +22,7 @@ package org.apache.plc4x.java.tools.plc4xserver;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -29,6 +30,7 @@ import java.util.concurrent.TimeoutException;
import org.apache.plc4x.java.DefaultPlcDriverManager;
import org.apache.plc4x.java.api.PlcConnection;
import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
import org.apache.plc4x.java.api.messages.PlcReadRequest;
import org.apache.plc4x.java.api.messages.PlcReadResponse;
import org.apache.plc4x.java.api.messages.PlcWriteRequest;
@@ -40,8 +42,13 @@ import org.junit.jupiter.api.Test;
public class Plc4xServerTest {
+ private static final String USERNAME = "test-user";
+ private static final String PASSWORD = "test-password";
+
private static final Plc4xServer SERVER = new Plc4xServer();
- private static final String CONNECTION_STRING_TEMPLATE = "plc4x://localhost:%d?remote-connection-string=%s";
+ // TLS (the default) with verify-ssl=false so the client trusts the server's self-signed cert.
+ private static final String CONNECTION_STRING_TEMPLATE =
+ "plc4x:tls://localhost:%d?remote-connection-string=%s&username=%s&password=%s&tls.verify-ssl=false";
private static final String CONNECTION_STRING_SIMULATED_ENCODED = "simulated%3A%2F%2Flocalhost";
private static final long TIMEOUT_VALUE = 10;
private static final TimeUnit TIMEOUT_UNIT = TimeUnit.SECONDS;
@@ -50,6 +57,8 @@ public class Plc4xServerTest {
@BeforeAll
public static void setUp() throws ExecutionException, InterruptedException, TimeoutException {
+ SERVER.setUsername(USERNAME);
+ SERVER.setPassword(PASSWORD);
SERVER.start().get(TIMEOUT_VALUE, TIMEOUT_UNIT);
}
@@ -58,18 +67,18 @@ public class Plc4xServerTest {
SERVER.stop();
}
+ private String connectionString(String username, String password) {
+ return String.format(CONNECTION_STRING_TEMPLATE, SERVER.getPort(),
+ CONNECTION_STRING_SIMULATED_ENCODED, username, password);
+ }
+
@Test
public void testWrite() throws Exception {
final PlcWriteResponse response;
- try (PlcConnection connection = connectionManager.getConnection(
- String.format(CONNECTION_STRING_TEMPLATE, SERVER.getPort(), CONNECTION_STRING_SIMULATED_ENCODED))) {
+ try (PlcConnection connection = connectionManager.getConnection(connectionString(USERNAME, PASSWORD))) {
final PlcWriteRequest request = connection.writeRequestBuilder()
- .addTagAddress(
- "foo",
- "STATE/foo:DINT",
- 42
- )
+ .addTagAddress("foo", "STATE/foo:DINT", 42)
.build();
response = request.execute().get(TIMEOUT_VALUE, TIMEOUT_UNIT);
}
@@ -81,13 +90,9 @@ public class Plc4xServerTest {
public void testRead() throws Exception {
final PlcReadResponse response;
- try (PlcConnection connection = connectionManager.getConnection(
- String.format(CONNECTION_STRING_TEMPLATE, SERVER.getPort(), CONNECTION_STRING_SIMULATED_ENCODED))) {
+ try (PlcConnection connection = connectionManager.getConnection(connectionString(USERNAME, PASSWORD))) {
final PlcReadRequest request = connection.readRequestBuilder()
- .addTagAddress(
- "foo",
- "RANDOM/foo:DINT"
- )
+ .addTagAddress("foo", "RANDOM/foo:DINT")
.build();
response = request.execute().get(TIMEOUT_VALUE, TIMEOUT_UNIT);
}
@@ -102,22 +107,14 @@ public class Plc4xServerTest {
final PlcWriteResponse writeResponse;
final PlcReadResponse readResponse;
- try (PlcConnection connection = connectionManager.getConnection(
- String.format(CONNECTION_STRING_TEMPLATE, SERVER.getPort(), CONNECTION_STRING_SIMULATED_ENCODED))) {
+ try (PlcConnection connection = connectionManager.getConnection(connectionString(USERNAME, PASSWORD))) {
final PlcWriteRequest writeRequest = connection.writeRequestBuilder()
- .addTagAddress(
- "foo",
- "STATE/foo:DINT",
- 21
- )
+ .addTagAddress("foo", "STATE/foo:DINT", 21)
.build();
writeResponse = writeRequest.execute().get(TIMEOUT_VALUE, TIMEOUT_UNIT);
final PlcReadRequest readRequest = connection.readRequestBuilder()
- .addTagAddress(
- "foo",
- "STATE/foo:DINT"
- )
+ .addTagAddress("foo", "STATE/foo:DINT")
.build();
readResponse = readRequest.execute().get(TIMEOUT_VALUE, TIMEOUT_UNIT);
}
@@ -128,4 +125,15 @@ public class Plc4xServerTest {
assertInstanceOf(Integer.class, readResponse.getPlcValue("foo").getObject());
assertEquals(21, readResponse.getInteger("foo"));
}
+
+ @Test
+ public void testWrongPasswordIsRejected() {
+ // A connection with bad credentials must fail during the mandatory auth handshake.
+ assertThrows(PlcConnectionException.class, () -> {
+ try (PlcConnection ignored = connectionManager.getConnection(
+ connectionString(USERNAME, "wrong-password"))) {
+ // Should never get here.
+ }
+ });
+ }
}
(plc4x-extras) 02/03: chore: Made the integration modules build again.
Posted by cd...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x-extras.git
commit 0859a6c8384681f15b9011406d914fe213a47fa1
Author: Christofer Dutz <ch...@c-ware.de>
AuthorDate: Tue Jun 23 19:12:46 2026 +0200
chore: Made the integration modules build again.
---
plc4j/integrations/apache-calcite/pom.xml | 6 ++++
plc4j/integrations/apache-nifi/README.md | 54 +++++++++++++++----------------
2 files changed, 33 insertions(+), 27 deletions(-)
diff --git a/plc4j/integrations/apache-calcite/pom.xml b/plc4j/integrations/apache-calcite/pom.xml
index bfb5e75..5530ac7 100644
--- a/plc4j/integrations/apache-calcite/pom.xml
+++ b/plc4j/integrations/apache-calcite/pom.xml
@@ -96,6 +96,12 @@
<version>${plc4x.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <version>1.5.34</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<dependencyManagement>
diff --git a/plc4j/integrations/apache-nifi/README.md b/plc4j/integrations/apache-nifi/README.md
index 04aeed8..cfecb66 100644
--- a/plc4j/integrations/apache-nifi/README.md
+++ b/plc4j/integrations/apache-nifi/README.md
@@ -71,33 +71,33 @@ When reading from a PLC the response is used to create a mapping between Plc typ
Table of data mapping between plc data and Avro types (as specified in [Avro specification](https://avro.apache.org/docs/1.11.1/specification/#primitive-types)).
-| PLC type | Avro Type |
-|----------:|-----------|
-| PlcBOOL | boolean |
-| PlcBYTE | bytes |
-| PlcSINT | int |
-| PlcINT | int |
-| PlcLINT | long |
-| PlcREAL | float |
-| PlcLREAL | double |
-| PlcCHAR | string |
-| PlcDATE_AND_TIME | string |
-| PlcDATE | string |
-| PlcDINT | string |
-| PlcDWORD | string |
-| PlcLTIME | string |
-| PlcLWORD | string |
-| PlcNull | string |
-| PlcSTRING | string |
-| PlcTIME_OF_DAY | string |
-| PlcTIME | string |
-| PlcUDINT | string |
-| PlcUINT | string |
-| PlcULINT | string |
-| PlcUSINT | string |
-| PlcWCHAR | string |
-| PlcWORD | string |
-| ELSE | string |
+| PLC type | Avro Type |
+|-----------------:|-----------|
+| PlcBOOL | boolean |
+| PlcBYTE | bytes |
+| PlcSINT | int |
+| PlcINT | int |
+| PlcLINT | long |
+| PlcREAL | float |
+| PlcLREAL | double |
+| PlcCHAR | string |
+| PlcDATE_AND_TIME | string |
+| PlcDATE | string |
+| PlcDINT | string |
+| PlcDWORD | string |
+| PlcLTIME | string |
+| PlcLWORD | string |
+| PlcNull | string |
+| PlcSTRING | string |
+| PlcTIME_OF_DAY | string |
+| PlcTIME | string |
+| PlcUDINT | string |
+| PlcUINT | string |
+| PlcULINT | string |
+| PlcUSINT | string |
+| PlcWCHAR | string |
+| PlcWORD | string |
+| ELSE | string |
Also, it is important to keep in mind the Processor Scheduling Configuration. Using the parameter **Run Schedule** (for example to *1 sec*), the reading frequency can be set. Note that by default, this value is defined to 0 sec (as fast as possible).