From 1d559d5e7b918cb9066f63bbade5b51acad1a914 Mon Sep 17 00:00:00 2001 From: vishwasvaidya-cloudsufi Date: Mon, 20 Jul 2026 17:53:21 +0530 Subject: [PATCH] Add SMBv2/v3 support to WindowsShareCopy with backward compatibility --- core-plugins/pom.xml | 24 +- .../plugin/batch/action/WindowsShareCopy.java | 219 +++++++++++++++--- .../cdap/plugin/batch/ETLBatchTestBase.java | 2 + .../action/WindowsShareCopyConfigTest.java | 111 +++++++++ .../widgets/WindowsShareCopy-action.json | 12 + 5 files changed, 338 insertions(+), 30 deletions(-) create mode 100644 core-plugins/src/test/java/io/cdap/plugin/batch/action/WindowsShareCopyConfigTest.java diff --git a/core-plugins/pom.xml b/core-plugins/pom.xml index 8bcee2fd8..74e7d5c49 100644 --- a/core-plugins/pom.xml +++ b/core-plugins/pom.xml @@ -111,13 +111,33 @@ commons-lang commons-lang - + jcifs jcifs 1.3.17 - + + + com.hierynomus + smbj + 0.11.5 + + + net.engio + mbassador + 1.3.0 + + + com.hierynomus + asn-one + 0.6.0 + + + org.bouncycastle + bcprov-jdk15to18 + 1.84 + dumbster diff --git a/core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java b/core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java index a683c65a2..3698021d6 100644 --- a/core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java +++ b/core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java @@ -20,6 +20,16 @@ import com.google.common.base.Throwables; import com.google.common.io.ByteStreams; import com.google.common.util.concurrent.ThreadFactoryBuilder; +import com.hierynomus.msdtyp.AccessMask; +import com.hierynomus.msfscc.fileinformation.FileIdBothDirectoryInformation; +import com.hierynomus.mssmb2.SMB2CreateDisposition; +import com.hierynomus.mssmb2.SMB2ShareAccess; +import com.hierynomus.smbj.SMBClient; +import com.hierynomus.smbj.auth.AuthenticationContext; +import com.hierynomus.smbj.connection.Connection; +import com.hierynomus.smbj.session.Session; +import com.hierynomus.smbj.share.DiskShare; +import com.hierynomus.smbj.share.File; import io.cdap.cdap.api.annotation.Description; import io.cdap.cdap.api.annotation.Macro; import io.cdap.cdap.api.annotation.Name; @@ -41,6 +51,9 @@ import java.io.BufferedOutputStream; import java.io.IOException; import java.io.InputStream; +import java.util.ArrayList; +import java.util.EnumSet; +import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.CompletionService; import java.util.concurrent.CountDownLatch; @@ -54,6 +67,7 @@ /** * {@link WindowsShareCopy} is an {@link Action} that will copy the data from a Windows share into an HDFS directory. + * Supports SMBv1 (via jcifs) and SMBv2/SMBv3 (via smbj) based on user configuration. */ @Plugin(type = Action.PLUGIN_TYPE) @Name("WindowsShareCopy") @@ -81,6 +95,32 @@ public void run(ActionContext context) throws Exception { config.numThreads; config.bufferSize = (config.bufferSize == null || config.bufferSize < MIN_BUFFER_SIZE) ? MIN_BUFFER_SIZE : config.bufferSize; + + // Determines the buffer size to be used during writing. + Configuration conf = new Configuration(); + conf.setLong("io.file.buffer.size", config.bufferSize); + conf.setLong("file.stream-buffer-size", config.bufferSize); + + // Create the HDFS directory if it doesn't exist. + final Path hdfsDir = new Path(config.destinationDirectory); + final FileSystem hdfs = hdfsDir.getFileSystem(conf); + if (!hdfs.exists(hdfsDir)) { + hdfs.mkdirs(hdfsDir); + } + + if (config.isSMBv1()) { + LOG.info("Executing copy using legacy SMBv1 protocol (jcifs)"); + executeSMBv1(hdfsDir, hdfs); + } else { + LOG.info("Executing copy using modern SMBv2/v3 protocol (smbj)"); + executeSMBv2(hdfsDir, hdfs); + } + } + + /** + * Executes the copy operation using the legacy jcifs (SMBv1) library. + */ + private void executeSMBv1(final Path hdfsDir, final FileSystem hdfs) throws Exception { StringBuilder sb = new StringBuilder("smb://"); sb.append(config.netBiosHostname); sb.append("/"); @@ -102,21 +142,9 @@ public void run(ActionContext context) throws Exception { // Authentication with NTLM and read the directory from the Windows Share. final NtlmPasswordAuthentication auth = new NtlmPasswordAuthentication(config.netBiosDomainName, - config.netBiosUsername, - config.netBiosPassword); + config.netBiosUsername, + config.netBiosPassword); SmbFile dir = new SmbFile(smbDirectory, auth); - - // Determines the buffer size to be used during writing. - Configuration conf = new Configuration(); - conf.setLong("io.file.buffer.size", config.bufferSize); - conf.setLong("file.stream-buffer-size", config.bufferSize); - // Create the HDFS directory if it doesn't exist. - final Path hdfsDir = new Path(config.destinationDirectory); - final FileSystem hdfs = hdfsDir.getFileSystem(conf); - if (!hdfs.exists(hdfsDir)) { - hdfs.mkdirs(hdfsDir); - } - String[] files = dir.list(); // Copies the files in a multithreaded way CountDownLatch executorTerminateLatch = new CountDownLatch(1); @@ -128,15 +156,10 @@ public void run(ActionContext context) throws Exception { completionService.submit(new Callable() { @Override public String call() throws Exception { - try { - if (smbDirectory.endsWith("/")) { - return copyFileToHDFS(hdfs, smbDirectory + file, hdfsDir, auth); - } else { - return copyFileToHDFS(hdfs, smbDirectory + "/" + file, hdfsDir, auth); - } - } catch (Exception e) { - LOG.warn("Exception while copying the file {}", file, e); - return null; + if (smbDirectory.endsWith("/")) { + return copyFileToHDFS(hdfs, smbDirectory + file, hdfsDir, auth); + } else { + return copyFileToHDFS(hdfs, smbDirectory + "/" + file, hdfsDir, auth); } } }); @@ -162,12 +185,12 @@ public String call() throws Exception { } private String copyFileToHDFS(FileSystem hdfs, String smbSourceFile, Path dest, NtlmPasswordAuthentication auth) - throws IOException { + throws IOException { SmbFile smbFile = new SmbFile(smbSourceFile, auth); String name = smbFile.getName(); Path destFile = new Path(dest, name); LOG.debug("Thread {} is copying source file {}, dest file {}", Thread.currentThread().getName(), - smbFile.getName(), destFile.getName()); + smbFile.getName(), destFile.getName()); // If file already exists, then we skip over. if (hdfs.exists(destFile) && !config.overwrite) { LOG.info("File {} already exists on HDFS, Skipping", destFile.getName()); @@ -177,8 +200,115 @@ private String copyFileToHDFS(FileSystem hdfs, String smbSourceFile, Path dest, try (InputStream in = smbFile.getInputStream(); BufferedOutputStream out = new BufferedOutputStream(hdfs.create(destFile), config.bufferSize)) { ByteStreams.copy(in, out); - } catch (IOException e) { - LOG.warn("Exception in copying the file {}", name, e); + } + return name; + } + + /** + * Executes the copy operation using the modern smbj library. + */ + private void executeSMBv2(final Path hdfsDir, final FileSystem hdfs) throws Exception { + String normalizedSourcePath = config.getSourceDirectory(); + + try (SMBClient client = new SMBClient()) { + AuthenticationContext authContext = new AuthenticationContext( + config.netBiosUsername, + config.netBiosPassword.toCharArray(), + config.netBiosDomainName == null ? "" : config.netBiosDomainName + ); + + List filesToCopy = new ArrayList<>(); + + try (Connection listConnection = client.connect(config.netBiosHostname); + Session listSession = listConnection.authenticate(authContext); + DiskShare listShare = (DiskShare) listSession.connectShare(config.netBiosSharename)) { + + if (listShare.folderExists(normalizedSourcePath)) { + for (FileIdBothDirectoryInformation f : listShare.list(normalizedSourcePath)) { + String fileName = f.getFileName(); + if (fileName.equals(".") || fileName.equals("..")) { + continue; + } + String fullPath = normalizedSourcePath.isEmpty() ? fileName : normalizedSourcePath + "\\" + fileName; + filesToCopy.add(fullPath); + } + } else if (listShare.fileExists(normalizedSourcePath)) { + filesToCopy.add(normalizedSourcePath); + } else { + throw new IllegalArgumentException( + String.format("The source path '%s' does not exist on share '%s'", + normalizedSourcePath, config.netBiosSharename) + ); + } + } + + CountDownLatch executorTerminateLatch = new CountDownLatch(1); + ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch); + CompletionService completionService = new ExecutorCompletionService<>(executorService); + + try { + for (final String filePath : filesToCopy) { + completionService.submit(new Callable() { + @Override + public String call() throws Exception { + try (Connection threadConnection = client.connect(config.netBiosHostname); + Session threadSession = threadConnection.authenticate(authContext); + DiskShare threadShare = (DiskShare) threadSession.connectShare(config.netBiosSharename)) { + + return copyFileToHDFS_SMBv2(hdfs, threadShare, filePath, hdfsDir); + } + } + }); + } + + int count = 0; + while (count < filesToCopy.size()) { + try { + Future fileWritten = completionService.take(); + String fileName = fileWritten.get(); + if (fileName != null) { + LOG.debug("{} is copied", fileName); + } + } catch (Throwable t) { + throw Throwables.propagate(t); + } + count++; + } + } finally { + executorService.shutdownNow(); + executorTerminateLatch.await(); + } + } + } + + /** + * The new SMBv2/v3 file copy method + */ + private String copyFileToHDFS_SMBv2(FileSystem hdfs, DiskShare share, String smbSourceFile, Path dest) + throws IOException { + int lastSlashIndex = smbSourceFile.lastIndexOf('\\'); + String name = lastSlashIndex >= 0 ? smbSourceFile.substring(lastSlashIndex + 1) : smbSourceFile; + + Path destFile = new Path(dest, name); + LOG.debug("Thread {} is copying source file {}, dest file {}", Thread.currentThread().getName(), + name, destFile.getName()); + + if (hdfs.exists(destFile) && !config.overwrite) { + LOG.info("File {} already exists on HDFS, Skipping", destFile.getName()); + return null; + } + LOG.info("Copying file {} to {}", smbSourceFile, destFile.toString()); + + try (File smbFile = share.openFile( + smbSourceFile, + EnumSet.of(AccessMask.GENERIC_READ), + null, + SMB2ShareAccess.ALL, + SMB2CreateDisposition.FILE_OPEN, + null); + InputStream in = smbFile.getInputStream(); + BufferedOutputStream out = new BufferedOutputStream(hdfs.create(destFile), config.bufferSize)) { + ByteStreams.copy(in, out); } return name; } @@ -263,9 +393,16 @@ public class WindowsShareCopyConfig extends PluginConfig { @Macro private Integer bufferSize; + @Description("Specifies the SMB protocol version. Use 'SMBv1' for legacy systems or 'SMBv2/v3' for modern " + + "secure shares.") + @Nullable + @Macro + private final String smbVersion; + WindowsShareCopyConfig(String netBiosDomainName, String netBiosHostname, String netBiosUsername, String netBiosPassword, String netBiosSharename, String sourceDirectory, - String destinationDirectory, Integer bufferSize, Integer numThreads, String overwrite) { + String destinationDirectory, Integer bufferSize, Integer numThreads, String overwrite + , String smbVersion) { this.netBiosDomainName = netBiosDomainName; this.netBiosHostname = netBiosHostname; @@ -277,6 +414,32 @@ public class WindowsShareCopyConfig extends PluginConfig { this.bufferSize = bufferSize; this.numThreads = numThreads; this.overwrite = !("false".equals(overwrite)); + this.smbVersion = smbVersion; + } + + /** + * Determines if the plugin should use legacy SMBv1. + * Defaults to true (SMBv1) if null, meaning old pipelines will work without modification. + */ + public boolean isSMBv1() { + return smbVersion == null || "SMBv1".equalsIgnoreCase(smbVersion); + } + + /** + * Normalizes the source directory by removing all leading slashes + * and converting to Windows-native backslashes for SMBv2/v3. + */ + public String getSourceDirectory() { + if (Strings.isNullOrEmpty(sourceDirectory)) { + return sourceDirectory; + } + String normalized = sourceDirectory; + + while (normalized.startsWith("/") || normalized.startsWith("\\")) { + normalized = normalized.substring(1); + } + + return normalized.replace('/', '\\'); } public void validate(FailureCollector collector) { diff --git a/core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java b/core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java index 8c393c377..378f73810 100644 --- a/core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java +++ b/core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java @@ -39,6 +39,7 @@ import io.cdap.plugin.alert.TMSAlertPublisher; import io.cdap.plugin.batch.action.EmailAction; import io.cdap.plugin.batch.action.SSHAction; +import io.cdap.plugin.batch.action.WindowsShareCopy; import io.cdap.plugin.batch.aggregator.DedupAggregator; import io.cdap.plugin.batch.aggregator.GroupByAggregator; import io.cdap.plugin.batch.aggregator.function.AggregateFunction; @@ -147,6 +148,7 @@ public static void setupTest() throws Exception { TMSAlertPublisher.class, ErrorCollector.class, FileSink.class, + WindowsShareCopy.class, // Spark needs this class to get exported to use the correct classloader AggregateFunction.class); // add format plugins diff --git a/core-plugins/src/test/java/io/cdap/plugin/batch/action/WindowsShareCopyConfigTest.java b/core-plugins/src/test/java/io/cdap/plugin/batch/action/WindowsShareCopyConfigTest.java new file mode 100644 index 000000000..a709c5d96 --- /dev/null +++ b/core-plugins/src/test/java/io/cdap/plugin/batch/action/WindowsShareCopyConfigTest.java @@ -0,0 +1,111 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * Licensed 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 io.cdap.plugin.batch.action; + +import io.cdap.cdap.etl.mock.validation.MockFailureCollector; +import org.junit.Assert; +import org.junit.Test; + +/** + * Unit tests for {@link WindowsShareCopy.WindowsShareCopyConfig} + */ +public class WindowsShareCopyConfigTest { + + private WindowsShareCopy.WindowsShareCopyConfig createConfig(String sourceDirectory, String smbVersion) { + return new WindowsShareCopy(null).new WindowsShareCopyConfig( + "testDomain", + "testHostname", + "testUser", + "testPassword", + "testShare", + sourceDirectory, + "/testDest", + 4096, + 1, + "true", + smbVersion + ); + } + + @Test + public void testPathNormalization() { + // 1. Test multiple leading forward slashes + WindowsShareCopy.WindowsShareCopyConfig config1 = createConfig("///folder/subfolder", "SMBv2/v3"); + Assert.assertEquals("folder\\subfolder", config1.getSourceDirectory()); + + // 2. Test multiple leading backward slashes + WindowsShareCopy.WindowsShareCopyConfig config2 = createConfig("\\\\\\folder\\subfolder", "SMBv2/v3"); + Assert.assertEquals("folder\\subfolder", config2.getSourceDirectory()); + + // 3. Test mixed/no leading slashes but internal forward slashes + WindowsShareCopy.WindowsShareCopyConfig config3 = createConfig("folder/subfolder/file.txt", "SMBv2/v3"); + Assert.assertEquals("folder\\subfolder\\file.txt", config3.getSourceDirectory()); + + // 4. Test empty string + WindowsShareCopy.WindowsShareCopyConfig config4 = createConfig("", "SMBv2/v3"); + Assert.assertEquals("", config4.getSourceDirectory()); + + // 5. Test null + WindowsShareCopy.WindowsShareCopyConfig config5 = createConfig(null, "SMBv2/v3"); + Assert.assertNull(config5.getSourceDirectory()); + } + + @Test + public void testSMBVersionFallback() { + // 1. Test null falls back to true (SMBv1) for backward compatibility + WindowsShareCopy.WindowsShareCopyConfig nullConfig = createConfig("/dir", null); + Assert.assertTrue(nullConfig.isSMBv1()); + + // 2. Test explicit SMBv1 + WindowsShareCopy.WindowsShareCopyConfig v1Config = createConfig("/dir", "SMBv1"); + Assert.assertTrue(v1Config.isSMBv1()); + + // 2b. Test explicit SMBv1 (case insensitive) + WindowsShareCopy.WindowsShareCopyConfig v1ConfigLower = createConfig("/dir", "smbv1"); + Assert.assertTrue(v1ConfigLower.isSMBv1()); + + // 3. Test explicit SMBv2/v3 + WindowsShareCopy.WindowsShareCopyConfig v2Config = createConfig("/dir", "SMBv2/v3"); + Assert.assertFalse(v2Config.isSMBv1()); + } + + @Test + public void testValidConfiguration() { + WindowsShareCopy.WindowsShareCopyConfig config = createConfig("/dir", "SMBv2/v3"); + MockFailureCollector collector = new MockFailureCollector(); + config.validate(collector); + + Assert.assertEquals(0, collector.getValidationFailures().size()); + } + + @Test + public void testInvalidConfiguration() { + // Create a config with null for all required fields + WindowsShareCopy.WindowsShareCopyConfig invalidConfig = + new WindowsShareCopy(null).new WindowsShareCopyConfig( + null, null, null, null, + null, null, null, 4096, + 1, "true", "SMBv2/v3" + ); + + MockFailureCollector collector = new MockFailureCollector(); + invalidConfig.validate(collector); + + // Should trigger 5 validation failures (hostname, username, password, sharename, sourcedir, destdir) + Assert.assertEquals(6, collector.getValidationFailures().size()); + } +} diff --git a/core-plugins/widgets/WindowsShareCopy-action.json b/core-plugins/widgets/WindowsShareCopy-action.json index 861297bf4..c47c60959 100644 --- a/core-plugins/widgets/WindowsShareCopy-action.json +++ b/core-plugins/widgets/WindowsShareCopy-action.json @@ -61,6 +61,18 @@ { "label": "Plugin Properties", "properties": [ + { + "widget-type": "select", + "label": "SMB Protocol Version", + "name": "smbVersion", + "widget-attributes": { + "values": [ + "SMBv2/v3", + "SMBv1" + ], + "default": "SMBv2/v3" + } + }, { "widget-type": "textbox", "label": "Number of parallel tasks",