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        # Get a file
154        elif command == "get":
155            self.command_get(arguments, command)
156
157        # SMB server info
158        elif command == "info":
159            self.command_info(arguments, command)
160
161        # List directory contents in a share
162        elif command in ["ls", "dir"]:
163            self.command_ls(arguments, command)
164
165        # Creates a new remote directory
166        elif command == "mkdir":
167            self.command_mkdir(arguments, command)
168
169        # Put a file
170        elif command == "put":
171            self.command_put(arguments, command)
172
173        # Changes the current local directory
174        elif command == "lcd":
175            self.command_lcd(arguments, command)
176
177        # Lists the contents of the current local directory
178        elif command == "lls":
179            self.command_lls(arguments, command)
180
181        # Creates a new local directory
182        elif command == "lmkdir":
183            self.command_lmkdir(arguments, command)
184
185        # Shows the current local directory
186        elif command == "lpwd":
187            self.command_lpwd(arguments, command)
188
189        # Removes a local file
190        elif command == "lrm":
191            self.command_lrm(arguments, command)
192
193        # Removes a local directory
194        elif command == "lrmdir":
195            self.command_lrmdir(arguments, command)
196
197        # Shows the current local directory
198        elif command == "ltree":
199            self.command_ltree(arguments, command)
200
201        # Modules
202        elif command == "module":
203            self.command_module(arguments, command)
204
205        # Reconnects the current SMB session
206        elif command in ["connect", "reconnect"]:
207            self.command_reconnect(arguments, command)
208
209        # Reset the TTY output
210        elif command == "reset":
211            self.command_reset(arguments, command)
212
213        # Removes a remote file
214        elif command == "rm":
215            self.command_rm(arguments, command)
216            
217        # Removes a remote directory
218        elif command == "rmdir":
219            self.command_rmdir(arguments, command)
220
221        # List shares
222        elif command == "shares":
223            self.command_shares(arguments, command)
224        
225        # Displays a tree view of the CWD
226        elif command == "tree":
227            self.command_tree(arguments, command)
228        
229        # Use a share
230        elif command == "use":
231            self.command_use(arguments, command)
232
233    # Commands ================================================================
234
235    @command_arguments_required
236    @active_smb_connection_needed
237    @smb_share_is_set
238    def command_bat(self, arguments, command):
239        # Command arguments required   : Yes
240        # Active SMB connection needed : Yes
241        # SMB share needed             : Yes
242
243        path = ' '.join(arguments)
244        try:
245            rawcontents = self.smbSession.read_file(path=path)
246            if rawcontents is not None:
247                encoding = charset_normalizer.detect(rawcontents)["encoding"]
248                if encoding is not None:
249                    filecontent = rawcontents.decode(encoding).rstrip()
250                    lexer = Syntax.guess_lexer(path=ntpath.basename(path), code=filecontent)
251                    # Some trickery for the files undetected by the lexer
252                    if lexer == "default":
253                        if '<?xml' in filecontent:
254                            lexer = "xml"
255                        elif '<html>' in filecontent:
256                            lexer = "html"
257                    syntax = Syntax(code=filecontent, line_numbers=True, lexer=lexer)
258                    Console().print(syntax)
259                else:
260                    print("[!] Could not detect charset of '%s'." % path)
261        except impacket.smbconnection.SessionError as e:
262            print("[!] SMB Error: %s" % e)
263
264    @command_arguments_required
265    @active_smb_connection_needed
266    @smb_share_is_set
267    def command_cd(self, arguments, command):
268        # Command arguments required   : Yes
269        # Active SMB connection needed : Yes
270        # SMB share needed             : Yes
271
272        path = ' '.join(arguments)
273        try:
274            self.smbSession.set_cwd(path=path)
275        except impacket.smbconnection.SessionError as e:
276            print("[!] SMB Error: %s" % e)
277
278    @command_arguments_required
279    @active_smb_connection_needed
280    @smb_share_is_set
281    def command_cat(self, arguments, command):
282        # Command arguments required   : Yes
283        # Active SMB connection needed : Yes
284        # SMB share needed             : Yes
285
286        path = ' '.join(arguments)
287        try:
288            rawcontents = self.smbSession.read_file(path=path)
289            if rawcontents is not None:
290                encoding = charset_normalizer.detect(rawcontents)["encoding"]
291                if encoding is not None:
292                    filecontent = rawcontents.decode(encoding).rstrip()
293                    print(filecontent)
294                else:
295                    print("[!] Could not detect charset of '%s'." % path)
296        except impacket.smbconnection.SessionError as e:
297            print("[!] SMB Error: %s" % e)
298
299    def command_close(self, arguments, command):
300        # Command arguments required   : No
301        # Active SMB connection needed : No
302        # SMB share needed             : No
303
304        self.smbSession.ping_smb_session()
305        if self.smbSession.connected:
306            self.smbSession.close_smb_session()
307
308    @command_arguments_required
309    @active_smb_connection_needed
310    @smb_share_is_set
311    def command_get(self, arguments, command):
312        # Command arguments required   : Yes
313        # Active SMB connection needed : Yes
314        # SMB share needed             : Yes
315
316        # Get files recursively
317        if arguments[0] == "-r":
318            path = ' '.join(arguments[1:]).replace('/', ntpath.sep)
319            try:
320                self.smbSession.get_file_recursively(path=path)
321            except impacket.smbconnection.SessionError as e:
322                print("[!] SMB Error: %s" % e)
323        # Get a single file
324        else:
325            path = ' '.join(arguments).replace('/', ntpath.sep)
326            try:
327                self.smbSession.get_file(path=path)
328            except impacket.smbconnection.SessionError as e:
329                print("[!] SMB Error: %s" % e)
330
331    def command_help(self, arguments, command):
332        # Command arguments required   : No
333        # Active SMB connection needed : No
334        # SMB share needed             : No
335
336        if len(arguments) != 0:
337            self.commandCompleterObject.print_help(command=arguments[0])
338        else:
339            self.commandCompleterObject.print_help(command=None)
340
341    @active_smb_connection_needed
342    def command_info(self, arguments, command):
343        # Command arguments required   : No
344        # Active SMB connection needed : Yes
345        # SMB share needed             : No
346
347        print_server_info = False
348        print_share_info = False
349        if len(arguments) != 0:
350            print_server_info = (arguments[0].lower() == "server")
351            print_share_info = (arguments[0].lower() == "share")
352        else:
353            print_server_info = True
354            print_share_info = True
355
356        try:
357            self.smbSession.info(
358                share=print_share_info,
359                server=print_server_info
360            )
361        except impacket.smbconnection.SessionError as e:
362            print("[!] SMB Error: %s" % e)
363
364    @command_arguments_required
365    def command_lcd(self, arguments, command):
366        # Command arguments required   : Yes
367        # Active SMB connection needed : No
368        # SMB share needed             : No
369
370        path = ' '.join(arguments)
371        if os.path.exists(path=path):
372            if os.path.isdir(s=path):
373                os.chdir(path=path)
374            else:
375                print("[!] Path '%s' is not a directory." % path)
376        else:
377            print("[!] Directory '%s' does not exists." % path)
378
379    def command_lls(self, arguments, command):
380        # Command arguments required   : No
381        # Active SMB connection needed : No
382        # SMB share needed             : No
383
384        if len(arguments) == 0:
385            path = '.'
386        else:
387            path = ' '.join(arguments)
388
389        # lls <directory>
390        if os.path.isdir(path):
391            directory_contents = os.listdir(path=path)
392            for entryname in sorted(directory_contents):
393                path_to_file = path + os.path.sep + entryname
394                rights_str = unix_permissions(path_to_file)
395                size_str = b_filesize(os.path.getsize(filename=path_to_file))
396                date_str = datetime.datetime.fromtimestamp(os.path.getmtime(filename=path_to_file)).strftime("%Y-%m-%d %H:%M")
397
398                if os.path.isdir(s=entryname):
399                    if self.config.no_colors:
400                        print("%s %10s  %s  %s%s" % (rights_str, size_str, date_str, entryname, os.path.sep))
401                    else:
402                        print("%s %10s  %s  \x1b[1;96m%s\x1b[0m%s" % (rights_str, size_str, date_str, entryname, os.path.sep))
403                else:
404                    if self.config.no_colors:
405                        print("%s %10s  %s  %s" % (rights_str, size_str, date_str, entryname))
406                    else:
407                        print("%s %10s  %s  \x1b[1m%s\x1b[0m" % (rights_str, size_str, date_str, entryname))
408        # lls <file>
409        elif os.path.isfile(path):
410            rights_str = unix_permissions(path)
411            size_str = b_filesize(os.path.getsize(filename=path))
412            date_str = datetime.datetime.fromtimestamp(os.path.getmtime(filename=path)).strftime("%Y-%m-%d %H:%M")
413            if self.config.no_colors:
414                print("%s %10s  %s  %s" % (rights_str, size_str, date_str, os.path.basename(path)))
415            else:
416               print("%s %10s  %s  \x1b[1m%s\x1b[0m" % (rights_str, size_str, date_str, os.path.basename(path))) 
417        else:
418            print("[!] No such file or directory.")
419
420    @command_arguments_required
421    def command_lmkdir(self, arguments, command):
422        # Command arguments required   : Yes
423        # Active SMB connection needed : No
424        # SMB share needed             : No
425
426        path = ' '.join(arguments)
427
428        # Split each dir
429        if os.path.sep in path:
430            path = path.strip(os.path.sep).split(os.path.sep)
431        else:
432            path = [path]
433
434        # Create each dir in the path
435        for depth in range(1, len(path)+1):
436            tmp_path = os.path.sep.join(path[:depth])
437            if not os.path.exists(tmp_path):
438                os.mkdir(path=tmp_path)
439
440    def command_lpwd(self, arguments, command):
441        # Command arguments required   : No
442        # Active SMB connection needed : No
443        # SMB share needed             : No
444
445        print(os.getcwd())
446
447    @command_arguments_required
448    def command_lrm(self, arguments, command):
449        # Command arguments required   : Yes
450        # Active SMB connection needed : No
451        # SMB share needed             : No
452
453        path = ' '.join(arguments)
454        if os.path.exists(path):
455            if not os.path.isdir(s=path):
456                try:
457                    os.remove(path=path)
458                except Exception as e:
459                    print("[!] Error removing file '%s' : %s" % path)
460            else:
461                print("[!] Cannot delete '%s'. It is a directory, use 'lrmdir <directory>' instead." % path)
462        else:
463            print("[!] Path '%s' does not exist." % path)
464
465    @command_arguments_required
466    def command_lrmdir(self, arguments, command):
467        # Command arguments required   : Yes
468        # Active SMB connection needed : No
469        # SMB share needed             : No
470
471        path = ' '.join(arguments)
472        if os.path.exists(path):
473            if os.path.isdir(s=path):
474                try:
475                    shutil.rmtree(path=path)
476                except Exception as e:
477                    print("[!] Error removing directory '%s' : %s" % path)
478            else:
479                print("[!] Cannot delete '%s'. It is a file, use 'lrm <file>' instead." % path)
480        else:
481            print("[!] Path '%s' does not exist." % path)
482
483    def command_ltree(self, arguments, command):
484        # Command arguments required   : No
485        # Active SMB connection needed : No
486        # SMB share needed             : No
487
488        if len(arguments) == 0:
489            local_tree(path='.', config=self.config)
490        else:
491            local_tree(path=' '.join(arguments), config=self.config)
492
493    @active_smb_connection_needed
494    @smb_share_is_set
495    def command_ls(self, arguments, command):
496        # Command arguments required   : No
497        # Active SMB connection needed : Yes
498        # SMB share needed             : Yes
499
500        # Read the files
501        directory_contents = self.smbSession.list_contents(path=' '.join(arguments))
502
503        for longname in sorted(directory_contents.keys(), key=lambda x:x.lower()):
504            windows_ls_entry(directory_contents[longname], self.config)
505            
506    @command_arguments_required
507    @active_smb_connection_needed
508    @smb_share_is_set
509    def command_mkdir(self, arguments, command):
510        # Command arguments required   : Yes
511        # Active SMB connection needed : Yes
512        # SMB share needed             : Yes
513
514        path = ' '.join(arguments)
515        self.smbSession.mkdir(path=path)
516
517    @command_arguments_required
518    @active_smb_connection_needed
519    @smb_share_is_set
520    def command_module(self, arguments, command):
521        module_name = arguments[0]
522
523        if module_name in self.modules.keys():
524            module = self.modules[module_name](self.smbSession, self.config)
525            module.run(' '.join(arguments[1:]))
526        else:
527            print("[!] Module '%s' does not exist." % module_name)
528
529    @command_arguments_required
530    @active_smb_connection_needed
531    @smb_share_is_set
532    def command_put(self, arguments, command):
533        # Command arguments required   : Yes
534        # Active SMB connection needed : Yes
535        # SMB share needed             : Yes
536
537        # Put files recursively
538        if arguments[0] == "-r":
539            localpath = ' '.join(arguments[1:])
540            try:
541                self.smbSession.put_file_recursively(localpath=localpath)
542            except impacket.smbconnection.SessionError as e:
543                print("[!] SMB Error: %s" % e)
544
545        # Put a single file
546        else:
547            localpath = ' '.join(arguments)
548            try:
549                self.smbSession.put_file(localpath=localpath)
550            except impacket.smbconnection.SessionError as e:
551                print("[!] SMB Error: %s" % e)
552
553    def command_reconnect(self, arguments, command):
554        # Command arguments required   : No
555        # Active SMB connection needed : No
556        # SMB share needed             : No
557
558        self.smbSession.ping_smb_session()
559        if self.smbSession.connected:
560            self.smbSession.close_smb_session()
561            self.smbSession.init_smb_session()
562        else:
563            self.smbSession.init_smb_session()
564
565    def command_reset(self, arguments, command):
566        # Command arguments required   : No
567        # Active SMB connection needed : No
568        # SMB share needed             : No
569        sys.stdout.write('\x1b[?25h') # Sets the cursor to on
570        sys.stdout.write('\x1b[v')  
571        sys.stdout.write('\x1b[o') # Reset
572        sys.stdout.flush()
573
574    @command_arguments_required
575    @active_smb_connection_needed
576    @smb_share_is_set
577    def command_rm(self, arguments, command):
578        # Command arguments required   : Yes
579        # Active SMB connection needed : Yes
580        # SMB share needed             : Yes
581
582        path = ' '.join(arguments)
583        if self.smbSession.path_exists(path):
584            if self.smbSession.path_isfile(path):
585                try:
586                    self.smbSession.rm(path=path)
587                except Exception as e:
588                    print("[!] Error removing file '%s' : %s" % path)
589            else:
590                print("[!] Cannot delete '%s': This is a directory, use 'rmdir <directory>' instead." % path)
591        else:
592            print("[!] Remote file '%s' does not exist." % path)
593
594    @command_arguments_required
595    @active_smb_connection_needed
596    @smb_share_is_set
597    def command_rmdir(self, arguments, command):
598        # Command arguments required   : Yes
599        # Active SMB connection needed : Yes
600        # SMB share needed             : Yes
601
602        path = ' '.join(arguments)
603        if self.smbSession.path_exists(path):
604            if self.smbSession.path_isdir(path):
605                try:
606                    self.smbSession.rmdir(path=path)
607                except Exception as e:
608                    print("[!] Error removing directory '%s' : %s" % path)
609            else:
610                print("[!] Cannot delete '%s': This is a file, use 'rm <file>' instead." % path)
611        else:
612            print("[!] Remote directory '%s' does not exist." % path)
613
614    @active_smb_connection_needed
615    def command_shares(self, arguments, command):
616        # Command arguments required   : No
617        # Active SMB connection needed : Yes
618        # SMB share needed             : No
619
620        shares = self.smbSession.list_shares()
621        if len(shares.keys()) != 0:
622            table = Table(title=None)
623            table.add_column("Share")
624            table.add_column("Hidden")
625            table.add_column("Type")
626            table.add_column("Description", justify="left")
627
628            for sharename in sorted(shares.keys()):
629                is_hidden = bool(sharename.endswith('$'))
630                types = ', '.join([s.replace("STYPE_","") for s in shares[sharename]["type"]])
631                if is_hidden:
632                    table.add_row(sharename, str(is_hidden), types, shares[sharename]["comment"])
633                else:
634                    table.add_row(sharename, str(is_hidden), types, shares[sharename]["comment"])
635
636            Console().print(table)
637        else:
638            print("[!] No share served on '%s'" % self.smbSession.address)
639
640    @active_smb_connection_needed
641    @smb_share_is_set
642    def command_tree(self, arguments, command):
643        # Command arguments required   : No
644        # Active SMB connection needed : Yes
645        # SMB share needed             : Yes
646
647        if len(arguments) == 0:
648            self.smbSession.tree(path='.')
649        else:
650            self.smbSession.tree(path=' '.join(arguments))
651
652    @command_arguments_required
653    @active_smb_connection_needed
654    def command_use(self, arguments, command):
655        # Command arguments required   : Yes
656        # Active SMB connection needed : Yes
657        # SMB share needed             : No
658
659        sharename = ' '.join(arguments)
660        # Reload the list of shares
661        shares = self.smbSession.list_shares()
662        shares = [s.lower() for s in shares.keys()]
663        if sharename.lower() in shares:
664            self.smbSession.set_share(sharename)
665        else:
666            print("[!] No share named '%s' on '%s'" % (sharename, self.smbSession.address))
667
668    # Private functions =======================================================
669
670    def __load_modules(self):
671
672
673        self.modules.clear()
674
675        modules_dir = os.path.normpath(os.path.dirname(__file__) + os.path.sep + ".." + os.path.sep + "modules")
676        if self.config.debug:
677            print("[debug] Loading modules from %s ..." % modules_dir)
678        sys.path.extend([modules_dir])
679
680        for file in os.listdir(modules_dir):
681            filepath = os.path.normpath(modules_dir + os.path.sep + file)
682            if file.endswith('.py'):
683                if os.path.isfile(filepath) and file not in ["__init__.py"]:
684                    try:
685                        module_file = import_module('smbclientng.modules.%s' % (file[:-3]))
686                        module = module_file.__getattribute__(file[:-3])
687                        self.modules[module.name.lower()] = module
688                    except AttributeError as e:
689                        pass
690
691        if self.config.debug:
692            if len(self.modules.keys()) == 0:
693                print("[debug] Loaded 0 modules.")
694            elif len(self.modules.keys()) == 1:
695                print("[debug] Loaded 1 module:")
696            else:
697                print("[debug] Loaded %d modules:" % len(self.modules.keys()))
698            for modulename in sorted(self.modules.keys()):
699                print("[debug] %s : \"%s\" (%s)" % (self.modules[modulename].name, self.modules[modulename].description, self.modules[modulename]))
700
701        if self.commandCompleterObject is not None:
702            self.commandCompleterObject.commands["module"]["subcommands"] = list(self.modules.keys())
703
704    def __prompt(self):
705        """
706        Prints the command prompt for the interactive shell.
707
708        This method constructs and returns the command prompt string based on the current state of the SMB session.
709        The prompt indicates the connection status with a visual symbol and displays the current working directory
710        or the SMB share path. The prompt appearance changes based on whether colors are enabled in the configuration.
711
712        Returns:
713            str: The formatted command prompt string.
714        """
715
716        self.smbSession.ping_smb_session()
717        if self.smbSession.connected:
718            if self.config.no_colors:
719                connected_dot = "[v]"
720            else:
721                connected_dot = "\x1b[1;92m⏺\x1b[0m"
722        else:
723            if self.config.no_colors:
724                connected_dot = "[x]"
725            else:
726                connected_dot = "\x1b[1;91m⏺\x1b[0m"
727        
728        if self.smbSession.smb_share is None:
729            if self.config.no_colors:
730                str_prompt = "%s[\\\\%s\\]> " % (connected_dot, self.smbSession.address)
731            else:
732                str_prompt = "%s[\x1b[1;94m\\\\%s\\\x1b[0m]> " % (connected_dot, self.smbSession.address)
733        else:
734            str_path = "\\\\%s\\%s\\%s" % (self.smbSession.address, self.smbSession.smb_share, self.smbSession.smb_cwd.lstrip(ntpath.sep))
735            if self.config.no_colors:
736                str_prompt = "%s[%s]> " % (connected_dot, str_path)
737            else:
738                str_prompt = "%s[\x1b[1;94m%s\x1b[0m]> " % (connected_dot, str_path)
739
740        return str_prompt
def command_arguments_required(func):
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
def active_smb_connection_needed(func):
38def active_smb_connection_needed(func):
39    def wrapper(*args, **kwargs):
40        self, arguments,command  = args[0], args[1], args[2]
41        #
42        self.smbSession.ping_smb_session()
43        if self.smbSession.connected:
44            return func(*args, **kwargs)
45        else:
46            print("[!] SMB Session is disconnected.")
47            return None
48    return wrapper
def smb_share_is_set(func):
50def smb_share_is_set(func):
51    def wrapper(*args, **kwargs):
52        self, arguments,command  = args[0], args[1], args[2]
53        if self.smbSession.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
class InteractiveShell:
 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    def __init__(self, smbSession, config):
 82        # Objects
 83        self.smbSession = smbSession
 84        self.config = config
 85        self.commandCompleterObject = CommandCompleter(smbSession=self.smbSession, config=self.config)
 86        readline.set_completer(self.commandCompleterObject.complete)
 87        readline.parse_and_bind("tab: complete")
 88        readline.set_completer_delims("\n")
 89        # Additional modules
 90        self.modules = {}
 91        self.__load_modules()
 92
 93    def run(self):
 94        running = True
 95        while running:
 96            try:
 97                user_input = input(self.__prompt()).strip().split(" ")
 98                command, arguments = user_input[0].lower(), user_input[1:]
 99                
100                # Exit the command line
101                if command == "exit":
102                    running = False
103
104                elif command in self.commandCompleterObject.commands.keys():
105                    self.process_command(
106                        command=command, 
107                        arguments=arguments
108                    )
109
110                elif command.strip() == "":
111                    pass
112
113                # Fallback to unknown command
114                else:
115                    print("Unknown command. Type \"help\" for help.")
116
117            except KeyboardInterrupt as e:
118                print()
119
120            except EOFError as e:
121                print()
122                running = False
123
124            except Exception as e:
125                if self.config.debug:
126                    traceback.print_exc()
127                print("[!] Error: %s" % str(e))
128
129    def process_command(self, command, arguments=[]):
130        # Skip
131        if command.strip() == "":
132            pass
133        
134        # Display help
135        elif command == "help":
136            self.command_help(arguments, command)
137
138        # Cat the contents of a file
139        elif command == "bat":
140            self.command_bat(arguments, command)
141
142        # Cat the contents of a file
143        elif command == "cat":
144            self.command_cat(arguments, command)
145
146        # Closes the current SMB session
147        elif command == "close":
148            self.command_close(arguments, command)
149               
150        # Change directory in the current share
151        elif command == "cd":
152            self.command_cd(arguments, command)
153
154        # Get a file
155        elif command == "get":
156            self.command_get(arguments, command)
157
158        # SMB server info
159        elif command == "info":
160            self.command_info(arguments, command)
161
162        # List directory contents in a share
163        elif command in ["ls", "dir"]:
164            self.command_ls(arguments, command)
165
166        # Creates a new remote directory
167        elif command == "mkdir":
168            self.command_mkdir(arguments, command)
169
170        # Put a file
171        elif command == "put":
172            self.command_put(arguments, command)
173
174        # Changes the current local directory
175        elif command == "lcd":
176            self.command_lcd(arguments, command)
177
178        # Lists the contents of the current local directory
179        elif command == "lls":
180            self.command_lls(arguments, command)
181
182        # Creates a new local directory
183        elif command == "lmkdir":
184            self.command_lmkdir(arguments, command)
185
186        # Shows the current local directory
187        elif command == "lpwd":
188            self.command_lpwd(arguments, command)
189
190        # Removes a local file
191        elif command == "lrm":
192            self.command_lrm(arguments, command)
193
194        # Removes a local directory
195        elif command == "lrmdir":
196            self.command_lrmdir(arguments, command)
197
198        # Shows the current local directory
199        elif command == "ltree":
200            self.command_ltree(arguments, command)
201
202        # Modules
203        elif command == "module":
204            self.command_module(arguments, command)
205
206        # Reconnects the current SMB session
207        elif command in ["connect", "reconnect"]:
208            self.command_reconnect(arguments, command)
209
210        # Reset the TTY output
211        elif command == "reset":
212            self.command_reset(arguments, command)
213
214        # Removes a remote file
215        elif command == "rm":
216            self.command_rm(arguments, command)
217            
218        # Removes a remote directory
219        elif command == "rmdir":
220            self.command_rmdir(arguments, command)
221
222        # List shares
223        elif command == "shares":
224            self.command_shares(arguments, command)
225        
226        # Displays a tree view of the CWD
227        elif command == "tree":
228            self.command_tree(arguments, command)
229        
230        # Use a share
231        elif command == "use":
232            self.command_use(arguments, command)
233
234    # Commands ================================================================
235
236    @command_arguments_required
237    @active_smb_connection_needed
238    @smb_share_is_set
239    def command_bat(self, arguments, command):
240        # Command arguments required   : Yes
241        # Active SMB connection needed : Yes
242        # SMB share needed             : Yes
243
244        path = ' '.join(arguments)
245        try:
246            rawcontents = self.smbSession.read_file(path=path)
247            if rawcontents is not None:
248                encoding = charset_normalizer.detect(rawcontents)["encoding"]
249                if encoding is not None:
250                    filecontent = rawcontents.decode(encoding).rstrip()
251                    lexer = Syntax.guess_lexer(path=ntpath.basename(path), code=filecontent)
252                    # Some trickery for the files undetected by the lexer
253                    if lexer == "default":
254                        if '<?xml' in filecontent:
255                            lexer = "xml"
256                        elif '<html>' in filecontent:
257                            lexer = "html"
258                    syntax = Syntax(code=filecontent, line_numbers=True, lexer=lexer)
259                    Console().print(syntax)
260                else:
261                    print("[!] Could not detect charset of '%s'." % path)
262        except impacket.smbconnection.SessionError as e:
263            print("[!] SMB Error: %s" % e)
264
265    @command_arguments_required
266    @active_smb_connection_needed
267    @smb_share_is_set
268    def command_cd(self, arguments, command):
269        # Command arguments required   : Yes
270        # Active SMB connection needed : Yes
271        # SMB share needed             : Yes
272
273        path = ' '.join(arguments)
274        try:
275            self.smbSession.set_cwd(path=path)
276        except impacket.smbconnection.SessionError as e:
277            print("[!] SMB Error: %s" % e)
278
279    @command_arguments_required
280    @active_smb_connection_needed
281    @smb_share_is_set
282    def command_cat(self, arguments, command):
283        # Command arguments required   : Yes
284        # Active SMB connection needed : Yes
285        # SMB share needed             : Yes
286
287        path = ' '.join(arguments)
288        try:
289            rawcontents = self.smbSession.read_file(path=path)
290            if rawcontents is not None:
291                encoding = charset_normalizer.detect(rawcontents)["encoding"]
292                if encoding is not None:
293                    filecontent = rawcontents.decode(encoding).rstrip()
294                    print(filecontent)
295                else:
296                    print("[!] Could not detect charset of '%s'." % path)
297        except impacket.smbconnection.SessionError as e:
298            print("[!] SMB Error: %s" % e)
299
300    def command_close(self, arguments, command):
301        # Command arguments required   : No
302        # Active SMB connection needed : No
303        # SMB share needed             : No
304
305        self.smbSession.ping_smb_session()
306        if self.smbSession.connected:
307            self.smbSession.close_smb_session()
308
309    @command_arguments_required
310    @active_smb_connection_needed
311    @smb_share_is_set
312    def command_get(self, arguments, command):
313        # Command arguments required   : Yes
314        # Active SMB connection needed : Yes
315        # SMB share needed             : Yes
316
317        # Get files recursively
318        if arguments[0] == "-r":
319            path = ' '.join(arguments[1:]).replace('/', ntpath.sep)
320            try:
321                self.smbSession.get_file_recursively(path=path)
322            except impacket.smbconnection.SessionError as e:
323                print("[!] SMB Error: %s" % e)
324        # Get a single file
325        else:
326            path = ' '.join(arguments).replace('/', ntpath.sep)
327            try:
328                self.smbSession.get_file(path=path)
329            except impacket.smbconnection.SessionError as e:
330                print("[!] SMB Error: %s" % e)
331
332    def command_help(self, arguments, command):
333        # Command arguments required   : No
334        # Active SMB connection needed : No
335        # SMB share needed             : No
336
337        if len(arguments) != 0:
338            self.commandCompleterObject.print_help(command=arguments[0])
339        else:
340            self.commandCompleterObject.print_help(command=None)
341
342    @active_smb_connection_needed
343    def command_info(self, arguments, command):
344        # Command arguments required   : No
345        # Active SMB connection needed : Yes
346        # SMB share needed             : No
347
348        print_server_info = False
349        print_share_info = False
350        if len(arguments) != 0:
351            print_server_info = (arguments[0].lower() == "server")
352            print_share_info = (arguments[0].lower() == "share")
353        else:
354            print_server_info = True
355            print_share_info = True
356
357        try:
358            self.smbSession.info(
359                share=print_share_info,
360                server=print_server_info
361            )
362        except impacket.smbconnection.SessionError as e:
363            print("[!] SMB Error: %s" % e)
364
365    @command_arguments_required
366    def command_lcd(self, arguments, command):
367        # Command arguments required   : Yes
368        # Active SMB connection needed : No
369        # SMB share needed             : No
370
371        path = ' '.join(arguments)
372        if os.path.exists(path=path):
373            if os.path.isdir(s=path):
374                os.chdir(path=path)
375            else:
376                print("[!] Path '%s' is not a directory." % path)
377        else:
378            print("[!] Directory '%s' does not exists." % path)
379
380    def command_lls(self, arguments, command):
381        # Command arguments required   : No
382        # Active SMB connection needed : No
383        # SMB share needed             : No
384
385        if len(arguments) == 0:
386            path = '.'
387        else:
388            path = ' '.join(arguments)
389
390        # lls <directory>
391        if os.path.isdir(path):
392            directory_contents = os.listdir(path=path)
393            for entryname in sorted(directory_contents):
394                path_to_file = path + os.path.sep + entryname
395                rights_str = unix_permissions(path_to_file)
396                size_str = b_filesize(os.path.getsize(filename=path_to_file))
397                date_str = datetime.datetime.fromtimestamp(os.path.getmtime(filename=path_to_file)).strftime("%Y-%m-%d %H:%M")
398
399                if os.path.isdir(s=entryname):
400                    if self.config.no_colors:
401                        print("%s %10s  %s  %s%s" % (rights_str, size_str, date_str, entryname, os.path.sep))
402                    else:
403                        print("%s %10s  %s  \x1b[1;96m%s\x1b[0m%s" % (rights_str, size_str, date_str, entryname, os.path.sep))
404                else:
405                    if self.config.no_colors:
406                        print("%s %10s  %s  %s" % (rights_str, size_str, date_str, entryname))
407                    else:
408                        print("%s %10s  %s  \x1b[1m%s\x1b[0m" % (rights_str, size_str, date_str, entryname))
409        # lls <file>
410        elif os.path.isfile(path):
411            rights_str = unix_permissions(path)
412            size_str = b_filesize(os.path.getsize(filename=path))
413            date_str = datetime.datetime.fromtimestamp(os.path.getmtime(filename=path)).strftime("%Y-%m-%d %H:%M")
414            if self.config.no_colors:
415                print("%s %10s  %s  %s" % (rights_str, size_str, date_str, os.path.basename(path)))
416            else:
417               print("%s %10s  %s  \x1b[1m%s\x1b[0m" % (rights_str, size_str, date_str, os.path.basename(path))) 
418        else:
419            print("[!] No such file or directory.")
420
421    @command_arguments_required
422    def command_lmkdir(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
429        # Split each dir
430        if os.path.sep in path:
431            path = path.strip(os.path.sep).split(os.path.sep)
432        else:
433            path = [path]
434
435        # Create each dir in the path
436        for depth in range(1, len(path)+1):
437            tmp_path = os.path.sep.join(path[:depth])
438            if not os.path.exists(tmp_path):
439                os.mkdir(path=tmp_path)
440
441    def command_lpwd(self, arguments, command):
442        # Command arguments required   : No
443        # Active SMB connection needed : No
444        # SMB share needed             : No
445
446        print(os.getcwd())
447
448    @command_arguments_required
449    def command_lrm(self, arguments, command):
450        # Command arguments required   : Yes
451        # Active SMB connection needed : No
452        # SMB share needed             : No
453
454        path = ' '.join(arguments)
455        if os.path.exists(path):
456            if not os.path.isdir(s=path):
457                try:
458                    os.remove(path=path)
459                except Exception as e:
460                    print("[!] Error removing file '%s' : %s" % path)
461            else:
462                print("[!] Cannot delete '%s'. It is a directory, use 'lrmdir <directory>' instead." % path)
463        else:
464            print("[!] Path '%s' does not exist." % path)
465
466    @command_arguments_required
467    def command_lrmdir(self, arguments, command):
468        # Command arguments required   : Yes
469        # Active SMB connection needed : No
470        # SMB share needed             : No
471
472        path = ' '.join(arguments)
473        if os.path.exists(path):
474            if os.path.isdir(s=path):
475                try:
476                    shutil.rmtree(path=path)
477                except Exception as e:
478                    print("[!] Error removing directory '%s' : %s" % path)
479            else:
480                print("[!] Cannot delete '%s'. It is a file, use 'lrm <file>' instead." % path)
481        else:
482            print("[!] Path '%s' does not exist." % path)
483
484    def command_ltree(self, arguments, command):
485        # Command arguments required   : No
486        # Active SMB connection needed : No
487        # SMB share needed             : No
488
489        if len(arguments) == 0:
490            local_tree(path='.', config=self.config)
491        else:
492            local_tree(path=' '.join(arguments), config=self.config)
493
494    @active_smb_connection_needed
495    @smb_share_is_set
496    def command_ls(self, arguments, command):
497        # Command arguments required   : No
498        # Active SMB connection needed : Yes
499        # SMB share needed             : Yes
500
501        # Read the files
502        directory_contents = self.smbSession.list_contents(path=' '.join(arguments))
503
504        for longname in sorted(directory_contents.keys(), key=lambda x:x.lower()):
505            windows_ls_entry(directory_contents[longname], self.config)
506            
507    @command_arguments_required
508    @active_smb_connection_needed
509    @smb_share_is_set
510    def command_mkdir(self, arguments, command):
511        # Command arguments required   : Yes
512        # Active SMB connection needed : Yes
513        # SMB share needed             : Yes
514
515        path = ' '.join(arguments)
516        self.smbSession.mkdir(path=path)
517
518    @command_arguments_required
519    @active_smb_connection_needed
520    @smb_share_is_set
521    def command_module(self, arguments, command):
522        module_name = arguments[0]
523
524        if module_name in self.modules.keys():
525            module = self.modules[module_name](self.smbSession, self.config)
526            module.run(' '.join(arguments[1:]))
527        else:
528            print("[!] Module '%s' does not exist." % module_name)
529
530    @command_arguments_required
531    @active_smb_connection_needed
532    @smb_share_is_set
533    def command_put(self, arguments, command):
534        # Command arguments required   : Yes
535        # Active SMB connection needed : Yes
536        # SMB share needed             : Yes
537
538        # Put files recursively
539        if arguments[0] == "-r":
540            localpath = ' '.join(arguments[1:])
541            try:
542                self.smbSession.put_file_recursively(localpath=localpath)
543            except impacket.smbconnection.SessionError as e:
544                print("[!] SMB Error: %s" % e)
545
546        # Put a single file
547        else:
548            localpath = ' '.join(arguments)
549            try:
550                self.smbSession.put_file(localpath=localpath)
551            except impacket.smbconnection.SessionError as e:
552                print("[!] SMB Error: %s" % e)
553
554    def command_reconnect(self, arguments, command):
555        # Command arguments required   : No
556        # Active SMB connection needed : No
557        # SMB share needed             : No
558
559        self.smbSession.ping_smb_session()
560        if self.smbSession.connected:
561            self.smbSession.close_smb_session()
562            self.smbSession.init_smb_session()
563        else:
564            self.smbSession.init_smb_session()
565
566    def command_reset(self, arguments, command):
567        # Command arguments required   : No
568        # Active SMB connection needed : No
569        # SMB share needed             : No
570        sys.stdout.write('\x1b[?25h') # Sets the cursor to on
571        sys.stdout.write('\x1b[v')  
572        sys.stdout.write('\x1b[o') # Reset
573        sys.stdout.flush()
574
575    @command_arguments_required
576    @active_smb_connection_needed
577    @smb_share_is_set
578    def command_rm(self, arguments, command):
579        # Command arguments required   : Yes
580        # Active SMB connection needed : Yes
581        # SMB share needed             : Yes
582
583        path = ' '.join(arguments)
584        if self.smbSession.path_exists(path):
585            if self.smbSession.path_isfile(path):
586                try:
587                    self.smbSession.rm(path=path)
588                except Exception as e:
589                    print("[!] Error removing file '%s' : %s" % path)
590            else:
591                print("[!] Cannot delete '%s': This is a directory, use 'rmdir <directory>' instead." % path)
592        else:
593            print("[!] Remote file '%s' does not exist." % path)
594
595    @command_arguments_required
596    @active_smb_connection_needed
597    @smb_share_is_set
598    def command_rmdir(self, arguments, command):
599        # Command arguments required   : Yes
600        # Active SMB connection needed : Yes
601        # SMB share needed             : Yes
602
603        path = ' '.join(arguments)
604        if self.smbSession.path_exists(path):
605            if self.smbSession.path_isdir(path):
606                try:
607                    self.smbSession.rmdir(path=path)
608                except Exception as e:
609                    print("[!] Error removing directory '%s' : %s" % path)
610            else:
611                print("[!] Cannot delete '%s': This is a file, use 'rm <file>' instead." % path)
612        else:
613            print("[!] Remote directory '%s' does not exist." % path)
614
615    @active_smb_connection_needed
616    def command_shares(self, arguments, command):
617        # Command arguments required   : No
618        # Active SMB connection needed : Yes
619        # SMB share needed             : No
620
621        shares = self.smbSession.list_shares()
622        if len(shares.keys()) != 0:
623            table = Table(title=None)
624            table.add_column("Share")
625            table.add_column("Hidden")
626            table.add_column("Type")
627            table.add_column("Description", justify="left")
628
629            for sharename in sorted(shares.keys()):
630                is_hidden = bool(sharename.endswith('$'))
631                types = ', '.join([s.replace("STYPE_","") for s in shares[sharename]["type"]])
632                if is_hidden:
633                    table.add_row(sharename, str(is_hidden), types, shares[sharename]["comment"])
634                else:
635                    table.add_row(sharename, str(is_hidden), types, shares[sharename]["comment"])
636
637            Console().print(table)
638        else:
639            print("[!] No share served on '%s'" % self.smbSession.address)
640
641    @active_smb_connection_needed
642    @smb_share_is_set
643    def command_tree(self, arguments, command):
644        # Command arguments required   : No
645        # Active SMB connection needed : Yes
646        # SMB share needed             : Yes
647
648        if len(arguments) == 0:
649            self.smbSession.tree(path='.')
650        else:
651            self.smbSession.tree(path=' '.join(arguments))
652
653    @command_arguments_required
654    @active_smb_connection_needed
655    def command_use(self, arguments, command):
656        # Command arguments required   : Yes
657        # Active SMB connection needed : Yes
658        # SMB share needed             : No
659
660        sharename = ' '.join(arguments)
661        # Reload the list of shares
662        shares = self.smbSession.list_shares()
663        shares = [s.lower() for s in shares.keys()]
664        if sharename.lower() in shares:
665            self.smbSession.set_share(sharename)
666        else:
667            print("[!] No share named '%s' on '%s'" % (sharename, self.smbSession.address))
668
669    # Private functions =======================================================
670
671    def __load_modules(self):
672
673
674        self.modules.clear()
675
676        modules_dir = os.path.normpath(os.path.dirname(__file__) + os.path.sep + ".." + os.path.sep + "modules")
677        if self.config.debug:
678            print("[debug] Loading modules from %s ..." % modules_dir)
679        sys.path.extend([modules_dir])
680
681        for file in os.listdir(modules_dir):
682            filepath = os.path.normpath(modules_dir + os.path.sep + file)
683            if file.endswith('.py'):
684                if os.path.isfile(filepath) and file not in ["__init__.py"]:
685                    try:
686                        module_file = import_module('smbclientng.modules.%s' % (file[:-3]))
687                        module = module_file.__getattribute__(file[:-3])
688                        self.modules[module.name.lower()] = module
689                    except AttributeError as e:
690                        pass
691
692        if self.config.debug:
693            if len(self.modules.keys()) == 0:
694                print("[debug] Loaded 0 modules.")
695            elif len(self.modules.keys()) == 1:
696                print("[debug] Loaded 1 module:")
697            else:
698                print("[debug] Loaded %d modules:" % len(self.modules.keys()))
699            for modulename in sorted(self.modules.keys()):
700                print("[debug] %s : \"%s\" (%s)" % (self.modules[modulename].name, self.modules[modulename].description, self.modules[modulename]))
701
702        if self.commandCompleterObject is not None:
703            self.commandCompleterObject.commands["module"]["subcommands"] = list(self.modules.keys())
704
705    def __prompt(self):
706        """
707        Prints the command prompt for the interactive shell.
708
709        This method constructs and returns the command prompt string based on the current state of the SMB session.
710        The prompt indicates the connection status with a visual symbol and displays the current working directory
711        or the SMB share path. The prompt appearance changes based on whether colors are enabled in the configuration.
712
713        Returns:
714            str: The formatted command prompt string.
715        """
716
717        self.smbSession.ping_smb_session()
718        if self.smbSession.connected:
719            if self.config.no_colors:
720                connected_dot = "[v]"
721            else:
722                connected_dot = "\x1b[1;92m⏺\x1b[0m"
723        else:
724            if self.config.no_colors:
725                connected_dot = "[x]"
726            else:
727                connected_dot = "\x1b[1;91m⏺\x1b[0m"
728        
729        if self.smbSession.smb_share is None:
730            if self.config.no_colors:
731                str_prompt = "%s[\\\\%s\\]> " % (connected_dot, self.smbSession.address)
732            else:
733                str_prompt = "%s[\x1b[1;94m\\\\%s\\\x1b[0m]> " % (connected_dot, self.smbSession.address)
734        else:
735            str_path = "\\\\%s\\%s\\%s" % (self.smbSession.address, self.smbSession.smb_share, self.smbSession.smb_cwd.lstrip(ntpath.sep))
736            if self.config.no_colors:
737                str_prompt = "%s[%s]> " % (connected_dot, str_path)
738            else:
739                str_prompt = "%s[\x1b[1;94m%s\x1b[0m]> " % (connected_dot, str_path)
740
741        return str_prompt

Class InteractiveShell is designed to manage the interactive command line interface for smbclient-ng.

This class handles user input, executes commands, and manages the state of the SMB session. It provides a command line interface for users to interact with SMB shares, execute commands like directory listing, file transfer, and more.

Attributes: smbSession (SMBConnection): The active SMB connection session. debug (bool): Flag to enable or disable debug mode. smb_share (str): The current SMB share in use. smb_path (str): The current path within the SMB share. commandCompleterObject (CommandCompleter): Object to handle command completion and help generation.

Methods: __init__(self, smbSession, debug=False): Initializes the InteractiveShell with the given SMB session and debug mode. run(self): Starts the command line interface loop, processing user input until exit.

InteractiveShell(smbSession, config)
81    def __init__(self, smbSession, config):
82        # Objects
83        self.smbSession = smbSession
84        self.config = config
85        self.commandCompleterObject = CommandCompleter(smbSession=self.smbSession, config=self.config)
86        readline.set_completer(self.commandCompleterObject.complete)
87        readline.parse_and_bind("tab: complete")
88        readline.set_completer_delims("\n")
89        # Additional modules
90        self.modules = {}
91        self.__load_modules()
smbSession
config
commandCompleterObject
modules
def run(self):
 93    def run(self):
 94        running = True
 95        while running:
 96            try:
 97                user_input = input(self.__prompt()).strip().split(" ")
 98                command, arguments = user_input[0].lower(), user_input[1:]
 99                
100                # Exit the command line
101                if command == "exit":
102                    running = False
103
104                elif command in self.commandCompleterObject.commands.keys():
105                    self.process_command(
106                        command=command, 
107                        arguments=arguments
108                    )
109
110                elif command.strip() == "":
111                    pass
112
113                # Fallback to unknown command
114                else:
115                    print("Unknown command. Type \"help\" for help.")
116
117            except KeyboardInterrupt as e:
118                print()
119
120            except EOFError as e:
121                print()
122                running = False
123
124            except Exception as e:
125                if self.config.debug:
126                    traceback.print_exc()
127                print("[!] Error: %s" % str(e))
def process_command(self, command, arguments=[]):
129    def process_command(self, command, arguments=[]):
130        # Skip
131        if command.strip() == "":
132            pass
133        
134        # Display help
135        elif command == "help":
136            self.command_help(arguments, command)
137
138        # Cat the contents of a file
139        elif command == "bat":
140            self.command_bat(arguments, command)
141
142        # Cat the contents of a file
143        elif command == "cat":
144            self.command_cat(arguments, command)
145
146        # Closes the current SMB session
147        elif command == "close":
148            self.command_close(arguments, command)
149               
150        # Change directory in the current share
151        elif command == "cd":
152            self.command_cd(arguments, command)
153
154        # Get a file
155        elif command == "get":
156            self.command_get(arguments, command)
157
158        # SMB server info
159        elif command == "info":
160            self.command_info(arguments, command)
161
162        # List directory contents in a share
163        elif command in ["ls", "dir"]:
164            self.command_ls(arguments, command)
165
166        # Creates a new remote directory
167        elif command == "mkdir":
168            self.command_mkdir(arguments, command)
169
170        # Put a file
171        elif command == "put":
172            self.command_put(arguments, command)
173
174        # Changes the current local directory
175        elif command == "lcd":
176            self.command_lcd(arguments, command)
177
178        # Lists the contents of the current local directory
179        elif command == "lls":
180            self.command_lls(arguments, command)
181
182        # Creates a new local directory
183        elif command == "lmkdir":
184            self.command_lmkdir(arguments, command)
185
186        # Shows the current local directory
187        elif command == "lpwd":
188            self.command_lpwd(arguments, command)
189
190        # Removes a local file
191        elif command == "lrm":
192            self.command_lrm(arguments, command)
193
194        # Removes a local directory
195        elif command == "lrmdir":
196            self.command_lrmdir(arguments, command)
197
198        # Shows the current local directory
199        elif command == "ltree":
200            self.command_ltree(arguments, command)
201
202        # Modules
203        elif command == "module":
204            self.command_module(arguments, command)
205
206        # Reconnects the current SMB session
207        elif command in ["connect", "reconnect"]:
208            self.command_reconnect(arguments, command)
209
210        # Reset the TTY output
211        elif command == "reset":
212            self.command_reset(arguments, command)
213
214        # Removes a remote file
215        elif command == "rm":
216            self.command_rm(arguments, command)
217            
218        # Removes a remote directory
219        elif command == "rmdir":
220            self.command_rmdir(arguments, command)
221
222        # List shares
223        elif command == "shares":
224            self.command_shares(arguments, command)
225        
226        # Displays a tree view of the CWD
227        elif command == "tree":
228            self.command_tree(arguments, command)
229        
230        # Use a share
231        elif command == "use":
232            self.command_use(arguments, command)
def command_bat(*args, **kwargs):
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
def command_cd(*args, **kwargs):
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
def command_cat(*args, **kwargs):
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
def command_close(self, arguments, command):
300    def command_close(self, arguments, command):
301        # Command arguments required   : No
302        # Active SMB connection needed : No
303        # SMB share needed             : No
304
305        self.smbSession.ping_smb_session()
306        if self.smbSession.connected:
307            self.smbSession.close_smb_session()
def command_get(*args, **kwargs):
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
def command_help(self, arguments, command):
332    def command_help(self, arguments, command):
333        # Command arguments required   : No
334        # Active SMB connection needed : No
335        # SMB share needed             : No
336
337        if len(arguments) != 0:
338            self.commandCompleterObject.print_help(command=arguments[0])
339        else:
340            self.commandCompleterObject.print_help(command=None)
def command_info(*args, **kwargs):
39    def wrapper(*args, **kwargs):
40        self, arguments,command  = args[0], args[1], args[2]
41        #
42        self.smbSession.ping_smb_session()
43        if self.smbSession.connected:
44            return func(*args, **kwargs)
45        else:
46            print("[!] SMB Session is disconnected.")
47            return None
def command_lcd(*args, **kwargs):
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
def command_lls(self, arguments, command):
380    def command_lls(self, arguments, command):
381        # Command arguments required   : No
382        # Active SMB connection needed : No
383        # SMB share needed             : No
384
385        if len(arguments) == 0:
386            path = '.'
387        else:
388            path = ' '.join(arguments)
389
390        # lls <directory>
391        if os.path.isdir(path):
392            directory_contents = os.listdir(path=path)
393            for entryname in sorted(directory_contents):
394                path_to_file = path + os.path.sep + entryname
395                rights_str = unix_permissions(path_to_file)
396                size_str = b_filesize(os.path.getsize(filename=path_to_file))
397                date_str = datetime.datetime.fromtimestamp(os.path.getmtime(filename=path_to_file)).strftime("%Y-%m-%d %H:%M")
398
399                if os.path.isdir(s=entryname):
400                    if self.config.no_colors:
401                        print("%s %10s  %s  %s%s" % (rights_str, size_str, date_str, entryname, os.path.sep))
402                    else:
403                        print("%s %10s  %s  \x1b[1;96m%s\x1b[0m%s" % (rights_str, size_str, date_str, entryname, os.path.sep))
404                else:
405                    if self.config.no_colors:
406                        print("%s %10s  %s  %s" % (rights_str, size_str, date_str, entryname))
407                    else:
408                        print("%s %10s  %s  \x1b[1m%s\x1b[0m" % (rights_str, size_str, date_str, entryname))
409        # lls <file>
410        elif os.path.isfile(path):
411            rights_str = unix_permissions(path)
412            size_str = b_filesize(os.path.getsize(filename=path))
413            date_str = datetime.datetime.fromtimestamp(os.path.getmtime(filename=path)).strftime("%Y-%m-%d %H:%M")
414            if self.config.no_colors:
415                print("%s %10s  %s  %s" % (rights_str, size_str, date_str, os.path.basename(path)))
416            else:
417               print("%s %10s  %s  \x1b[1m%s\x1b[0m" % (rights_str, size_str, date_str, os.path.basename(path))) 
418        else:
419            print("[!] No such file or directory.")
def command_lmkdir(*args, **kwargs):
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
def command_lpwd(self, arguments, command):
441    def command_lpwd(self, arguments, command):
442        # Command arguments required   : No
443        # Active SMB connection needed : No
444        # SMB share needed             : No
445
446        print(os.getcwd())
def command_lrm(*args, **kwargs):
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
def command_lrmdir(*args, **kwargs):
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
def command_ltree(self, arguments, command):
484    def command_ltree(self, arguments, command):
485        # Command arguments required   : No
486        # Active SMB connection needed : No
487        # SMB share needed             : No
488
489        if len(arguments) == 0:
490            local_tree(path='.', config=self.config)
491        else:
492            local_tree(path=' '.join(arguments), config=self.config)
def command_ls(*args, **kwargs):
39    def wrapper(*args, **kwargs):
40        self, arguments,command  = args[0], args[1], args[2]
41        #
42        self.smbSession.ping_smb_session()
43        if self.smbSession.connected:
44            return func(*args, **kwargs)
45        else:
46            print("[!] SMB Session is disconnected.")
47            return None
def command_mkdir(*args, **kwargs):
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
def command_module(*args, **kwargs):
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
def command_put(*args, **kwargs):
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
def command_reconnect(self, arguments, command):
554    def command_reconnect(self, arguments, command):
555        # Command arguments required   : No
556        # Active SMB connection needed : No
557        # SMB share needed             : No
558
559        self.smbSession.ping_smb_session()
560        if self.smbSession.connected:
561            self.smbSession.close_smb_session()
562            self.smbSession.init_smb_session()
563        else:
564            self.smbSession.init_smb_session()
def command_reset(self, arguments, command):
566    def command_reset(self, arguments, command):
567        # Command arguments required   : No
568        # Active SMB connection needed : No
569        # SMB share needed             : No
570        sys.stdout.write('\x1b[?25h') # Sets the cursor to on
571        sys.stdout.write('\x1b[v')  
572        sys.stdout.write('\x1b[o') # Reset
573        sys.stdout.flush()
def command_rm(*args, **kwargs):
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
def command_rmdir(*args, **kwargs):
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
def command_shares(*args, **kwargs):
39    def wrapper(*args, **kwargs):
40        self, arguments,command  = args[0], args[1], args[2]
41        #
42        self.smbSession.ping_smb_session()
43        if self.smbSession.connected:
44            return func(*args, **kwargs)
45        else:
46            print("[!] SMB Session is disconnected.")
47            return None
def command_tree(*args, **kwargs):
39    def wrapper(*args, **kwargs):
40        self, arguments,command  = args[0], args[1], args[2]
41        #
42        self.smbSession.ping_smb_session()
43        if self.smbSession.connected:
44            return func(*args, **kwargs)
45        else:
46            print("[!] SMB Session is disconnected.")
47            return None
def command_use(*args, **kwargs):
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