smbclientng.core.SMBSession

  1#!/usr/bin/env python3
  2# -*- coding: utf-8 -*-
  3# File name          : smbclient-ng.py
  4# Author             : Podalirius (@podalirius_)
  5# Date created       : 20 may 2024
  6
  7
  8import io
  9import impacket.smbconnection
 10import ntpath
 11import os
 12import re
 13import traceback
 14from smbclientng.core.LocalFileIO import LocalFileIO
 15from smbclientng.core.utils import b_filesize, STYPE_MASK
 16
 17
 18class SMBSession(object):
 19    """
 20    Class SMBSession is designed to handle the session management for SMB (Server Message Block) protocol connections.
 21    It provides functionalities to connect to an SMB server, authenticate using either NTLM or Kerberos, and manage SMB shares.
 22
 23    Attributes:
 24        address (str): The IP address or hostname of the SMB server.
 25        domain (str): The domain name for SMB server authentication.
 26        username (str): The username for SMB server authentication.
 27        password (str): The password for SMB server authentication.
 28        lmhash (str): The LM hash of the user's password, if available.
 29        nthash (str): The NT hash of the user's password, if available.
 30        use_kerberos (bool): A flag to determine whether to use Kerberos for authentication.
 31        kdcHost (str): The Key Distribution Center (KDC) host for Kerberos authentication.
 32        debug (bool): A flag to enable debug output.
 33        smbClient (object): The SMB client object used for the connection.
 34        connected (bool): A flag to check the status of the connection.
 35        smb_share (str): The current SMB share in use.
 36        smb_path (str): The current path within the SMB share.
 37
 38    Methods:
 39        __init__(address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, debug=False):
 40            Initializes the SMBSession with the specified parameters.
 41        init_smb_session():
 42            Initializes the SMB session by connecting to the server and authenticating using the specified method.
 43    """
 44
 45    def __init__(self, address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None):
 46        super(SMBSession, self).__init__()
 47        # Objects
 48        self.config = config
 49
 50        # Target server
 51        self.address = address
 52
 53        # Credentials
 54        self.domain = domain
 55        self.username = username
 56        self.password = password 
 57        self.lmhash = lmhash
 58        self.nthash = nthash
 59        self.use_kerberos = use_kerberos
 60        self.kdcHost = kdcHost
 61
 62        self.smbClient = None
 63        self.connected = False
 64
 65        self.available_shares = {}
 66        self.smb_share = None
 67        self.smb_cwd = ""
 68
 69        self.list_shares()
 70
 71    # Connect and disconnect SMB session
 72
 73    def init_smb_session(self):
 74        """
 75        Initializes and establishes a session with the SMB server.
 76
 77        This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration.
 78        It attempts to connect to the SMB server specified by the `address` attribute and authenticate using the credentials provided during the object's initialization.
 79
 80        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.
 81
 82        Returns:
 83            bool: True if the connection and authentication are successful, False otherwise.
 84        """
 85
 86        self.connected = False
 87
 88        if self.config.debug:
 89            print("[debug] [>] Connecting to remote SMB server '%s' ... " % self.address)
 90        try:
 91            self.smbClient = impacket.smbconnection.SMBConnection(
 92                remoteName=self.address,
 93                remoteHost=self.address,
 94                sess_port=int(445)
 95            )
 96        except OSError as err:
 97            print("[!] %s" % err)
 98            self.smbClient = None
 99
100        if self.smbClient is not None:
101            if self.use_kerberos:
102                if self.config.debug:
103                    print("[debug] [>] Authenticating as '%s\\%s' with kerberos ... " % (self.domain, self.username))
104                try:
105                    self.connected = self.smbClient.kerberosLogin(
106                        user=self.username,
107                        password=self.password,
108                        domain=self.domain,
109                        lmhash=self.lmhash,
110                        nthash=self.nthash,
111                        aesKey=self.aesKey,
112                        kdcHost=self.kdcHost
113                    )
114                except impacket.smbconnection.SessionError as err:
115                    if self.config.debug:
116                        traceback.print_exc()
117                    print("[!] Could not login: %s" % err)
118                    self.connected = False
119
120            else:
121                if self.config.debug:
122                    print("[debug] [>] Authenticating as '%s\\%s' with NTLM ... " % (self.domain, self.username))
123                try:
124                    self.connected = self.smbClient.login(
125                        user=self.username,
126                        password=self.password,
127                        domain=self.domain,
128                        lmhash=self.lmhash,
129                        nthash=self.nthash
130                    )
131                except impacket.smbconnection.SessionError as err:
132                    if self.config.debug:
133                        traceback.print_exc()
134                    print("[!] Could not login: %s" % err)
135                    self.connected = False
136
137            if self.connected:
138                print("[+] Successfully authenticated to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username))
139            else:
140                print("[!] Failed to authenticate to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username))
141
142        return self.connected
143
144    def close_smb_session(self):
145        """
146        Closes the current SMB session by disconnecting the SMB client.
147
148        This method ensures that the SMB client connection is properly closed. It checks if the client is connected
149        and if so, it closes the connection and resets the connection status.
150
151        Raises:
152            Exception: If the SMB client is not initialized or if there's an error during the disconnection process.
153        """
154
155        if self.smbClient is not None:
156            if self.connected:
157                self.smbClient.close()
158                self.connected = False
159                if self.config.debug:
160                    print("[+] SMB connection closed successfully.")
161            else:
162                if self.config.debug:
163                    print("[!] No active SMB connection to close.")
164        else:
165            raise Exception("SMB client is not initialized.")
166
167    # Operations
168
169    def read_file(self, path=None):
170        if self.path_isfile(path=path):
171            tmp_file_path = self.smb_cwd + ntpath.sep + path
172            matches = self.smbClient.listPath(
173                shareName=self.smb_share, 
174                path=tmp_file_path
175            )
176
177            fh = io.BytesIO()
178            try:
179                # opening the files in streams instead of mounting shares allows 
180                # for running the script from unprivileged containers
181                self.smbClient.getFile(self.smb_share, tmp_file_path, fh.write)
182            except impacket.smbconnection.SessionError as e:
183                return None
184            rawdata = fh.getvalue()
185            fh.close()
186            return rawdata
187        else:
188            print("[!] Remote path '%s' is not a file." % path)
189
190    def find(self, paths=[], callback=None):
191        def recurse_action(paths=[], depth=0, callback=None):
192            if callback is None:
193                return []
194            next_directories_to_explore = []
195            for path in paths:
196                remote_smb_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
197                entries = []
198                
199                try:
200                    entries = self.smbClient.listPath(
201                        shareName=self.smb_share, 
202                        path=(remote_smb_path + ntpath.sep + '*')
203                    )
204                except impacket.smbconnection.SessionError as err:
205                    continue 
206                # Remove dot names
207                entries = [e for e in entries if e.get_longname() not in [".", ".."]]
208                # Sort the entries ignoring case
209                entries = sorted(entries, key=lambda x:x.get_longname().lower())
210                
211                for entry in entries:
212                    if entry.is_directory():
213                        callback(entry, path + entry.get_longname() + ntpath.sep, depth)
214                    else:
215                        callback(entry, path + entry.get_longname(), depth)
216
217                # Next directories to explore
218                for entry in entries:
219                    if entry.is_directory():
220                        next_directories_to_explore.append(path + entry.get_longname() + ntpath.sep)
221            
222            return next_directories_to_explore
223        # 
224        if callback is not None:
225            depth = 0
226            while len(paths) != 0:
227                paths = recurse_action(
228                    paths=paths,
229                    depth=depth,
230                    callback=callback
231                )
232                depth = depth + 1
233        else:
234            print("[!] SMBSession.find(), callback function cannot be None.")
235
236    def get_file(self, path=None, keepRemotePath=False):
237        """
238        Retrieves a file from the specified path on the SMB share.
239
240        This method attempts to retrieve a file from the given path within the currently connected SMB share.
241        If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local
242        file object and writing the contents of the remote file to it using the SMB client's getFile method.
243
244        Parameters:
245            path (str): The path of the file to retrieve. If None, uses the current smb_path.
246
247        Returns:
248            None
249        """
250
251        tmp_file_path = self.smb_cwd + ntpath.sep + path
252        matches = self.smbClient.listPath(
253            shareName=self.smb_share, 
254            path=tmp_file_path
255        )
256        
257        for entry in matches:
258            if entry.is_directory():
259                print("[>] Skipping '%s' because it is a directory." % tmp_file_path)
260            else:
261                try:
262                    if ntpath.sep in path:
263                        outputfile = ntpath.dirname(path) + ntpath.sep + entry.get_longname()
264                    else:
265                        outputfile = entry.get_longname()
266                    f = LocalFileIO(
267                        mode="wb", 
268                        path=outputfile,
269                        expected_size=entry.get_filesize(), 
270                        debug=self.config.debug,
271                        keepRemotePath=keepRemotePath
272                    )
273                    self.smbClient.getFile(
274                        shareName=self.smb_share, 
275                        pathName=tmp_file_path, 
276                        callback=f.write
277                    )
278                    f.close()
279                except (BrokenPipeError, KeyboardInterrupt) as e:
280                    f.close()
281                    print("\x1b[v\x1b[o\r[!] Interrupted.")
282                    self.close_smb_session()
283                    self.init_smb_session()
284                        
285        return None
286
287    def get_file_recursively(self, path=None):
288        """
289        Recursively retrieves files from a specified path on the SMB share.
290
291        This method navigates through all directories starting from the given path,
292        and downloads all files found. It handles directories recursively, ensuring
293        that all nested files are retrieved. The method skips over directory entries
294        and handles errors gracefully, attempting to continue the operation where possible.
295
296        Parameters:
297            path (str): The initial directory path from which to start the recursive file retrieval.
298                        If None, it starts from the root of the configured SMB share.
299        """
300        
301        def recurse_action(base_dir="", path=[]):
302            remote_smb_path = base_dir + ntpath.sep.join(path)
303            entries = self.smbClient.listPath(
304                shareName=self.smb_share, 
305                path=remote_smb_path + '\\*'
306            )
307            if len(entries) != 0:
308                files = [entry for entry in entries if not entry.is_directory()]
309                directories = [entry for entry in entries if entry.is_directory() and entry.get_longname() not in [".", ".."]]
310
311                # Files
312                if len(files) != 0:
313                    print("[>] Retrieving files of '%s'" % remote_smb_path)
314                for entry_file in files:
315                    if not entry_file.is_directory():
316                        f = LocalFileIO(
317                            mode="wb",
318                            path=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
319                            expected_size=entry_file.get_filesize(),
320                            debug=self.config.debug
321                        )
322                        try:
323                            self.smbClient.getFile(
324                                shareName=self.smb_share, 
325                                pathName=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
326                                callback=f.write
327                            )
328                            f.close()
329                        except BrokenPipeError as err:
330                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
331                            f.close(remove=True)
332                            break
333                        except Exception as err:
334                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
335                            f.close(remove=True)
336                
337                # Directories
338                for entry_directory in directories:
339                    if entry_directory.is_directory():
340                        recurse_action(
341                            base_dir=self.smb_cwd, 
342                            path=path+[entry_directory.get_longname()]
343                        )                   
344        # Entrypoint
345        try:
346            recurse_action(
347                base_dir=self.smb_cwd, 
348                path=[path]
349            )
350        except (BrokenPipeError, KeyboardInterrupt) as e:
351            print("\x1b[v\x1b[o\r[!] Interrupted.")
352            self.close_smb_session()
353            self.init_smb_session()
354
355    def info(self, share=True, server=True):
356        """
357        Displays information about the server and optionally the shares.
358
359        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.
360
361        Parameters:
362            share (bool): If True, display information about the current share.
363            server (bool): If True, display information about the server.
364
365        Returns:
366            None
367        """
368
369        if server:
370            if self.config.no_colors:
371                print("[+] Server:")
372                print("  ├─NetBIOS:")
373                print("  │ ├─ NetBIOS Hostname ──────── : %s" % (self.smbClient.getServerName()))
374                print("  │ └─ NetBIOS Domain ────────── : %s" % (self.smbClient.getServerDomain()))
375                print("  ├─DNS:")
376                print("  │ ├─ DNS Hostname ──────────── : %s" % (self.smbClient.getServerDNSHostName()))
377                print("  │ └─ DNS Domain ────────────── : %s" % (self.smbClient.getServerDNSDomainName()))
378                print("  ├─OS:")
379                print("  │ ├─ OS Name ───────────────── : %s" % (self.smbClient.getServerOS()))
380                print("  │ └─ OS Version ────────────── : %s.%s.%s" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild()))
381                print("  ├─Server:")
382                print("  │ ├─ Signing Required ──────── : %s" % (self.smbClient.isSigningRequired()))
383                print("  │ ├─ Login Required ────────── : %s" % (self.smbClient.isLoginRequired()))
384                print("  │ ├─ Supports NTLMv2 ───────── : %s" % (self.smbClient.doesSupportNTLMv2()))
385                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
386                print("  │ ├─ Max size of read chunk ── : %d bytes (%s)" % (MaxReadSize, b_filesize(MaxReadSize)))
387                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
388                print("  │ └─ Max size of write chunk ─ : %d bytes (%s)" % (MaxWriteSize, b_filesize(MaxWriteSize)))
389                print("  └─")
390            else:
391                print("[+] Server:")
392                print("  ├─NetBIOS:")
393                print("  │ ├─ \x1b[94mNetBIOS Hostname\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerName()))
394                print("  │ └─ \x1b[94mNetBIOS Domain\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDomain()))
395                print("  ├─DNS:")
396                print("  │ ├─ \x1b[94mDNS Hostname\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSHostName()))
397                print("  │ └─ \x1b[94mDNS Domain\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSDomainName()))
398                print("  ├─OS:")
399                print("  │ ├─ \x1b[94mOS Name\x1b[0m \x1b[90m─────────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerOS()))
400                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()))
401                print("  ├─Server:")
402                print("  │ ├─ \x1b[94mSigning Required\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isSigningRequired()))
403                print("  │ ├─ \x1b[94mLogin Required\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isLoginRequired()))
404                print("  │ ├─ \x1b[94mSupports NTLMv2\x1b[0m \x1b[90m─────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.doesSupportNTLMv2()))
405                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
406                print("  │ ├─ \x1b[94mMax size of read chunk\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxReadSize, b_filesize(MaxReadSize)))
407                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
408                print("  │ └─ \x1b[94mMax size of write chunk\x1b[0m \x1b[90m─\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxWriteSize, b_filesize(MaxWriteSize)))
409                print("  └─")
410
411        if share and self.smb_share is not None:
412            share_name = self.available_shares.get(self.smb_share.lower(), "")["name"]
413            share_comment = self.available_shares.get(self.smb_share.lower(), "")["comment"]
414            share_type = self.available_shares.get(self.smb_share.lower(), "")["type"]
415            share_type =', '.join([s.replace("STYPE_","") for s in share_type])
416            share_rawtype = self.available_shares.get(self.smb_share.lower(), "")["rawtype"]
417            if self.config.no_colors:
418                print("\n[+] Share:")
419                print("  ├─ Name ──────────── : %s" % (share_name))
420                print("  ├─ Description ───── : %s" % (share_comment))
421                print("  ├─ Type ──────────── : %s" % (share_type))
422                print("  └─ Raw type value ── : %s" % (share_rawtype))
423            else:
424                print("\n[+] Share:")
425                print("  ├─ \x1b[94mName\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_name))
426                print("  ├─ \x1b[94mDescription\x1b[0m \x1b[90m─────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_comment))
427                print("  ├─ \x1b[94mType\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_type))
428                print("  └─ \x1b[94mRaw type value\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%s\x1b[0m" % (share_rawtype))
429
430    def list_contents(self, path=None):
431        """
432        Lists the contents of a specified directory on the SMB share.
433
434        This method retrieves the contents of a directory specified by `shareName` and `path`. If `shareName` or `path`
435        is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with
436        the long names of the files and directories as keys and their respective SMB entry objects as values.
437
438        Args:
439            shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None.
440            path (str, optional): The directory path to list contents from. Defaults to the current path if None.
441
442        Returns:
443            dict: A dictionary with file and directory names as keys and their SMB entry objects as values.
444        """
445        
446        if path is None or len(path) == 0:
447            path = self.smb_cwd
448        path = path.rstrip(ntpath.sep) + ntpath.sep + "*"
449
450        contents = {}
451        entries = self.smbClient.listPath(
452            shareName=self.smb_share, 
453            path=path
454        )
455        for entry in entries:
456            contents[entry.get_longname()] = entry
457
458        return contents
459
460    def list_shares(self):
461        """
462        Lists all the shares available on the connected SMB server.
463
464        This method queries the SMB server to retrieve a list of all available shares. It populates the `shares` dictionary
465        with key-value pairs where the key is the share name and the value is a dictionary containing details about the share
466        such as its name, type, raw type, and any comments associated with the share.
467
468        Returns:
469            dict: A dictionary containing information about each share available on the server.
470        """
471
472        self.available_shares = {}
473
474        if self.connected:
475            if self.smbClient is not None:
476                resp = self.smbClient.listShares()
477
478                for share in resp:
479                    # SHARE_INFO_1 structure (lmshare.h)
480                    # https://learn.microsoft.com/en-us/windows/win32/api/lmshare/ns-lmshare-share_info_1
481                    sharename = share["shi1_netname"][:-1]
482                    sharecomment = share["shi1_remark"][:-1]
483                    sharetype = share["shi1_type"]
484
485                    self.available_shares[sharename.lower()] = {
486                        "name": sharename, 
487                        "type": STYPE_MASK(sharetype), 
488                        "rawtype": sharetype, 
489                        "comment": sharecomment
490                    }
491            else:
492                print("[!] Error: SMBSession.smbClient is None.")
493
494        return self.available_shares
495
496    def mkdir(self, path=None):
497        """
498        Creates a directory at the specified path on the SMB share.
499
500        This method takes a path and attempts to create the directory structure on the SMB share. If the path includes
501        nested directories, it will create each directory in the sequence. If a directory already exists, it will skip
502        the creation for that directory without raising an error.
503
504        Args:
505            path (str, optional): The full path of the directory to create on the SMB share. Defaults to None.
506
507        Note:
508            The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility.
509        """
510
511        if path is not None:
512            # Prepare path
513            path = path.replace('/',ntpath.sep)
514            if ntpath.sep in path:
515                path = path.strip(ntpath.sep).split(ntpath.sep)
516            else:
517                path = [path]
518
519            # Create each dir in the path
520            for depth in range(1, len(path)+1):
521                tmp_path = ntpath.sep.join(path[:depth])
522                try:
523                    self.smbClient.createDirectory(
524                        shareName=self.smb_share, 
525                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + tmp_path + ntpath.sep)
526                    )
527                except impacket.smbconnection.SessionError as err:
528                    if err.getErrorCode() == 0xc0000035:
529                        # STATUS_OBJECT_NAME_COLLISION
530                        # Remote directory already created, this is normal
531                        # Src: https://github.com/fortra/impacket/blob/269ce69872f0e8f2188a80addb0c39fedfa6dcb8/impacket/nt_errors.py#L268C9-L268C19
532                        pass
533                    else:
534                        print("[!] Failed to create directory '%s': %s" % (tmp_path, err))
535                        if self.config.debug:
536                            traceback.print_exc()
537        else:
538            pass
539
540    def path_exists(self, path=None):
541        """
542        Checks if the specified path exists on the SMB share.
543
544        This method determines if a given path exists on the SMB share by attempting to list the contents of the path.
545        If the path listing is successful and returns one or more entries, the path is considered to exist.
546
547        Args:
548            path (str, optional): The path to check on the SMB share. Defaults to None.
549
550        Returns:
551            bool: True if the path exists, False otherwise or if an error occurs.
552        """
553
554        if path is not None:
555            path = path.replace('*','')
556            try:
557                contents = self.smbClient.listPath(
558                    shareName=self.smb_share,
559                    path=ntpath.normpath(self.smb_cwd + ntpath.sep + path + ntpath.sep)
560                )
561                return (len(contents) != 0)
562            except Exception as e:
563                return False
564        else:
565            return False
566   
567    def path_isdir(self, pathFromRoot=None):
568        """
569        Checks if the specified path is a directory on the SMB share.
570
571        This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the
572        contents of the path and filtering for entries that match the basename of the path and are marked as directories.
573
574        Args:
575            path (str, optional): The path to check on the SMB share. Defaults to None.
576
577        Returns:
578            bool: True if the path is a directory, False otherwise or if an error occurs.
579        """
580
581        if pathFromRoot is not None:
582            # Replace slashes if any
583            path = pathFromRoot.replace('/', ntpath.sep)
584            
585            # Strip wildcards to avoid injections
586            path = path.replace('*','')
587
588            # Normalize path and strip leading backslash
589            path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep)
590
591            if path.strip() in ['', '.', '..']:
592                # By defininition they exist on the filesystem
593                return True
594            else:
595                try:
596                    contents = self.smbClient.listPath(
597                        shareName=self.smb_share,
598                        path=path+'*'
599                    )
600                    # Filter on directories
601                    contents = [
602                        c for c in contents
603                        if c.get_longname() == ntpath.basename(path) and c.is_directory()
604                    ]
605                    return (len(contents) != 0)
606                except Exception as e:
607                    return False
608        else:
609            return False
610
611    def path_isfile(self, path=None):
612        """
613        Checks if the specified path is a file on the SMB share.
614
615        This method determines if a given path corresponds to a file on the SMB share. It does this by listing the
616        contents of the path and filtering for entries that match the basename of the path and are not marked as directories.
617
618        Args:
619            path (str, optional): The path to check on the SMB share. Defaults to None.
620
621        Returns:
622            bool: True if the path is a file, False otherwise or if an error occurs.
623        """
624
625        if path is not None:
626            path = path.replace('*','')
627            search_dir = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
628            search_dir = ntpath.dirname(search_dir) + ntpath.sep + '*'
629            try:
630                contents = self.smbClient.listPath(
631                    shareName=self.smb_share,
632                    path=search_dir
633                )
634                # Filter on files
635                contents = [
636                    c for c in contents
637                    if c.get_longname() == ntpath.basename(path) and not c.is_directory()
638                ]
639                return (len(contents) != 0)
640            except Exception as e:
641                return False
642        else:
643            return False
644
645    def ping_smb_session(self):
646        """
647        Tests the connectivity to the SMB server by sending an echo command.
648
649        This method attempts to send an echo command to the SMB server to check if the session is still active.
650        It updates the `connected` attribute of the class based on the success or failure of the echo command.
651
652        Returns:
653            bool: True if the echo command succeeds (indicating the session is active), False otherwise.
654        """
655
656        try:
657            self.smbClient.getSMBServer().echo()
658        except Exception as e:
659            self.connected = False
660        return self.connected
661
662    def put_file(self, localpath=None):
663        """
664        Uploads a single file to the SMB share.
665
666        This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path.
667        It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session.
668        General exceptions are caught and logged, with a traceback provided if debugging is enabled.
669
670        Args:
671            localpath (str, optional): The local file path of the file to be uploaded. Defaults to None.
672        """
673
674        if os.path.exists(localpath):
675            if os.path.isfile(localpath):
676                try:
677                    localfile = os.path.basename(localpath)
678                    f = LocalFileIO(
679                        mode="rb", 
680                        path=localpath, 
681                        debug=self.config.debug
682                    )
683                    self.smbClient.putFile(
684                        shareName=self.smb_share, 
685                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + localfile + ntpath.sep), 
686                        callback=f.read
687                    )
688                    f.close()
689                except (BrokenPipeError, KeyboardInterrupt) as err:
690                    print("[!] Interrupted.")
691                    self.close_smb_session()
692                    self.init_smb_session()
693                except Exception as err:
694                    print("[!] Failed to upload '%s': %s" % (localfile, err))
695                    if self.config.debug:
696                        traceback.print_exc()
697            else:
698                print("[!] The specified localpath is a directory. Use 'put -r <directory>' instead.")
699        else:
700            print("[!] The specified localpath does not exist.")
701
702    def put_file_recursively(self, localpath=None):
703        """
704        Recursively uploads files from a specified local directory to the SMB share.
705
706        This method walks through the given local directory and all its subdirectories, uploading each file to the
707        corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is,
708        it iterates over all files and directories within the local path, creating necessary directories on the SMB share
709        and uploading files. If the local path is not a directory, it prints an error message.
710
711        Args:
712            localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None.
713        """
714
715        if os.path.exists(localpath):
716            if os.path.isfile(localpath):
717                # Iterate over all files and directories within the local path
718                local_files = {}
719                for root, dirs, files in os.walk(localpath):
720                    if len(files) != 0:
721                        local_files[root] = files
722
723                # Iterate over the found files
724                for local_dir_path in sorted(local_files.keys()):
725                    print("[>] Putting files of '%s'" % local_dir_path)
726
727                    # Create remote directory
728                    remote_dir_path = local_dir_path.replace(os.path.sep, ntpath.sep)
729                    self.mkdir(
730                        path=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep)
731                    )
732
733                    for local_file_path in local_files[local_dir_path]:
734                        try:
735                            f = LocalFileIO(
736                                mode="rb", 
737                                path=local_dir_path + os.path.sep + local_file_path, 
738                                debug=self.config.debug
739                            )
740                            self.smbClient.putFile(
741                                shareName=self.smb_share, 
742                                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep + local_file_path), 
743                                callback=f.read
744                            )
745                            f.close()
746
747                        except BrokenPipeError as err:
748                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
749                            f.close(remove=True)
750                            break
751                        except Exception as err:
752                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
753                            f.close(remove=True)
754                else:
755                    print("[!] The specified localpath is a file. Use 'put <file>' instead.")
756        else:
757            print("[!] The specified localpath does not exist.")
758
759    def rmdir(self, path=None):
760        """
761        Removes a directory from the SMB share at the specified path.
762
763        This method attempts to delete a directory located at the given path on the SMB share. If the operation fails,
764        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
765        the stack trace of the exception.
766
767        Args:
768            path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None.
769        """
770        try:
771            self.smbClient.deleteDirectory(
772                shareName=self.smb_share, 
773                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
774            )
775        except Exception as err:
776            print("[!] Failed to remove directory '%s': %s" % (path, err))
777            if self.config.debug:
778                traceback.print_exc()
779
780    def rm(self, path=None):
781        """
782        Removes a file from the SMB share at the specified path.
783
784        This method attempts to delete a file located at the given path on the SMB share. If the operation fails,
785        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
786        the stack trace of the exception.
787
788        Args:
789            path (str, optional): The path of the file to be removed on the SMB share. Defaults to None.
790        """
791        try:
792            self.smbClient.deleteFile(
793                shareName=self.smb_share, 
794                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
795            )
796        except Exception as err:
797            print("[!] Failed to remove file '%s': %s" % (path, err))
798            if self.config.debug:
799                traceback.print_exc()
800
801    def tree(self, path=None):
802        """
803        Recursively lists the directory structure of the SMB share starting from the specified path.
804
805        This function prints a visual representation of the directory tree of the remote SMB share. It uses
806        recursion to navigate through directories and lists all files and subdirectories in each directory.
807        The output is color-coded and formatted to enhance readability, with directories highlighted in cyan.
808
809        Args:
810            path (str, optional): The starting path on the SMB share from which to begin listing the tree.
811                                  Defaults to the root of the current share.
812        """
813        
814        def recurse_action(base_dir="", path=[], prompt=[]):
815            bars = ["│   ", "├── ", "└── "]
816
817            remote_smb_path = ntpath.normpath(base_dir + ntpath.sep + ntpath.sep.join(path))
818
819            entries = []
820            try:
821                entries = self.smbClient.listPath(
822                    shareName=self.smb_share, 
823                    path=remote_smb_path+'\\*'
824                )
825            except impacket.smbconnection.SessionError as err:
826                code, const, text = err.getErrorCode(), err.getErrorString()[0], err.getErrorString()[1]
827                errmsg = "Error 0x%08x (%s): %s" % (code, const, text)
828                if self.config.no_colors:
829                    print("%s%s" % (''.join(prompt+[bars[2]]), errmsg))
830                else:
831                    print("%s\x1b[1;91m%s\x1b[0m" % (''.join(prompt+[bars[2]]), errmsg))
832                return 
833
834            entries = [e for e in entries if e.get_longname() not in [".", ".."]]
835            entries = sorted(entries, key=lambda x:x.get_longname())
836
837            # 
838            if len(entries) > 1:
839                index = 0
840                for entry in entries:
841                    index += 1
842                    # This is the first entry 
843                    if index == 0:
844                        if entry.is_directory():
845                            if self.config.no_colors:
846                                print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
847                            else:
848                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
849                            recurse_action(
850                                base_dir=base_dir, 
851                                path=path+[entry.get_longname()],
852                                prompt=prompt+["│   "]
853                            )
854                        else:
855                            if self.config.no_colors:
856                                print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
857                            else:
858                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
859
860                    # This is the last entry
861                    elif index == len(entries):
862                        if entry.is_directory():
863                            if self.config.no_colors:
864                                print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
865                            else:
866                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
867                            recurse_action(
868                                base_dir=base_dir, 
869                                path=path+[entry.get_longname()],
870                                prompt=prompt+["    "]
871                            )
872                        else:
873                            if self.config.no_colors:
874                                print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
875                            else:
876                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
877                        
878                    # These are entries in the middle
879                    else:
880                        if entry.is_directory():
881                            if self.config.no_colors:
882                                print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
883                            else:
884                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
885                            recurse_action(
886                                base_dir=base_dir, 
887                                path=path+[entry.get_longname()],
888                                prompt=prompt+["│   "]
889                            )
890                        else:
891                            if self.config.no_colors:
892                                print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
893                            else:
894                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
895
896            # 
897            elif len(entries) == 1:
898                entry = entries[0]
899                if entry.is_directory():
900                    if self.config.no_colors:
901                        print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
902                    else:
903                        print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
904                    recurse_action(
905                        base_dir=base_dir, 
906                        path=path+[entry.get_longname()],
907                        prompt=prompt+["    "]
908                    )
909                else:
910                    if self.config.no_colors:
911                        print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
912                    else:
913                        print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
914
915        # Entrypoint
916        try:
917            if self.config.no_colors:
918                print("%s\\" % path)
919            else:
920                print("\x1b[1;96m%s\x1b[0m\\" % path)
921            recurse_action(
922                base_dir=self.smb_cwd, 
923                path=[path],
924                prompt=[""]
925            )
926        except (BrokenPipeError, KeyboardInterrupt) as e:
927            print("[!] Interrupted.")
928            self.close_smb_session()
929            self.init_smb_session()
930
931    # Setter / Getter
932
933    def set_share(self, shareName):
934        """
935        Sets the current SMB share to the specified share name.
936
937        This method updates the SMB session to use the specified share name. It checks if the share name is valid
938        and updates the smb_share attribute of the SMBSession instance.
939
940        Parameters:
941            shareName (str): The name of the share to set as the current SMB share.
942
943        Raises:
944            ValueError: If the shareName is None or an empty string.
945        """
946
947        if shareName is not None:
948            self.list_shares()
949            if shareName.lower() in self.available_shares.keys():
950                # Doing this in order to keep the case of the share adevertised by the remote machine
951                self.smb_share = self.available_shares[shareName.lower()]["name"]
952            else:
953                print("[!] Could not set share '%s', it does not exist remotely." % shareName)
954
955    def set_cwd(self, path=None):
956        """
957        Sets the current working directory on the SMB share to the specified path.
958
959        This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory.
960        If the specified path is not a directory, the cwd remains unchanged.
961
962        Parameters:
963            path (str): The path to set as the current working directory.
964
965        Raises:
966            ValueError: If the specified path is not a directory.
967        """
968
969        if path is not None:
970            # Set path separators to ntpath sep 
971            if '/' in path:
972                path = path.replace('/', ntpath.sep)
973
974            if path.startswith(ntpath.sep):
975                # Absolute path
976                path = path + ntpath.sep
977            else:
978                # Relative path to the CWD
979                if len(self.smb_cwd) == 0:
980                    path = path + ntpath.sep
981                else:
982                    path = self.smb_cwd + ntpath.sep + path
983            
984            # Path normalization
985            path = ntpath.normpath(path)
986            path = re.sub(r'\\+', r'\\', path)
987
988            if path in ["", ".", ".."]:
989                self.smb_cwd = ""
990            else:
991                if self.path_isdir(pathFromRoot=path.strip(ntpath.sep)):
992                    # Path exists on the remote 
993                    self.smb_cwd = ntpath.normpath(path)
994                else:
995                    # Path does not exists or is not a directory on the remote 
996                    print("[!] Remote directory '%s' does not exist." % path)
class SMBSession:
 19class SMBSession(object):
 20    """
 21    Class SMBSession is designed to handle the session management for SMB (Server Message Block) protocol connections.
 22    It provides functionalities to connect to an SMB server, authenticate using either NTLM or Kerberos, and manage SMB shares.
 23
 24    Attributes:
 25        address (str): The IP address or hostname of the SMB server.
 26        domain (str): The domain name for SMB server authentication.
 27        username (str): The username for SMB server authentication.
 28        password (str): The password for SMB server authentication.
 29        lmhash (str): The LM hash of the user's password, if available.
 30        nthash (str): The NT hash of the user's password, if available.
 31        use_kerberos (bool): A flag to determine whether to use Kerberos for authentication.
 32        kdcHost (str): The Key Distribution Center (KDC) host for Kerberos authentication.
 33        debug (bool): A flag to enable debug output.
 34        smbClient (object): The SMB client object used for the connection.
 35        connected (bool): A flag to check the status of the connection.
 36        smb_share (str): The current SMB share in use.
 37        smb_path (str): The current path within the SMB share.
 38
 39    Methods:
 40        __init__(address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, debug=False):
 41            Initializes the SMBSession with the specified parameters.
 42        init_smb_session():
 43            Initializes the SMB session by connecting to the server and authenticating using the specified method.
 44    """
 45
 46    def __init__(self, address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None):
 47        super(SMBSession, self).__init__()
 48        # Objects
 49        self.config = config
 50
 51        # Target server
 52        self.address = address
 53
 54        # Credentials
 55        self.domain = domain
 56        self.username = username
 57        self.password = password 
 58        self.lmhash = lmhash
 59        self.nthash = nthash
 60        self.use_kerberos = use_kerberos
 61        self.kdcHost = kdcHost
 62
 63        self.smbClient = None
 64        self.connected = False
 65
 66        self.available_shares = {}
 67        self.smb_share = None
 68        self.smb_cwd = ""
 69
 70        self.list_shares()
 71
 72    # Connect and disconnect SMB session
 73
 74    def init_smb_session(self):
 75        """
 76        Initializes and establishes a session with the SMB server.
 77
 78        This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration.
 79        It attempts to connect to the SMB server specified by the `address` attribute and authenticate using the credentials provided during the object's initialization.
 80
 81        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.
 82
 83        Returns:
 84            bool: True if the connection and authentication are successful, False otherwise.
 85        """
 86
 87        self.connected = False
 88
 89        if self.config.debug:
 90            print("[debug] [>] Connecting to remote SMB server '%s' ... " % self.address)
 91        try:
 92            self.smbClient = impacket.smbconnection.SMBConnection(
 93                remoteName=self.address,
 94                remoteHost=self.address,
 95                sess_port=int(445)
 96            )
 97        except OSError as err:
 98            print("[!] %s" % err)
 99            self.smbClient = None
100
101        if self.smbClient is not None:
102            if self.use_kerberos:
103                if self.config.debug:
104                    print("[debug] [>] Authenticating as '%s\\%s' with kerberos ... " % (self.domain, self.username))
105                try:
106                    self.connected = self.smbClient.kerberosLogin(
107                        user=self.username,
108                        password=self.password,
109                        domain=self.domain,
110                        lmhash=self.lmhash,
111                        nthash=self.nthash,
112                        aesKey=self.aesKey,
113                        kdcHost=self.kdcHost
114                    )
115                except impacket.smbconnection.SessionError as err:
116                    if self.config.debug:
117                        traceback.print_exc()
118                    print("[!] Could not login: %s" % err)
119                    self.connected = False
120
121            else:
122                if self.config.debug:
123                    print("[debug] [>] Authenticating as '%s\\%s' with NTLM ... " % (self.domain, self.username))
124                try:
125                    self.connected = self.smbClient.login(
126                        user=self.username,
127                        password=self.password,
128                        domain=self.domain,
129                        lmhash=self.lmhash,
130                        nthash=self.nthash
131                    )
132                except impacket.smbconnection.SessionError as err:
133                    if self.config.debug:
134                        traceback.print_exc()
135                    print("[!] Could not login: %s" % err)
136                    self.connected = False
137
138            if self.connected:
139                print("[+] Successfully authenticated to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username))
140            else:
141                print("[!] Failed to authenticate to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username))
142
143        return self.connected
144
145    def close_smb_session(self):
146        """
147        Closes the current SMB session by disconnecting the SMB client.
148
149        This method ensures that the SMB client connection is properly closed. It checks if the client is connected
150        and if so, it closes the connection and resets the connection status.
151
152        Raises:
153            Exception: If the SMB client is not initialized or if there's an error during the disconnection process.
154        """
155
156        if self.smbClient is not None:
157            if self.connected:
158                self.smbClient.close()
159                self.connected = False
160                if self.config.debug:
161                    print("[+] SMB connection closed successfully.")
162            else:
163                if self.config.debug:
164                    print("[!] No active SMB connection to close.")
165        else:
166            raise Exception("SMB client is not initialized.")
167
168    # Operations
169
170    def read_file(self, path=None):
171        if self.path_isfile(path=path):
172            tmp_file_path = self.smb_cwd + ntpath.sep + path
173            matches = self.smbClient.listPath(
174                shareName=self.smb_share, 
175                path=tmp_file_path
176            )
177
178            fh = io.BytesIO()
179            try:
180                # opening the files in streams instead of mounting shares allows 
181                # for running the script from unprivileged containers
182                self.smbClient.getFile(self.smb_share, tmp_file_path, fh.write)
183            except impacket.smbconnection.SessionError as e:
184                return None
185            rawdata = fh.getvalue()
186            fh.close()
187            return rawdata
188        else:
189            print("[!] Remote path '%s' is not a file." % path)
190
191    def find(self, paths=[], callback=None):
192        def recurse_action(paths=[], depth=0, callback=None):
193            if callback is None:
194                return []
195            next_directories_to_explore = []
196            for path in paths:
197                remote_smb_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
198                entries = []
199                
200                try:
201                    entries = self.smbClient.listPath(
202                        shareName=self.smb_share, 
203                        path=(remote_smb_path + ntpath.sep + '*')
204                    )
205                except impacket.smbconnection.SessionError as err:
206                    continue 
207                # Remove dot names
208                entries = [e for e in entries if e.get_longname() not in [".", ".."]]
209                # Sort the entries ignoring case
210                entries = sorted(entries, key=lambda x:x.get_longname().lower())
211                
212                for entry in entries:
213                    if entry.is_directory():
214                        callback(entry, path + entry.get_longname() + ntpath.sep, depth)
215                    else:
216                        callback(entry, path + entry.get_longname(), depth)
217
218                # Next directories to explore
219                for entry in entries:
220                    if entry.is_directory():
221                        next_directories_to_explore.append(path + entry.get_longname() + ntpath.sep)
222            
223            return next_directories_to_explore
224        # 
225        if callback is not None:
226            depth = 0
227            while len(paths) != 0:
228                paths = recurse_action(
229                    paths=paths,
230                    depth=depth,
231                    callback=callback
232                )
233                depth = depth + 1
234        else:
235            print("[!] SMBSession.find(), callback function cannot be None.")
236
237    def get_file(self, path=None, keepRemotePath=False):
238        """
239        Retrieves a file from the specified path on the SMB share.
240
241        This method attempts to retrieve a file from the given path within the currently connected SMB share.
242        If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local
243        file object and writing the contents of the remote file to it using the SMB client's getFile method.
244
245        Parameters:
246            path (str): The path of the file to retrieve. If None, uses the current smb_path.
247
248        Returns:
249            None
250        """
251
252        tmp_file_path = self.smb_cwd + ntpath.sep + path
253        matches = self.smbClient.listPath(
254            shareName=self.smb_share, 
255            path=tmp_file_path
256        )
257        
258        for entry in matches:
259            if entry.is_directory():
260                print("[>] Skipping '%s' because it is a directory." % tmp_file_path)
261            else:
262                try:
263                    if ntpath.sep in path:
264                        outputfile = ntpath.dirname(path) + ntpath.sep + entry.get_longname()
265                    else:
266                        outputfile = entry.get_longname()
267                    f = LocalFileIO(
268                        mode="wb", 
269                        path=outputfile,
270                        expected_size=entry.get_filesize(), 
271                        debug=self.config.debug,
272                        keepRemotePath=keepRemotePath
273                    )
274                    self.smbClient.getFile(
275                        shareName=self.smb_share, 
276                        pathName=tmp_file_path, 
277                        callback=f.write
278                    )
279                    f.close()
280                except (BrokenPipeError, KeyboardInterrupt) as e:
281                    f.close()
282                    print("\x1b[v\x1b[o\r[!] Interrupted.")
283                    self.close_smb_session()
284                    self.init_smb_session()
285                        
286        return None
287
288    def get_file_recursively(self, path=None):
289        """
290        Recursively retrieves files from a specified path on the SMB share.
291
292        This method navigates through all directories starting from the given path,
293        and downloads all files found. It handles directories recursively, ensuring
294        that all nested files are retrieved. The method skips over directory entries
295        and handles errors gracefully, attempting to continue the operation where possible.
296
297        Parameters:
298            path (str): The initial directory path from which to start the recursive file retrieval.
299                        If None, it starts from the root of the configured SMB share.
300        """
301        
302        def recurse_action(base_dir="", path=[]):
303            remote_smb_path = base_dir + ntpath.sep.join(path)
304            entries = self.smbClient.listPath(
305                shareName=self.smb_share, 
306                path=remote_smb_path + '\\*'
307            )
308            if len(entries) != 0:
309                files = [entry for entry in entries if not entry.is_directory()]
310                directories = [entry for entry in entries if entry.is_directory() and entry.get_longname() not in [".", ".."]]
311
312                # Files
313                if len(files) != 0:
314                    print("[>] Retrieving files of '%s'" % remote_smb_path)
315                for entry_file in files:
316                    if not entry_file.is_directory():
317                        f = LocalFileIO(
318                            mode="wb",
319                            path=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
320                            expected_size=entry_file.get_filesize(),
321                            debug=self.config.debug
322                        )
323                        try:
324                            self.smbClient.getFile(
325                                shareName=self.smb_share, 
326                                pathName=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
327                                callback=f.write
328                            )
329                            f.close()
330                        except BrokenPipeError as err:
331                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
332                            f.close(remove=True)
333                            break
334                        except Exception as err:
335                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
336                            f.close(remove=True)
337                
338                # Directories
339                for entry_directory in directories:
340                    if entry_directory.is_directory():
341                        recurse_action(
342                            base_dir=self.smb_cwd, 
343                            path=path+[entry_directory.get_longname()]
344                        )                   
345        # Entrypoint
346        try:
347            recurse_action(
348                base_dir=self.smb_cwd, 
349                path=[path]
350            )
351        except (BrokenPipeError, KeyboardInterrupt) as e:
352            print("\x1b[v\x1b[o\r[!] Interrupted.")
353            self.close_smb_session()
354            self.init_smb_session()
355
356    def info(self, share=True, server=True):
357        """
358        Displays information about the server and optionally the shares.
359
360        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.
361
362        Parameters:
363            share (bool): If True, display information about the current share.
364            server (bool): If True, display information about the server.
365
366        Returns:
367            None
368        """
369
370        if server:
371            if self.config.no_colors:
372                print("[+] Server:")
373                print("  ├─NetBIOS:")
374                print("  │ ├─ NetBIOS Hostname ──────── : %s" % (self.smbClient.getServerName()))
375                print("  │ └─ NetBIOS Domain ────────── : %s" % (self.smbClient.getServerDomain()))
376                print("  ├─DNS:")
377                print("  │ ├─ DNS Hostname ──────────── : %s" % (self.smbClient.getServerDNSHostName()))
378                print("  │ └─ DNS Domain ────────────── : %s" % (self.smbClient.getServerDNSDomainName()))
379                print("  ├─OS:")
380                print("  │ ├─ OS Name ───────────────── : %s" % (self.smbClient.getServerOS()))
381                print("  │ └─ OS Version ────────────── : %s.%s.%s" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild()))
382                print("  ├─Server:")
383                print("  │ ├─ Signing Required ──────── : %s" % (self.smbClient.isSigningRequired()))
384                print("  │ ├─ Login Required ────────── : %s" % (self.smbClient.isLoginRequired()))
385                print("  │ ├─ Supports NTLMv2 ───────── : %s" % (self.smbClient.doesSupportNTLMv2()))
386                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
387                print("  │ ├─ Max size of read chunk ── : %d bytes (%s)" % (MaxReadSize, b_filesize(MaxReadSize)))
388                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
389                print("  │ └─ Max size of write chunk ─ : %d bytes (%s)" % (MaxWriteSize, b_filesize(MaxWriteSize)))
390                print("  └─")
391            else:
392                print("[+] Server:")
393                print("  ├─NetBIOS:")
394                print("  │ ├─ \x1b[94mNetBIOS Hostname\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerName()))
395                print("  │ └─ \x1b[94mNetBIOS Domain\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDomain()))
396                print("  ├─DNS:")
397                print("  │ ├─ \x1b[94mDNS Hostname\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSHostName()))
398                print("  │ └─ \x1b[94mDNS Domain\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSDomainName()))
399                print("  ├─OS:")
400                print("  │ ├─ \x1b[94mOS Name\x1b[0m \x1b[90m─────────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerOS()))
401                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()))
402                print("  ├─Server:")
403                print("  │ ├─ \x1b[94mSigning Required\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isSigningRequired()))
404                print("  │ ├─ \x1b[94mLogin Required\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isLoginRequired()))
405                print("  │ ├─ \x1b[94mSupports NTLMv2\x1b[0m \x1b[90m─────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.doesSupportNTLMv2()))
406                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
407                print("  │ ├─ \x1b[94mMax size of read chunk\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxReadSize, b_filesize(MaxReadSize)))
408                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
409                print("  │ └─ \x1b[94mMax size of write chunk\x1b[0m \x1b[90m─\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxWriteSize, b_filesize(MaxWriteSize)))
410                print("  └─")
411
412        if share and self.smb_share is not None:
413            share_name = self.available_shares.get(self.smb_share.lower(), "")["name"]
414            share_comment = self.available_shares.get(self.smb_share.lower(), "")["comment"]
415            share_type = self.available_shares.get(self.smb_share.lower(), "")["type"]
416            share_type =', '.join([s.replace("STYPE_","") for s in share_type])
417            share_rawtype = self.available_shares.get(self.smb_share.lower(), "")["rawtype"]
418            if self.config.no_colors:
419                print("\n[+] Share:")
420                print("  ├─ Name ──────────── : %s" % (share_name))
421                print("  ├─ Description ───── : %s" % (share_comment))
422                print("  ├─ Type ──────────── : %s" % (share_type))
423                print("  └─ Raw type value ── : %s" % (share_rawtype))
424            else:
425                print("\n[+] Share:")
426                print("  ├─ \x1b[94mName\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_name))
427                print("  ├─ \x1b[94mDescription\x1b[0m \x1b[90m─────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_comment))
428                print("  ├─ \x1b[94mType\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_type))
429                print("  └─ \x1b[94mRaw type value\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%s\x1b[0m" % (share_rawtype))
430
431    def list_contents(self, path=None):
432        """
433        Lists the contents of a specified directory on the SMB share.
434
435        This method retrieves the contents of a directory specified by `shareName` and `path`. If `shareName` or `path`
436        is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with
437        the long names of the files and directories as keys and their respective SMB entry objects as values.
438
439        Args:
440            shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None.
441            path (str, optional): The directory path to list contents from. Defaults to the current path if None.
442
443        Returns:
444            dict: A dictionary with file and directory names as keys and their SMB entry objects as values.
445        """
446        
447        if path is None or len(path) == 0:
448            path = self.smb_cwd
449        path = path.rstrip(ntpath.sep) + ntpath.sep + "*"
450
451        contents = {}
452        entries = self.smbClient.listPath(
453            shareName=self.smb_share, 
454            path=path
455        )
456        for entry in entries:
457            contents[entry.get_longname()] = entry
458
459        return contents
460
461    def list_shares(self):
462        """
463        Lists all the shares available on the connected SMB server.
464
465        This method queries the SMB server to retrieve a list of all available shares. It populates the `shares` dictionary
466        with key-value pairs where the key is the share name and the value is a dictionary containing details about the share
467        such as its name, type, raw type, and any comments associated with the share.
468
469        Returns:
470            dict: A dictionary containing information about each share available on the server.
471        """
472
473        self.available_shares = {}
474
475        if self.connected:
476            if self.smbClient is not None:
477                resp = self.smbClient.listShares()
478
479                for share in resp:
480                    # SHARE_INFO_1 structure (lmshare.h)
481                    # https://learn.microsoft.com/en-us/windows/win32/api/lmshare/ns-lmshare-share_info_1
482                    sharename = share["shi1_netname"][:-1]
483                    sharecomment = share["shi1_remark"][:-1]
484                    sharetype = share["shi1_type"]
485
486                    self.available_shares[sharename.lower()] = {
487                        "name": sharename, 
488                        "type": STYPE_MASK(sharetype), 
489                        "rawtype": sharetype, 
490                        "comment": sharecomment
491                    }
492            else:
493                print("[!] Error: SMBSession.smbClient is None.")
494
495        return self.available_shares
496
497    def mkdir(self, path=None):
498        """
499        Creates a directory at the specified path on the SMB share.
500
501        This method takes a path and attempts to create the directory structure on the SMB share. If the path includes
502        nested directories, it will create each directory in the sequence. If a directory already exists, it will skip
503        the creation for that directory without raising an error.
504
505        Args:
506            path (str, optional): The full path of the directory to create on the SMB share. Defaults to None.
507
508        Note:
509            The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility.
510        """
511
512        if path is not None:
513            # Prepare path
514            path = path.replace('/',ntpath.sep)
515            if ntpath.sep in path:
516                path = path.strip(ntpath.sep).split(ntpath.sep)
517            else:
518                path = [path]
519
520            # Create each dir in the path
521            for depth in range(1, len(path)+1):
522                tmp_path = ntpath.sep.join(path[:depth])
523                try:
524                    self.smbClient.createDirectory(
525                        shareName=self.smb_share, 
526                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + tmp_path + ntpath.sep)
527                    )
528                except impacket.smbconnection.SessionError as err:
529                    if err.getErrorCode() == 0xc0000035:
530                        # STATUS_OBJECT_NAME_COLLISION
531                        # Remote directory already created, this is normal
532                        # Src: https://github.com/fortra/impacket/blob/269ce69872f0e8f2188a80addb0c39fedfa6dcb8/impacket/nt_errors.py#L268C9-L268C19
533                        pass
534                    else:
535                        print("[!] Failed to create directory '%s': %s" % (tmp_path, err))
536                        if self.config.debug:
537                            traceback.print_exc()
538        else:
539            pass
540
541    def path_exists(self, path=None):
542        """
543        Checks if the specified path exists on the SMB share.
544
545        This method determines if a given path exists on the SMB share by attempting to list the contents of the path.
546        If the path listing is successful and returns one or more entries, the path is considered to exist.
547
548        Args:
549            path (str, optional): The path to check on the SMB share. Defaults to None.
550
551        Returns:
552            bool: True if the path exists, False otherwise or if an error occurs.
553        """
554
555        if path is not None:
556            path = path.replace('*','')
557            try:
558                contents = self.smbClient.listPath(
559                    shareName=self.smb_share,
560                    path=ntpath.normpath(self.smb_cwd + ntpath.sep + path + ntpath.sep)
561                )
562                return (len(contents) != 0)
563            except Exception as e:
564                return False
565        else:
566            return False
567   
568    def path_isdir(self, pathFromRoot=None):
569        """
570        Checks if the specified path is a directory on the SMB share.
571
572        This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the
573        contents of the path and filtering for entries that match the basename of the path and are marked as directories.
574
575        Args:
576            path (str, optional): The path to check on the SMB share. Defaults to None.
577
578        Returns:
579            bool: True if the path is a directory, False otherwise or if an error occurs.
580        """
581
582        if pathFromRoot is not None:
583            # Replace slashes if any
584            path = pathFromRoot.replace('/', ntpath.sep)
585            
586            # Strip wildcards to avoid injections
587            path = path.replace('*','')
588
589            # Normalize path and strip leading backslash
590            path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep)
591
592            if path.strip() in ['', '.', '..']:
593                # By defininition they exist on the filesystem
594                return True
595            else:
596                try:
597                    contents = self.smbClient.listPath(
598                        shareName=self.smb_share,
599                        path=path+'*'
600                    )
601                    # Filter on directories
602                    contents = [
603                        c for c in contents
604                        if c.get_longname() == ntpath.basename(path) and c.is_directory()
605                    ]
606                    return (len(contents) != 0)
607                except Exception as e:
608                    return False
609        else:
610            return False
611
612    def path_isfile(self, path=None):
613        """
614        Checks if the specified path is a file on the SMB share.
615
616        This method determines if a given path corresponds to a file on the SMB share. It does this by listing the
617        contents of the path and filtering for entries that match the basename of the path and are not marked as directories.
618
619        Args:
620            path (str, optional): The path to check on the SMB share. Defaults to None.
621
622        Returns:
623            bool: True if the path is a file, False otherwise or if an error occurs.
624        """
625
626        if path is not None:
627            path = path.replace('*','')
628            search_dir = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
629            search_dir = ntpath.dirname(search_dir) + ntpath.sep + '*'
630            try:
631                contents = self.smbClient.listPath(
632                    shareName=self.smb_share,
633                    path=search_dir
634                )
635                # Filter on files
636                contents = [
637                    c for c in contents
638                    if c.get_longname() == ntpath.basename(path) and not c.is_directory()
639                ]
640                return (len(contents) != 0)
641            except Exception as e:
642                return False
643        else:
644            return False
645
646    def ping_smb_session(self):
647        """
648        Tests the connectivity to the SMB server by sending an echo command.
649
650        This method attempts to send an echo command to the SMB server to check if the session is still active.
651        It updates the `connected` attribute of the class based on the success or failure of the echo command.
652
653        Returns:
654            bool: True if the echo command succeeds (indicating the session is active), False otherwise.
655        """
656
657        try:
658            self.smbClient.getSMBServer().echo()
659        except Exception as e:
660            self.connected = False
661        return self.connected
662
663    def put_file(self, localpath=None):
664        """
665        Uploads a single file to the SMB share.
666
667        This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path.
668        It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session.
669        General exceptions are caught and logged, with a traceback provided if debugging is enabled.
670
671        Args:
672            localpath (str, optional): The local file path of the file to be uploaded. Defaults to None.
673        """
674
675        if os.path.exists(localpath):
676            if os.path.isfile(localpath):
677                try:
678                    localfile = os.path.basename(localpath)
679                    f = LocalFileIO(
680                        mode="rb", 
681                        path=localpath, 
682                        debug=self.config.debug
683                    )
684                    self.smbClient.putFile(
685                        shareName=self.smb_share, 
686                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + localfile + ntpath.sep), 
687                        callback=f.read
688                    )
689                    f.close()
690                except (BrokenPipeError, KeyboardInterrupt) as err:
691                    print("[!] Interrupted.")
692                    self.close_smb_session()
693                    self.init_smb_session()
694                except Exception as err:
695                    print("[!] Failed to upload '%s': %s" % (localfile, err))
696                    if self.config.debug:
697                        traceback.print_exc()
698            else:
699                print("[!] The specified localpath is a directory. Use 'put -r <directory>' instead.")
700        else:
701            print("[!] The specified localpath does not exist.")
702
703    def put_file_recursively(self, localpath=None):
704        """
705        Recursively uploads files from a specified local directory to the SMB share.
706
707        This method walks through the given local directory and all its subdirectories, uploading each file to the
708        corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is,
709        it iterates over all files and directories within the local path, creating necessary directories on the SMB share
710        and uploading files. If the local path is not a directory, it prints an error message.
711
712        Args:
713            localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None.
714        """
715
716        if os.path.exists(localpath):
717            if os.path.isfile(localpath):
718                # Iterate over all files and directories within the local path
719                local_files = {}
720                for root, dirs, files in os.walk(localpath):
721                    if len(files) != 0:
722                        local_files[root] = files
723
724                # Iterate over the found files
725                for local_dir_path in sorted(local_files.keys()):
726                    print("[>] Putting files of '%s'" % local_dir_path)
727
728                    # Create remote directory
729                    remote_dir_path = local_dir_path.replace(os.path.sep, ntpath.sep)
730                    self.mkdir(
731                        path=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep)
732                    )
733
734                    for local_file_path in local_files[local_dir_path]:
735                        try:
736                            f = LocalFileIO(
737                                mode="rb", 
738                                path=local_dir_path + os.path.sep + local_file_path, 
739                                debug=self.config.debug
740                            )
741                            self.smbClient.putFile(
742                                shareName=self.smb_share, 
743                                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep + local_file_path), 
744                                callback=f.read
745                            )
746                            f.close()
747
748                        except BrokenPipeError as err:
749                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
750                            f.close(remove=True)
751                            break
752                        except Exception as err:
753                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
754                            f.close(remove=True)
755                else:
756                    print("[!] The specified localpath is a file. Use 'put <file>' instead.")
757        else:
758            print("[!] The specified localpath does not exist.")
759
760    def rmdir(self, path=None):
761        """
762        Removes a directory from the SMB share at the specified path.
763
764        This method attempts to delete a directory located at the given path on the SMB share. If the operation fails,
765        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
766        the stack trace of the exception.
767
768        Args:
769            path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None.
770        """
771        try:
772            self.smbClient.deleteDirectory(
773                shareName=self.smb_share, 
774                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
775            )
776        except Exception as err:
777            print("[!] Failed to remove directory '%s': %s" % (path, err))
778            if self.config.debug:
779                traceback.print_exc()
780
781    def rm(self, path=None):
782        """
783        Removes a file from the SMB share at the specified path.
784
785        This method attempts to delete a file located at the given path on the SMB share. If the operation fails,
786        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
787        the stack trace of the exception.
788
789        Args:
790            path (str, optional): The path of the file to be removed on the SMB share. Defaults to None.
791        """
792        try:
793            self.smbClient.deleteFile(
794                shareName=self.smb_share, 
795                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
796            )
797        except Exception as err:
798            print("[!] Failed to remove file '%s': %s" % (path, err))
799            if self.config.debug:
800                traceback.print_exc()
801
802    def tree(self, path=None):
803        """
804        Recursively lists the directory structure of the SMB share starting from the specified path.
805
806        This function prints a visual representation of the directory tree of the remote SMB share. It uses
807        recursion to navigate through directories and lists all files and subdirectories in each directory.
808        The output is color-coded and formatted to enhance readability, with directories highlighted in cyan.
809
810        Args:
811            path (str, optional): The starting path on the SMB share from which to begin listing the tree.
812                                  Defaults to the root of the current share.
813        """
814        
815        def recurse_action(base_dir="", path=[], prompt=[]):
816            bars = ["│   ", "├── ", "└── "]
817
818            remote_smb_path = ntpath.normpath(base_dir + ntpath.sep + ntpath.sep.join(path))
819
820            entries = []
821            try:
822                entries = self.smbClient.listPath(
823                    shareName=self.smb_share, 
824                    path=remote_smb_path+'\\*'
825                )
826            except impacket.smbconnection.SessionError as err:
827                code, const, text = err.getErrorCode(), err.getErrorString()[0], err.getErrorString()[1]
828                errmsg = "Error 0x%08x (%s): %s" % (code, const, text)
829                if self.config.no_colors:
830                    print("%s%s" % (''.join(prompt+[bars[2]]), errmsg))
831                else:
832                    print("%s\x1b[1;91m%s\x1b[0m" % (''.join(prompt+[bars[2]]), errmsg))
833                return 
834
835            entries = [e for e in entries if e.get_longname() not in [".", ".."]]
836            entries = sorted(entries, key=lambda x:x.get_longname())
837
838            # 
839            if len(entries) > 1:
840                index = 0
841                for entry in entries:
842                    index += 1
843                    # This is the first entry 
844                    if index == 0:
845                        if entry.is_directory():
846                            if self.config.no_colors:
847                                print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
848                            else:
849                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
850                            recurse_action(
851                                base_dir=base_dir, 
852                                path=path+[entry.get_longname()],
853                                prompt=prompt+["│   "]
854                            )
855                        else:
856                            if self.config.no_colors:
857                                print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
858                            else:
859                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
860
861                    # This is the last entry
862                    elif index == len(entries):
863                        if entry.is_directory():
864                            if self.config.no_colors:
865                                print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
866                            else:
867                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
868                            recurse_action(
869                                base_dir=base_dir, 
870                                path=path+[entry.get_longname()],
871                                prompt=prompt+["    "]
872                            )
873                        else:
874                            if self.config.no_colors:
875                                print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
876                            else:
877                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
878                        
879                    # These are entries in the middle
880                    else:
881                        if entry.is_directory():
882                            if self.config.no_colors:
883                                print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
884                            else:
885                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
886                            recurse_action(
887                                base_dir=base_dir, 
888                                path=path+[entry.get_longname()],
889                                prompt=prompt+["│   "]
890                            )
891                        else:
892                            if self.config.no_colors:
893                                print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
894                            else:
895                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
896
897            # 
898            elif len(entries) == 1:
899                entry = entries[0]
900                if entry.is_directory():
901                    if self.config.no_colors:
902                        print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
903                    else:
904                        print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
905                    recurse_action(
906                        base_dir=base_dir, 
907                        path=path+[entry.get_longname()],
908                        prompt=prompt+["    "]
909                    )
910                else:
911                    if self.config.no_colors:
912                        print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
913                    else:
914                        print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
915
916        # Entrypoint
917        try:
918            if self.config.no_colors:
919                print("%s\\" % path)
920            else:
921                print("\x1b[1;96m%s\x1b[0m\\" % path)
922            recurse_action(
923                base_dir=self.smb_cwd, 
924                path=[path],
925                prompt=[""]
926            )
927        except (BrokenPipeError, KeyboardInterrupt) as e:
928            print("[!] Interrupted.")
929            self.close_smb_session()
930            self.init_smb_session()
931
932    # Setter / Getter
933
934    def set_share(self, shareName):
935        """
936        Sets the current SMB share to the specified share name.
937
938        This method updates the SMB session to use the specified share name. It checks if the share name is valid
939        and updates the smb_share attribute of the SMBSession instance.
940
941        Parameters:
942            shareName (str): The name of the share to set as the current SMB share.
943
944        Raises:
945            ValueError: If the shareName is None or an empty string.
946        """
947
948        if shareName is not None:
949            self.list_shares()
950            if shareName.lower() in self.available_shares.keys():
951                # Doing this in order to keep the case of the share adevertised by the remote machine
952                self.smb_share = self.available_shares[shareName.lower()]["name"]
953            else:
954                print("[!] Could not set share '%s', it does not exist remotely." % shareName)
955
956    def set_cwd(self, path=None):
957        """
958        Sets the current working directory on the SMB share to the specified path.
959
960        This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory.
961        If the specified path is not a directory, the cwd remains unchanged.
962
963        Parameters:
964            path (str): The path to set as the current working directory.
965
966        Raises:
967            ValueError: If the specified path is not a directory.
968        """
969
970        if path is not None:
971            # Set path separators to ntpath sep 
972            if '/' in path:
973                path = path.replace('/', ntpath.sep)
974
975            if path.startswith(ntpath.sep):
976                # Absolute path
977                path = path + ntpath.sep
978            else:
979                # Relative path to the CWD
980                if len(self.smb_cwd) == 0:
981                    path = path + ntpath.sep
982                else:
983                    path = self.smb_cwd + ntpath.sep + path
984            
985            # Path normalization
986            path = ntpath.normpath(path)
987            path = re.sub(r'\\+', r'\\', path)
988
989            if path in ["", ".", ".."]:
990                self.smb_cwd = ""
991            else:
992                if self.path_isdir(pathFromRoot=path.strip(ntpath.sep)):
993                    # Path exists on the remote 
994                    self.smb_cwd = ntpath.normpath(path)
995                else:
996                    # Path does not exists or is not a directory on the remote 
997                    print("[!] Remote directory '%s' does not exist." % path)

Class SMBSession is designed to handle the session management for SMB (Server Message Block) protocol connections. It provides functionalities to connect to an SMB server, authenticate using either NTLM or Kerberos, and manage SMB shares.

Attributes: address (str): The IP address or hostname of the SMB server. domain (str): The domain name for SMB server authentication. username (str): The username for SMB server authentication. password (str): The password for SMB server authentication. lmhash (str): The LM hash of the user's password, if available. nthash (str): The NT hash of the user's password, if available. use_kerberos (bool): A flag to determine whether to use Kerberos for authentication. kdcHost (str): The Key Distribution Center (KDC) host for Kerberos authentication. debug (bool): A flag to enable debug output. smbClient (object): The SMB client object used for the connection. connected (bool): A flag to check the status of the connection. smb_share (str): The current SMB share in use. smb_path (str): The current path within the SMB share.

Methods: __init__(address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, debug=False): Initializes the SMBSession with the specified parameters. init_smb_session(): Initializes the SMB session by connecting to the server and authenticating using the specified method.

SMBSession( address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None)
46    def __init__(self, address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None):
47        super(SMBSession, self).__init__()
48        # Objects
49        self.config = config
50
51        # Target server
52        self.address = address
53
54        # Credentials
55        self.domain = domain
56        self.username = username
57        self.password = password 
58        self.lmhash = lmhash
59        self.nthash = nthash
60        self.use_kerberos = use_kerberos
61        self.kdcHost = kdcHost
62
63        self.smbClient = None
64        self.connected = False
65
66        self.available_shares = {}
67        self.smb_share = None
68        self.smb_cwd = ""
69
70        self.list_shares()
config
address
domain
username
password
lmhash
nthash
use_kerberos
kdcHost
smbClient
connected
available_shares
smb_share
smb_cwd
def init_smb_session(self):
 74    def init_smb_session(self):
 75        """
 76        Initializes and establishes a session with the SMB server.
 77
 78        This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration.
 79        It attempts to connect to the SMB server specified by the `address` attribute and authenticate using the credentials provided during the object's initialization.
 80
 81        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.
 82
 83        Returns:
 84            bool: True if the connection and authentication are successful, False otherwise.
 85        """
 86
 87        self.connected = False
 88
 89        if self.config.debug:
 90            print("[debug] [>] Connecting to remote SMB server '%s' ... " % self.address)
 91        try:
 92            self.smbClient = impacket.smbconnection.SMBConnection(
 93                remoteName=self.address,
 94                remoteHost=self.address,
 95                sess_port=int(445)
 96            )
 97        except OSError as err:
 98            print("[!] %s" % err)
 99            self.smbClient = None
100
101        if self.smbClient is not None:
102            if self.use_kerberos:
103                if self.config.debug:
104                    print("[debug] [>] Authenticating as '%s\\%s' with kerberos ... " % (self.domain, self.username))
105                try:
106                    self.connected = self.smbClient.kerberosLogin(
107                        user=self.username,
108                        password=self.password,
109                        domain=self.domain,
110                        lmhash=self.lmhash,
111                        nthash=self.nthash,
112                        aesKey=self.aesKey,
113                        kdcHost=self.kdcHost
114                    )
115                except impacket.smbconnection.SessionError as err:
116                    if self.config.debug:
117                        traceback.print_exc()
118                    print("[!] Could not login: %s" % err)
119                    self.connected = False
120
121            else:
122                if self.config.debug:
123                    print("[debug] [>] Authenticating as '%s\\%s' with NTLM ... " % (self.domain, self.username))
124                try:
125                    self.connected = self.smbClient.login(
126                        user=self.username,
127                        password=self.password,
128                        domain=self.domain,
129                        lmhash=self.lmhash,
130                        nthash=self.nthash
131                    )
132                except impacket.smbconnection.SessionError as err:
133                    if self.config.debug:
134                        traceback.print_exc()
135                    print("[!] Could not login: %s" % err)
136                    self.connected = False
137
138            if self.connected:
139                print("[+] Successfully authenticated to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username))
140            else:
141                print("[!] Failed to authenticate to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username))
142
143        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):
145    def close_smb_session(self):
146        """
147        Closes the current SMB session by disconnecting the SMB client.
148
149        This method ensures that the SMB client connection is properly closed. It checks if the client is connected
150        and if so, it closes the connection and resets the connection status.
151
152        Raises:
153            Exception: If the SMB client is not initialized or if there's an error during the disconnection process.
154        """
155
156        if self.smbClient is not None:
157            if self.connected:
158                self.smbClient.close()
159                self.connected = False
160                if self.config.debug:
161                    print("[+] SMB connection closed successfully.")
162            else:
163                if self.config.debug:
164                    print("[!] No active SMB connection to close.")
165        else:
166            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):
170    def read_file(self, path=None):
171        if self.path_isfile(path=path):
172            tmp_file_path = self.smb_cwd + ntpath.sep + path
173            matches = self.smbClient.listPath(
174                shareName=self.smb_share, 
175                path=tmp_file_path
176            )
177
178            fh = io.BytesIO()
179            try:
180                # opening the files in streams instead of mounting shares allows 
181                # for running the script from unprivileged containers
182                self.smbClient.getFile(self.smb_share, tmp_file_path, fh.write)
183            except impacket.smbconnection.SessionError as e:
184                return None
185            rawdata = fh.getvalue()
186            fh.close()
187            return rawdata
188        else:
189            print("[!] Remote path '%s' is not a file." % path)
def find(self, paths=[], callback=None):
191    def find(self, paths=[], callback=None):
192        def recurse_action(paths=[], depth=0, callback=None):
193            if callback is None:
194                return []
195            next_directories_to_explore = []
196            for path in paths:
197                remote_smb_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
198                entries = []
199                
200                try:
201                    entries = self.smbClient.listPath(
202                        shareName=self.smb_share, 
203                        path=(remote_smb_path + ntpath.sep + '*')
204                    )
205                except impacket.smbconnection.SessionError as err:
206                    continue 
207                # Remove dot names
208                entries = [e for e in entries if e.get_longname() not in [".", ".."]]
209                # Sort the entries ignoring case
210                entries = sorted(entries, key=lambda x:x.get_longname().lower())
211                
212                for entry in entries:
213                    if entry.is_directory():
214                        callback(entry, path + entry.get_longname() + ntpath.sep, depth)
215                    else:
216                        callback(entry, path + entry.get_longname(), depth)
217
218                # Next directories to explore
219                for entry in entries:
220                    if entry.is_directory():
221                        next_directories_to_explore.append(path + entry.get_longname() + ntpath.sep)
222            
223            return next_directories_to_explore
224        # 
225        if callback is not None:
226            depth = 0
227            while len(paths) != 0:
228                paths = recurse_action(
229                    paths=paths,
230                    depth=depth,
231                    callback=callback
232                )
233                depth = depth + 1
234        else:
235            print("[!] SMBSession.find(), callback function cannot be None.")
def get_file(self, path=None, keepRemotePath=False):
237    def get_file(self, path=None, keepRemotePath=False):
238        """
239        Retrieves a file from the specified path on the SMB share.
240
241        This method attempts to retrieve a file from the given path within the currently connected SMB share.
242        If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local
243        file object and writing the contents of the remote file to it using the SMB client's getFile method.
244
245        Parameters:
246            path (str): The path of the file to retrieve. If None, uses the current smb_path.
247
248        Returns:
249            None
250        """
251
252        tmp_file_path = self.smb_cwd + ntpath.sep + path
253        matches = self.smbClient.listPath(
254            shareName=self.smb_share, 
255            path=tmp_file_path
256        )
257        
258        for entry in matches:
259            if entry.is_directory():
260                print("[>] Skipping '%s' because it is a directory." % tmp_file_path)
261            else:
262                try:
263                    if ntpath.sep in path:
264                        outputfile = ntpath.dirname(path) + ntpath.sep + entry.get_longname()
265                    else:
266                        outputfile = entry.get_longname()
267                    f = LocalFileIO(
268                        mode="wb", 
269                        path=outputfile,
270                        expected_size=entry.get_filesize(), 
271                        debug=self.config.debug,
272                        keepRemotePath=keepRemotePath
273                    )
274                    self.smbClient.getFile(
275                        shareName=self.smb_share, 
276                        pathName=tmp_file_path, 
277                        callback=f.write
278                    )
279                    f.close()
280                except (BrokenPipeError, KeyboardInterrupt) as e:
281                    f.close()
282                    print("\x1b[v\x1b[o\r[!] Interrupted.")
283                    self.close_smb_session()
284                    self.init_smb_session()
285                        
286        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):
288    def get_file_recursively(self, path=None):
289        """
290        Recursively retrieves files from a specified path on the SMB share.
291
292        This method navigates through all directories starting from the given path,
293        and downloads all files found. It handles directories recursively, ensuring
294        that all nested files are retrieved. The method skips over directory entries
295        and handles errors gracefully, attempting to continue the operation where possible.
296
297        Parameters:
298            path (str): The initial directory path from which to start the recursive file retrieval.
299                        If None, it starts from the root of the configured SMB share.
300        """
301        
302        def recurse_action(base_dir="", path=[]):
303            remote_smb_path = base_dir + ntpath.sep.join(path)
304            entries = self.smbClient.listPath(
305                shareName=self.smb_share, 
306                path=remote_smb_path + '\\*'
307            )
308            if len(entries) != 0:
309                files = [entry for entry in entries if not entry.is_directory()]
310                directories = [entry for entry in entries if entry.is_directory() and entry.get_longname() not in [".", ".."]]
311
312                # Files
313                if len(files) != 0:
314                    print("[>] Retrieving files of '%s'" % remote_smb_path)
315                for entry_file in files:
316                    if not entry_file.is_directory():
317                        f = LocalFileIO(
318                            mode="wb",
319                            path=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
320                            expected_size=entry_file.get_filesize(),
321                            debug=self.config.debug
322                        )
323                        try:
324                            self.smbClient.getFile(
325                                shareName=self.smb_share, 
326                                pathName=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
327                                callback=f.write
328                            )
329                            f.close()
330                        except BrokenPipeError as err:
331                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
332                            f.close(remove=True)
333                            break
334                        except Exception as err:
335                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
336                            f.close(remove=True)
337                
338                # Directories
339                for entry_directory in directories:
340                    if entry_directory.is_directory():
341                        recurse_action(
342                            base_dir=self.smb_cwd, 
343                            path=path+[entry_directory.get_longname()]
344                        )                   
345        # Entrypoint
346        try:
347            recurse_action(
348                base_dir=self.smb_cwd, 
349                path=[path]
350            )
351        except (BrokenPipeError, KeyboardInterrupt) as e:
352            print("\x1b[v\x1b[o\r[!] Interrupted.")
353            self.close_smb_session()
354            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 info(self, share=True, server=True):
356    def info(self, share=True, server=True):
357        """
358        Displays information about the server and optionally the shares.
359
360        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.
361
362        Parameters:
363            share (bool): If True, display information about the current share.
364            server (bool): If True, display information about the server.
365
366        Returns:
367            None
368        """
369
370        if server:
371            if self.config.no_colors:
372                print("[+] Server:")
373                print("  ├─NetBIOS:")
374                print("  │ ├─ NetBIOS Hostname ──────── : %s" % (self.smbClient.getServerName()))
375                print("  │ └─ NetBIOS Domain ────────── : %s" % (self.smbClient.getServerDomain()))
376                print("  ├─DNS:")
377                print("  │ ├─ DNS Hostname ──────────── : %s" % (self.smbClient.getServerDNSHostName()))
378                print("  │ └─ DNS Domain ────────────── : %s" % (self.smbClient.getServerDNSDomainName()))
379                print("  ├─OS:")
380                print("  │ ├─ OS Name ───────────────── : %s" % (self.smbClient.getServerOS()))
381                print("  │ └─ OS Version ────────────── : %s.%s.%s" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild()))
382                print("  ├─Server:")
383                print("  │ ├─ Signing Required ──────── : %s" % (self.smbClient.isSigningRequired()))
384                print("  │ ├─ Login Required ────────── : %s" % (self.smbClient.isLoginRequired()))
385                print("  │ ├─ Supports NTLMv2 ───────── : %s" % (self.smbClient.doesSupportNTLMv2()))
386                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
387                print("  │ ├─ Max size of read chunk ── : %d bytes (%s)" % (MaxReadSize, b_filesize(MaxReadSize)))
388                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
389                print("  │ └─ Max size of write chunk ─ : %d bytes (%s)" % (MaxWriteSize, b_filesize(MaxWriteSize)))
390                print("  └─")
391            else:
392                print("[+] Server:")
393                print("  ├─NetBIOS:")
394                print("  │ ├─ \x1b[94mNetBIOS Hostname\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerName()))
395                print("  │ └─ \x1b[94mNetBIOS Domain\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDomain()))
396                print("  ├─DNS:")
397                print("  │ ├─ \x1b[94mDNS Hostname\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSHostName()))
398                print("  │ └─ \x1b[94mDNS Domain\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSDomainName()))
399                print("  ├─OS:")
400                print("  │ ├─ \x1b[94mOS Name\x1b[0m \x1b[90m─────────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerOS()))
401                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()))
402                print("  ├─Server:")
403                print("  │ ├─ \x1b[94mSigning Required\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isSigningRequired()))
404                print("  │ ├─ \x1b[94mLogin Required\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isLoginRequired()))
405                print("  │ ├─ \x1b[94mSupports NTLMv2\x1b[0m \x1b[90m─────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.doesSupportNTLMv2()))
406                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
407                print("  │ ├─ \x1b[94mMax size of read chunk\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxReadSize, b_filesize(MaxReadSize)))
408                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
409                print("  │ └─ \x1b[94mMax size of write chunk\x1b[0m \x1b[90m─\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxWriteSize, b_filesize(MaxWriteSize)))
410                print("  └─")
411
412        if share and self.smb_share is not None:
413            share_name = self.available_shares.get(self.smb_share.lower(), "")["name"]
414            share_comment = self.available_shares.get(self.smb_share.lower(), "")["comment"]
415            share_type = self.available_shares.get(self.smb_share.lower(), "")["type"]
416            share_type =', '.join([s.replace("STYPE_","") for s in share_type])
417            share_rawtype = self.available_shares.get(self.smb_share.lower(), "")["rawtype"]
418            if self.config.no_colors:
419                print("\n[+] Share:")
420                print("  ├─ Name ──────────── : %s" % (share_name))
421                print("  ├─ Description ───── : %s" % (share_comment))
422                print("  ├─ Type ──────────── : %s" % (share_type))
423                print("  └─ Raw type value ── : %s" % (share_rawtype))
424            else:
425                print("\n[+] Share:")
426                print("  ├─ \x1b[94mName\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_name))
427                print("  ├─ \x1b[94mDescription\x1b[0m \x1b[90m─────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_comment))
428                print("  ├─ \x1b[94mType\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_type))
429                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):
431    def list_contents(self, path=None):
432        """
433        Lists the contents of a specified directory on the SMB share.
434
435        This method retrieves the contents of a directory specified by `shareName` and `path`. If `shareName` or `path`
436        is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with
437        the long names of the files and directories as keys and their respective SMB entry objects as values.
438
439        Args:
440            shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None.
441            path (str, optional): The directory path to list contents from. Defaults to the current path if None.
442
443        Returns:
444            dict: A dictionary with file and directory names as keys and their SMB entry objects as values.
445        """
446        
447        if path is None or len(path) == 0:
448            path = self.smb_cwd
449        path = path.rstrip(ntpath.sep) + ntpath.sep + "*"
450
451        contents = {}
452        entries = self.smbClient.listPath(
453            shareName=self.smb_share, 
454            path=path
455        )
456        for entry in entries:
457            contents[entry.get_longname()] = entry
458
459        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):
461    def list_shares(self):
462        """
463        Lists all the shares available on the connected SMB server.
464
465        This method queries the SMB server to retrieve a list of all available shares. It populates the `shares` dictionary
466        with key-value pairs where the key is the share name and the value is a dictionary containing details about the share
467        such as its name, type, raw type, and any comments associated with the share.
468
469        Returns:
470            dict: A dictionary containing information about each share available on the server.
471        """
472
473        self.available_shares = {}
474
475        if self.connected:
476            if self.smbClient is not None:
477                resp = self.smbClient.listShares()
478
479                for share in resp:
480                    # SHARE_INFO_1 structure (lmshare.h)
481                    # https://learn.microsoft.com/en-us/windows/win32/api/lmshare/ns-lmshare-share_info_1
482                    sharename = share["shi1_netname"][:-1]
483                    sharecomment = share["shi1_remark"][:-1]
484                    sharetype = share["shi1_type"]
485
486                    self.available_shares[sharename.lower()] = {
487                        "name": sharename, 
488                        "type": STYPE_MASK(sharetype), 
489                        "rawtype": sharetype, 
490                        "comment": sharecomment
491                    }
492            else:
493                print("[!] Error: SMBSession.smbClient is None.")
494
495        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):
497    def mkdir(self, path=None):
498        """
499        Creates a directory at the specified path on the SMB share.
500
501        This method takes a path and attempts to create the directory structure on the SMB share. If the path includes
502        nested directories, it will create each directory in the sequence. If a directory already exists, it will skip
503        the creation for that directory without raising an error.
504
505        Args:
506            path (str, optional): The full path of the directory to create on the SMB share. Defaults to None.
507
508        Note:
509            The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility.
510        """
511
512        if path is not None:
513            # Prepare path
514            path = path.replace('/',ntpath.sep)
515            if ntpath.sep in path:
516                path = path.strip(ntpath.sep).split(ntpath.sep)
517            else:
518                path = [path]
519
520            # Create each dir in the path
521            for depth in range(1, len(path)+1):
522                tmp_path = ntpath.sep.join(path[:depth])
523                try:
524                    self.smbClient.createDirectory(
525                        shareName=self.smb_share, 
526                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + tmp_path + ntpath.sep)
527                    )
528                except impacket.smbconnection.SessionError as err:
529                    if err.getErrorCode() == 0xc0000035:
530                        # STATUS_OBJECT_NAME_COLLISION
531                        # Remote directory already created, this is normal
532                        # Src: https://github.com/fortra/impacket/blob/269ce69872f0e8f2188a80addb0c39fedfa6dcb8/impacket/nt_errors.py#L268C9-L268C19
533                        pass
534                    else:
535                        print("[!] Failed to create directory '%s': %s" % (tmp_path, err))
536                        if self.config.debug:
537                            traceback.print_exc()
538        else:
539            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 path_exists(self, path=None):
541    def path_exists(self, path=None):
542        """
543        Checks if the specified path exists on the SMB share.
544
545        This method determines if a given path exists on the SMB share by attempting to list the contents of the path.
546        If the path listing is successful and returns one or more entries, the path is considered to exist.
547
548        Args:
549            path (str, optional): The path to check on the SMB share. Defaults to None.
550
551        Returns:
552            bool: True if the path exists, False otherwise or if an error occurs.
553        """
554
555        if path is not None:
556            path = path.replace('*','')
557            try:
558                contents = self.smbClient.listPath(
559                    shareName=self.smb_share,
560                    path=ntpath.normpath(self.smb_cwd + ntpath.sep + path + ntpath.sep)
561                )
562                return (len(contents) != 0)
563            except Exception as e:
564                return False
565        else:
566            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):
568    def path_isdir(self, pathFromRoot=None):
569        """
570        Checks if the specified path is a directory on the SMB share.
571
572        This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the
573        contents of the path and filtering for entries that match the basename of the path and are marked as directories.
574
575        Args:
576            path (str, optional): The path to check on the SMB share. Defaults to None.
577
578        Returns:
579            bool: True if the path is a directory, False otherwise or if an error occurs.
580        """
581
582        if pathFromRoot is not None:
583            # Replace slashes if any
584            path = pathFromRoot.replace('/', ntpath.sep)
585            
586            # Strip wildcards to avoid injections
587            path = path.replace('*','')
588
589            # Normalize path and strip leading backslash
590            path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep)
591
592            if path.strip() in ['', '.', '..']:
593                # By defininition they exist on the filesystem
594                return True
595            else:
596                try:
597                    contents = self.smbClient.listPath(
598                        shareName=self.smb_share,
599                        path=path+'*'
600                    )
601                    # Filter on directories
602                    contents = [
603                        c for c in contents
604                        if c.get_longname() == ntpath.basename(path) and c.is_directory()
605                    ]
606                    return (len(contents) != 0)
607                except Exception as e:
608                    return False
609        else:
610            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):
612    def path_isfile(self, path=None):
613        """
614        Checks if the specified path is a file on the SMB share.
615
616        This method determines if a given path corresponds to a file on the SMB share. It does this by listing the
617        contents of the path and filtering for entries that match the basename of the path and are not marked as directories.
618
619        Args:
620            path (str, optional): The path to check on the SMB share. Defaults to None.
621
622        Returns:
623            bool: True if the path is a file, False otherwise or if an error occurs.
624        """
625
626        if path is not None:
627            path = path.replace('*','')
628            search_dir = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
629            search_dir = ntpath.dirname(search_dir) + ntpath.sep + '*'
630            try:
631                contents = self.smbClient.listPath(
632                    shareName=self.smb_share,
633                    path=search_dir
634                )
635                # Filter on files
636                contents = [
637                    c for c in contents
638                    if c.get_longname() == ntpath.basename(path) and not c.is_directory()
639                ]
640                return (len(contents) != 0)
641            except Exception as e:
642                return False
643        else:
644            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):
646    def ping_smb_session(self):
647        """
648        Tests the connectivity to the SMB server by sending an echo command.
649
650        This method attempts to send an echo command to the SMB server to check if the session is still active.
651        It updates the `connected` attribute of the class based on the success or failure of the echo command.
652
653        Returns:
654            bool: True if the echo command succeeds (indicating the session is active), False otherwise.
655        """
656
657        try:
658            self.smbClient.getSMBServer().echo()
659        except Exception as e:
660            self.connected = False
661        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):
663    def put_file(self, localpath=None):
664        """
665        Uploads a single file to the SMB share.
666
667        This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path.
668        It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session.
669        General exceptions are caught and logged, with a traceback provided if debugging is enabled.
670
671        Args:
672            localpath (str, optional): The local file path of the file to be uploaded. Defaults to None.
673        """
674
675        if os.path.exists(localpath):
676            if os.path.isfile(localpath):
677                try:
678                    localfile = os.path.basename(localpath)
679                    f = LocalFileIO(
680                        mode="rb", 
681                        path=localpath, 
682                        debug=self.config.debug
683                    )
684                    self.smbClient.putFile(
685                        shareName=self.smb_share, 
686                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + localfile + ntpath.sep), 
687                        callback=f.read
688                    )
689                    f.close()
690                except (BrokenPipeError, KeyboardInterrupt) as err:
691                    print("[!] Interrupted.")
692                    self.close_smb_session()
693                    self.init_smb_session()
694                except Exception as err:
695                    print("[!] Failed to upload '%s': %s" % (localfile, err))
696                    if self.config.debug:
697                        traceback.print_exc()
698            else:
699                print("[!] The specified localpath is a directory. Use 'put -r <directory>' instead.")
700        else:
701            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):
703    def put_file_recursively(self, localpath=None):
704        """
705        Recursively uploads files from a specified local directory to the SMB share.
706
707        This method walks through the given local directory and all its subdirectories, uploading each file to the
708        corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is,
709        it iterates over all files and directories within the local path, creating necessary directories on the SMB share
710        and uploading files. If the local path is not a directory, it prints an error message.
711
712        Args:
713            localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None.
714        """
715
716        if os.path.exists(localpath):
717            if os.path.isfile(localpath):
718                # Iterate over all files and directories within the local path
719                local_files = {}
720                for root, dirs, files in os.walk(localpath):
721                    if len(files) != 0:
722                        local_files[root] = files
723
724                # Iterate over the found files
725                for local_dir_path in sorted(local_files.keys()):
726                    print("[>] Putting files of '%s'" % local_dir_path)
727
728                    # Create remote directory
729                    remote_dir_path = local_dir_path.replace(os.path.sep, ntpath.sep)
730                    self.mkdir(
731                        path=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep)
732                    )
733
734                    for local_file_path in local_files[local_dir_path]:
735                        try:
736                            f = LocalFileIO(
737                                mode="rb", 
738                                path=local_dir_path + os.path.sep + local_file_path, 
739                                debug=self.config.debug
740                            )
741                            self.smbClient.putFile(
742                                shareName=self.smb_share, 
743                                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep + local_file_path), 
744                                callback=f.read
745                            )
746                            f.close()
747
748                        except BrokenPipeError as err:
749                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
750                            f.close(remove=True)
751                            break
752                        except Exception as err:
753                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
754                            f.close(remove=True)
755                else:
756                    print("[!] The specified localpath is a file. Use 'put <file>' instead.")
757        else:
758            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):
760    def rmdir(self, path=None):
761        """
762        Removes a directory from the SMB share at the specified path.
763
764        This method attempts to delete a directory located at the given path on the SMB share. If the operation fails,
765        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
766        the stack trace of the exception.
767
768        Args:
769            path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None.
770        """
771        try:
772            self.smbClient.deleteDirectory(
773                shareName=self.smb_share, 
774                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
775            )
776        except Exception as err:
777            print("[!] Failed to remove directory '%s': %s" % (path, err))
778            if self.config.debug:
779                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):
781    def rm(self, path=None):
782        """
783        Removes a file from the SMB share at the specified path.
784
785        This method attempts to delete a file located at the given path on the SMB share. If the operation fails,
786        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
787        the stack trace of the exception.
788
789        Args:
790            path (str, optional): The path of the file to be removed on the SMB share. Defaults to None.
791        """
792        try:
793            self.smbClient.deleteFile(
794                shareName=self.smb_share, 
795                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
796            )
797        except Exception as err:
798            print("[!] Failed to remove file '%s': %s" % (path, err))
799            if self.config.debug:
800                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):
802    def tree(self, path=None):
803        """
804        Recursively lists the directory structure of the SMB share starting from the specified path.
805
806        This function prints a visual representation of the directory tree of the remote SMB share. It uses
807        recursion to navigate through directories and lists all files and subdirectories in each directory.
808        The output is color-coded and formatted to enhance readability, with directories highlighted in cyan.
809
810        Args:
811            path (str, optional): The starting path on the SMB share from which to begin listing the tree.
812                                  Defaults to the root of the current share.
813        """
814        
815        def recurse_action(base_dir="", path=[], prompt=[]):
816            bars = ["│   ", "├── ", "└── "]
817
818            remote_smb_path = ntpath.normpath(base_dir + ntpath.sep + ntpath.sep.join(path))
819
820            entries = []
821            try:
822                entries = self.smbClient.listPath(
823                    shareName=self.smb_share, 
824                    path=remote_smb_path+'\\*'
825                )
826            except impacket.smbconnection.SessionError as err:
827                code, const, text = err.getErrorCode(), err.getErrorString()[0], err.getErrorString()[1]
828                errmsg = "Error 0x%08x (%s): %s" % (code, const, text)
829                if self.config.no_colors:
830                    print("%s%s" % (''.join(prompt+[bars[2]]), errmsg))
831                else:
832                    print("%s\x1b[1;91m%s\x1b[0m" % (''.join(prompt+[bars[2]]), errmsg))
833                return 
834
835            entries = [e for e in entries if e.get_longname() not in [".", ".."]]
836            entries = sorted(entries, key=lambda x:x.get_longname())
837
838            # 
839            if len(entries) > 1:
840                index = 0
841                for entry in entries:
842                    index += 1
843                    # This is the first entry 
844                    if index == 0:
845                        if entry.is_directory():
846                            if self.config.no_colors:
847                                print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
848                            else:
849                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
850                            recurse_action(
851                                base_dir=base_dir, 
852                                path=path+[entry.get_longname()],
853                                prompt=prompt+["│   "]
854                            )
855                        else:
856                            if self.config.no_colors:
857                                print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
858                            else:
859                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
860
861                    # This is the last entry
862                    elif index == len(entries):
863                        if entry.is_directory():
864                            if self.config.no_colors:
865                                print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
866                            else:
867                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
868                            recurse_action(
869                                base_dir=base_dir, 
870                                path=path+[entry.get_longname()],
871                                prompt=prompt+["    "]
872                            )
873                        else:
874                            if self.config.no_colors:
875                                print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
876                            else:
877                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
878                        
879                    # These are entries in the middle
880                    else:
881                        if entry.is_directory():
882                            if self.config.no_colors:
883                                print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
884                            else:
885                                print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
886                            recurse_action(
887                                base_dir=base_dir, 
888                                path=path+[entry.get_longname()],
889                                prompt=prompt+["│   "]
890                            )
891                        else:
892                            if self.config.no_colors:
893                                print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
894                            else:
895                                print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
896
897            # 
898            elif len(entries) == 1:
899                entry = entries[0]
900                if entry.is_directory():
901                    if self.config.no_colors:
902                        print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
903                    else:
904                        print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
905                    recurse_action(
906                        base_dir=base_dir, 
907                        path=path+[entry.get_longname()],
908                        prompt=prompt+["    "]
909                    )
910                else:
911                    if self.config.no_colors:
912                        print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
913                    else:
914                        print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
915
916        # Entrypoint
917        try:
918            if self.config.no_colors:
919                print("%s\\" % path)
920            else:
921                print("\x1b[1;96m%s\x1b[0m\\" % path)
922            recurse_action(
923                base_dir=self.smb_cwd, 
924                path=[path],
925                prompt=[""]
926            )
927        except (BrokenPipeError, KeyboardInterrupt) as e:
928            print("[!] Interrupted.")
929            self.close_smb_session()
930            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 set_share(self, shareName):
934    def set_share(self, shareName):
935        """
936        Sets the current SMB share to the specified share name.
937
938        This method updates the SMB session to use the specified share name. It checks if the share name is valid
939        and updates the smb_share attribute of the SMBSession instance.
940
941        Parameters:
942            shareName (str): The name of the share to set as the current SMB share.
943
944        Raises:
945            ValueError: If the shareName is None or an empty string.
946        """
947
948        if shareName is not None:
949            self.list_shares()
950            if shareName.lower() in self.available_shares.keys():
951                # Doing this in order to keep the case of the share adevertised by the remote machine
952                self.smb_share = self.available_shares[shareName.lower()]["name"]
953            else:
954                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):
956    def set_cwd(self, path=None):
957        """
958        Sets the current working directory on the SMB share to the specified path.
959
960        This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory.
961        If the specified path is not a directory, the cwd remains unchanged.
962
963        Parameters:
964            path (str): The path to set as the current working directory.
965
966        Raises:
967            ValueError: If the specified path is not a directory.
968        """
969
970        if path is not None:
971            # Set path separators to ntpath sep 
972            if '/' in path:
973                path = path.replace('/', ntpath.sep)
974
975            if path.startswith(ntpath.sep):
976                # Absolute path
977                path = path + ntpath.sep
978            else:
979                # Relative path to the CWD
980                if len(self.smb_cwd) == 0:
981                    path = path + ntpath.sep
982                else:
983                    path = self.smb_cwd + ntpath.sep + path
984            
985            # Path normalization
986            path = ntpath.normpath(path)
987            path = re.sub(r'\\+', r'\\', path)
988
989            if path in ["", ".", ".."]:
990                self.smb_cwd = ""
991            else:
992                if self.path_isdir(pathFromRoot=path.strip(ntpath.sep)):
993                    # Path exists on the remote 
994                    self.smb_cwd = ntpath.normpath(path)
995                else:
996                    # Path does not exists or is not a directory on the remote 
997                    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.