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 random
  13import re
  14import sys
  15import traceback
  16from smbclientng.core.LocalFileIO import LocalFileIO
  17from smbclientng.core.utils import b_filesize, STYPE_MASK, is_port_open
  18
  19
  20class SMBSession(object):
  21    """
  22    Represents an SMB session for interacting with an SMB server.
  23
  24    This class provides methods to manage and interact with an SMB server, including
  25    connecting to the server, listing shares, uploading and downloading files, and
  26    managing directories and files on the server. It handles session initialization,
  27    authentication, and cleanup.
  28
  29    Attributes:
  30        host (str): The hostname or IP address of the SMB server.
  31        port (int): The port number on which the SMB server is listening.
  32        credentials (dict): Authentication credentials for the SMB server.
  33        config (dict, optional): Configuration options for the SMB session.
  34        smbClient (impacket.smbconnection.SMBConnection): The SMB connection instance.
  35        connected (bool): Connection status to the SMB server.
  36        available_shares (dict): A dictionary of available SMB shares.
  37        smb_share (str): The current SMB share in use.
  38        smb_cwd (str): The current working directory on the SMB share.
  39        smb_tree_id (int): The tree ID of the connected SMB share.
  40
  41    Methods:
  42        close_smb_session(): Closes the current SMB session.
  43        init_smb_session(): Initializes the SMB session with the server.
  44        list_shares(): Lists all shares available on the SMB server.
  45        set_share(shareName): Sets the current SMB share.
  46        set_cwd(path): Sets the current working directory on the SMB share.
  47        put_file(localpath): Uploads a file to the current SMB share.
  48        get_file(remotepath, localpath): Downloads a file from the SMB share.
  49        mkdir(path): Creates a directory on the SMB share.
  50        rmdir(path): Removes a directory from the SMB share.
  51        rm(path): Removes a file from the SMB share.
  52        read_file(path): Reads a file from the SMB share.
  53        test_rights(sharename): Tests read and write access rights on a share.
  54    """
  55
  56    def __init__(self, host, port, credentials, config=None, logger=None):
  57        super(SMBSession, self).__init__()
  58        # Objects
  59        self.config = config
  60        self.logger = logger
  61
  62        # Target server
  63        self.host = host
  64        # Target port (by default on 445)
  65        self.port = port
  66
  67        # Credentials
  68        self.credentials = credentials
  69
  70        self.smbClient = None
  71        self.connected = False
  72
  73        self.available_shares = {}
  74        self.smb_share = None
  75        self.smb_cwd = ""
  76        self.smb_tree_id = None
  77
  78        self.list_shares()
  79
  80    # Connect and disconnect SMB session
  81
  82    def close_smb_session(self):
  83        """
  84        Closes the current SMB session by disconnecting the SMB client.
  85
  86        This method ensures that the SMB client connection is properly closed. It checks if the client is connected
  87        and if so, it closes the connection and resets the connection status.
  88
  89        Raises:
  90            Exception: If the SMB client is not initialized or if there's an error during the disconnection process.
  91        """
  92
  93        if self.smbClient is not None:
  94            if self.connected:
  95                self.smbClient.close()
  96                self.connected = False
  97                self.logger.debug("[+] SMB connection closed successfully.")
  98            else:
  99                self.logger.debug("[!] No active SMB connection to close.")
 100        else:
 101            raise Exception("SMB client is not initialized.")
 102
 103    def init_smb_session(self):
 104        """
 105        Initializes and establishes a session with the SMB server.
 106
 107        This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration.
 108        It attempts to connect to the SMB server specified by the `address` attribute and authenticate using the credentials provided during the object's initialization.
 109
 110        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.
 111
 112        Returns:
 113            bool: True if the connection and authentication are successful, False otherwise.
 114        """
 115
 116        self.connected = False
 117
 118        self.logger.debug("[>] Connecting to remote SMB server '%s' ... " % self.host)
 119        
 120        try:
 121            if is_port_open(self.host, self.port):
 122                self.smbClient = impacket.smbconnection.SMBConnection(
 123                    remoteName=self.host,
 124                    remoteHost=self.host,
 125                    sess_port=int(self.port)
 126                )
 127            else:
 128                self.connected = False
 129        except OSError as err:
 130            if self.config.debug:
 131                traceback.print_exc()
 132            self.logger.error(err)
 133            self.smbClient = None
 134
 135        if self.smbClient is not None:
 136            if self.credentials.use_kerberos:
 137                self.logger.debug("[>] Authenticating as '%s\\%s' with kerberos ... " % (self.credentials.domain, self.credentials.username))
 138                try:
 139                    self.connected = self.smbClient.kerberosLogin(
 140                        user=self.credentials.username,
 141                        password=self.credentials.password,
 142                        domain=self.credentials.domain,
 143                        lmhash=self.credentials.lm_hex,
 144                        nthash=self.credentials.nt_hex,
 145                        aesKey=self.credentials.aesKey,
 146                        kdcHost=self.credentials.kdcHost
 147                    )
 148                except impacket.smbconnection.SessionError as err:
 149                    if self.config.debug:
 150                        traceback.print_exc()
 151                    self.logger.error("Could not login: %s" % err)
 152                    self.connected = False
 153
 154            else:
 155                self.logger.debug("[>] Authenticating as '%s\\%s' with NTLM ... " % (self.credentials.domain, self.credentials.username))
 156                
 157                try:
 158                    self.connected = self.smbClient.login(
 159                        user=self.credentials.username,
 160                        password=self.credentials.password,
 161                        domain=self.credentials.domain,
 162                        lmhash=self.credentials.lm_hex,
 163                        nthash=self.credentials.nt_hex
 164                    )
 165                except impacket.smbconnection.SessionError as err:
 166                    if self.config.debug:
 167                        traceback.print_exc()
 168                    self.logger.error("Could not login: %s" % err)
 169                    self.connected = False
 170
 171            if self.connected:
 172                self.logger.print("[+] Successfully authenticated to '%s' as '%s\\%s'!" % (self.host, self.credentials.domain, self.credentials.username))
 173            else:
 174                self.logger.error("Failed to authenticate to '%s' as '%s\\%s'!" % (self.host, self.credentials.domain, self.credentials.username))
 175
 176        return self.connected
 177
 178    def ping_smb_session(self):
 179        """
 180        Tests the connectivity to the SMB server by sending an echo command.
 181
 182        This method attempts to send an echo command to the SMB server to check if the session is still active.
 183        It updates the `connected` attribute of the class based on the success or failure of the echo command.
 184
 185        Returns:
 186            bool: True if the echo command succeeds (indicating the session is active), False otherwise.
 187        """
 188
 189        if not is_port_open(self.host, self.port):
 190            self.connected = False
 191        else:
 192            try:
 193                self.smbClient.getSMBServer().echo()
 194            except Exception as e:
 195                self.connected = False
 196
 197        return self.connected
 198
 199    # Operations
 200
 201    def find(self, paths=[], callback=None):
 202        """
 203        Finds files and directories on the SMB share based on the provided paths and executes a callback function on each entry.
 204
 205        This method traverses the specified paths on the SMB share, recursively exploring directories and invoking the callback
 206        function on each file or directory found. The callback function is called with three arguments: the entry object, the
 207        full path of the entry, and the current depth of recursion.
 208
 209        Args:
 210            paths (list, optional): A list of paths to start the search from. Defaults to an empty list.
 211            callback (function, optional): A function to be called on each entry found. The function should accept three arguments:
 212                                           the entry object, the full path of the entry, and the current depth of recursion. Defaults to None.
 213
 214        Note:
 215            If the callback function is None, the method will print an error message and return without performing any action.
 216        """
 217
 218        def recurse_action(paths=[], depth=0, callback=None):
 219            if callback is None:
 220                return []
 221            
 222            next_directories_to_explore = []
 223
 224            for path in paths:
 225                remote_smb_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
 226                entries = []
 227                
 228                try:
 229                    entries = self.smbClient.listPath(
 230                        shareName=self.smb_share, 
 231                        path=(remote_smb_path + ntpath.sep + '*')
 232                    )
 233                except impacket.smbconnection.SessionError as err:
 234                    continue 
 235                # Remove dot names
 236                entries = [e for e in entries if e.get_longname() not in [".", ".."]]
 237                # Sort the entries ignoring case
 238                entries = sorted(entries, key=lambda x:x.get_longname().lower())
 239                
 240                for entry in entries:
 241                    if entry.is_directory():
 242                        fullpath = path + ntpath.sep + entry.get_longname() + ntpath.sep
 243                        next_directories_to_explore.append(fullpath)
 244                    else:
 245                        fullpath = path + ntpath.sep + entry.get_longname()
 246                    fullpath = re.sub(r'\\\\+', r'\\', fullpath)
 247                    callback(entry, fullpath, depth)
 248                     
 249            return next_directories_to_explore
 250        # 
 251        if callback is not None:
 252            depth = 0
 253            while len(paths) != 0:
 254                paths = recurse_action(
 255                    paths=paths,
 256                    depth=depth,
 257                    callback=callback
 258                )
 259                depth = depth + 1
 260        else:
 261            self.logger.error("SMBSession.find(), callback function cannot be None.")
 262
 263    def get_file(self, path=None, keepRemotePath=False):
 264        """
 265        Retrieves a file from the specified path on the SMB share.
 266
 267        This method attempts to retrieve a file from the given path within the currently connected SMB share.
 268        If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local
 269        file object and writing the contents of the remote file to it using the SMB client's getFile method.
 270
 271        Parameters:
 272            path (str): The path of the file to retrieve. If None, uses the current smb_path.
 273
 274        Returns:
 275            None
 276        """
 277
 278        # Parse path
 279        path = path.replace('/', ntpath.sep)
 280        if ntpath.sep in path:
 281            tmp_search_path = ntpath.normpath(self.smb_cwd + ntpath.sep + ntpath.dirname(path))
 282        else:
 283            tmp_search_path = ntpath.normpath(self.smb_cwd + ntpath.sep)
 284        # Parse filename
 285        filename = ntpath.basename(path)
 286
 287        # Search for the file
 288        matches = self.smbClient.listPath(
 289            shareName=self.smb_share, 
 290            path=tmp_search_path + ntpath.sep + '*'
 291        )   
 292
 293        # Filter the entries
 294        matching_entries = []
 295        for entry in matches:
 296            if entry.is_directory():
 297                # Skip directories
 298                continue
 299            if entry.get_longname() == filename:
 300                matching_entries.append(entry)
 301            elif '*' in filename:
 302                regexp = filename.replace('.', '\\.').replace('*', '.*')
 303                if re.match(regexp, entry.get_longname()):
 304                    matching_entries.append(entry)
 305        
 306        matching_entries = sorted(list(set(matching_entries)), key=lambda x: x.get_longname())
 307
 308        for entry in matching_entries:
 309            if entry.is_directory():
 310                self.logger.debug("[>] Skipping '%s' because it is a directory." % (tmp_search_path + ntpath.sep + entry.get_longname()))
 311            else:
 312                try:
 313                    if ntpath.sep in path:
 314                        outputfile = ntpath.dirname(path) + ntpath.sep + entry.get_longname()
 315                    else:
 316                        outputfile = entry.get_longname()
 317                    f = LocalFileIO(
 318                        mode="wb", 
 319                        path=outputfile,
 320                        expected_size=entry.get_filesize(), 
 321                        debug=self.config.debug,
 322                        keepRemotePath=keepRemotePath
 323                    )
 324                    self.smbClient.getFile(
 325                        shareName=self.smb_share, 
 326                        pathName=tmp_search_path + ntpath.sep + entry.get_longname(), 
 327                        callback=f.write
 328                    )
 329                    f.close()
 330                except (BrokenPipeError, KeyboardInterrupt) as e:
 331                    f.close()
 332                    print("\x1b[v\x1b[o\r[!] Interrupted.")
 333                    self.close_smb_session()
 334                    self.init_smb_session()
 335                        
 336        return None
 337
 338    def get_file_recursively(self, path=None):
 339        """
 340        Recursively retrieves files from a specified path on the SMB share.
 341
 342        This method navigates through all directories starting from the given path,
 343        and downloads all files found. It handles directories recursively, ensuring
 344        that all nested files are retrieved. The method skips over directory entries
 345        and handles errors gracefully, attempting to continue the operation where possible.
 346
 347        Parameters:
 348            path (str): The initial directory path from which to start the recursive file retrieval.
 349                        If None, it starts from the root of the configured SMB share.
 350        """
 351        
 352        def recurse_action(base_dir="", path=[]):
 353            if len(base_dir) == 0:
 354                remote_smb_path = ntpath.sep.join(path)
 355            else:
 356                remote_smb_path = base_dir + ntpath.sep + ntpath.sep.join(path)
 357            remote_smb_path = ntpath.normpath(remote_smb_path)
 358
 359            entries = self.smbClient.listPath(
 360                shareName=self.smb_share, 
 361                path=remote_smb_path + '\\*'
 362            )
 363            if len(entries) != 0:
 364                files = [entry for entry in entries if not entry.is_directory()]
 365                directories = [entry for entry in entries if entry.is_directory() and entry.get_longname() not in [".", ".."]]
 366
 367                # Files
 368                if len(files) != 0:
 369                    self.logger.print("[>] Retrieving files of '%s'" % remote_smb_path)
 370                for entry_file in files:
 371                    if not entry_file.is_directory():
 372                        f = LocalFileIO(
 373                            mode="wb",
 374                            path=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
 375                            expected_size=entry_file.get_filesize(),
 376                            keepRemotePath=True,
 377                            debug=self.config.debug
 378                        )
 379                        try:
 380                            self.smbClient.getFile(
 381                                shareName=self.smb_share, 
 382                                pathName=remote_smb_path + ntpath.sep + entry_file.get_longname(), 
 383                                callback=f.write
 384                            )
 385                            f.close()
 386                        except BrokenPipeError as err:
 387                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
 388                            f.close(remove=True)
 389                            break
 390                        except Exception as err:
 391                            f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err))
 392                            f.close(remove=True)
 393                
 394                # Directories
 395                for entry_directory in directories:
 396                    if entry_directory.is_directory():
 397                        recurse_action(
 398                            base_dir=self.smb_cwd, 
 399                            path=path+[entry_directory.get_longname()]
 400                        )                   
 401        # Entrypoint
 402        try:
 403            recurse_action(
 404                base_dir=self.smb_cwd, 
 405                path=[path]
 406            )
 407        except (BrokenPipeError, KeyboardInterrupt) as e:
 408            print("\x1b[v\x1b[o\r[!] Interrupted.")
 409            self.close_smb_session()
 410            self.init_smb_session()
 411
 412    def get_entry(self, path=None):
 413        """
 414        Retrieves information about a specific entry located at the provided path on the SMB share.
 415
 416        This method checks if the specified path exists on the SMB share. If the path exists, it retrieves the details of the entry at that path, including the directory name and file name. If the entry is found, it returns the entry object; otherwise, it returns None.
 417
 418        Args:
 419            path (str): The path of the entry to retrieve information about.
 420
 421        Returns:
 422            Entry: An object representing the entry at the specified path, or None if the entry is not found.
 423        """
 424
 425        if self.path_exists(path=path):
 426            matches = self.smbClient.listPath(
 427                shareName=self.smb_share,
 428                path=path
 429            )
 430
 431            if len(matches) == 1:
 432                return matches[0]
 433            else:
 434                return None
 435        else:
 436            return None 
 437
 438    def info(self, share=True, server=True):
 439        """
 440        Displays information about the server and optionally the shares.
 441
 442        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.
 443
 444        Parameters:
 445            share (bool): If True, display information about the current share.
 446            server (bool): If True, display information about the server.
 447
 448        Returns:
 449            None
 450        """
 451
 452        if server:
 453            if self.config.no_colors:
 454                self.logger.print("[+] Server:")
 455                self.logger.print("  ├─NetBIOS:")
 456                self.logger.print("  │ ├─ NetBIOS Hostname ──────── : %s" % (self.smbClient.getServerName()))
 457                self.logger.print("  │ └─ NetBIOS Domain ────────── : %s" % (self.smbClient.getServerDomain()))
 458                self.logger.print("  ├─DNS:")
 459                self.logger.print("  │ ├─ DNS Hostname ──────────── : %s" % (self.smbClient.getServerDNSHostName()))
 460                self.logger.print("  │ └─ DNS Domain ────────────── : %s" % (self.smbClient.getServerDNSDomainName()))
 461                self.logger.print("  ├─OS:")
 462                self.logger.print("  │ ├─ OS Name ───────────────── : %s" % (self.smbClient.getServerOS()))
 463                self.logger.print("  │ └─ OS Version ────────────── : %s.%s.%s" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild()))
 464                self.logger.print("  ├─Server:")
 465                self.logger.print("  │ ├─ Signing Required ──────── : %s" % (self.smbClient.isSigningRequired()))
 466                self.logger.print("  │ ├─ Login Required ────────── : %s" % (self.smbClient.isLoginRequired()))
 467                self.logger.print("  │ ├─ Supports NTLMv2 ───────── : %s" % (self.smbClient.doesSupportNTLMv2()))
 468                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
 469                self.logger.print("  │ ├─ Max size of read chunk ── : %d bytes (%s)" % (MaxReadSize, b_filesize(MaxReadSize)))
 470                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
 471                self.logger.print("  │ └─ Max size of write chunk ─ : %d bytes (%s)" % (MaxWriteSize, b_filesize(MaxWriteSize)))
 472                self.logger.print("  └─")
 473            else:
 474                self.logger.print("[+] Server:")
 475                self.logger.print("  ├─NetBIOS:")
 476                self.logger.print("  │ ├─ \x1b[94mNetBIOS Hostname\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerName()))
 477                self.logger.print("  │ └─ \x1b[94mNetBIOS Domain\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDomain()))
 478                self.logger.print("  ├─DNS:")
 479                self.logger.print("  │ ├─ \x1b[94mDNS Hostname\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSHostName()))
 480                self.logger.print("  │ └─ \x1b[94mDNS Domain\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSDomainName()))
 481                self.logger.print("  ├─OS:")
 482                self.logger.print("  │ ├─ \x1b[94mOS Name\x1b[0m \x1b[90m─────────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerOS()))
 483                self.logger.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()))
 484                self.logger.print("  ├─Server:")
 485                self.logger.print("  │ ├─ \x1b[94mSigning Required\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isSigningRequired()))
 486                self.logger.print("  │ ├─ \x1b[94mLogin Required\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isLoginRequired()))
 487                self.logger.print("  │ ├─ \x1b[94mSupports NTLMv2\x1b[0m \x1b[90m─────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.doesSupportNTLMv2()))
 488                MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"]
 489                self.logger.print("  │ ├─ \x1b[94mMax size of read chunk\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxReadSize, b_filesize(MaxReadSize)))
 490                MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"]
 491                self.logger.print("  │ └─ \x1b[94mMax size of write chunk\x1b[0m \x1b[90m─\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxWriteSize, b_filesize(MaxWriteSize)))
 492                self.logger.print("  └─")
 493
 494        if share and self.smb_share is not None:
 495            share_name = self.available_shares.get(self.smb_share.lower(), "")["name"]
 496            share_comment = self.available_shares.get(self.smb_share.lower(), "")["comment"]
 497            share_type = self.available_shares.get(self.smb_share.lower(), "")["type"]
 498            share_type =', '.join([s.replace("STYPE_","") for s in share_type])
 499            share_rawtype = self.available_shares.get(self.smb_share.lower(), "")["rawtype"]
 500            if self.config.no_colors:
 501                self.logger.print("\n[+] Share:")
 502                self.logger.print("  ├─ Name ──────────── : %s" % (share_name))
 503                self.logger.print("  ├─ Description ───── : %s" % (share_comment))
 504                self.logger.print("  ├─ Type ──────────── : %s" % (share_type))
 505                self.logger.print("  └─ Raw type value ── : %s" % (share_rawtype))
 506            else:
 507                self.logger.print("\n[+] Share:")
 508                self.logger.print("  ├─ \x1b[94mName\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_name))
 509                self.logger.print("  ├─ \x1b[94mDescription\x1b[0m \x1b[90m─────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_comment))
 510                self.logger.print("  ├─ \x1b[94mType\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_type))
 511                self.logger.print("  └─ \x1b[94mRaw type value\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%s\x1b[0m" % (share_rawtype))
 512
 513    def list_contents(self, path=None):
 514        """
 515        Lists the contents of a specified directory on the SMB share.
 516
 517        This method retrieves the contents of a directory specified by `shareName` and `path`. If `shareName` or `path`
 518        is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with
 519        the long names of the files and directories as keys and their respective SMB entry objects as values.
 520
 521        Args:
 522            shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None.
 523            path (str, optional): The directory path to list contents from. Defaults to the current path if None.
 524
 525        Returns:
 526            dict: A dictionary with file and directory names as keys and their SMB entry objects as values.
 527        """
 528        
 529        dest_path = [self.smb_cwd.rstrip(ntpath.sep),]
 530        if path is not None and len(path) > 0:
 531            dest_path.append(path.rstrip(ntpath.sep))
 532        dest_path.append('*')
 533        path = ntpath.sep.join(dest_path)
 534
 535        contents = {}
 536        entries = self.smbClient.listPath(
 537            shareName=self.smb_share, 
 538            path=path
 539        )
 540        for entry in entries:
 541            contents[entry.get_longname()] = entry
 542
 543        return contents
 544
 545    def list_shares(self):
 546        """
 547        Lists all the shares available on the connected SMB server.
 548
 549        This method queries the SMB server to retrieve a list of all available shares. It populates the `shares` dictionary
 550        with key-value pairs where the key is the share name and the value is a dictionary containing details about the share
 551        such as its name, type, raw type, and any comments associated with the share.
 552
 553        Returns:
 554            dict: A dictionary containing information about each share available on the server.
 555        """
 556
 557        self.available_shares = {}
 558
 559        if self.connected:
 560            if self.smbClient is not None:
 561                resp = self.smbClient.listShares()
 562
 563                for share in resp:
 564                    # SHARE_INFO_1 structure (lmshare.h)
 565                    # https://learn.microsoft.com/en-us/windows/win32/api/lmshare/ns-lmshare-share_info_1
 566                    sharename = share["shi1_netname"][:-1]
 567                    sharecomment = share["shi1_remark"][:-1]
 568                    sharetype = share["shi1_type"]
 569
 570                    self.available_shares[sharename.lower()] = {
 571                        "name": sharename, 
 572                        "type": STYPE_MASK(sharetype), 
 573                        "rawtype": sharetype, 
 574                        "comment": sharecomment
 575                    }
 576            else:
 577                self.logger.error("Error: SMBSession.smbClient is None.")
 578
 579        return self.available_shares
 580
 581    def mkdir(self, path=None):
 582        """
 583        Creates a directory at the specified path on the SMB share.
 584
 585        This method takes a path and attempts to create the directory structure on the SMB share. If the path includes
 586        nested directories, it will create each directory in the sequence. If a directory already exists, it will skip
 587        the creation for that directory without raising an error.
 588
 589        Args:
 590            path (str, optional): The full path of the directory to create on the SMB share. Defaults to None.
 591
 592        Note:
 593            The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility.
 594        """
 595
 596        if path is not None:
 597            # Prepare path
 598            path = path.replace('/',ntpath.sep)
 599            if ntpath.sep in path:
 600                path = path.strip(ntpath.sep).split(ntpath.sep)
 601            else:
 602                path = [path]
 603
 604            # Create each dir in the path
 605            for depth in range(1, len(path)+1):
 606                tmp_path = ntpath.sep.join(path[:depth])
 607                try:
 608                    self.smbClient.createDirectory(
 609                        shareName=self.smb_share, 
 610                        pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + tmp_path + ntpath.sep)
 611                    )
 612                except impacket.smbconnection.SessionError as err:
 613                    if err.getErrorCode() == 0xc0000035:
 614                        # STATUS_OBJECT_NAME_COLLISION
 615                        # Remote directory already created, this is normal
 616                        # Src: https://github.com/fortra/impacket/blob/269ce69872f0e8f2188a80addb0c39fedfa6dcb8/impacket/nt_errors.py#L268C9-L268C19
 617                        pass
 618                    else:
 619                        self.logger.error("Failed to create directory '%s': %s" % (tmp_path, err))
 620                        if self.config.debug:
 621                            traceback.print_exc()
 622        else:
 623            pass
 624
 625    def mount(self, local_mount_point, remote_path):
 626        """
 627        Generates the command to mount an SMB share on different platforms.
 628
 629        This method takes the local mount point and the remote path of the SMB share and generates the appropriate mount command based on the platform.
 630        It constructs the mount command using the provided parameters and executes it using the os.system() function.
 631
 632        Args:
 633            local_mount_point (str): The local directory where the SMB share will be mounted.
 634            remote_path (str): The remote path on the SMB share to be mounted.
 635
 636        Note:
 637            - For Windows platform, the command uses 'net use' to mount the share.
 638            - For Linux platform, the command uses 'mount' to mount the share.
 639            - For macOS platform, the command uses 'mount_smbfs' to mount the share.
 640            - If the platform is not supported, an error message is displayed.
 641
 642        Returns:
 643            None
 644        """
 645
 646        if not os.path.exists(local_mount_point):
 647            pass
 648
 649        if sys.platform.startswith('win'):
 650            remote_path = remote_path.replace('/',ntpath.sep)
 651            command = f"net use {local_mount_point} \\\\{self.host}\\{self.smb_share}\\{remote_path}"
 652        
 653        elif sys.platform.startswith('linux'):
 654            remote_path = remote_path.replace(ntpath.sep,'/')
 655            command = f"mount -t cifs //{self.host}/{self.smb_share}/{remote_path} {local_mount_point} -o username={self.credentials.username},password={self.credentials.password}"
 656        
 657        elif sys.platform.startswith('darwin'):
 658            remote_path = remote_path.replace(ntpath.sep,'/')
 659            command = f"mount_smbfs //{self.credentials.username}:{self.credentials.password}@{self.host}/{self.smb_share}/{remote_path} {local_mount_point}"
 660        
 661        else:
 662            command = None
 663            self.logger.error("Unsupported platform for mounting SMB share.")
 664        
 665        if command is not None:
 666            if self.config.debug:
 667                self.logger.debug("Executing: %s" % command)
 668            os.system(command)
 669
 670    def path_exists(self, path=None):
 671        """
 672        Checks if the specified path exists on the SMB share.
 673
 674        This method determines if a given path exists on the SMB share by attempting to list the contents of the path.
 675        If the path listing is successful and returns one or more entries, the path is considered to exist.
 676
 677        Args:
 678            path (str, optional): The path to check on the SMB share. Defaults to None.
 679
 680        Returns:
 681            bool: True if the path exists, False otherwise or if an error occurs.
 682        """
 683
 684        if path is not None:
 685            path = path.replace('*','')
 686            path = path.replace('/', ntpath.sep)
 687            try:
 688                contents = self.smbClient.listPath(
 689                    shareName=self.smb_share,
 690                    path=ntpath.normpath(self.smb_cwd + ntpath.sep + path + ntpath.sep)
 691                )
 692                return (len(contents) != 0)
 693            except Exception as e:
 694                return False
 695        else:
 696            return False
 697   
 698    def path_isdir(self, pathFromRoot=None):
 699        """
 700        Checks if the specified path is a directory on the SMB share.
 701
 702        This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the
 703        contents of the path and filtering for entries that match the basename of the path and are marked as directories.
 704
 705        Args:
 706            path (str, optional): The path to check on the SMB share. Defaults to None.
 707
 708        Returns:
 709            bool: True if the path is a directory, False otherwise or if an error occurs.
 710        """
 711
 712        if pathFromRoot is not None: 
 713            # Strip wildcards to avoid injections
 714            path = pathFromRoot.replace('*','')
 715            # Replace slashes if any
 716            path = path.replace('/', ntpath.sep)
 717
 718            # Normalize path and strip leading backslash
 719            path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep)
 720
 721            if path.strip() in ['', '.', '..']:
 722                # By defininition they exist on the filesystem
 723                return True
 724            else:
 725                try:
 726                    contents = self.smbClient.listPath(
 727                        shareName=self.smb_share,
 728                        path=path+'*'
 729                    )
 730                    # Filter on directories
 731                    contents = [
 732                        c for c in contents
 733                        if c.get_longname() == ntpath.basename(path) and c.is_directory()
 734                    ]
 735                    return (len(contents) != 0)
 736                except Exception as e:
 737                    return False
 738        else:
 739            return False
 740
 741    def path_isfile(self, pathFromRoot=None):
 742        """
 743        Checks if the specified path is a file on the SMB share.
 744
 745        This method determines if a given path corresponds to a file on the SMB share. It does this by listing the
 746        contents of the path and filtering for entries that match the basename of the path and are not marked as directories.
 747
 748        Args:
 749            path (str, optional): The path to check on the SMB share. Defaults to None.
 750
 751        Returns:
 752            bool: True if the path is a file, False otherwise or if an error occurs.
 753        """
 754
 755        if pathFromRoot is not None: 
 756            # Strip wildcards to avoid injections
 757            path = pathFromRoot.replace('*','')
 758            # Replace slashes if any
 759            path = path.replace('/', ntpath.sep)
 760
 761            # Normalize path and strip leading backslash
 762            path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep)
 763
 764            try:
 765                contents = self.smbClient.listPath(
 766                    shareName=self.smb_share,
 767                    path=ntpath.dirname(path) + ntpath.sep + '*'
 768                )
 769                # Filter on files
 770                contents = [
 771                    c for c in contents
 772                    if c.get_longname() == ntpath.basename(path) and not c.is_directory()
 773                ]
 774                return (len(contents) != 0)
 775            except Exception as e:
 776                return False
 777        else:
 778            return False
 779
 780    def put_file(self, localpath=None):
 781        """
 782        Uploads a single file to the SMB share.
 783
 784        This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path.
 785        It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session.
 786        General exceptions are caught and logged, with a traceback provided if debugging is enabled.
 787
 788        Args:
 789            localpath (str, optional): The local file path of the file to be uploaded. Defaults to None.
 790        """
 791
 792        # Parse path
 793        localpath = localpath.replace('/', os.path.sep)
 794        if os.path.sep in localpath:
 795            if localpath.startswith(os.path.sep):
 796                # Absolute path
 797                tmp_search_path = os.path.normpath(localpath)
 798            else:
 799                # Relative path
 800                tmp_search_path = os.path.normpath(os.getcwd() + os.path.sep + os.path.dirname(localpath))
 801        else:
 802            tmp_search_path = os.path.normpath(os.getcwd() + os.path.sep)
 803
 804        # Parse filename
 805        filename = os.path.basename(localpath)
 806
 807        # Search for the file
 808        matches = os.listdir(tmp_search_path)
 809        # Filter the entries
 810        matching_entries = []
 811        for entry in matches:
 812            if entry == filename:
 813                matching_entries.append(entry)
 814            elif '*' in filename:
 815                regexp = filename.replace('.', '\\.').replace('*', '.*')
 816                if re.match(regexp, entry):
 817                    matching_entries.append(entry)
 818
 819        matching_entries = sorted(list(set(matching_entries)))
 820
 821        # Loop and upload
 822        for localpath in matching_entries:
 823            if os.path.exists(localpath):
 824                if os.path.isfile(localpath):
 825                    try:
 826                        localfile = os.path.basename(localpath)
 827                        f = LocalFileIO(
 828                            mode="rb", 
 829                            path=localpath, 
 830                            debug=self.config.debug
 831                        )
 832                        self.smbClient.putFile(
 833                            shareName=self.smb_share, 
 834                            pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + localfile + ntpath.sep), 
 835                            callback=f.read
 836                        )
 837                        f.close()
 838
 839                    except (BrokenPipeError, KeyboardInterrupt) as err:
 840                        self.logger.error("Interrupted.")
 841                        self.close_smb_session()
 842                        self.init_smb_session()
 843
 844                    except (Exception, PermissionError) as err:
 845                        f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
 846                        f.close(remove=False)
 847                        if self.config.debug:
 848                            traceback.print_exc()
 849                else:
 850                    # [!] The specified localpath is a directory. Use 'put -r <directory>' instead.
 851                    pass
 852            else:
 853                # [!] The specified localpath does not exist.
 854                pass
 855
 856    def put_file_recursively(self, localpath=None):
 857        """
 858        Recursively uploads files from a specified local directory to the SMB share.
 859
 860        This method walks through the given local directory and all its subdirectories, uploading each file to the
 861        corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is,
 862        it iterates over all files and directories within the local path, creating necessary directories on the SMB share
 863        and uploading files. If the local path is not a directory, it prints an error message.
 864
 865        Args:
 866            localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None.
 867        """
 868
 869        if os.path.exists(localpath):
 870            if os.path.isdir(localpath):
 871                # Iterate over all files and directories within the local path
 872                local_files = {}
 873                for root, dirs, files in os.walk(localpath):
 874                    if len(files) != 0:
 875                        local_files[root] = files
 876
 877                # Iterate over the found files
 878                for local_dir_path in sorted(local_files.keys()):
 879                    self.logger.print("[>] Putting files of '%s'" % local_dir_path)
 880
 881                    # Create remote directory
 882                    remote_dir_path = local_dir_path.replace(os.path.sep, ntpath.sep)
 883                    self.mkdir(
 884                        path=ntpath.normpath(remote_dir_path + ntpath.sep)
 885                    )
 886
 887                    for local_file_path in local_files[local_dir_path]:
 888                        try:
 889                            f = LocalFileIO(
 890                                mode="rb", 
 891                                path=local_dir_path + os.path.sep + local_file_path, 
 892                                debug=self.config.debug
 893                            )
 894                            self.smbClient.putFile(
 895                                shareName=self.smb_share, 
 896                                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep + local_file_path), 
 897                                callback=f.read
 898                            )
 899                            f.close()
 900
 901                        except (BrokenPipeError, KeyboardInterrupt) as err:
 902                            self.logger.error("Interrupted.")
 903                            self.close_smb_session()
 904                            self.init_smb_session()
 905                            
 906                        except (Exception, PermissionError) as err:
 907                            f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err))
 908                            f.close(remove=False)
 909                            if self.config.debug:
 910                                traceback.print_exc()
 911                else:
 912                    self.logger.error("The specified localpath is a file. Use 'put <file>' instead.")
 913        else:
 914            self.logger.error("The specified localpath does not exist.")
 915
 916    def read_file(self, path=None):
 917        """
 918        Reads a file from the SMB share.
 919
 920        This method attempts to read the contents of a file specified by the `path` parameter from the SMB share.
 921        It constructs the full path to the file, checks if the path is a valid file, and then reads the file content
 922        into a byte stream which is returned to the caller.
 923
 924        Args:
 925            path (str, optional): The path of the file to be read from the SMB share. Defaults to None.
 926
 927        Returns:
 928            bytes: The content of the file as a byte stream, or None if the file does not exist or an error occurs.
 929        """
 930
 931        if self.path_isfile(pathFromRoot=path):
 932            path = path.replace('/', ntpath.sep)
 933            if path.startswith(ntpath.sep):
 934                # Absolute path
 935                tmp_file_path = ntpath.normpath(path)
 936            else:
 937                # Relative path
 938                tmp_file_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path)
 939            tmp_file_path = tmp_file_path.lstrip(ntpath.sep)
 940
 941            fh = io.BytesIO()
 942            try:
 943                # opening the files in streams instead of mounting shares allows 
 944                # for running the script from unprivileged containers
 945                self.smbClient.getFile(self.smb_share, tmp_file_path, fh.write)
 946            except impacket.smbconnection.SessionError as e:
 947                return None
 948            rawdata = fh.getvalue()
 949            fh.close()
 950            return rawdata
 951        else:
 952            return None
 953
 954    def rmdir(self, path=None):
 955        """
 956        Removes a directory from the SMB share at the specified path.
 957
 958        This method attempts to delete a directory located at the given path on the SMB share. If the operation fails,
 959        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
 960        the stack trace of the exception.
 961
 962        Args:
 963            path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None.
 964        """
 965        try:
 966            self.smbClient.deleteDirectory(
 967                shareName=self.smb_share, 
 968                pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 
 969            )
 970        except Exception as err:
 971            self.logger.error("Failed to remove directory '%s': %s" % (path, err))
 972            if self.config.debug:
 973                traceback.print_exc()
 974
 975    def rm(self, path=None):
 976        """
 977        Removes a file from the SMB share at the specified path.
 978
 979        This method attempts to delete a file located at the given path on the SMB share. If the operation fails,
 980        it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints
 981        the stack trace of the exception.
 982
 983        Args:
 984            path (str, optional): The path of the file to be removed on the SMB share. Defaults to None.
 985        """
 986
 987        # Parse path
 988        path = path.replace('/', ntpath.sep)
 989        if ntpath.sep in path:
 990            tmp_search_path = ntpath.normpath(self.smb_cwd + ntpath.sep + ntpath.dirname(path))
 991        else:
 992            tmp_search_path = ntpath.normpath(self.smb_cwd + ntpath.sep)
 993        # Parse filename
 994        filename = ntpath.basename(path)
 995
 996        # Search for the file
 997        matches = self.smbClient.listPath(
 998            shareName=self.smb_share, 
 999            path=tmp_search_path + ntpath.sep + '*'
1000        )   
1001
1002        # Filter the entries
1003        matching_entries = []
1004        for entry in matches:
1005            if entry.is_directory():
1006                # Skip directories
1007                continue
1008            if entry.get_longname() == filename:
1009                matching_entries.append(entry)
1010            elif '*' in filename:
1011                regexp = filename.replace('.', '\\.').replace('*', '.*')
1012                if re.match(regexp, entry.get_longname()):
1013                    matching_entries.append(entry)
1014        
1015        matching_entries = sorted(list(set(matching_entries)), key=lambda x: x.get_longname())
1016
1017        for entry in matching_entries:
1018            try:
1019                self.smbClient.deleteFile(
1020                    shareName=self.smb_share, 
1021                    pathName=ntpath.normpath(tmp_search_path + ntpath.sep + entry.get_longname()), 
1022                )
1023            except Exception as err:
1024                self.logger.error("Failed to remove file '%s': %s" % (path, err))
1025                if self.config.debug:
1026                    traceback.print_exc()
1027
1028    def tree(self, path=None):
1029        """
1030        Recursively lists the directory structure of the SMB share starting from the specified path.
1031
1032        This function prints a visual representation of the directory tree of the remote SMB share. It uses
1033        recursion to navigate through directories and lists all files and subdirectories in each directory.
1034        The output is color-coded and formatted to enhance readability, with directories highlighted in cyan.
1035
1036        Args:
1037            path (str, optional): The starting path on the SMB share from which to begin listing the tree.
1038                                  Defaults to the root of the current share.
1039        """
1040        
1041        def recurse_action(base_dir="", path=[], prompt=[]):
1042            bars = ["│   ", "├── ", "└── "]
1043
1044            remote_smb_path = ntpath.normpath(base_dir + ntpath.sep + ntpath.sep.join(path))
1045
1046            entries = []
1047            try:
1048                entries = self.smbClient.listPath(
1049                    shareName=self.smb_share, 
1050                    path=remote_smb_path+'\\*'
1051                )
1052            except impacket.smbconnection.SessionError as err:
1053                code, const, text = err.getErrorCode(), err.getErrorString()[0], err.getErrorString()[1]
1054                errmsg = "Error 0x%08x (%s): %s" % (code, const, text)
1055                if self.config.no_colors:
1056                    self.logger.print("%s%s" % (''.join(prompt+[bars[2]]), errmsg))
1057                else:
1058                    self.logger.print("%s\x1b[1;91m%s\x1b[0m" % (''.join(prompt+[bars[2]]), errmsg))
1059                return 
1060
1061            entries = [e for e in entries if e.get_longname() not in [".", ".."]]
1062            entries = sorted(entries, key=lambda x:x.get_longname())
1063
1064            # 
1065            if len(entries) > 1:
1066                index = 0
1067                for entry in entries:
1068                    index += 1
1069                    # This is the first entry 
1070                    if index == 0:
1071                        if entry.is_directory():
1072                            if self.config.no_colors:
1073                                self.logger.print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
1074                            else:
1075                                self.logger.print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
1076                            recurse_action(
1077                                base_dir=base_dir, 
1078                                path=path+[entry.get_longname()],
1079                                prompt=prompt+["│   "]
1080                            )
1081                        else:
1082                            if self.config.no_colors:
1083                                self.logger.print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
1084                            else:
1085                                self.logger.print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
1086
1087                    # This is the last entry
1088                    elif index == len(entries):
1089                        if entry.is_directory():
1090                            if self.config.no_colors:
1091                                self.logger.print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
1092                            else:
1093                                self.logger.print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
1094                            recurse_action(
1095                                base_dir=base_dir, 
1096                                path=path+[entry.get_longname()],
1097                                prompt=prompt+["    "]
1098                            )
1099                        else:
1100                            if self.config.no_colors:
1101                                self.logger.print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
1102                            else:
1103                                self.logger.print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
1104                        
1105                    # These are entries in the middle
1106                    else:
1107                        if entry.is_directory():
1108                            if self.config.no_colors:
1109                                self.logger.print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
1110                            else:
1111                                self.logger.print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname()))
1112                            recurse_action(
1113                                base_dir=base_dir, 
1114                                path=path+[entry.get_longname()],
1115                                prompt=prompt+["│   "]
1116                            )
1117                        else:
1118                            if self.config.no_colors:
1119                                self.logger.print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname()))
1120                            else:
1121                                self.logger.print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname()))
1122
1123            # 
1124            elif len(entries) == 1:
1125                entry = entries[0]
1126                if entry.is_directory():
1127                    if self.config.no_colors:
1128                        self.logger.print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
1129                    else:
1130                        self.logger.print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname()))
1131                    recurse_action(
1132                        base_dir=base_dir, 
1133                        path=path+[entry.get_longname()],
1134                        prompt=prompt+["    "]
1135                    )
1136                else:
1137                    if self.config.no_colors:
1138                        self.logger.print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname()))
1139                    else:
1140                        self.logger.print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname()))
1141
1142        # Entrypoint
1143        try:
1144            if self.config.no_colors:
1145                self.logger.print("%s\\" % path)
1146            else:
1147                self.logger.print("\x1b[1;96m%s\x1b[0m\\" % path)
1148            recurse_action(
1149                base_dir=self.smb_cwd, 
1150                path=[path],
1151                prompt=[""]
1152            )
1153        except (BrokenPipeError, KeyboardInterrupt) as e:
1154            self.logger.error("Interrupted.")
1155            self.close_smb_session()
1156            self.init_smb_session()
1157
1158    def umount(self, local_mount_point):
1159        """
1160        Unmounts the specified local mount point of the remote share.
1161
1162        This method unmounts the specified local mount point of the remote share based on the platform.
1163        It supports Windows, Linux, and macOS platforms for unmounting.
1164
1165        Parameters:
1166            local_mount_point (str): The local mount point to unmount.
1167
1168        Raises:
1169            None
1170        """
1171
1172        if os.path.exists(local_mount_point):
1173            if sys.platform.startswith('win'):
1174                command = f"net use {local_mount_point} /delete"
1175
1176            elif sys.platform.startswith('linux') or sys.platform.startswith('darwin'):
1177                command = f"umount {local_mount_point}"
1178
1179            else:
1180                command = None
1181                self.logger.error("Unsupported platform for unmounting SMB share.")
1182        
1183            if command is not None:
1184                self.logger.debug("Executing: %s" % command)
1185                os.system(command)
1186        else:
1187            self.logger.error("Cannot unmount a non existing path.")        
1188
1189    # Other functions
1190
1191    def test_rights(self, sharename): 
1192        """
1193        Tests the read and write access rights of the current SMB session.
1194
1195        This method checks the read and write access rights of the current SMB session by attempting to list paths and create/delete temporary directories.
1196        
1197        Returns:
1198            dict: A dictionary containing the read and write access rights status.
1199                - "readable" (bool): Indicates if the session has read access rights.
1200                - "writable" (bool): Indicates if the session has write access rights.
1201        """
1202
1203        # Restore the current share
1204        current_share = self.smb_share
1205        self.set_share(shareName=sharename)
1206
1207        access_rights = {"readable": False, "writable": False}
1208        try:
1209            self.smbClient.listPath(self.smb_share, '*', password=None)
1210            access_rights["readable"] = True
1211        except impacket.smbconnection.SessionError as e:
1212            access_rights["readable"] = False
1213
1214        try:
1215            temp_dir = ntpath.normpath("\\" + ''.join([random.choice("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPRSTUVWXYZ0123456759") for k in range(16)]))
1216            self.smbClient.createDirectory(self.smb_share, temp_dir)
1217            self.smbClient.deleteDirectory(self.smb_share, temp_dir)
1218            access_rights["writable"] = True
1219        except impacket.smbconnection.SessionError as e:
1220            access_rights["writable"] = False
1221
1222        # Restore the current share
1223        self.set_share(shareName=current_share)
1224
1225        return access_rights
1226
1227    # Setter / Getter
1228
1229    def set_share(self, shareName):
1230        """
1231        Sets the current SMB share to the specified share name.
1232
1233        This method updates the SMB session to use the specified share name. It checks if the share name is valid
1234        and updates the smb_share attribute of the SMBSession instance.
1235
1236        Parameters:
1237            shareName (str): The name of the share to set as the current SMB share.
1238
1239        Raises:
1240            ValueError: If the shareName is None or an empty string.
1241        """
1242
1243        if shareName is not None:
1244            self.list_shares()
1245            if shareName.lower() in self.available_shares.keys():
1246                # Doing this in order to keep the case of the share adevertised by the remote machine
1247                self.smb_share = self.available_shares[shareName.lower()]["name"]
1248                self.smb_cwd = ""
1249                # Connects the tree
1250                self.smb_tree_id = self.smbClient.connectTree(self.smb_share)
1251            else:
1252                self.logger.error("Could not set share '%s', it does not exist remotely." % shareName)
1253        else:
1254            self.smb_share = None
1255            
1256    def set_cwd(self, path=None):
1257        """
1258        Sets the current working directory on the SMB share to the specified path.
1259
1260        This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory.
1261        If the specified path is not a directory, the cwd remains unchanged.
1262
1263        Parameters:
1264            path (str): The path to set as the current working directory.
1265
1266        Raises:
1267            ValueError: If the specified path is not a directory.
1268        """
1269
1270        if path is not None:
1271            # Set path separators to ntpath sep 
1272            if '/' in path:
1273                path = path.replace('/', ntpath.sep)
1274
1275            if path.startswith(ntpath.sep):
1276                # Absolute path
1277                path = path + ntpath.sep
1278            else:
1279                # Relative path to the CWD
1280                if len(self.smb_cwd) == 0:
1281                    path = path + ntpath.sep
1282                else:
1283                    path = self.smb_cwd + ntpath.sep + path
1284            
1285            # Path normalization
1286            path = ntpath.normpath(path)
1287            path = re.sub(r'\\+', r'\\', path)
1288
1289            if path in ["", ".", ".."]:
1290                self.smb_cwd = ""
1291            else:
1292                if self.path_isdir(pathFromRoot=path.strip(ntpath.sep)):
1293                    # Path exists on the remote 
1294                    self.smb_cwd = ntpath.normpath(path)
1295                else:
1296                    # Path does not exists or is not a directory on the remote 
1297                    self.logger.error("Remote directory '%s' does not exist." % path)