diff --git a/core-plugins/pom.xml b/core-plugins/pom.xml index 8bcee2fd8..73e378750 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.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 a683c65a2..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 @@ -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); @@ -162,12 +190,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()); @@ -183,6 +211,122 @@ private String copyFileToHDFS(FileSystem hdfs, String smbSourceFile, Path dest, 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( + config.netBiosUsername, + config.netBiosPassword.toCharArray(), + config.netBiosDomainName == null ? "" : config.netBiosDomainName + ); + + Session session = connection.authenticate(authContext); + + try (final DiskShare share = (DiskShare) session.connectShare(config.netBiosSharename)) { + + List filesToCopy = new ArrayList<>(); + + 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) + ); + } + + 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 { + return copyFileToHDFS_SMBv2(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++; + } + } 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); + } catch (IOException e) { + LOG.warn("Exception in copying the file {}", name, e); + } + return name; + } + /** * Creates an {@link ExecutorService} that has the given number of threads. * @@ -263,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; @@ -277,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",