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