smbclientng.core.InteractiveShell

   1#!/usr/bin/env python3
   2# -*- coding: utf-8 -*-
   3# File name          : InteractiveShell.py
   4# Author             : Podalirius (@podalirius_)
   5# Date created       : 23 may 2024
   6
   7
   8import charset_normalizer
   9import datetime
  10import impacket
  11from importlib import import_module
  12import ntpath
  13import os
  14import readline
  15import shutil
  16import shlex
  17import sys
  18import traceback
  19from rich.console import Console
  20from rich.table import Table
  21from rich.syntax import Syntax
  22from smbclientng.core.CommandCompleter import CommandCompleter
  23from smbclientng.core.utils import b_filesize, unix_permissions, windows_ls_entry, local_tree, resolve_local_files, resolve_remote_files
  24
  25
  26## Decorators
  27
  28def command_arguments_required(func):
  29    def wrapper(*args, **kwargs):
  30        self, arguments,command  = args[0], args[1], args[2]
  31        if len(arguments) != 0:
  32            return func(*args, **kwargs)
  33        else:
  34            self.commandCompleterObject.print_help(command=command)
  35            return None
  36    return wrapper
  37
  38def active_smb_connection_needed(func):
  39    def wrapper(*args, **kwargs):
  40        self, arguments,command  = args[0], args[1], args[2]
  41        #
  42        self.sessionsManager.current_session.ping_smb_session()
  43        if self.sessionsManager.current_session.connected:
  44            return func(*args, **kwargs)
  45        else:
  46            print("[!] SMB Session is disconnected.")
  47            return None
  48    return wrapper
  49
  50def smb_share_is_set(func):
  51    def wrapper(*args, **kwargs):
  52        self, arguments,command  = args[0], args[1], args[2]
  53        if self.sessionsManager.current_session.smb_share is not None:
  54            return func(*args, **kwargs)
  55        else:
  56            print("[!] You must open a share first, try the 'use <share>' command.")
  57            return None
  58    return wrapper
  59
  60
  61class InteractiveShell(object):
  62    """
  63    Class InteractiveShell is designed to manage the interactive command line interface for smbclient-ng.
  64    
  65    This class handles user input, executes commands, and manages the state of the SMB session. It provides
  66    a command line interface for users to interact with SMB shares, execute commands like directory listing,
  67    file transfer, and more.
  68
  69    Attributes:
  70        smbSession (SMBConnection): The active SMB connection session.
  71        debug (bool): Flag to enable or disable debug mode.
  72        smb_share (str): The current SMB share in use.
  73        smb_path (str): The current path within the SMB share.
  74        commandCompleterObject (CommandCompleter): Object to handle command completion and help generation.
  75
  76    Methods:
  77        __init__(self, smbSession, debug=False): Initializes the InteractiveShell with the given SMB session and debug mode.
  78        run(self): Starts the command line interface loop, processing user input until exit.
  79    """
  80
  81    running = True
  82    modules = {}
  83    
  84    def __init__(self, sessionsManager, config, logger):
  85        # Objects
  86        self.sessionsManager = sessionsManager
  87        self.config = config
  88        self.logger = logger
  89        # Internals
  90        self.commandCompleterObject = CommandCompleter(
  91            smbSession=self.sessionsManager.current_session,
  92            config=self.config,
  93            logger=self.logger,
  94        )
  95        readline.set_completer(self.commandCompleterObject.complete)
  96        readline.parse_and_bind("tab: complete")
  97        readline.set_completer_delims("\n")
  98        # Additional modules
  99        self.__load_modules()
 100
 101    def run(self):
 102        # Read commands from script file first
 103        if self.config.startup_script:
 104            f = open(self.config.startup_script, 'r')
 105            for line in f.readlines():
 106                try:
 107                    self.logger.print("%s%s" % (self.__prompt(), line.strip()))
 108                    readline.add_history(line.strip())
 109                    self.process_line(commandLine=line.strip())
 110                except KeyboardInterrupt as e:
 111                    self.logger.print()
 112
 113                except EOFError as e:
 114                    self.logger.print()
 115                    running = False
 116
 117                except Exception as err:
 118                    if self.config.debug:
 119                        traceback.print_exc()
 120                    self.logger.error(str(err))
 121
 122        # Then interactive console
 123        if not self.config.not_interactive:
 124            while self.running:
 125                try:
 126                    user_input = input(self.__prompt()).strip()
 127                    self.process_line(commandLine=user_input)
 128                except KeyboardInterrupt as e:
 129                    self.logger.print()
 130
 131                except EOFError as e:
 132                    self.logger.print()
 133                    running = False
 134
 135                except Exception as err:
 136                    if self.config.debug:
 137                        traceback.print_exc()
 138                    self.logger.error(str(err))
 139
 140    def process_line(self, commandLine):
 141        # Split and parse the commandLine
 142        tokens = shlex.split(commandLine)
 143        if len(tokens) == 0:
 144            command = ""
 145            arguments = []
 146        elif len(tokens) == 1:
 147            command = tokens[0].lower()
 148            arguments = []
 149        else:
 150            command = tokens[0].lower()
 151            arguments = tokens[1:]
 152        
 153        # Skip
 154        if command.strip() == "":
 155            pass
 156        # Execute the command
 157        elif command in self.commandCompleterObject.commands.keys():
 158
 159            # Exit the command line
 160            if command in ["exit", "quit"]:
 161                self.running = False
 162            
 163            # Display help
 164            elif command == "help":
 165                self.command_help(arguments, command)
 166
 167            # Cat the contents of a file
 168            elif command == "bat":
 169                self.command_bat(arguments, command)
 170
 171            # Cat the contents of a file
 172            elif command == "cat":
 173                self.command_cat(arguments, command)
 174
 175            # Closes the current SMB session
 176            elif command == "close":
 177                self.command_close(arguments, command)
 178                
 179            # Change directory in the current share
 180            elif command == "cd":
 181                self.command_cd(arguments, command)
 182            
 183            # debug
 184            elif command == "debug":
 185                self.command_debug(arguments, command)
 186
 187            # Get a file
 188            elif command == "get":
 189                self.command_get(arguments, command)
 190
 191            # SMB server info
 192            elif command == "info":
 193                self.command_info(arguments, command)
 194
 195            # List directory contents in a share
 196            elif command in ["ls", "dir"]:
 197                self.command_ls(arguments, command)
 198
 199            # Creates a new remote directory
 200            elif command == "mkdir":
 201                self.command_mkdir(arguments, command)
 202
 203            # Put a file
 204            elif command == "put":
 205                self.command_put(arguments, command)
 206
 207            # Shows the content of a local file
 208            elif command == "lcat":
 209                self.command_lcat(arguments, command)
 210
 211            # Changes the current local directory
 212            elif command == "lcd":
 213                self.command_lcd(arguments, command)
 214
 215            # Creates a copy of a local file
 216            elif command == "lcp":
 217                self.command_lcp(arguments, command)
 218
 219            # Pretty prints the content of a local file
 220            elif command == "lbat":
 221                self.command_lbat(arguments, command)
 222
 223            # Lists the contents of the current local directory
 224            elif command == "lls":
 225                self.command_lls(arguments, command)
 226
 227            # Creates a new local directory
 228            elif command == "lmkdir":
 229                self.command_lmkdir(arguments, command)
 230
 231            # Shows the current local directory
 232            elif command == "lpwd":
 233                self.command_lpwd(arguments, command)
 234
 235            # Renames a local file
 236            elif command == "lrename":
 237                self.command_lrename(arguments, command)
 238            
 239            # Removes a local file
 240            elif command == "lrm":
 241                self.command_lrm(arguments, command)
 242
 243            # Removes a local directory
 244            elif command == "lrmdir":
 245                self.command_lrmdir(arguments, command)
 246
 247            # Shows the current local directory
 248            elif command == "ltree":
 249                self.command_ltree(arguments, command)
 250
 251            # Modules
 252            elif command == "module":
 253                self.command_module(arguments, command)
 254
 255            # Creates a mount point of the remote share on the local machine
 256            elif command == "mount":
 257                self.command_mount(arguments, command)
 258
 259            # Reconnects the current SMB session
 260            elif command in ["connect", "reconnect"]:
 261                self.command_reconnect(arguments, command)
 262
 263            # Reset the TTY output
 264            elif command == "reset":
 265                self.command_reset(arguments, command)
 266
 267            # Removes a remote file
 268            elif command == "rm":
 269                self.command_rm(arguments, command)
 270                
 271            # Removes a remote directory
 272            elif command == "rmdir":
 273                self.command_rmdir(arguments, command)
 274
 275            # Sessions management
 276            elif command == "sessions":
 277                self.sessionsManager.process_command_line(arguments)
 278
 279            # List shares
 280            elif command == "sizeof":
 281                self.command_sizeof(arguments, command)
 282
 283            # List shares
 284            elif command == "shares":
 285                self.command_shares(arguments, command)
 286            
 287            # Displays a tree view of the CWD
 288            elif command == "tree":
 289                self.command_tree(arguments, command)
 290            
 291            # Use a share
 292            elif command == "use":
 293                self.command_use(arguments, command)
 294        
 295        # Fallback to unknown command   
 296        else:
 297            self.logger.print("Unknown command. Type \"help\" for help.")
 298
 299    # Commands ================================================================
 300
 301    def command_debug(self, arguments, command):
 302        try:
 303            self.logger.print("[debug] command    = '%s'" % command)
 304            self.logger.print("[debug] arguments  = %s" % arguments)
 305        except Exception as e:
 306            traceback.print_exc()
 307
 308    @command_arguments_required
 309    @active_smb_connection_needed
 310    @smb_share_is_set
 311    def command_bat(self, arguments, command):
 312        # Command arguments required   : Yes
 313        # Active SMB connection needed : Yes
 314        # SMB share needed             : Yes
 315
 316        # Parse wildcards
 317        files_and_directories = resolve_local_files(arguments)
 318
 319        for path_to_file in files_and_directories:
 320            if self.sessionsManager.current_session.path_isfile(pathFromRoot=path_to_file):
 321                # Read the file
 322                try:    
 323                    rawcontents = self.sessionsManager.current_session.read_file(path=path_to_file)
 324                    if rawcontents is not None:
 325                        encoding = charset_normalizer.detect(rawcontents)["encoding"]
 326                        if encoding is not None:
 327                            filecontent = rawcontents.decode(encoding).rstrip()
 328                            lexer = Syntax.guess_lexer(path=ntpath.basename(path_to_file), code=filecontent)
 329                            # Some trickery for the files undetected by the lexer
 330                            if lexer == "default":
 331                                if '<?xml' in filecontent:
 332                                    lexer = "xml"
 333                                elif '<html>' in filecontent:
 334                                    lexer = "html"
 335                            syntax = Syntax(code=filecontent, line_numbers=True, lexer=lexer)
 336                            if len(files_and_directories) > 1:
 337                                self.logger.print("\x1b[1;93m[>] %s\x1b[0m" % (path_to_file+' ').ljust(80,'='))
 338                            Console().print(syntax)
 339                        else:
 340                            self.logger.error("[!] Could not detect charset of '%s'." % path_to_file)
 341                except impacket.smbconnection.SessionError as e:
 342                    self.logger.error("[!] SMB Error: %s" % e)
 343
 344    @command_arguments_required
 345    @active_smb_connection_needed
 346    @smb_share_is_set
 347    def command_cd(self, arguments, command):
 348        # Command arguments required   : Yes
 349        # Active SMB connection needed : Yes
 350        # SMB share needed             : Yes
 351
 352        try:
 353            self.sessionsManager.current_session.set_cwd(path=arguments[0])
 354        except impacket.smbconnection.SessionError as e:
 355            self.logger.error("[!] SMB Error: %s" % e)
 356
 357    @command_arguments_required
 358    @active_smb_connection_needed
 359    @smb_share_is_set
 360    def command_cat(self, arguments, command):
 361        # Command arguments required   : Yes
 362        # Active SMB connection needed : Yes
 363        # SMB share needed             : Yes
 364
 365        # Parse wildcards
 366        files_and_directories = resolve_local_files(arguments)
 367
 368        for path_to_file in files_and_directories:
 369            if self.sessionsManager.current_session.path_isfile(pathFromRoot=path_to_file):
 370                # Read the file
 371                try:
 372                    rawcontents = self.sessionsManager.current_session.read_file(path=path_to_file)
 373                    if rawcontents is not None:
 374                        encoding = charset_normalizer.detect(rawcontents)["encoding"]
 375                        if encoding is not None:
 376                            filecontent = rawcontents.decode(encoding).rstrip()
 377                            if len(files_and_directories) > 1:
 378                                self.logger.print("\x1b[1;93m[>] %s\x1b[0m" % (path_to_file+' ').ljust(80,'='))
 379                            self.logger.print(filecontent)
 380                        else:
 381                            self.logger.error("[!] Could not detect charset of '%s'." % path_to_file)
 382                except impacket.smbconnection.SessionError as e:
 383                    self.logger.error("[!] SMB Error: %s" % e)
 384
 385    def command_close(self, arguments, command):
 386        # Command arguments required   : No
 387        # Active SMB connection needed : No
 388        # SMB share needed             : No
 389
 390        self.sessionsManager.current_session.ping_smb_session()
 391        if self.sessionsManager.current_session.connected:
 392            self.sessionsManager.current_session.close_smb_session()
 393
 394    @command_arguments_required
 395    @active_smb_connection_needed
 396    @smb_share_is_set
 397    def command_get(self, arguments, command):
 398        # Command arguments required   : Yes
 399        # Active SMB connection needed : Yes
 400        # SMB share needed             : Yes
 401
 402        is_recursive = False
 403        while '-r' in arguments:
 404            is_recursive = True
 405            arguments.remove('-r')
 406
 407        # Parse wildcards
 408        files_and_directories = resolve_remote_files(self.sessionsManager.current_session, arguments)
 409
 410        # 
 411        for remotepath in files_and_directories:
 412            try:
 413                if is_recursive and self.sessionsManager.current_session.path_isdir(remotepath):
 414                    # Get files recursively
 415                    self.sessionsManager.current_session.get_file_recursively(path=remotepath)
 416                else:
 417                    # Get this single file
 418                    self.sessionsManager.current_session.get_file(path=remotepath)
 419            except impacket.smbconnection.SessionError as e:
 420                self.logger.error("[!] SMB Error: %s" % e)
 421
 422    def command_help(self, arguments, command):
 423        # Command arguments required   : No
 424        # Active SMB connection needed : No
 425        # SMB share needed             : No
 426
 427        if len(arguments) != 0:
 428            self.commandCompleterObject.print_help(command=arguments[0])
 429        else:
 430            self.commandCompleterObject.print_help(command=None)
 431
 432    @active_smb_connection_needed
 433    def command_info(self, arguments, command):
 434        # Command arguments required   : No
 435        # Active SMB connection needed : Yes
 436        # SMB share needed             : No
 437
 438        print_server_info = False
 439        print_share_info = False
 440        if len(arguments) != 0:
 441            print_server_info = (arguments[0].lower() == "server")
 442            print_share_info = (arguments[0].lower() == "share")
 443        else:
 444            print_server_info = True
 445            print_share_info = True
 446
 447        try:
 448            self.sessionsManager.current_session.info(
 449                share=print_share_info,
 450                server=print_server_info
 451            )
 452        except impacket.smbconnection.SessionError as e:
 453            self.logger.error("[!] SMB Error: %s" % e)
 454
 455    @command_arguments_required
 456    def command_lbat(self, arguments, command):
 457        # Command arguments required   : Yes
 458        # Active SMB connection needed : No
 459        # SMB share needed             : No
 460
 461        # Parse wildcards
 462        files_and_directories = resolve_remote_files(self.sessionsManager.current_session, arguments)
 463
 464        for path_to_file in files_and_directories:
 465            # Read the file
 466            try:
 467                if os.path.exists(path=path_to_file):
 468                    f = open(path_to_file, 'rb')
 469                    rawcontents = f.read()
 470                    #
 471                    if rawcontents is not None:
 472                        encoding = charset_normalizer.detect(rawcontents)["encoding"]
 473                        if encoding is not None:
 474                            filecontent = rawcontents.decode(encoding).rstrip()
 475                            lexer = Syntax.guess_lexer(path=ntpath.basename(path_to_file), code=filecontent)
 476                            # Some trickery for the files undetected by the lexer
 477                            if lexer == "default":
 478                                if '<?xml' in filecontent:
 479                                    lexer = "xml"
 480                                elif '<html>' in filecontent:
 481                                    lexer = "html"
 482                            syntax = Syntax(code=filecontent, line_numbers=True, lexer=lexer)
 483                            if len(files_and_directories) > 1:
 484                                self.logger.print("\x1b[1;93m[>] %s\x1b[0m" % (path_to_file+' ').ljust(80,'='))
 485                            Console().print(syntax)
 486                        else:
 487                            self.logger.error("[!] Could not detect charset of '%s'." % path_to_file)
 488                else:
 489                    self.logger.error("[!] Local file '%s' does not exist." % path_to_file)
 490            except impacket.smbconnection.SessionError as e:
 491                self.logger.error("[!] SMB Error: %s" % e)
 492
 493    @command_arguments_required
 494    def command_lcat(self, arguments, command):
 495        # Command arguments required   : Yes
 496        # Active SMB connection needed : No
 497        # SMB share needed             : No
 498
 499        # Parse wildcards
 500        files_and_directories = resolve_remote_files(self.sessionsManager.current_session, arguments)
 501
 502        for path_to_file in files_and_directories:
 503            # Read the file 
 504            try:
 505                if os.path.exists(path=path_to_file):
 506                    f = open(path_to_file, 'rb')
 507                    rawcontents = f.read()
 508                    #
 509                    if rawcontents is not None:
 510                        encoding = charset_normalizer.detect(rawcontents)["encoding"]
 511                        if encoding is not None:
 512                            filecontent = rawcontents.decode(encoding).rstrip()
 513                            if len(files_and_directories) > 1:
 514                                self.logger.print("\x1b[1;93m[>] %s\x1b[0m" % (path_to_file+' ').ljust(80,'='))
 515                            self.logger.print(filecontent)
 516                        else:
 517                            self.logger.error("[!] Could not detect charset of '%s'." % path_to_file)
 518                else:
 519                    self.logger.error("[!] Local file '%s' does not exist." % path_to_file)
 520            except impacket.smbconnection.SessionError as e:
 521                self.logger.error("[!] SMB Error: %s" % e)
 522
 523    @command_arguments_required
 524    def command_lcd(self, arguments, command):
 525        # Command arguments required   : Yes
 526        # Active SMB connection needed : No
 527        # SMB share needed             : No
 528        
 529        path = arguments[0]
 530
 531        if os.path.exists(path=path):
 532            if os.path.isdir(s=path):
 533                os.chdir(path=path)
 534            else:
 535                self.logger.error("Path '%s' is not a directory." % path)
 536        else:
 537            self.logger.error("Directory '%s' does not exists." % path)
 538
 539    @command_arguments_required
 540    def command_lcp(self, arguments, command):
 541        # Command arguments required   : Yes
 542        # Active SMB connection needed : No
 543        # SMB share needed             : No
 544
 545        if len(arguments) == 2:
 546            src_path = arguments[0]
 547            dst_path = arguments[1]
 548            if os.path.exists(path=src_path):
 549                try:
 550                    shutil.copyfile(src=src_path, dst=dst_path)
 551                except shutil.SameFileError as err:
 552                    self.logger.error("[!] Error: %s" % err)
 553            else:
 554                self.logger.error("[!] File '%s' does not exists." % src_path)
 555        else:
 556            self.commandCompleterObject.print_help(command=command)
 557
 558    def command_lls(self, arguments, command):
 559        # Command arguments required   : No
 560        # Active SMB connection needed : No
 561        # SMB share needed             : No
 562
 563        if len(arguments) == 0:
 564            arguments = ['.']
 565        else:
 566            arguments = resolve_local_files(arguments)
 567
 568        for path in arguments:
 569            if len(arguments) > 1:
 570                self.logger.print("%s:" % path)
 571            # lls <directory>
 572            if os.path.isdir(path):
 573                directory_contents = os.listdir(path=path)
 574                for entryname in sorted(directory_contents):
 575                    path_to_file = path + os.path.sep + entryname
 576                    rights_str = unix_permissions(path_to_file)
 577                    size_str = b_filesize(os.path.getsize(filename=path_to_file))
 578                    date_str = datetime.datetime.fromtimestamp(os.path.getmtime(filename=path_to_file)).strftime("%Y-%m-%d %H:%M")
 579
 580                    if os.path.isdir(s=entryname):
 581                        if self.config.no_colors:
 582                            self.logger.print("%s %10s  %s  %s%s" % (rights_str, size_str, date_str, entryname, os.path.sep))
 583                        else:
 584                            self.logger.print("%s %10s  %s  \x1b[1;96m%s\x1b[0m%s" % (rights_str, size_str, date_str, entryname, os.path.sep))
 585                    else:
 586                        if self.config.no_colors:
 587                            self.logger.print("%s %10s  %s  %s" % (rights_str, size_str, date_str, entryname))
 588                        else:
 589                            self.logger.print("%s %10s  %s  \x1b[1m%s\x1b[0m" % (rights_str, size_str, date_str, entryname))
 590            # lls <file>
 591            elif os.path.isfile(path):
 592                rights_str = unix_permissions(path)
 593                size_str = b_filesize(os.path.getsize(filename=path))
 594                date_str = datetime.datetime.fromtimestamp(os.path.getmtime(filename=path)).strftime("%Y-%m-%d %H:%M")
 595                if self.config.no_colors:
 596                    self.logger.print("%s %10s  %s  %s" % (rights_str, size_str, date_str, os.path.basename(path)))
 597                else:
 598                    self.logger.print("%s %10s  %s  \x1b[1m%s\x1b[0m" % (rights_str, size_str, date_str, os.path.basename(path))) 
 599            
 600            if len(arguments) > 1:
 601                self.logger.print()
 602
 603    @command_arguments_required
 604    def command_lmkdir(self, arguments, command):
 605        # Command arguments required   : Yes
 606        # Active SMB connection needed : No
 607        # SMB share needed             : No
 608
 609        for path in arguments:
 610            if os.path.sep in path:
 611                path = path.strip(os.path.sep).split(os.path.sep)
 612            else:
 613                path = [path]
 614
 615            # Create each dir in the path
 616            for depth in range(1, len(path)+1):
 617                tmp_path = os.path.sep.join(path[:depth])
 618                if not os.path.exists(tmp_path):
 619                    os.mkdir(path=tmp_path)
 620
 621    def command_lpwd(self, arguments, command):
 622        # Command arguments required   : No
 623        # Active SMB connection needed : No
 624        # SMB share needed             : No
 625
 626        self.logger.print(os.getcwd())
 627
 628    @command_arguments_required
 629    def command_lrename(self, arguments, command):
 630        # Command arguments required   : Yes
 631        # Active SMB connection needed : No
 632        # SMB share needed             : No
 633
 634        if len(arguments) == 2:
 635            os.rename(src=arguments[0], dst=arguments[1])
 636        else:
 637            self.commandCompleterObject.print_help(command=command)
 638
 639    @command_arguments_required
 640    def command_lrm(self, arguments, command):
 641        # Command arguments required   : Yes
 642        # Active SMB connection needed : No
 643        # SMB share needed             : No
 644
 645        path = arguments[0]
 646
 647        if os.path.exists(path):
 648            if not os.path.isdir(s=path):
 649                try:
 650                    os.remove(path=path)
 651                except Exception as e:
 652                    self.logger.error("Error removing file '%s' : %s" % path)
 653            else:
 654                self.logger.error("Cannot delete '%s'. It is a directory, use 'lrmdir <directory>' instead." % path)
 655        else:
 656            self.logger.error("Path '%s' does not exist." % path)
 657
 658    @command_arguments_required
 659    def command_lrmdir(self, arguments, command):
 660        # Command arguments required   : Yes
 661        # Active SMB connection needed : No
 662        # SMB share needed             : No
 663
 664        if len(arguments) == 0:
 665            path = '.'
 666        else:
 667            path = arguments[0]
 668
 669        if os.path.exists(path):
 670            if os.path.isdir(s=path):
 671                try:
 672                    shutil.rmtree(path=path)
 673                except Exception as e:
 674                    self.logger.error("Error removing directory '%s' : %s" % path)
 675            else:
 676                self.logger.error("Cannot delete '%s'. It is a file, use 'lrm <file>' instead." % path)
 677        else:
 678            self.logger.error("Path '%s' does not exist." % path)
 679
 680    def command_ltree(self, arguments, command):
 681        # Command arguments required   : No
 682        # Active SMB connection needed : No
 683        # SMB share needed             : No
 684
 685        if len(arguments) == 0:
 686            path = '.'
 687        else:
 688            path = arguments[0]
 689
 690        if len(arguments) == 0:
 691            local_tree(path='.', config=self.config)
 692        else:
 693            local_tree(path=path, config=self.config)
 694
 695    @active_smb_connection_needed
 696    @smb_share_is_set
 697    def command_ls(self, arguments, command):
 698        # Command arguments required   : No
 699        # Active SMB connection needed : Yes
 700        # SMB share needed             : Yes
 701
 702        if len(arguments) == 0:
 703            arguments = ['.']
 704        else:
 705            arguments = resolve_remote_files(self.sessionsManager.current_session, arguments)
 706
 707        for path in arguments:
 708            if len(arguments) > 1:
 709                self.logger.print("%s:" % path)
 710
 711            if self.sessionsManager.current_session.path_isdir(pathFromRoot=path):
 712                # Read the files
 713                directory_contents = self.sessionsManager.current_session.list_contents(path=path)
 714            else:
 715                entry = self.sessionsManager.current_session.get_entry(path=path)
 716                if entry is not None:
 717                    directory_contents = {entry.get_longname(): entry}
 718                else:
 719                    directory_contents = {}
 720
 721            for longname in sorted(directory_contents.keys(), key=lambda x:x.lower()):
 722                windows_ls_entry(directory_contents[longname], self.config)
 723
 724            if len(arguments) > 1:
 725                self.logger.print()
 726            
 727    @command_arguments_required
 728    @active_smb_connection_needed
 729    @smb_share_is_set
 730    def command_mkdir(self, arguments, command):
 731        # Command arguments required   : Yes
 732        # Active SMB connection needed : Yes
 733        # SMB share needed             : Yes
 734
 735        self.sessionsManager.current_session.mkdir(path=arguments[0])
 736
 737    @command_arguments_required
 738    @active_smb_connection_needed
 739    @smb_share_is_set
 740    def command_module(self, arguments, command):
 741        module_name = arguments[0]
 742
 743        if module_name in self.modules.keys():
 744            module = self.modules[module_name](self.sessionsManager.current_session, self.config, self.logger)
 745            arguments_string = ' '.join(arguments[1:])
 746            module.run(arguments_string)
 747        else:
 748            self.logger.error("Module '%s' does not exist." % module_name)
 749
 750    @command_arguments_required
 751    @active_smb_connection_needed
 752    @smb_share_is_set
 753    def command_mount(self, arguments, command):
 754        # Command arguments required   : Yes
 755        # Active SMB connection needed : Yes
 756        # SMB share needed             : Yes
 757
 758        if len(arguments) == 2:
 759            remote_path = arguments[0]
 760            if not remote_path.startswith(ntpath.sep):
 761                remote_path = self.sessionsManager.current_session.smb_cwd + ntpath.sep + remote_path
 762
 763            local_mount_point = arguments[1]
 764
 765            self.logger.debug("Trying to mount remote '%s' onto local '%s'" % (remote_path, local_mount_point))
 766
 767            try:
 768                self.sessionsManager.current_session.mount(local_mount_point, remote_path)
 769            except (impacket.smbconnection.SessionError, impacket.smb3.SessionError) as e:
 770                self.sessionsManager.current_session.umount(local_mount_point)
 771        else:
 772            self.commandCompleterObject.print_help(command=command)
 773
 774    @command_arguments_required
 775    @active_smb_connection_needed
 776    @smb_share_is_set
 777    def command_put(self, arguments, command):
 778        # Command arguments required   : Yes
 779        # Active SMB connection needed : Yes
 780        # SMB share needed             : Yes
 781        
 782        is_recursive = False
 783        while '-r' in arguments:
 784            is_recursive = True
 785            arguments.remove('-r')
 786
 787        # Parse wildcards
 788        files_and_directories = resolve_local_files(arguments)
 789
 790        # 
 791        for localpath in files_and_directories:
 792            try:
 793                self.logger.print(localpath)
 794                if is_recursive and os.path.isdir(s=localpath):
 795                    # Put files recursively
 796                    self.sessionsManager.current_session.put_file_recursively(localpath=localpath)
 797                else:
 798                    # Put this single file
 799                    self.sessionsManager.current_session.put_file(localpath=localpath)
 800            except impacket.smbconnection.SessionError as e:
 801                self.logger.error("[!] SMB Error: %s" % e)
 802
 803    def command_reconnect(self, arguments, command):
 804        # Command arguments required   : No
 805        # Active SMB connection needed : No
 806        # SMB share needed             : No
 807
 808        self.sessionsManager.current_session.ping_smb_session()
 809        if self.sessionsManager.current_session.connected:
 810            self.sessionsManager.current_session.close_smb_session()
 811            self.sessionsManager.current_session.init_smb_session()
 812        else:
 813            self.sessionsManager.current_session.init_smb_session()
 814
 815    def command_reset(self, arguments, command):
 816        # Command arguments required   : No
 817        # Active SMB connection needed : No
 818        # SMB share needed             : No
 819        sys.stdout.write('\x1b[?25h') # Sets the cursor to on
 820        sys.stdout.write('\x1b[v')  
 821        sys.stdout.write('\x1b[o') # Reset
 822        sys.stdout.flush()
 823
 824    @command_arguments_required
 825    @active_smb_connection_needed
 826    @smb_share_is_set
 827    def command_rm(self, arguments, command):
 828        # Command arguments required   : Yes
 829        # Active SMB connection needed : Yes
 830        # SMB share needed             : Yes
 831
 832        for path_to_file in arguments:
 833            # Wildcard
 834            if '*' in path_to_file:
 835                self.sessionsManager.current_session.rm(path=path_to_file)
 836            # File
 837            elif self.sessionsManager.current_session.path_exists(path_to_file):
 838                if self.sessionsManager.current_session.path_isfile(path_to_file):
 839                    try:
 840                        self.sessionsManager.current_session.rm(path=path_to_file)
 841                    except Exception as e:
 842                        self.logger.error("Error removing file '%s' : %s" % path_to_file)
 843                else:
 844                    self.logger.error("Cannot delete '%s': This is a directory, use 'rmdir <directory>' instead." % path_to_file)
 845            # File does not exist
 846            else:
 847                self.logger.error("Remote file '%s' does not exist." % path_to_file)
 848
 849    @command_arguments_required
 850    @active_smb_connection_needed
 851    @smb_share_is_set
 852    def command_rmdir(self, arguments, command):
 853        # Command arguments required   : Yes
 854        # Active SMB connection needed : Yes
 855        # SMB share needed             : Yes
 856
 857        for path_to_directory in arguments:
 858            if self.sessionsManager.current_session.path_exists(path_to_directory):
 859                if self.sessionsManager.current_session.path_isdir(path_to_directory):
 860                    try:
 861                        self.sessionsManager.current_session.rmdir(path=path_to_directory)
 862                    except Exception as e:
 863                        self.logger.error("Error removing directory '%s' : %s" % path_to_directory)
 864                else:
 865                    self.logger.error("Cannot delete '%s': This is a file, use 'rm <file>' instead." % path_to_directory)
 866            else:
 867                self.logger.error("Remote directory '%s' does not exist." % path_to_directory)
 868
 869    @active_smb_connection_needed
 870    @smb_share_is_set
 871    def command_sizeof(self, arguments, command):
 872        # Command arguments required   : Yes
 873        # Active SMB connection needed : Yes
 874        # SMB share needed             : Yes
 875
 876        class RecursiveSizeOfPrint(object):
 877            def __init__(self, entry, sessionsManager, config, logger):
 878                self.entry = entry
 879                self.config = config
 880                self.logger = logger
 881                self.sessionsManager = sessionsManager
 882                self.size = 0
 883            
 884            def update(self, entry, fullpath, depth):
 885                self.size += entry.get_filesize()
 886                self.print(end='\r')
 887            
 888            def print(self, end='\n'):
 889                # Directory
 890                if self.entry.is_directory():
 891                    if self.config.no_colors:
 892                        path = "%s\\" % self.entry.get_longname()
 893                    else:
 894                        path = "\x1b[1;96m%s\x1b[0m\\" % self.entry.get_longname()
 895                # File
 896                else:
 897                    if self.config.no_colors:
 898                        path = "%s" % self.entry.get_longname()
 899                    else:
 900                        path = "\x1b[1m%s\x1b[0m" % self.entry.get_longname()
 901                self.logger.print("%10s  %s" % (b_filesize(self.size), path), end=end)
 902
 903        entries = []
 904        if len(arguments) == 0:
 905            entries = self.sessionsManager.current_session.list_contents()
 906            entries = [entry for name, entry in entries.items() if name not in ['.', '..']]
 907        else:
 908            entry = self.sessionsManager.current_session.get_entry(path=' '.join(arguments))
 909            entries = []
 910            if entry is not None:
 911                entries = [entry]
 912            else:
 913                self.logger.print("[!] Path '%s' does not exist." % ' '.join(arguments))
 914
 915        total = 0
 916        for entry in entries:
 917            rsop = RecursiveSizeOfPrint(
 918                entry=entry, 
 919                sessionsManager=self.sessionsManager, 
 920                config=self.config,
 921                logger=self.logger
 922            )
 923            # Directory
 924            if entry.is_directory():
 925                self.sessionsManager.current_session.find(
 926                    paths=[entry.get_longname()],
 927                    callback=rsop.update
 928                )
 929            # File
 930            else:
 931                rsop.update(entry=entry, fullpath=entry.get_longname(), depth=0)
 932            # Close the print
 933            rsop.print()
 934            total += rsop.size
 935        
 936        if len(entries) > 1:
 937            self.logger.print("──────────────────────")
 938            self.logger.print("     total  %s" % b_filesize(total))
 939
 940    @active_smb_connection_needed
 941    def command_shares(self, arguments, command):
 942        # Command arguments required   : No
 943        # Active SMB connection needed : Yes
 944        # SMB share needed             : No
 945
 946        do_check_rights = False
 947        if len(arguments) != 0:
 948            if arguments[0] == "rights":
 949                do_check_rights = True
 950
 951        shares = self.sessionsManager.current_session.list_shares()
 952        if len(shares.keys()) != 0:
 953            table = Table(title=None)
 954            table.add_column("Share")
 955            table.add_column("Visibility")
 956            table.add_column("Type")
 957            table.add_column("Description", justify="left")
 958            if do_check_rights:
 959                table.add_column("Rights")
 960
 961            for sharename in sorted(shares.keys()):
 962                types = ', '.join([s.replace("STYPE_","") for s in shares[sharename]["type"]])
 963
 964                is_hidden = bool(sharename.endswith('$'))
 965                if is_hidden:
 966                    str_hidden = "[bold bright_blue]Hidden[/bold bright_blue]"
 967                    str_sharename = "[bold bright_blue]" + shares[sharename]["name"] + "[/bold bright_blue]"
 968                    str_types = "[bold bright_blue]" + types + "[/bold bright_blue]"
 969                    str_comment = "[bold bright_blue]" + shares[sharename]["comment"] + "[/bold bright_blue]"
 970                else:
 971                    str_hidden = "[bold bright_yellow]Visible[/bold bright_yellow]"
 972                    str_sharename = "[bold bright_yellow]" + shares[sharename]["name"] + "[/bold bright_yellow]"
 973                    str_types = "[bold bright_yellow]" + types + "[/bold bright_yellow]"
 974                    str_comment = "[bold bright_yellow]" + shares[sharename]["comment"] + "[/bold bright_yellow]"
 975
 976                if do_check_rights:
 977                    access_rights = self.sessionsManager.current_session.test_rights(sharename=shares[sharename]["name"])
 978                    str_access_rights = "[bold yellow]NO ACCESS[/bold yellow]"
 979                    if access_rights["readable"] and access_rights["writable"]:
 980                        str_access_rights = "[bold green]READ[/bold green], [bold red]WRITE[/bold red]"
 981                    elif access_rights["readable"]:
 982                        str_access_rights = "[bold green]READ[/bold green]"
 983                    elif access_rights["writable"]:
 984                        # Without READ?? This should not happen IMHO
 985                        str_access_rights = "[bold red]WRITE[/bold red]"
 986                    else:
 987                        str_access_rights = "[bold yellow]NO ACCESS[/bold yellow]"
 988
 989                if do_check_rights:
 990                    table.add_row(str_sharename, str_hidden, str_types, str_comment, str_access_rights)
 991                else:
 992                    table.add_row(str_sharename, str_hidden, str_types, str_comment)
 993
 994            Console().print(table)
 995        else:
 996            self.logger.error("No share served on '%s'" % self.sessionsManager.current_session.host)
 997
 998    @active_smb_connection_needed
 999    @smb_share_is_set
1000    def command_tree(self, arguments, command):
1001        # Command arguments required   : No
1002        # Active SMB connection needed : Yes
1003        # SMB share needed             : Yes
1004
1005        if len(arguments) == 0:
1006            self.sessionsManager.current_session.tree(path='.')
1007        else:
1008            self.sessionsManager.current_session.tree(path=arguments[0])
1009
1010    @command_arguments_required
1011    @active_smb_connection_needed
1012    @smb_share_is_set
1013    def command_umount(self, arguments, command):
1014        # Command arguments required   : Yes
1015        # Active SMB connection needed : Yes
1016        # SMB share needed             : Yes
1017
1018        local_mount_point = arguments[0]
1019
1020        self.logger.debug("Trying to unmount local mount point '%s'" % (local_mount_point))
1021        
1022        self.sessionsManager.current_session.umount(local_mount_point)
1023        
1024    @command_arguments_required
1025    @active_smb_connection_needed
1026    def command_use(self, arguments, command):
1027        # Command arguments required   : Yes
1028        # Active SMB connection needed : Yes
1029        # SMB share needed             : No
1030
1031        sharename = arguments[0]
1032
1033        # Reload the list of shares
1034        shares = self.sessionsManager.current_session.list_shares()
1035        shares = [s.lower() for s in shares.keys()]
1036
1037        if sharename.lower() in shares:
1038            self.sessionsManager.current_session.set_share(sharename)
1039        else:
1040            self.logger.error("No share named '%s' on '%s'" % (sharename, self.sessionsManager.current_session.host))
1041
1042    # Private functions =======================================================
1043
1044    def __load_modules(self):
1045        """
1046        Dynamically loads all Python modules from the 'modules' directory and stores them in the 'modules' dictionary.
1047        Each module is expected to be a Python file that contains a class with the same name as the file (minus the .py extension).
1048        The class must have at least two attributes: 'name' and 'description'.
1049        
1050        This method clears any previously loaded modules, constructs the path to the modules directory, and iterates over
1051        each file in that directory. If the file is a Python file (ends with .py and is not '__init__.py'), it attempts to
1052        import the module and access the class within it to add to the 'modules' dictionary.
1053        
1054        If debug mode is enabled in the configuration, it prints debug information about the loading process and the loaded modules.
1055        """
1056
1057        self.modules.clear()
1058
1059        modules_dir = os.path.normpath(os.path.dirname(__file__) + os.path.sep + ".." + os.path.sep + "modules")
1060        self.logger.debug("Loading modules from %s ..." % modules_dir)
1061        sys.path.extend([modules_dir])
1062
1063        for file in os.listdir(modules_dir):
1064            filepath = os.path.normpath(modules_dir + os.path.sep + file)
1065            if file.endswith('.py'):
1066                if os.path.isfile(filepath) and file not in ["__init__.py"]:
1067                    try:
1068                        module_file = import_module('smbclientng.modules.%s' % (file[:-3]))
1069                        module = module_file.__getattribute__(file[:-3])
1070                        self.modules[module.name.lower()] = module
1071                    except AttributeError as e:
1072                        pass
1073
1074        if self.config.debug:
1075            if len(self.modules.keys()) == 0:
1076                self.logger.debug("Loaded 0 modules.")
1077            elif len(self.modules.keys()) == 1:
1078                self.logger.debug("Loaded 1 module:")
1079            else:
1080                self.logger.debug("Loaded %d modules:" % len(self.modules.keys()))
1081            for modulename in sorted(self.modules.keys()):
1082                self.logger.debug("%s : \"%s\" (%s)" % (self.modules[modulename].name, self.modules[modulename].description, self.modules[modulename]))
1083
1084        if self.commandCompleterObject is not None:
1085            self.commandCompleterObject.commands["module"]["subcommands"] = list(self.modules.keys())
1086
1087    def __prompt(self):
1088        """
1089        Prints the command prompt for the interactive shell.
1090
1091        This method constructs and returns the command prompt string based on the current state of the SMB session.
1092        The prompt indicates the connection status with a visual symbol and displays the current working directory
1093        or the SMB share path. The prompt appearance changes based on whether colors are enabled in the configuration.
1094
1095        Returns:
1096            str: The formatted command prompt string.
1097        """
1098
1099        # A session exists
1100        if self.sessionsManager.current_session is not None:
1101            # Check if the session is still active
1102            self.sessionsManager.current_session.ping_smb_session()
1103            if self.sessionsManager.current_session.connected:
1104                if self.config.no_colors:
1105                    connected_dot = "[v]"
1106                else:
1107                    connected_dot = "\x1b[1;92m⏺\x1b[0m"
1108            else:
1109                if self.config.no_colors:
1110                    connected_dot = "[x]"
1111                else:
1112                    connected_dot = "\x1b[1;91m⏺\x1b[0m"
1113            
1114            # Session ID if 
1115            session_prompt = ""
1116            if len(self.sessionsManager.sessions.keys()) >= 2:
1117                session_prompt = "[#%d]" % self.sessionsManager.current_session_id
1118
1119            # No share set yet
1120            if self.sessionsManager.current_session.smb_share is None:
1121                str_path = "\\\\%s\\" % self.sessionsManager.current_session.host
1122            # A share is set
1123            else:
1124                if len(self.sessionsManager.current_session.smb_cwd) == 0:
1125                    current_path = ""
1126                else:
1127                    current_path = self.sessionsManager.current_session.smb_cwd.strip(ntpath.sep) + ntpath.sep
1128                
1129                str_path = "\\\\%s\\%s\\%s" % (self.sessionsManager.current_session.host, self.sessionsManager.current_session.smb_share, current_path)
1130        # No active session
1131        else:
1132            connected_dot = ""
1133            session_prompt = ""
1134            str_path = "No active session"
1135
1136        # Build final prompt string
1137        if self.config.no_colors:
1138            str_prompt = "%s%s[%s]> " % (connected_dot, session_prompt, str_path)
1139        else:
1140            str_prompt = "%s%s[\x1b[1;94m%s\x1b[0m]> " % (connected_dot, session_prompt, str_path)
1141
1142        return str_prompt