From d173f328cebda7660653ebb61a1ab70fe5de4ad7 Mon Sep 17 00:00:00 2001 From: vishwasvaidya-cloudsufi Date: Mon, 20 Jul 2026 17:53:21 +0530 Subject: [PATCH 1/2] Migrate WindowsShareCopy from jcifs to smbj for SMBv2/v3 support --- core-plugins/pom.xml | 24 ++- .../plugin/batch/action/WindowsShareCopy.java | 159 +++++++++++------- 2 files changed, 115 insertions(+), 68 deletions(-) diff --git a/core-plugins/pom.xml b/core-plugins/pom.xml index 8bcee2fd8..1b6f59574 100644 --- a/core-plugins/pom.xml +++ b/core-plugins/pom.xml @@ -111,11 +111,27 @@ 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.68 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..1f13ed95f 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; @@ -30,8 +40,6 @@ import io.cdap.cdap.etl.api.StageConfigurer; import io.cdap.cdap.etl.api.action.Action; import io.cdap.cdap.etl.api.action.ActionContext; -import jcifs.smb.NtlmPasswordAuthentication; -import jcifs.smb.SmbFile; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -41,6 +49,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 +65,7 @@ /** * {@link WindowsShareCopy} is an {@link Action} that will copy the data from a Windows share into an HDFS directory. + * Supports SMBv2 and SMBv3 via the smbj library. */ @Plugin(type = Action.PLUGIN_TYPE) @Name("WindowsShareCopy") @@ -81,35 +93,18 @@ public void run(ActionContext context) throws Exception { config.numThreads; config.bufferSize = (config.bufferSize == null || config.bufferSize < MIN_BUFFER_SIZE) ? MIN_BUFFER_SIZE : config.bufferSize; - StringBuilder sb = new StringBuilder("smb://"); - sb.append(config.netBiosHostname); - sb.append("/"); - sb.append(config.netBiosSharename); - sb.append("/"); - if (config.sourceDirectory.startsWith("/")) { - if (config.sourceDirectory.length() > 1) { - sb.append(config.sourceDirectory.substring(1)); - } - } else { - sb.append(config.sourceDirectory); + // Normalize source directory to be compatible with smbj (remove leading slashes, use backslashes) + String normalizedSourcePath = config.sourceDirectory; + if (normalizedSourcePath.startsWith("/") || normalizedSourcePath.startsWith("\\")) { + normalizedSourcePath = normalizedSourcePath.substring(1); } - final String smbDirectory = sb.toString(); - - // Register the SMB File handler. - jcifs.Config.registerSmbURLHandler(); - // Set Jcifs Log level to log debug information also - jcifs.Config.setProperty("jcifs.util.loglevel", "4"); - - // Authentication with NTLM and read the directory from the Windows Share. - final NtlmPasswordAuthentication auth = new NtlmPasswordAuthentication(config.netBiosDomainName, - config.netBiosUsername, - config.netBiosPassword); - SmbFile dir = new SmbFile(smbDirectory, auth); + normalizedSourcePath = normalizedSourcePath.replace('/', '\\'); // 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); @@ -117,64 +112,100 @@ public void run(ActionContext context) throws Exception { hdfs.mkdirs(hdfsDir); } - String[] files = dir.list(); - // Copies the files in a multithreaded way - CountDownLatch executorTerminateLatch = new CountDownLatch(1); - ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch); - CompletionService completionService = new ExecutorCompletionService<>(executorService); - - try { - for (final String file : files) { - 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; + SMBClient client = new SMBClient(); + try (Connection connection = client.connect(config.netBiosHostname)) { + AuthenticationContext authContext = new AuthenticationContext( + config.netBiosUsername, + config.netBiosPassword.toCharArray(), + config.netBiosDomainName == null ? "" : config.netBiosDomainName + ); + + Session session = connection.authenticate(authContext); + + try (DiskShare share = (DiskShare) session.connectShare(config.netBiosSharename)) { + + List filesToCopy = new ArrayList<>(); + + // Determine if path is a folder or single file + if (share.folderExists(normalizedSourcePath)) { + for (FileIdBothDirectoryInformation f : share.list(normalizedSourcePath)) { + String fileName = f.getFileName(); + if (fileName.equals(".") || fileName.equals("..")) { + continue; } + String fullPath = normalizedSourcePath.isEmpty() ? fileName : normalizedSourcePath + "\\" + fileName; + filesToCopy.add(fullPath); } - }); - } + } else if (share.fileExists(normalizedSourcePath)) { + filesToCopy.add(normalizedSourcePath); + } else { + throw new IllegalArgumentException( + String.format("The source path '%s' does not exist on share '%s'", normalizedSourcePath, + config.netBiosSharename) + ); + } + // Copies the files in a multithreaded way + CountDownLatch executorTerminateLatch = new CountDownLatch(1); + ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch); + CompletionService completionService = new ExecutorCompletionService<>(executorService); - int count = 0; - while (count < files.length) { try { - Future fileWritten = completionService.take(); - String fileName = fileWritten.get(); - if (fileName != null) { - LOG.debug("{} is copied", fileName); + for (final String filePath : filesToCopy) { + completionService.submit(new Callable() { + @Override + public String call() throws Exception { + try { + return copyFileToHDFS(hdfs, share, filePath, hdfsDir); + } catch (Exception e) { + LOG.warn("Exception while copying the file {}", filePath, e); + return null; + } + } + }); + } + + 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++; } - } catch (Throwable t) { - throw Throwables.propagate(t); + } finally { + executorService.shutdownNow(); + executorTerminateLatch.await(); } - count++; } - } finally { - executorService.shutdownNow(); - executorTerminateLatch.await(); } } - private String copyFileToHDFS(FileSystem hdfs, String smbSourceFile, Path dest, NtlmPasswordAuthentication auth) + private String copyFileToHDFS(FileSystem hdfs, DiskShare share, String smbSourceFile, Path dest) throws IOException { - SmbFile smbFile = new SmbFile(smbSourceFile, auth); - String name = smbFile.getName(); + 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(), - smbFile.getName(), destFile.getName()); + name, 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()); return null; } LOG.info("Copying file {} to {}", smbSourceFile, destFile.toString()); - try (InputStream in = smbFile.getInputStream(); + 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); } catch (IOException e) { From 5b8b73d8bda4ebd7641ce7cc839cf6a78c258512 Mon Sep 17 00:00:00 2001 From: vishwasvaidya-cloudsufi Date: Fri, 24 Jul 2026 16:01:37 +0530 Subject: [PATCH 2/2] Add SMBv2/v3 support to WindowsShareCopy with backward compatibility --- core-plugins/pom.xml | 10 +- .../plugin/batch/action/WindowsShareCopy.java | 163 ++++++++++++++++-- .../widgets/WindowsShareCopy-action.json | 12 ++ 3 files changed, 165 insertions(+), 20 deletions(-) diff --git a/core-plugins/pom.xml b/core-plugins/pom.xml index 1b6f59574..73e378750 100644 --- a/core-plugins/pom.xml +++ b/core-plugins/pom.xml @@ -111,13 +111,18 @@ commons-lang commons-lang - + + + jcifs + jcifs + 1.3.17 + + com.hierynomus smbj 0.11.5 - net.engio mbassador @@ -133,7 +138,6 @@ bcprov-jdk15to18 1.68 - 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 1f13ed95f..e401d7384 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 @@ -40,6 +40,8 @@ import io.cdap.cdap.etl.api.StageConfigurer; import io.cdap.cdap.etl.api.action.Action; import io.cdap.cdap.etl.api.action.ActionContext; +import jcifs.smb.NtlmPasswordAuthentication; +import jcifs.smb.SmbFile; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -65,7 +67,7 @@ /** * {@link WindowsShareCopy} is an {@link Action} that will copy the data from a Windows share into an HDFS directory. - * Supports SMBv2 and SMBv3 via the smbj library. + * Supports SMBv1 (via jcifs) and SMBv2/SMBv3 (via smbj) based on user configuration. */ @Plugin(type = Action.PLUGIN_TYPE) @Name("WindowsShareCopy") @@ -93,12 +95,6 @@ public void run(ActionContext context) throws Exception { config.numThreads; config.bufferSize = (config.bufferSize == null || config.bufferSize < MIN_BUFFER_SIZE) ? MIN_BUFFER_SIZE : config.bufferSize; - // Normalize source directory to be compatible with smbj (remove leading slashes, use backslashes) - String normalizedSourcePath = config.sourceDirectory; - if (normalizedSourcePath.startsWith("/") || normalizedSourcePath.startsWith("\\")) { - normalizedSourcePath = normalizedSourcePath.substring(1); - } - normalizedSourcePath = normalizedSourcePath.replace('/', '\\'); // Determines the buffer size to be used during writing. Configuration conf = new Configuration(); @@ -112,6 +108,119 @@ public void run(ActionContext context) throws Exception { 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("/"); + sb.append(config.netBiosSharename); + sb.append("/"); + if (config.sourceDirectory.startsWith("/")) { + if (config.sourceDirectory.length() > 1) { + sb.append(config.sourceDirectory.substring(1)); + } + } else { + sb.append(config.sourceDirectory); + } + final String smbDirectory = sb.toString(); + + // Register the SMB File handler. + jcifs.Config.registerSmbURLHandler(); + // Set Jcifs Log level to log debug information also + jcifs.Config.setProperty("jcifs.util.loglevel", "4"); + + // Authentication with NTLM and read the directory from the Windows Share. + final NtlmPasswordAuthentication auth = new NtlmPasswordAuthentication(config.netBiosDomainName, + config.netBiosUsername, + config.netBiosPassword); + SmbFile dir = new SmbFile(smbDirectory, auth); + String[] files = dir.list(); + // Copies the files in a multithreaded way + CountDownLatch executorTerminateLatch = new CountDownLatch(1); + ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch); + CompletionService completionService = new ExecutorCompletionService<>(executorService); + + try { + for (final String file : files) { + 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; + } + } + }); + } + + int count = 0; + while (count < files.length) { + 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(); + } + } + + private String copyFileToHDFS(FileSystem hdfs, String smbSourceFile, Path dest, NtlmPasswordAuthentication auth) + 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()); + // If file already exists, then we skip over. + 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 (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; + } + + /** + * The new SMBv2/v3 logic + */ + private void executeSMBv2(final Path hdfsDir, final FileSystem hdfs) throws Exception { + String normalizedSourcePath = config.sourceDirectory; + if (normalizedSourcePath.startsWith("/") || normalizedSourcePath.startsWith("\\")) { + normalizedSourcePath = normalizedSourcePath.substring(1); + } + normalizedSourcePath = normalizedSourcePath.replace('/', '\\'); + SMBClient client = new SMBClient(); try (Connection connection = client.connect(config.netBiosHostname)) { AuthenticationContext authContext = new AuthenticationContext( @@ -122,11 +231,10 @@ public void run(ActionContext context) throws Exception { Session session = connection.authenticate(authContext); - try (DiskShare share = (DiskShare) session.connectShare(config.netBiosSharename)) { + try (final DiskShare share = (DiskShare) session.connectShare(config.netBiosSharename)) { List filesToCopy = new ArrayList<>(); - // Determine if path is a folder or single file if (share.folderExists(normalizedSourcePath)) { for (FileIdBothDirectoryInformation f : share.list(normalizedSourcePath)) { String fileName = f.getFileName(); @@ -140,11 +248,11 @@ public void run(ActionContext context) throws Exception { filesToCopy.add(normalizedSourcePath); } else { throw new IllegalArgumentException( - String.format("The source path '%s' does not exist on share '%s'", normalizedSourcePath, - config.netBiosSharename) + String.format("The source path '%s' does not exist on share '%s'", + normalizedSourcePath, config.netBiosSharename) ); } - // Copies the files in a multithreaded way + CountDownLatch executorTerminateLatch = new CountDownLatch(1); ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch); CompletionService completionService = new ExecutorCompletionService<>(executorService); @@ -155,7 +263,7 @@ public void run(ActionContext context) throws Exception { @Override public String call() throws Exception { try { - return copyFileToHDFS(hdfs, share, filePath, hdfsDir); + return copyFileToHDFS_SMBv2(hdfs, share, filePath, hdfsDir); } catch (Exception e) { LOG.warn("Exception while copying the file {}", filePath, e); return null; @@ -185,19 +293,24 @@ public String call() throws Exception { } } - private String copyFileToHDFS(FileSystem hdfs, DiskShare share, String smbSourceFile, Path dest) - throws IOException { + /** + * 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 file already exists, then we skip over. + 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), @@ -294,9 +407,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; @@ -308,6 +428,15 @@ 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); } public void validate(FailureCollector collector) { 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",