smbclientng.core.SMBSession API documentation

SMBSession( address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None) View Source

47    def __init__(self, address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None):
48        super(SMBSession, self).__init__()
49        # Objects
50        self.config = config
51
52        # Target server
53        self.address = address
54
55        # Credentials
56        self.domain = domain
57        self.username = username
58        self.password = password 
59        self.lmhash = lmhash
60        self.nthash = nthash
61        self.use_kerberos = use_kerberos
62        self.kdcHost = kdcHost
63
64        self.smbClient = None
65        self.connected = False
66
67        self.available_shares = {}
68        self.smb_share = None
69        self.smb_cwd = ""
70        self.smb_tree_id = None
71
72        self.list_shares()

config

address

domain

username

password

lmhash

nthash

use_kerberos

kdcHost

smbClient

connected

available_shares

smb_share

smb_cwd

smb_tree_id

def init_smb_session(self): View Source

 76    def init_smb_session(self):
 77        """
 78        Initializes and establishes a session with the SMB server.
 79
 80        This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration.
 81        It attempts to connect to the SMB server specified by the `address` attribute and authenticate using the credentials provided during the object's initialization.
 82
 83        The method will print debug information if the `debug` attribute is set to True. Upon successful connection and authentication, it sets the `connected` attribute to True.
 84
 85        Returns:
 86            bool: True if the connection and authentication are successful, False otherwise.
 87        """
 88
 89        self.connected = False
 90
 91        if self.config.debug:
 92            print("[debug] [>] Connecting to remote SMB server '%s' ... " % self.address)
 93        try:
 94            self.smbClient = impacket.smbconnection.SMBConnection(
 95                remoteName=self.address,
 96                remoteHost=self.address,
 97                sess_port=int(445)
 98            )
 99        except OSError as err:
100            print("[!] %s" % err)
101            self.smbClient = None
102
103        if self.smbClient is not None:
104            if self.use_kerberos:
105                if self.config.debug:
106                    print("[debug] [>] Authenticating as '%s\\%s' with kerberos ... " % (self.domain, self.username))
107                try:
108                    self.connected = self.smbClient.kerberosLogin(
109                        user=self.username,
110                        password=self.password,
111                        domain=self.domain,
112                        lmhash=self.lmhash,
113                        nthash=self.nthash,
114                        aesKey=self.aesKey,
115                        kdcHost=self.kdcHost
116                    )
117                except impacket.smbconnection.SessionError as err:
118                    if self.config.debug:
119                        traceback.print_exc()
120                    print("[!] Could not login: %s" % err)
121                    self.connected = False
122
123            else:
124                if self.config.debug:
125                    print("[debug] [>] Authenticating as '%s\\%s' with NTLM ... " % (self.domain, self.username))
126                try:
127                    self.connected = self.smbClient.login(
128                        user=self.username,
129                        password=self.password,
130                        domain=self.domain,
131                        lmhash=self.lmhash,
132                        nthash=self.nthash
133                    )
134                except impacket.smbconnection.SessionError as err:
135                    if self.config.debug:
136                        traceback.print_exc()
137                    print("[!] Could not login: %s" % err)
138                    self.connected = False
139
140            if self.connected:
141                print("[+] Successfully authenticated to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username))
142            else:
143                print("[!] Failed to authenticate to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username))
144
145        return self.connected

Initializes and establishes a session with the SMB server.

This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration. It attempts to connect to the SMB server specified by the address attribute and authenticate using the credentials provided during the object's initialization.

The method will print debug information if the debug attribute is set to True. Upon successful connection and authentication, it sets the connected attribute to True.

Returns: bool: True if the connection and authentication are successful, False otherwise.

def close_smb_session(self): View Source

147    def close_smb_session(self):
148        """
149        Closes the current SMB session by disconnecting the SMB client.
150
151        This method ensures that the SMB client connection is properly closed. It checks if the client is connected
152        and if so, it closes the connection and resets the connection status.
153
154        Raises:
155            Exception: If the SMB client is not initialized or if there's an error during the disconnection process.
156        """
157
158        if self.smbClient is not None:
159            if self.connected:
160                self.smbClient.close()
161                self.connected = False
162                if self.config.debug:
163                    print("[+] SMB connection closed successfully.")
164            else:
165                if self.config.debug:
166                    print("[!] No active SMB connection to close.")
167        else:
168            raise Exception("SMB client is not initialized.")

Closes the current SMB session by disconnecting the SMB client.

This method ensures that the SMB client connection is properly closed. It checks if the client is connected and if so, it closes the connection and resets the connection status.

Raises: Exception: If the SMB client is not initialized or if there's an error during the disconnection process.

def read_file(self, path=None): View Source

172    def read_file(self, path=None):
173        if self.path_isfile(path=path):
174            tmp_file_path = self.smb_cwd + ntpath.sep + path
175            matches = self.smbClient.listPath(
176                shareName=self.smb_share, 
177                path=tmp_file_path
178            )
179
180            fh = io.BytesIO()
181            try:
182                # opening the files in streams instead of mounting shares allows 
183                # for running the script from unprivileged containers
184                self.smbClient.getFile(self.smb_share, tmp_file_path, fh.write)
185            except impacket.smbconnection.SessionError as e:
186                return None
187            rawdata = fh.getvalue()
188            fh.close()
189            return rawdata
190        else:
191            print("[!] Remote path '%s' is not a file." % path)

def find(self, paths=[], callback=None): View Source

193    def find(self, paths=[], callback=None):
194        def recurse_action(paths=[], depth=0, callback=None):
195            if callback is None:
196                return []
197            
198            next_directories_to_explore = []
199
200            for path in paths:
201                remote_smb_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
202                entries = []
203                
204                try:
205                    entries = self.smbClient.listPath(
206                        shareName=self.smb_share, 
207                        path=(remote_smb_path + ntpath.sep + '*')
208                    )
209                except impacket.smbconnection.SessionError as err:
210                    continue 
211                # Remove dot names
212                entries = [e for e in entries if e.get_longname() not in [".", ".."]]
213                # Sort the entries ignoring case
214                entries = sorted(entries, key=lambda x:x.get_longname().lower())
215                
216                for entry in entries:
217                    if entry.is_directory():
218                        callback(entry, path + ntpath.sep + entry.get_longname() + ntpath.sep, depth)
219                    else:
220                        callback(entry, path + ntpath.sep + entry.get_longname(), depth)
221
222                # Next directories to explore
223                for entry in entries:
224                    if entry.is_directory():
225                        next_directories_to_explore.append(path + ntpath.sep + entry.get_longname() + ntpath.sep)
226            
227            return next_directories_to_explore
228        # 
229        if callback is not None:
230            depth = 0
231            while len(paths) != 0:
232                paths = recurse_action(
233                    paths=paths,
234                    depth=depth,
235                    callback=callback
236                )
237                depth = depth + 1
238        else:
239            print("[!] SMBSession.find(), callback function cannot be None.")

def get_file(self, path=None, keepRemotePath=False): View Source

241    def get_file(self, path=None, keepRemotePath=False):
242        """
243        Retrieves a file from the specified path on the SMB share.
244
245        This method attempts to retrieve a file from the given path within the currently connected SMB share.
246        If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local
247        file object and writing the contents of the remote file to it using the SMB client's getFile method.
248
249        Parameters:
250            path (str): The path of the file to retrieve. If None, uses the current smb_path.
251
252        Returns:
253            None
254        """
255
256        tmp_file_path = self.smb_cwd + ntpath.sep + path
257        matches = self.smbClient.listPath(
258            shareName=self.smb_share, 
259            path=tmp_file_path
260        )
261        
262        for entry in matches:
263            if entry.is_directory():
264                print("[>] Skipping '%s' because it is a directory." % tmp_file_path)
265            else:
266                try:
267                    if ntpath.sep in path:
268                        outputfile = ntpath.dirname(path) + ntpath.sep + entry.get_longname()
269                    else:
270                        outputfile = entry.get_longname()
271                    f = LocalFileIO(
272                        mode="wb", 
273                        path=outputfile,
274                        expected_size=entry.get_filesize(), 
275                        debug=self.config.debug,
276                        keepRemotePath=keepRemotePath
277                    )
278                    self.smbClient.getFile(
279                        shareName=self.smb_share, 
280                        pathName=tmp_file_path, 
281                        callback=f.write
282                    )
283                    f.close()
284                except (BrokenPipeError, KeyboardInterrupt) as e:
285                    f.close()
286                    print("\x1b[v\x1b[o\r[!] Interrupted.")
287                    self.close_smb_session()
288                    self.init_smb_session()
289                        
290        return None

Retrieves a file from the specified path on the SMB share.

This method attempts to retrieve a file from the given path within the currently connected SMB share. If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local file object and writing the contents of the remote file to it using the SMB client's getFile method.

Parameters: path (str): The path of the file to retrieve. If None, uses the current smb_path.

Returns: None

def get_file_recursively(self, path=None): View Source

292    def get_file_recursively(self, path=None):
293        """
294        Recursively retrieves files from a specified path on the SMB share.
295
296        This method navigates through all directories starting from the given path,
297        and downloads all files found. It handles directories recursively, ensuring
298        that all nested files are retrieved. The method skips over directory entries
299        and handles errors gracefully, attempting to continue the operation where possible.
300
301        Parameters:
302            path (str): The initial directory path from which to start the recursive file retrieval.
303                        If None, it starts from the root of the configured SMB share.
304        """
305        
306        def recurse_action(base_dir="", path=[]):
307            if len(base_dir) == 0:
308                remote_smb_path = ntpath.sep.join(path)
309            else:
310                remote_smb_path = base_dir + ntpath.sep + ntpath.sep.join(path)
311            remote_smb_path = ntpath.normpath(remote_smb_path)
312
313            entries = self.smbClient.listPath(
314                shareName=self.smb_share, 
315                path=remote_smb_path + '\\*'
316            )
317            if len(entries) != 0:
318                files = [entry for entry in entries if not entry.is_directory()]
319                directories = [entry for entry in entries if entry.is_directory() and entry.get_longname() not in [".", ".."]]
320
321                # Files
322                if len(files) != 0:
323                    print("[>] Retrieving files of '%s'" % remote_smb_path)
324                for entry_file in files:
325                    if not entry_file.is_directory():
326                        f = LocalFileIO(
327                            mode="wb",
328                            path=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
329                            expected_size=entry_file.get_filesize(),
330                            keepRemotePath=True,
331                            debug=self.config.debug
332                        )
333                        try:
334                            self.smbClient.getFile(
335                                shareName=self.smb_share, 
336                                pathName=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
337                                callback=f.write
338                            )
339                            f.close()
340                        except BrokenPipeError as err:
341                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
342                            f.close(remove=True)
343                            break
344                        except Exception as err:
345                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
346                            f.close(remove=True)
347                
348                # Directories
349                for entry_directory in directories:
350                    if entry_directory.is_directory():
351                        recurse_action(
352                            base_dir=self.smb_cwd, 
353                            path=path+[entry_directory.get_longname()]
354                        )                   
355        # Entrypoint
356        try:
357            recurse_action(
358                base_dir=self.smb_cwd, 
359                path=[path]
360            )
361        except (BrokenPipeError, KeyboardInterrupt) as e:
362            print("\x1b[v\x1b[o\r[!] Interrupted.")
363            self.close_smb_session()
364            self.init_smb_session()

Recursively retrieves files from a specified path on the SMB share.

This method navigates through all directories starting from the given path, and downloads all files found. It handles directories recursively, ensuring that all nested files are retrieved. The method skips over directory entries and handles errors gracefully, attempting to continue the operation where possible.

Parameters: path (str): The initial directory path from which to start the recursive file retrieval. If None, it starts from the root of the configured SMB share.

def get_entry(self, path=None): View Source

366    def get_entry(self, path=None):
367        """
368        Retrieves information about a specific entry located at the provided path on the SMB share.
369
370        This method checks if the specified path exists on the SMB share. If the path exists, it retrieves the details of the entry at that path, including the directory name and file name. If the entry is found, it returns the entry object; otherwise, it returns None.
371
372        Args:
373            path (str): The path of the entry to retrieve information about.
374
375        Returns:
376            Entry: An object representing the entry at the specified path, or None if the entry is not found.
377        """
378
379        if self.path_exists(path=path):
380            matches = self.smbClient.listPath(shareName=self.smb_share, path=path)
381
382            if len(matches) == 1:
383                return matches[0]
384            else:
385                return None
386            
387        else:
388            return None

Retrieves information about a specific entry located at the provided path on the SMB share.

This method checks if the specified path exists on the SMB share. If the path exists, it retrieves the details of the entry at that path, including the directory name and file name. If the entry is found, it returns the entry object; otherwise, it returns None.

Args: path (str): The path of the entry to retrieve information about.

Returns: Entry: An object representing the entry at the specified path, or None if the entry is not found.

def info(self, share=True, server=True): View Source

390    def info(self, share=True, server=True):
391        """
392        Displays information about the server and optionally the shares.
393
394        This method prints detailed information about the server's characteristics such as NetBIOS names, DNS details, OS information, and SMB capabilities. If the `share` parameter is set to True and a share is currently set, it will also attempt to display information about the share.
395
396        Parameters:
397            share (bool): If True, display information about the current share.
398            server (bool): If True, display information about the server.
399
400        Returns:
401            None
402        """
403
404        if server:
405            if self.config.no_colors:
406                print("[+] Server:")
407                print("  ├─NetBIOS:")
408                print("  │ ├─ NetBIOS Hostname ──────── : %s" % (self.smbClient.getServerName()))
409                print("  │ └─ NetBIOS Domain ────────── : %s" % (self.smbClient.getServerDomain()))
410                print("  ├─DNS:")
411                print("  │ ├─ DNS Hostname ──────────── : %s" % (self.smbClient.getServerDNSHostName()))
412                print("  │ └─ DNS Domain ────────────── : %s" % (self.smbClient.getServerDNSDomainName()))
413                print("  ├─OS:")
414                print("  │ ├─ OS Name ───────────────── : %s" % (self.smbClient.getServerOS()))
415                print("  │ └─ OS Version ────────────── : %s.%s.%s" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild()))
416                print("  ├─Server:")
417                print("  │ ├─ Signing Required ──────── : %s" % (self.smbClient.isSigningRequired()))
418                print("  │ ├─ Login Required ────────── : %s" % (self.smbClient.isLoginRequired()))
419                print("  │ ├─ Supports NTLMv2 ───────── : %s" % (self.smbClient.doesSupportNTLMv2()))
420                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
421                print("  │ ├─ Max size of read chunk ── : %d bytes (%s)" % (MaxReadSize, b_filesize(MaxReadSize)))
422                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
423                print("  │ └─ Max size of write chunk ─ : %d bytes (%s)" % (MaxWriteSize, b_filesize(MaxWriteSize)))
424                print("  └─")
425            else:
426                print("[+] Server:")
427                print("  ├─NetBIOS:")
428                print("  │ ├─ \x1b[94mNetBIOS Hostname\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerName()))
429                print("  │ └─ \x1b[94mNetBIOS Domain\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDomain()))
430                print("  ├─DNS:")
431                print("  │ ├─ \x1b[94mDNS Hostname\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSHostName()))
432                print("  │ └─ \x1b[94mDNS Domain\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSDomainName()))
433                print("  ├─OS:")
434                print("  │ ├─ \x1b[94mOS Name\x1b[0m \x1b[90m─────────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerOS()))
435                print("  │ └─ \x1b[94mOS Version\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s.%s.%s\x1b[0m" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild()))
436                print("  ├─Server:")
437                print("  │ ├─ \x1b[94mSigning Required\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isSigningRequired()))
438                print("  │ ├─ \x1b[94mLogin Required\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isLoginRequired()))
439                print("  │ ├─ \x1b[94mSupports NTLMv2\x1b[0m \x1b[90m─────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.doesSupportNTLMv2()))
440                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
441                print("  │ ├─ \x1b[94mMax size of read chunk\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxReadSize, b_filesize(MaxReadSize)))
442                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
443                print("  │ └─ \x1b[94mMax size of write chunk\x1b[0m \x1b[90m─\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxWriteSize, b_filesize(MaxWriteSize)))
444                print("  └─")
445
446        if share and self.smb_share is not None:
447            share_name = self.available_shares.get(self.smb_share.lower(), "")["name"]
448            share_comment = self.available_shares.get(self.smb_share.lower(), "")["comment"]
449            share_type = self.available_shares.get(self.smb_share.lower(), "")["type"]
450            share_type =', '.join([s.replace("STYPE_","") for s in share_type])
451            share_rawtype = self.available_shares.get(self.smb_share.lower(), "")["rawtype"]
452            if self.config.no_colors:
453                print("\n[+] Share:")
454                print("  ├─ Name ──────────── : %s" % (share_name))
455                print("  ├─ Description ───── : %s" % (share_comment))
456                print("  ├─ Type ──────────── : %s" % (share_type))
457                print("  └─ Raw type value ── : %s" % (share_rawtype))
458            else:
459                print("\n[+] Share:")
460                print("  ├─ \x1b[94mName\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_name))
461                print("  ├─ \x1b[94mDescription\x1b[0m \x1b[90m─────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_comment))
462                print("  ├─ \x1b[94mType\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_type))
463                print("  └─ \x1b[94mRaw type value\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%s\x1b[0m" % (share_rawtype))

Displays information about the server and optionally the shares.

This method prints detailed information about the server's characteristics such as NetBIOS names, DNS details, OS information, and SMB capabilities. If the share parameter is set to True and a share is currently set, it will also attempt to display information about the share.

Parameters: share (bool): If True, display information about the current share. server (bool): If True, display information about the server.

Returns: None

def list_contents(self, path=None): View Source

465    def list_contents(self, path=None):
466        """
467        Lists the contents of a specified directory on the SMB share.
468
469        This method retrieves the contents of a directory specified by `shareName` and `path`. If `shareName` or `path`
470        is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with
471        the long names of the files and directories as keys and their respective SMB entry objects as values.
472
473        Args:
474            shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None.
475            path (str, optional): The directory path to list contents from. Defaults to the current path if None.
476
477        Returns:
478            dict: A dictionary with file and directory names as keys and their SMB entry objects as values.
479        """
480        
481        dest_path = [self.smb_cwd.rstrip(ntpath.sep),]
482        if path is not None and len(path) > 0:
483            dest_path.append(path.rstrip(ntpath.sep))
484        dest_path.append('*')
485        path = ntpath.sep.join(dest_path)
486
487        contents = {}
488        entries = self.smbClient.listPath(
489            shareName=self.smb_share, 
490            path=path
491        )
492        for entry in entries:
493            contents[entry.get_longname()] = entry
494
495        return contents

Lists the contents of a specified directory on the SMB share.

This method retrieves the contents of a directory specified by shareName and path. If shareName or path is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with the long names of the files and directories as keys and their respective SMB entry objects as values.

Args: shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None. path (str, optional): The directory path to list contents from. Defaults to the current path if None.

Returns: dict: A dictionary with file and directory names as keys and their SMB entry objects as values.

def list_shares(self): View Source

497    def list_shares(self):
498        """
499        Lists all the shares available on the connected SMB server.
500
501        This method queries the SMB server to retrieve a list of all available shares. It populates the `shares` dictionary
502        with key-value pairs where the key is the share name and the value is a dictionary containing details about the share
503        such as its name, type, raw type, and any comments associated with the share.
504
505        Returns:
506            dict: A dictionary containing information about each share available on the server.
507        """
508
509        self.available_shares = {}
510
511        if self.connected:
512            if self.smbClient is not None:
513                resp = self.smbClient.listShares()
514
515                for share in resp:
516                    # SHARE_INFO_1 structure (lmshare.h)
517                    # https://learn.microsoft.com/en-us/windows/win32/api/lmshare/ns-lmshare-share_info_1
518                    sharename = share["shi1_netname"][:-1]
519                    sharecomment = share["shi1_remark"][:-1]
520                    sharetype = share["shi1_type"]
521
522                    self.available_shares[sharename.lower()] = {
523                        "name": sharename, 
524                        "type": STYPE_MASK(sharetype), 
525                        "rawtype": sharetype, 
526                        "comment": sharecomment
527                    }
528            else:
529                print("[!] Error: SMBSession.smbClient is None.")
530
531        return self.available_shares

Lists all the shares available on the connected SMB server.

This method queries the SMB server to retrieve a list of all available shares. It populates the shares dictionary with key-value pairs where the key is the share name and the value is a dictionary containing details about the share such as its name, type, raw type, and any comments associated with the share.

Returns: dict: A dictionary containing information about each share available on the server.

def mkdir(self, path=None): View Source

533    def mkdir(self, path=None):
534        """
535        Creates a directory at the specified path on the SMB share.
536
537        This method takes a path and attempts to create the directory structure on the SMB share. If the path includes
538        nested directories, it will create each directory in the sequence. If a directory already exists, it will skip
539        the creation for that directory without raising an error.
540
541        Args:
542            path (str, optional): The full path of the directory to create on the SMB share. Defaults to None.
543
544        Note:
545            The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility.
546        """
547
548        if path is not None:
549            # Prepare path
550            path = path.replace('/',ntpath.sep)
551            if ntpath.sep in path:
552                path = path.strip(ntpath.sep).split(ntpath.sep)
553            else:
554                path = [path]
555
556            # Create each dir in the path
557            for depth in range(1, len(path)+1):
558                tmp_path = ntpath.sep.join(path[:depth])
559                try:
560                    self.smbClient.createDirectory(
561                        shareName=self.smb_share, 
562                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + tmp_path + ntpath.sep)
563                    )
564                except impacket.smbconnection.SessionError as err:
565                    if err.getErrorCode() == 0xc0000035:
566                        # STATUS_OBJECT_NAME_COLLISION
567                        # Remote directory already created, this is normal
568                        # Src: https://github.com/fortra/impacket/blob/269ce69872f0e8f2188a80addb0c39fedfa6dcb8/impacket/nt_errors.py#L268C9-L268C19
569                        pass
570                    else:
571                        print("[!] Failed to create directory '%s': %s" % (tmp_path, err))
572                        if self.config.debug:
573                            traceback.print_exc()
574        else:
575            pass

Creates a directory at the specified path on the SMB share.

This method takes a path and attempts to create the directory structure on the SMB share. If the path includes nested directories, it will create each directory in the sequence. If a directory already exists, it will skip the creation for that directory without raising an error.

Args: path (str, optional): The full path of the directory to create on the SMB share. Defaults to None.

Note: The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility.

def mount(self, local_mount_point, remote_path): View Source

577    def mount(self, local_mount_point, remote_path):
578
579        if not os.path.exists(local_mount_point):
580            pass
581
582        if sys.platform.startswith('win'):
583            remote_path = remote_path.replace('/',ntpath.sep)
584            command = f"net use {local_mount_point} \\\\{self.address}\\{self.smb_share}\\{remote_path}"
585        
586        elif sys.platform.startswith('linux'):
587            remote_path = remote_path.replace(ntpath.sep,'/')
588            command = f"mount -t cifs //{self.address}/{self.smb_share}/{remote_path} {local_mount_point} -o username={self.username},password={self.password}"
589        
590        elif sys.platform.startswith('darwin'):
591            remote_path = remote_path.replace(ntpath.sep,'/')
592            command = f"mount_smbfs //{self.username}:{self.password}@{self.address}/{self.smb_share}/{remote_path} {local_mount_point}"
593        
594        else:
595            command = None
596            print("[!] Unsupported platform for mounting SMB share.")
597        
598        if command is not None:
599            if self.config.debug:
600                print("[debug] Executing: %s" % command)
601            os.system(command)

def path_exists(self, path=None): View Source

603    def path_exists(self, path=None):
604        """
605        Checks if the specified path exists on the SMB share.
606
607        This method determines if a given path exists on the SMB share by attempting to list the contents of the path.
608        If the path listing is successful and returns one or more entries, the path is considered to exist.
609
610        Args:
611            path (str, optional): The path to check on the SMB share. Defaults to None.
612
613        Returns:
614            bool: True if the path exists, False otherwise or if an error occurs.
615        """
616
617        if path is not None:
618            path = path.replace('*','')
619            try:
620                contents = self.smbClient.listPath(
621                    shareName=self.smb_share,
622                    path=ntpath.normpath(self.smb_cwd + ntpath.sep + path + ntpath.sep)
623                )
624                return (len(contents) != 0)
625            except Exception as e:
626                return False
627        else:
628            return False

Checks if the specified path exists on the SMB share.

This method determines if a given path exists on the SMB share by attempting to list the contents of the path. If the path listing is successful and returns one or more entries, the path is considered to exist.

Args: path (str, optional): The path to check on the SMB share. Defaults to None.

Returns: bool: True if the path exists, False otherwise or if an error occurs.

def path_isdir(self, pathFromRoot=None): View Source

630    def path_isdir(self, pathFromRoot=None):
631        """
632        Checks if the specified path is a directory on the SMB share.
633
634        This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the
635        contents of the path and filtering for entries that match the basename of the path and are marked as directories.
636
637        Args:
638            path (str, optional): The path to check on the SMB share. Defaults to None.
639
640        Returns:
641            bool: True if the path is a directory, False otherwise or if an error occurs.
642        """
643
644        if pathFromRoot is not None:
645            # Replace slashes if any
646            path = pathFromRoot.replace('/', ntpath.sep)
647            
648            # Strip wildcards to avoid injections
649            path = path.replace('*','')
650
651            # Normalize path and strip leading backslash
652            path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep)
653
654            if path.strip() in ['', '.', '..']:
655                # By defininition they exist on the filesystem
656                return True
657            else:
658                try:
659                    contents = self.smbClient.listPath(
660                        shareName=self.smb_share,
661                        path=path+'*'
662                    )
663                    # Filter on directories
664                    contents = [
665                        c for c in contents
666                        if c.get_longname() == ntpath.basename(path) and c.is_directory()
667                    ]
668                    return (len(contents) != 0)
669                except Exception as e:
670                    return False
671        else:
672            return False

Checks if the specified path is a directory on the SMB share.

This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the contents of the path and filtering for entries that match the basename of the path and are marked as directories.

Args: path (str, optional): The path to check on the SMB share. Defaults to None.

Returns: bool: True if the path is a directory, False otherwise or if an error occurs.

def path_isfile(self, path=None): View Source

674    def path_isfile(self, path=None):
675        """
676        Checks if the specified path is a file on the SMB share.
677
678        This method determines if a given path corresponds to a file on the SMB share. It does this by listing the
679        contents of the path and filtering for entries that match the basename of the path and are not marked as directories.
680
681        Args:
682            path (str, optional): The path to check on the SMB share. Defaults to None.
683
684        Returns:
685            bool: True if the path is a file, False otherwise or if an error occurs.
686        """
687
688        if path is not None:
689            path = path.replace('*','')
690            search_dir = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
691            search_dir = ntpath.dirname(search_dir) + ntpath.sep + '*'
692            try:
693                contents = self.smbClient.listPath(
694                    shareName=self.smb_share,
695                    path=search_dir
696                )
697                # Filter on files
698                contents = [
699                    c for c in contents
700                    if c.get_longname() == ntpath.basename(path) and not c.is_directory()
701                ]
702                return (len(contents) != 0)
703            except Exception as e:
704                return False
705        else:
706            return False

Checks if the specified path is a file on the SMB share.

This method determines if a given path corresponds to a file on the SMB share. It does this by listing the contents of the path and filtering for entries that match the basename of the path and are not marked as directories.

Args: path (str, optional): The path to check on the SMB share. Defaults to None.

Returns: bool: True if the path is a file, False otherwise or if an error occurs.

def ping_smb_session(self): View Source

708    def ping_smb_session(self):
709        """
710        Tests the connectivity to the SMB server by sending an echo command.
711
712        This method attempts to send an echo command to the SMB server to check if the session is still active.
713        It updates the `connected` attribute of the class based on the success or failure of the echo command.
714
715        Returns:
716            bool: True if the echo command succeeds (indicating the session is active), False otherwise.
717        """
718
719        try:
720            self.smbClient.getSMBServer().echo()
721        except Exception as e:
722            self.connected = False
723        return self.connected

Tests the connectivity to the SMB server by sending an echo command.

This method attempts to send an echo command to the SMB server to check if the session is still active. It updates the connected attribute of the class based on the success or failure of the echo command.

Returns: bool: True if the echo command succeeds (indicating the session is active), False otherwise.

def put_file(self, localpath=None): View Source

725    def put_file(self, localpath=None):
726        """
727        Uploads a single file to the SMB share.
728
729        This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path.
730        It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session.
731        General exceptions are caught and logged, with a traceback provided if debugging is enabled.
732
733        Args:
734            localpath (str, optional): The local file path of the file to be uploaded. Defaults to None.
735        """
736
737        if os.path.exists(localpath):
738            if os.path.isfile(localpath):
739                try:
740                    localfile = os.path.basename(localpath)
741                    f = LocalFileIO(
742                        mode="rb", 
743                        path=localpath, 
744                        debug=self.config.debug
745                    )
746                    self.smbClient.putFile(
747                        shareName=self.smb_share, 
748                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + localfile + ntpath.sep), 
749                        callback=f.read
750                    )
751                    f.close()
752                except (BrokenPipeError, KeyboardInterrupt) as err:
753                    print("[!] Interrupted.")
754                    self.close_smb_session()
755                    self.init_smb_session()
756                except Exception as err:
757                    print("[!] Failed to upload '%s': %s" % (localfile, err))
758                    if self.config.debug:
759                        traceback.print_exc()
760            else:
761                print("[!] The specified localpath is a directory. Use 'put -r <directory>' instead.")
762        else:
763            print("[!] The specified localpath does not exist.")

Uploads a single file to the SMB share.

This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path. It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session. General exceptions are caught and logged, with a traceback provided if debugging is enabled.

Args: localpath (str, optional): The local file path of the file to be uploaded. Defaults to None.

def put_file_recursively(self, localpath=None): View Source

765    def put_file_recursively(self, localpath=None):
766        """
767        Recursively uploads files from a specified local directory to the SMB share.
768
769        This method walks through the given local directory and all its subdirectories, uploading each file to the
770        corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is,
771        it iterates over all files and directories within the local path, creating necessary directories on the SMB share
772        and uploading files. If the local path is not a directory, it prints an error message.
773
774        Args:
775            localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None.
776        """
777
778        if os.path.exists(localpath):
779            if os.path.isfile(localpath):
780                # Iterate over all files and directories within the local path
781                local_files = {}
782                for root, dirs, files in os.walk(localpath):
783                    if len(files) != 0:
784                        local_files[root] = files
785
786                # Iterate over the found files
787                for local_dir_path in sorted(local_files.keys()):
788                    print("[>] Putting files of '%s'" % local_dir_path)
789
790                    # Create remote directory
791                    remote_dir_path = local_dir_path.replace(os.path.sep, ntpath.sep)
792                    self.mkdir(
793                        path=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep)
794                    )
795
796                    for local_file_path in local_files[local_dir_path]:
797                        try:
798                            f = LocalFileIO(
799                                mode="rb", 
800                                path=local_dir_path + os.path.sep + local_file_path, 
801                                debug=self.config.debug
802                            )
803                            self.smbClient.putFile(
804                                shareName=self.smb_share, 
805                                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep + local_file_path), 
806                                callback=f.read
807                            )
808                            f.close()
809
810                        except BrokenPipeError as err:
811                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
812                            f.close(remove=True)
813                            break
814                        except Exception as err:
815                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
816                            f.close(remove=True)
817                else:
818                    print("[!] The specified localpath is a file. Use 'put <file>' instead.")
819        else:
820            print("[!] The specified localpath does not exist.")

Recursively uploads files from a specified local directory to the SMB share.

This method walks through the given local directory and all its subdirectories, uploading each file to the corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is, it iterates over all files and directories within the local path, creating necessary directories on the SMB share and uploading files. If the local path is not a directory, it prints an error message.

Args: localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None.

def rmdir(self, path=None): View Source

822    def rmdir(self, path=None):
823        """
824        Removes a directory from the SMB share at the specified path.
825
826        This method attempts to delete a directory located at the given path on the SMB share. If the operation fails,
827        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
828        the stack trace of the exception.
829
830        Args:
831            path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None.
832        """
833        try:
834            self.smbClient.deleteDirectory(
835                shareName=self.smb_share, 
836                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
837            )
838        except Exception as err:
839            print("[!] Failed to remove directory '%s': %s" % (path, err))
840            if self.config.debug:
841                traceback.print_exc()

Removes a directory from the SMB share at the specified path.

This method attempts to delete a directory located at the given path on the SMB share. If the operation fails, it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints the stack trace of the exception.

Args: path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None.

def rm(self, path=None): View Source

843    def rm(self, path=None):
844        """
845        Removes a file from the SMB share at the specified path.
846
847        This method attempts to delete a file located at the given path on the SMB share. If the operation fails,
848        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
849        the stack trace of the exception.
850
851        Args:
852            path (str, optional): The path of the file to be removed on the SMB share. Defaults to None.
853        """
854        try:
855            self.smbClient.deleteFile(
856                shareName=self.smb_share, 
857                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
858            )
859        except Exception as err:
860            print("[!] Failed to remove file '%s': %s" % (path, err))
861            if self.config.debug:
862                traceback.print_exc()

Removes a file from the SMB share at the specified path.

This method attempts to delete a file located at the given path on the SMB share. If the operation fails, it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints the stack trace of the exception.

Args: path (str, optional): The path of the file to be removed on the SMB share. Defaults to None.

def tree(self, path=None): View Source

864    def tree(self, path=None):
865        """
866        Recursively lists the directory structure of the SMB share starting from the specified path.
867
868        This function prints a visual representation of the directory tree of the remote SMB share. It uses
869        recursion to navigate through directories and lists all files and subdirectories in each directory.
870        The output is color-coded and formatted to enhance readability, with directories highlighted in cyan.
871
872        Args:
873            path (str, optional): The starting path on the SMB share from which to begin listing the tree.
874                                  Defaults to the root of the current share.
875        """
876        
877        def recurse_action(base_dir="", path=[], prompt=[]):
878            bars = ["│   ", "├── ", "└── "]
879
880            remote_smb_path = ntpath.normpath(base_dir + ntpath.sep + ntpath.sep.join(path))
881
882            entries = []
883            try:
884                entries = self.smbClient.listPath(
885                    shareName=self.smb_share, 
886                    path=remote_smb_path+'\\*'
887                )
888            except impacket.smbconnection.SessionError as err:
889                code, const, text = err.getErrorCode(), err.getErrorString()[0], err.getErrorString()[1]
890                errmsg = "Error 0x%08x (%s): %s" % (code, const, text)
891                if self.config.no_colors:
892                    print("%s%s" % (''.join(prompt+[bars[2]]), errmsg))
893                else:
894                    print("%s\x1b[1;91m%s\x1b[0m" % (''.join(prompt+[bars[2]]), errmsg))
895                return 
896
897            entries = [e for e in entries if e.get_longname() not in [".", ".."]]
898            entries = sorted(entries, key=lambda x:x.get_longname())
899
900            # 
901            if len(entries) > 1:
902                index = 0
903                for entry in entries:
904                    index += 1
905                    # This is the first entry 
906                    if index == 0:
907                        if entry.is_directory():
908                            if self.config.no_colors:
909                                print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
910                            else:
911                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
912                            recurse_action(
913                                base_dir=base_dir, 
914                                path=path+[entry.get_longname()],
915                                prompt=prompt+["│   "]
916                            )
917                        else:
918                            if self.config.no_colors:
919                                print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
920                            else:
921                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
922
923                    # This is the last entry
924                    elif index == len(entries):
925                        if entry.is_directory():
926                            if self.config.no_colors:
927                                print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
928                            else:
929                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
930                            recurse_action(
931                                base_dir=base_dir, 
932                                path=path+[entry.get_longname()],
933                                prompt=prompt+["    "]
934                            )
935                        else:
936                            if self.config.no_colors:
937                                print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
938                            else:
939                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
940                        
941                    # These are entries in the middle
942                    else:
943                        if entry.is_directory():
944                            if self.config.no_colors:
945                                print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
946                            else:
947                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
948                            recurse_action(
949                                base_dir=base_dir, 
950                                path=path+[entry.get_longname()],
951                                prompt=prompt+["│   "]
952                            )
953                        else:
954                            if self.config.no_colors:
955                                print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
956                            else:
957                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
958
959            # 
960            elif len(entries) == 1:
961                entry = entries[0]
962                if entry.is_directory():
963                    if self.config.no_colors:
964                        print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
965                    else:
966                        print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
967                    recurse_action(
968                        base_dir=base_dir, 
969                        path=path+[entry.get_longname()],
970                        prompt=prompt+["    "]
971                    )
972                else:
973                    if self.config.no_colors:
974                        print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
975                    else:
976                        print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
977
978        # Entrypoint
979        try:
980            if self.config.no_colors:
981                print("%s\\" % path)
982            else:
983                print("\x1b[1;96m%s\x1b[0m\\" % path)
984            recurse_action(
985                base_dir=self.smb_cwd, 
986                path=[path],
987                prompt=[""]
988            )
989        except (BrokenPipeError, KeyboardInterrupt) as e:
990            print("[!] Interrupted.")
991            self.close_smb_session()
992            self.init_smb_session()

Recursively lists the directory structure of the SMB share starting from the specified path.

This function prints a visual representation of the directory tree of the remote SMB share. It uses recursion to navigate through directories and lists all files and subdirectories in each directory. The output is color-coded and formatted to enhance readability, with directories highlighted in cyan.

Args: path (str, optional): The starting path on the SMB share from which to begin listing the tree. Defaults to the root of the current share.

def umount(self, local_mount_point): View Source

 994    def umount(self, local_mount_point):
 995        if os.path.exists(local_mount_point):
 996            if sys.platform.startswith('win'):
 997                command = f"net use {local_mount_point} /delete"
 998
 999            elif sys.platform.startswith('linux') or sys.platform.startswith('darwin'):
1000                command = f"umount {local_mount_point}"
1001
1002            else:
1003                command = None
1004                print("[!] Unsupported platform for unmounting SMB share.")
1005        
1006            if command is not None:
1007                if self.config.debug:
1008                    print("[debug] Executing: %s" % command)
1009                os.system(command)
1010        else:
1011            print("[!] Cannot unmount a non existing path.")

def set_share(self, shareName): View Source

1015    def set_share(self, shareName):
1016        """
1017        Sets the current SMB share to the specified share name.
1018
1019        This method updates the SMB session to use the specified share name. It checks if the share name is valid
1020        and updates the smb_share attribute of the SMBSession instance.
1021
1022        Parameters:
1023            shareName (str): The name of the share to set as the current SMB share.
1024
1025        Raises:
1026            ValueError: If the shareName is None or an empty string.
1027        """
1028
1029        if shareName is not None:
1030            self.list_shares()
1031            if shareName.lower() in self.available_shares.keys():
1032                # Doing this in order to keep the case of the share adevertised by the remote machine
1033                self.smb_share = self.available_shares[shareName.lower()]["name"]
1034                # Connects the tree
1035                self.smb_tree_id = self.smbClient.connectTree(self.smb_share)
1036            else:
1037                print("[!] Could not set share '%s', it does not exist remotely." % shareName)

Sets the current SMB share to the specified share name.

This method updates the SMB session to use the specified share name. It checks if the share name is valid and updates the smb_share attribute of the SMBSession instance.

Parameters: shareName (str): The name of the share to set as the current SMB share.

Raises: ValueError: If the shareName is None or an empty string.

def set_cwd(self, path=None): View Source

1039    def set_cwd(self, path=None):
1040        """
1041        Sets the current working directory on the SMB share to the specified path.
1042
1043        This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory.
1044        If the specified path is not a directory, the cwd remains unchanged.
1045
1046        Parameters:
1047            path (str): The path to set as the current working directory.
1048
1049        Raises:
1050            ValueError: If the specified path is not a directory.
1051        """
1052
1053        if path is not None:
1054            # Set path separators to ntpath sep 
1055            if '/' in path:
1056                path = path.replace('/', ntpath.sep)
1057
1058            if path.startswith(ntpath.sep):
1059                # Absolute path
1060                path = path + ntpath.sep
1061            else:
1062                # Relative path to the CWD
1063                if len(self.smb_cwd) == 0:
1064                    path = path + ntpath.sep
1065                else:
1066                    path = self.smb_cwd + ntpath.sep + path
1067            
1068            # Path normalization
1069            path = ntpath.normpath(path)
1070            path = re.sub(r'\\+', r'\\', path)
1071
1072            if path in ["", ".", ".."]:
1073                self.smb_cwd = ""
1074            else:
1075                if self.path_isdir(pathFromRoot=path.strip(ntpath.sep)):
1076                    # Path exists on the remote 
1077                    self.smb_cwd = ntpath.normpath(path)
1078                else:
1079                    # Path does not exists or is not a directory on the remote 
1080                    print("[!] Remote directory '%s' does not exist." % path)

Sets the current working directory on the SMB share to the specified path.

This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory. If the specified path is not a directory, the cwd remains unchanged.

Parameters: path (str): The path to set as the current working directory.

Raises: ValueError: If the specified path is not a directory.