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