From 348fa727b198ba8ce9ef41f8833bc26be98bc9e1 Mon Sep 17 00:00:00 2001 From: "Remi GASCOU (Podalirius)" <79218792+p0dalirius@users.noreply.github.com> Date: Wed, 18 Dec 2024 17:35:27 +0100 Subject: [PATCH] Fixed umount and fixed #119 --- smbclientng/core/InteractiveShell.py | 4 ++ smbclientng/core/SMBSession.py | 90 +++++++++++++++++++++------- 2 files changed, 74 insertions(+), 20 deletions(-) diff --git a/smbclientng/core/InteractiveShell.py b/smbclientng/core/InteractiveShell.py index 9e7cf32..fbcc796 100644 --- a/smbclientng/core/InteractiveShell.py +++ b/smbclientng/core/InteractiveShell.py @@ -314,6 +314,10 @@ def process_line(self, commandLine: str): elif command == "tree": self.command_tree(arguments, command) + # Removes a local mount point + elif command == "umount": + self.command_umount(arguments, command) + # Use a share elif command == "use": self.command_use(arguments, command) diff --git a/smbclientng/core/SMBSession.py b/smbclientng/core/SMBSession.py index 0b01d89..2913377 100644 --- a/smbclientng/core/SMBSession.py +++ b/smbclientng/core/SMBSession.py @@ -6,7 +6,6 @@ from __future__ import annotations import io -import impacket.smbconnection from typing import Optional from impacket.smbconnection import SMBConnection, SessionError from impacket.dcerpc.v5 import transport, rpcrt, srvs @@ -18,6 +17,7 @@ import re import sys import traceback +import subprocess from smbclientng.core.LocalFileIO import LocalFileIO from smbclientng.core.utils import b_filesize, STYPE_MASK, is_port_open, smb_entry_iterator from smbclientng.core.SIDResolver import SIDResolver @@ -830,29 +830,63 @@ def mount(self, local_mount_point: str, remote_path: str): None """ + executable = None + args = None + if not os.path.exists(local_mount_point): pass - if sys.platform.startswith('win'): - remote_path = remote_path.replace('/',ntpath.sep) - command = f"net use {local_mount_point} \\\\{self.host}\\{self.smb_share}\\{remote_path}" + if sys.platform.startswith("win"): + remote_path = remote_path.replace('/', ntpath.sep) + + executable = "net" + args = [ + "use", + f"{local_mount_point}", + f"\\\\{self.host}\\{self.smb_share}\\{remote_path}", + ] - elif sys.platform.startswith('linux'): - remote_path = remote_path.replace(ntpath.sep,'/') - command = f"mount -t cifs //{self.host}/{self.smb_share}/{remote_path} {local_mount_point} -o username={self.credentials.username},password={self.credentials.password}" + elif sys.platform.startswith("linux"): + remote_path = remote_path.replace(ntpath.sep, '/') + password = self.credentials.password.replace('"', '\\"') + + executable = "mount" + args = [ + "-t", + "cifs", + f"//{self.host}/{self.smb_share}/{remote_path}", + f"{local_mount_point}", + "-o", + f"username={self.credentials.username},password={password}" + ] - elif sys.platform.startswith('darwin'): - remote_path = remote_path.replace(ntpath.sep,'/') - command = f"mount_smbfs //{self.credentials.username}:{self.credentials.password}@{self.host}/{self.smb_share}/{remote_path} {local_mount_point}" + elif sys.platform.startswith("darwin"): + remote_path = remote_path.replace(ntpath.sep, '/') + password = self.credentials.password.replace('"', '\\"') + + executable = "mount_smbfs" + args = [ + f"//{self.credentials.username}:{self.credentials.password}@{self.host}/{self.smb_share}/{remote_path}", + f"{local_mount_point}", + ] else: - command = None + executable = None + args = None self.logger.error("Unsupported platform for mounting SMB share.") - if command is not None: + if executable is not None: if self.config.debug: - self.logger.debug("Executing: %s" % command) - os.system(command) + self.logger.debug("Executing '%s' with arguments: %s" % (executable, args)) + process = subprocess.Popen(executable=executable, args=args, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + stdout, stderr = process.communicate() + if process.returncode != 0: + self.logger.error("Command failed with error: %s" % stderr.decode()) + else: + output = stdout.decode() + if len(output) != 0: + self.logger.debug("Command output: %s" % stdout.decode()) + self.logger.info(f"Successfully mounted '\\\\{self.host}\\{self.smb_share}\\{remote_path}' to local '{local_mount_point}'.") def path_exists(self, path: Optional[str] = None): """ @@ -1324,18 +1358,34 @@ def umount(self, local_mount_point: str): if os.path.exists(local_mount_point): if sys.platform.startswith('win'): - command = f"net use {local_mount_point} /delete" + executable = "net" + args = [ + "use", + f"{local_mount_point}", + "/delete" + ] elif sys.platform.startswith('linux') or sys.platform.startswith('darwin'): - command = f"umount {local_mount_point}" + executable = "umount" + args = [f"{local_mount_point}"] else: - command = None + executable = None + args = None self.logger.error("Unsupported platform for unmounting SMB share.") - if command is not None: - self.logger.debug("Executing: %s" % command) - os.system(command) + if executable is not None: + if self.config.debug: + self.logger.debug("Executing '%s' with arguments: %s" % (executable, args)) + process = subprocess.Popen(executable=executable, args=args, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + stdout, stderr = process.communicate() + if process.returncode != 0: + self.logger.error("Command failed with error: %s" % stderr.decode()) + else: + output = stdout.decode() + if len(output) != 0: + self.logger.debug("Command output: %s" % stdout.decode()) + self.logger.info(f"Successfully unmounted local path '{local_mount_point}'.") else: self.logger.error("Cannot unmount a non existing path.")