smbclientng.core.SMBSession
1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# File name : smbclient-ng.py 4# Author : Podalirius (@podalirius_) 5# Date created : 20 may 2024 6 7 8import io 9import impacket.smbconnection 10import ntpath 11import os 12import re 13import traceback 14from smbclientng.core.LocalFileIO import LocalFileIO 15from smbclientng.core.utils import b_filesize, STYPE_MASK 16 17 18class SMBSession(object): 19 """ 20 Class SMBSession is designed to handle the session management for SMB (Server Message Block) protocol connections. 21 It provides functionalities to connect to an SMB server, authenticate using either NTLM or Kerberos, and manage SMB shares. 22 23 Attributes: 24 address (str): The IP address or hostname of the SMB server. 25 domain (str): The domain name for SMB server authentication. 26 username (str): The username for SMB server authentication. 27 password (str): The password for SMB server authentication. 28 lmhash (str): The LM hash of the user's password, if available. 29 nthash (str): The NT hash of the user's password, if available. 30 use_kerberos (bool): A flag to determine whether to use Kerberos for authentication. 31 kdcHost (str): The Key Distribution Center (KDC) host for Kerberos authentication. 32 debug (bool): A flag to enable debug output. 33 smbClient (object): The SMB client object used for the connection. 34 connected (bool): A flag to check the status of the connection. 35 smb_share (str): The current SMB share in use. 36 smb_path (str): The current path within the SMB share. 37 38 Methods: 39 __init__(address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, debug=False): 40 Initializes the SMBSession with the specified parameters. 41 init_smb_session(): 42 Initializes the SMB session by connecting to the server and authenticating using the specified method. 43 """ 44 45 def __init__(self, address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None): 46 super(SMBSession, self).__init__() 47 # Objects 48 self.config = config 49 50 # Target server 51 self.address = address 52 53 # Credentials 54 self.domain = domain 55 self.username = username 56 self.password = password 57 self.lmhash = lmhash 58 self.nthash = nthash 59 self.use_kerberos = use_kerberos 60 self.kdcHost = kdcHost 61 62 self.smbClient = None 63 self.connected = False 64 65 self.available_shares = {} 66 self.smb_share = None 67 self.smb_cwd = "" 68 69 self.list_shares() 70 71 # Connect and disconnect SMB session 72 73 def init_smb_session(self): 74 """ 75 Initializes and establishes a session with the SMB server. 76 77 This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration. 78 It attempts to connect to the SMB server specified by the `address` attribute and authenticate using the credentials provided during the object's initialization. 79 80 The method will print debug information if the `debug` attribute is set to True. Upon successful connection and authentication, it sets the `connected` attribute to True. 81 82 Returns: 83 bool: True if the connection and authentication are successful, False otherwise. 84 """ 85 86 self.connected = False 87 88 if self.config.debug: 89 print("[debug] [>] Connecting to remote SMB server '%s' ... " % self.address) 90 try: 91 self.smbClient = impacket.smbconnection.SMBConnection( 92 remoteName=self.address, 93 remoteHost=self.address, 94 sess_port=int(445) 95 ) 96 except OSError as err: 97 print("[!] %s" % err) 98 self.smbClient = None 99 100 if self.smbClient is not None: 101 if self.use_kerberos: 102 if self.config.debug: 103 print("[debug] [>] Authenticating as '%s\\%s' with kerberos ... " % (self.domain, self.username)) 104 try: 105 self.connected = self.smbClient.kerberosLogin( 106 user=self.username, 107 password=self.password, 108 domain=self.domain, 109 lmhash=self.lmhash, 110 nthash=self.nthash, 111 aesKey=self.aesKey, 112 kdcHost=self.kdcHost 113 ) 114 except impacket.smbconnection.SessionError as err: 115 if self.config.debug: 116 traceback.print_exc() 117 print("[!] Could not login: %s" % err) 118 self.connected = False 119 120 else: 121 if self.config.debug: 122 print("[debug] [>] Authenticating as '%s\\%s' with NTLM ... " % (self.domain, self.username)) 123 try: 124 self.connected = self.smbClient.login( 125 user=self.username, 126 password=self.password, 127 domain=self.domain, 128 lmhash=self.lmhash, 129 nthash=self.nthash 130 ) 131 except impacket.smbconnection.SessionError as err: 132 if self.config.debug: 133 traceback.print_exc() 134 print("[!] Could not login: %s" % err) 135 self.connected = False 136 137 if self.connected: 138 print("[+] Successfully authenticated to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username)) 139 else: 140 print("[!] Failed to authenticate to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username)) 141 142 return self.connected 143 144 def close_smb_session(self): 145 """ 146 Closes the current SMB session by disconnecting the SMB client. 147 148 This method ensures that the SMB client connection is properly closed. It checks if the client is connected 149 and if so, it closes the connection and resets the connection status. 150 151 Raises: 152 Exception: If the SMB client is not initialized or if there's an error during the disconnection process. 153 """ 154 155 if self.smbClient is not None: 156 if self.connected: 157 self.smbClient.close() 158 self.connected = False 159 if self.config.debug: 160 print("[+] SMB connection closed successfully.") 161 else: 162 if self.config.debug: 163 print("[!] No active SMB connection to close.") 164 else: 165 raise Exception("SMB client is not initialized.") 166 167 # Operations 168 169 def read_file(self, path=None): 170 if self.path_isfile(path=path): 171 tmp_file_path = self.smb_cwd + ntpath.sep + path 172 matches = self.smbClient.listPath( 173 shareName=self.smb_share, 174 path=tmp_file_path 175 ) 176 177 fh = io.BytesIO() 178 try: 179 # opening the files in streams instead of mounting shares allows 180 # for running the script from unprivileged containers 181 self.smbClient.getFile(self.smb_share, tmp_file_path, fh.write) 182 except impacket.smbconnection.SessionError as e: 183 return None 184 rawdata = fh.getvalue() 185 fh.close() 186 return rawdata 187 else: 188 print("[!] Remote path '%s' is not a file." % path) 189 190 def find(self, paths=[], callback=None): 191 def recurse_action(paths=[], depth=0, callback=None): 192 if callback is None: 193 return [] 194 next_directories_to_explore = [] 195 for path in paths: 196 remote_smb_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path) 197 entries = [] 198 199 try: 200 entries = self.smbClient.listPath( 201 shareName=self.smb_share, 202 path=(remote_smb_path + ntpath.sep + '*') 203 ) 204 except impacket.smbconnection.SessionError as err: 205 continue 206 # Remove dot names 207 entries = [e for e in entries if e.get_longname() not in [".", ".."]] 208 # Sort the entries ignoring case 209 entries = sorted(entries, key=lambda x:x.get_longname().lower()) 210 211 for entry in entries: 212 if entry.is_directory(): 213 callback(entry, path + entry.get_longname() + ntpath.sep, depth) 214 else: 215 callback(entry, path + entry.get_longname(), depth) 216 217 # Next directories to explore 218 for entry in entries: 219 if entry.is_directory(): 220 next_directories_to_explore.append(path + entry.get_longname() + ntpath.sep) 221 222 return next_directories_to_explore 223 # 224 if callback is not None: 225 depth = 0 226 while len(paths) != 0: 227 paths = recurse_action( 228 paths=paths, 229 depth=depth, 230 callback=callback 231 ) 232 depth = depth + 1 233 else: 234 print("[!] SMBSession.find(), callback function cannot be None.") 235 236 def get_file(self, path=None, keepRemotePath=False): 237 """ 238 Retrieves a file from the specified path on the SMB share. 239 240 This method attempts to retrieve a file from the given path within the currently connected SMB share. 241 If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local 242 file object and writing the contents of the remote file to it using the SMB client's getFile method. 243 244 Parameters: 245 path (str): The path of the file to retrieve. If None, uses the current smb_path. 246 247 Returns: 248 None 249 """ 250 251 tmp_file_path = self.smb_cwd + ntpath.sep + path 252 matches = self.smbClient.listPath( 253 shareName=self.smb_share, 254 path=tmp_file_path 255 ) 256 257 for entry in matches: 258 if entry.is_directory(): 259 print("[>] Skipping '%s' because it is a directory." % tmp_file_path) 260 else: 261 try: 262 if ntpath.sep in path: 263 outputfile = ntpath.dirname(path) + ntpath.sep + entry.get_longname() 264 else: 265 outputfile = entry.get_longname() 266 f = LocalFileIO( 267 mode="wb", 268 path=outputfile, 269 expected_size=entry.get_filesize(), 270 debug=self.config.debug, 271 keepRemotePath=keepRemotePath 272 ) 273 self.smbClient.getFile( 274 shareName=self.smb_share, 275 pathName=tmp_file_path, 276 callback=f.write 277 ) 278 f.close() 279 except (BrokenPipeError, KeyboardInterrupt) as e: 280 f.close() 281 print("\x1b[v\x1b[o\r[!] Interrupted.") 282 self.close_smb_session() 283 self.init_smb_session() 284 285 return None 286 287 def get_file_recursively(self, path=None): 288 """ 289 Recursively retrieves files from a specified path on the SMB share. 290 291 This method navigates through all directories starting from the given path, 292 and downloads all files found. It handles directories recursively, ensuring 293 that all nested files are retrieved. The method skips over directory entries 294 and handles errors gracefully, attempting to continue the operation where possible. 295 296 Parameters: 297 path (str): The initial directory path from which to start the recursive file retrieval. 298 If None, it starts from the root of the configured SMB share. 299 """ 300 301 def recurse_action(base_dir="", path=[]): 302 remote_smb_path = base_dir + ntpath.sep.join(path) 303 entries = self.smbClient.listPath( 304 shareName=self.smb_share, 305 path=remote_smb_path + '\\*' 306 ) 307 if len(entries) != 0: 308 files = [entry for entry in entries if not entry.is_directory()] 309 directories = [entry for entry in entries if entry.is_directory() and entry.get_longname() not in [".", ".."]] 310 311 # Files 312 if len(files) != 0: 313 print("[>] Retrieving files of '%s'" % remote_smb_path) 314 for entry_file in files: 315 if not entry_file.is_directory(): 316 f = LocalFileIO( 317 mode="wb", 318 path=remote_smb_path + ntpath.sep + entry_file.get_longname(), 319 expected_size=entry_file.get_filesize(), 320 debug=self.config.debug 321 ) 322 try: 323 self.smbClient.getFile( 324 shareName=self.smb_share, 325 pathName=remote_smb_path + ntpath.sep + entry_file.get_longname(), 326 callback=f.write 327 ) 328 f.close() 329 except BrokenPipeError as err: 330 f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err)) 331 f.close(remove=True) 332 break 333 except Exception as err: 334 f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err)) 335 f.close(remove=True) 336 337 # Directories 338 for entry_directory in directories: 339 if entry_directory.is_directory(): 340 recurse_action( 341 base_dir=self.smb_cwd, 342 path=path+[entry_directory.get_longname()] 343 ) 344 # Entrypoint 345 try: 346 recurse_action( 347 base_dir=self.smb_cwd, 348 path=[path] 349 ) 350 except (BrokenPipeError, KeyboardInterrupt) as e: 351 print("\x1b[v\x1b[o\r[!] Interrupted.") 352 self.close_smb_session() 353 self.init_smb_session() 354 355 def info(self, share=True, server=True): 356 """ 357 Displays information about the server and optionally the shares. 358 359 This method prints detailed information about the server's characteristics such as NetBIOS names, DNS details, OS information, and SMB capabilities. If the `share` parameter is set to True and a share is currently set, it will also attempt to display information about the share. 360 361 Parameters: 362 share (bool): If True, display information about the current share. 363 server (bool): If True, display information about the server. 364 365 Returns: 366 None 367 """ 368 369 if server: 370 if self.config.no_colors: 371 print("[+] Server:") 372 print(" ├─NetBIOS:") 373 print(" │ ├─ NetBIOS Hostname ──────── : %s" % (self.smbClient.getServerName())) 374 print(" │ └─ NetBIOS Domain ────────── : %s" % (self.smbClient.getServerDomain())) 375 print(" ├─DNS:") 376 print(" │ ├─ DNS Hostname ──────────── : %s" % (self.smbClient.getServerDNSHostName())) 377 print(" │ └─ DNS Domain ────────────── : %s" % (self.smbClient.getServerDNSDomainName())) 378 print(" ├─OS:") 379 print(" │ ├─ OS Name ───────────────── : %s" % (self.smbClient.getServerOS())) 380 print(" │ └─ OS Version ────────────── : %s.%s.%s" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild())) 381 print(" ├─Server:") 382 print(" │ ├─ Signing Required ──────── : %s" % (self.smbClient.isSigningRequired())) 383 print(" │ ├─ Login Required ────────── : %s" % (self.smbClient.isLoginRequired())) 384 print(" │ ├─ Supports NTLMv2 ───────── : %s" % (self.smbClient.doesSupportNTLMv2())) 385 MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"] 386 print(" │ ├─ Max size of read chunk ── : %d bytes (%s)" % (MaxReadSize, b_filesize(MaxReadSize))) 387 MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"] 388 print(" │ └─ Max size of write chunk ─ : %d bytes (%s)" % (MaxWriteSize, b_filesize(MaxWriteSize))) 389 print(" └─") 390 else: 391 print("[+] Server:") 392 print(" ├─NetBIOS:") 393 print(" │ ├─ \x1b[94mNetBIOS Hostname\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerName())) 394 print(" │ └─ \x1b[94mNetBIOS Domain\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDomain())) 395 print(" ├─DNS:") 396 print(" │ ├─ \x1b[94mDNS Hostname\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSHostName())) 397 print(" │ └─ \x1b[94mDNS Domain\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSDomainName())) 398 print(" ├─OS:") 399 print(" │ ├─ \x1b[94mOS Name\x1b[0m \x1b[90m─────────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerOS())) 400 print(" │ └─ \x1b[94mOS Version\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s.%s.%s\x1b[0m" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild())) 401 print(" ├─Server:") 402 print(" │ ├─ \x1b[94mSigning Required\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isSigningRequired())) 403 print(" │ ├─ \x1b[94mLogin Required\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isLoginRequired())) 404 print(" │ ├─ \x1b[94mSupports NTLMv2\x1b[0m \x1b[90m─────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.doesSupportNTLMv2())) 405 MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"] 406 print(" │ ├─ \x1b[94mMax size of read chunk\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxReadSize, b_filesize(MaxReadSize))) 407 MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"] 408 print(" │ └─ \x1b[94mMax size of write chunk\x1b[0m \x1b[90m─\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxWriteSize, b_filesize(MaxWriteSize))) 409 print(" └─") 410 411 if share and self.smb_share is not None: 412 share_name = self.available_shares.get(self.smb_share.lower(), "")["name"] 413 share_comment = self.available_shares.get(self.smb_share.lower(), "")["comment"] 414 share_type = self.available_shares.get(self.smb_share.lower(), "")["type"] 415 share_type =', '.join([s.replace("STYPE_","") for s in share_type]) 416 share_rawtype = self.available_shares.get(self.smb_share.lower(), "")["rawtype"] 417 if self.config.no_colors: 418 print("\n[+] Share:") 419 print(" ├─ Name ──────────── : %s" % (share_name)) 420 print(" ├─ Description ───── : %s" % (share_comment)) 421 print(" ├─ Type ──────────── : %s" % (share_type)) 422 print(" └─ Raw type value ── : %s" % (share_rawtype)) 423 else: 424 print("\n[+] Share:") 425 print(" ├─ \x1b[94mName\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_name)) 426 print(" ├─ \x1b[94mDescription\x1b[0m \x1b[90m─────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_comment)) 427 print(" ├─ \x1b[94mType\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_type)) 428 print(" └─ \x1b[94mRaw type value\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%s\x1b[0m" % (share_rawtype)) 429 430 def list_contents(self, path=None): 431 """ 432 Lists the contents of a specified directory on the SMB share. 433 434 This method retrieves the contents of a directory specified by `shareName` and `path`. If `shareName` or `path` 435 is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with 436 the long names of the files and directories as keys and their respective SMB entry objects as values. 437 438 Args: 439 shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None. 440 path (str, optional): The directory path to list contents from. Defaults to the current path if None. 441 442 Returns: 443 dict: A dictionary with file and directory names as keys and their SMB entry objects as values. 444 """ 445 446 if path is None or len(path) == 0: 447 path = self.smb_cwd 448 path = path.rstrip(ntpath.sep) + ntpath.sep + "*" 449 450 contents = {} 451 entries = self.smbClient.listPath( 452 shareName=self.smb_share, 453 path=path 454 ) 455 for entry in entries: 456 contents[entry.get_longname()] = entry 457 458 return contents 459 460 def list_shares(self): 461 """ 462 Lists all the shares available on the connected SMB server. 463 464 This method queries the SMB server to retrieve a list of all available shares. It populates the `shares` dictionary 465 with key-value pairs where the key is the share name and the value is a dictionary containing details about the share 466 such as its name, type, raw type, and any comments associated with the share. 467 468 Returns: 469 dict: A dictionary containing information about each share available on the server. 470 """ 471 472 self.available_shares = {} 473 474 if self.connected: 475 if self.smbClient is not None: 476 resp = self.smbClient.listShares() 477 478 for share in resp: 479 # SHARE_INFO_1 structure (lmshare.h) 480 # https://learn.microsoft.com/en-us/windows/win32/api/lmshare/ns-lmshare-share_info_1 481 sharename = share["shi1_netname"][:-1] 482 sharecomment = share["shi1_remark"][:-1] 483 sharetype = share["shi1_type"] 484 485 self.available_shares[sharename.lower()] = { 486 "name": sharename, 487 "type": STYPE_MASK(sharetype), 488 "rawtype": sharetype, 489 "comment": sharecomment 490 } 491 else: 492 print("[!] Error: SMBSession.smbClient is None.") 493 494 return self.available_shares 495 496 def mkdir(self, path=None): 497 """ 498 Creates a directory at the specified path on the SMB share. 499 500 This method takes a path and attempts to create the directory structure on the SMB share. If the path includes 501 nested directories, it will create each directory in the sequence. If a directory already exists, it will skip 502 the creation for that directory without raising an error. 503 504 Args: 505 path (str, optional): The full path of the directory to create on the SMB share. Defaults to None. 506 507 Note: 508 The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility. 509 """ 510 511 if path is not None: 512 # Prepare path 513 path = path.replace('/',ntpath.sep) 514 if ntpath.sep in path: 515 path = path.strip(ntpath.sep).split(ntpath.sep) 516 else: 517 path = [path] 518 519 # Create each dir in the path 520 for depth in range(1, len(path)+1): 521 tmp_path = ntpath.sep.join(path[:depth]) 522 try: 523 self.smbClient.createDirectory( 524 shareName=self.smb_share, 525 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + tmp_path + ntpath.sep) 526 ) 527 except impacket.smbconnection.SessionError as err: 528 if err.getErrorCode() == 0xc0000035: 529 # STATUS_OBJECT_NAME_COLLISION 530 # Remote directory already created, this is normal 531 # Src: https://github.com/fortra/impacket/blob/269ce69872f0e8f2188a80addb0c39fedfa6dcb8/impacket/nt_errors.py#L268C9-L268C19 532 pass 533 else: 534 print("[!] Failed to create directory '%s': %s" % (tmp_path, err)) 535 if self.config.debug: 536 traceback.print_exc() 537 else: 538 pass 539 540 def path_exists(self, path=None): 541 """ 542 Checks if the specified path exists on the SMB share. 543 544 This method determines if a given path exists on the SMB share by attempting to list the contents of the path. 545 If the path listing is successful and returns one or more entries, the path is considered to exist. 546 547 Args: 548 path (str, optional): The path to check on the SMB share. Defaults to None. 549 550 Returns: 551 bool: True if the path exists, False otherwise or if an error occurs. 552 """ 553 554 if path is not None: 555 path = path.replace('*','') 556 try: 557 contents = self.smbClient.listPath( 558 shareName=self.smb_share, 559 path=ntpath.normpath(self.smb_cwd + ntpath.sep + path + ntpath.sep) 560 ) 561 return (len(contents) != 0) 562 except Exception as e: 563 return False 564 else: 565 return False 566 567 def path_isdir(self, pathFromRoot=None): 568 """ 569 Checks if the specified path is a directory on the SMB share. 570 571 This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the 572 contents of the path and filtering for entries that match the basename of the path and are marked as directories. 573 574 Args: 575 path (str, optional): The path to check on the SMB share. Defaults to None. 576 577 Returns: 578 bool: True if the path is a directory, False otherwise or if an error occurs. 579 """ 580 581 if pathFromRoot is not None: 582 # Replace slashes if any 583 path = pathFromRoot.replace('/', ntpath.sep) 584 585 # Strip wildcards to avoid injections 586 path = path.replace('*','') 587 588 # Normalize path and strip leading backslash 589 path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep) 590 591 if path.strip() in ['', '.', '..']: 592 # By defininition they exist on the filesystem 593 return True 594 else: 595 try: 596 contents = self.smbClient.listPath( 597 shareName=self.smb_share, 598 path=path+'*' 599 ) 600 # Filter on directories 601 contents = [ 602 c for c in contents 603 if c.get_longname() == ntpath.basename(path) and c.is_directory() 604 ] 605 return (len(contents) != 0) 606 except Exception as e: 607 return False 608 else: 609 return False 610 611 def path_isfile(self, path=None): 612 """ 613 Checks if the specified path is a file on the SMB share. 614 615 This method determines if a given path corresponds to a file on the SMB share. It does this by listing the 616 contents of the path and filtering for entries that match the basename of the path and are not marked as directories. 617 618 Args: 619 path (str, optional): The path to check on the SMB share. Defaults to None. 620 621 Returns: 622 bool: True if the path is a file, False otherwise or if an error occurs. 623 """ 624 625 if path is not None: 626 path = path.replace('*','') 627 search_dir = ntpath.normpath(self.smb_cwd + ntpath.sep + path) 628 search_dir = ntpath.dirname(search_dir) + ntpath.sep + '*' 629 try: 630 contents = self.smbClient.listPath( 631 shareName=self.smb_share, 632 path=search_dir 633 ) 634 # Filter on files 635 contents = [ 636 c for c in contents 637 if c.get_longname() == ntpath.basename(path) and not c.is_directory() 638 ] 639 return (len(contents) != 0) 640 except Exception as e: 641 return False 642 else: 643 return False 644 645 def ping_smb_session(self): 646 """ 647 Tests the connectivity to the SMB server by sending an echo command. 648 649 This method attempts to send an echo command to the SMB server to check if the session is still active. 650 It updates the `connected` attribute of the class based on the success or failure of the echo command. 651 652 Returns: 653 bool: True if the echo command succeeds (indicating the session is active), False otherwise. 654 """ 655 656 try: 657 self.smbClient.getSMBServer().echo() 658 except Exception as e: 659 self.connected = False 660 return self.connected 661 662 def put_file(self, localpath=None): 663 """ 664 Uploads a single file to the SMB share. 665 666 This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path. 667 It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session. 668 General exceptions are caught and logged, with a traceback provided if debugging is enabled. 669 670 Args: 671 localpath (str, optional): The local file path of the file to be uploaded. Defaults to None. 672 """ 673 674 if os.path.exists(localpath): 675 if os.path.isfile(localpath): 676 try: 677 localfile = os.path.basename(localpath) 678 f = LocalFileIO( 679 mode="rb", 680 path=localpath, 681 debug=self.config.debug 682 ) 683 self.smbClient.putFile( 684 shareName=self.smb_share, 685 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + localfile + ntpath.sep), 686 callback=f.read 687 ) 688 f.close() 689 except (BrokenPipeError, KeyboardInterrupt) as err: 690 print("[!] Interrupted.") 691 self.close_smb_session() 692 self.init_smb_session() 693 except Exception as err: 694 print("[!] Failed to upload '%s': %s" % (localfile, err)) 695 if self.config.debug: 696 traceback.print_exc() 697 else: 698 print("[!] The specified localpath is a directory. Use 'put -r <directory>' instead.") 699 else: 700 print("[!] The specified localpath does not exist.") 701 702 def put_file_recursively(self, localpath=None): 703 """ 704 Recursively uploads files from a specified local directory to the SMB share. 705 706 This method walks through the given local directory and all its subdirectories, uploading each file to the 707 corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is, 708 it iterates over all files and directories within the local path, creating necessary directories on the SMB share 709 and uploading files. If the local path is not a directory, it prints an error message. 710 711 Args: 712 localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None. 713 """ 714 715 if os.path.exists(localpath): 716 if os.path.isfile(localpath): 717 # Iterate over all files and directories within the local path 718 local_files = {} 719 for root, dirs, files in os.walk(localpath): 720 if len(files) != 0: 721 local_files[root] = files 722 723 # Iterate over the found files 724 for local_dir_path in sorted(local_files.keys()): 725 print("[>] Putting files of '%s'" % local_dir_path) 726 727 # Create remote directory 728 remote_dir_path = local_dir_path.replace(os.path.sep, ntpath.sep) 729 self.mkdir( 730 path=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep) 731 ) 732 733 for local_file_path in local_files[local_dir_path]: 734 try: 735 f = LocalFileIO( 736 mode="rb", 737 path=local_dir_path + os.path.sep + local_file_path, 738 debug=self.config.debug 739 ) 740 self.smbClient.putFile( 741 shareName=self.smb_share, 742 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep + local_file_path), 743 callback=f.read 744 ) 745 f.close() 746 747 except BrokenPipeError as err: 748 f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err)) 749 f.close(remove=True) 750 break 751 except Exception as err: 752 f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err)) 753 f.close(remove=True) 754 else: 755 print("[!] The specified localpath is a file. Use 'put <file>' instead.") 756 else: 757 print("[!] The specified localpath does not exist.") 758 759 def rmdir(self, path=None): 760 """ 761 Removes a directory from the SMB share at the specified path. 762 763 This method attempts to delete a directory located at the given path on the SMB share. If the operation fails, 764 it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints 765 the stack trace of the exception. 766 767 Args: 768 path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None. 769 """ 770 try: 771 self.smbClient.deleteDirectory( 772 shareName=self.smb_share, 773 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 774 ) 775 except Exception as err: 776 print("[!] Failed to remove directory '%s': %s" % (path, err)) 777 if self.config.debug: 778 traceback.print_exc() 779 780 def rm(self, path=None): 781 """ 782 Removes a file from the SMB share at the specified path. 783 784 This method attempts to delete a file located at the given path on the SMB share. If the operation fails, 785 it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints 786 the stack trace of the exception. 787 788 Args: 789 path (str, optional): The path of the file to be removed on the SMB share. Defaults to None. 790 """ 791 try: 792 self.smbClient.deleteFile( 793 shareName=self.smb_share, 794 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 795 ) 796 except Exception as err: 797 print("[!] Failed to remove file '%s': %s" % (path, err)) 798 if self.config.debug: 799 traceback.print_exc() 800 801 def tree(self, path=None): 802 """ 803 Recursively lists the directory structure of the SMB share starting from the specified path. 804 805 This function prints a visual representation of the directory tree of the remote SMB share. It uses 806 recursion to navigate through directories and lists all files and subdirectories in each directory. 807 The output is color-coded and formatted to enhance readability, with directories highlighted in cyan. 808 809 Args: 810 path (str, optional): The starting path on the SMB share from which to begin listing the tree. 811 Defaults to the root of the current share. 812 """ 813 814 def recurse_action(base_dir="", path=[], prompt=[]): 815 bars = ["│ ", "├── ", "└── "] 816 817 remote_smb_path = ntpath.normpath(base_dir + ntpath.sep + ntpath.sep.join(path)) 818 819 entries = [] 820 try: 821 entries = self.smbClient.listPath( 822 shareName=self.smb_share, 823 path=remote_smb_path+'\\*' 824 ) 825 except impacket.smbconnection.SessionError as err: 826 code, const, text = err.getErrorCode(), err.getErrorString()[0], err.getErrorString()[1] 827 errmsg = "Error 0x%08x (%s): %s" % (code, const, text) 828 if self.config.no_colors: 829 print("%s%s" % (''.join(prompt+[bars[2]]), errmsg)) 830 else: 831 print("%s\x1b[1;91m%s\x1b[0m" % (''.join(prompt+[bars[2]]), errmsg)) 832 return 833 834 entries = [e for e in entries if e.get_longname() not in [".", ".."]] 835 entries = sorted(entries, key=lambda x:x.get_longname()) 836 837 # 838 if len(entries) > 1: 839 index = 0 840 for entry in entries: 841 index += 1 842 # This is the first entry 843 if index == 0: 844 if entry.is_directory(): 845 if self.config.no_colors: 846 print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 847 else: 848 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 849 recurse_action( 850 base_dir=base_dir, 851 path=path+[entry.get_longname()], 852 prompt=prompt+["│ "] 853 ) 854 else: 855 if self.config.no_colors: 856 print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname())) 857 else: 858 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname())) 859 860 # This is the last entry 861 elif index == len(entries): 862 if entry.is_directory(): 863 if self.config.no_colors: 864 print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 865 else: 866 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 867 recurse_action( 868 base_dir=base_dir, 869 path=path+[entry.get_longname()], 870 prompt=prompt+[" "] 871 ) 872 else: 873 if self.config.no_colors: 874 print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname())) 875 else: 876 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname())) 877 878 # These are entries in the middle 879 else: 880 if entry.is_directory(): 881 if self.config.no_colors: 882 print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 883 else: 884 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 885 recurse_action( 886 base_dir=base_dir, 887 path=path+[entry.get_longname()], 888 prompt=prompt+["│ "] 889 ) 890 else: 891 if self.config.no_colors: 892 print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname())) 893 else: 894 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname())) 895 896 # 897 elif len(entries) == 1: 898 entry = entries[0] 899 if entry.is_directory(): 900 if self.config.no_colors: 901 print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 902 else: 903 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 904 recurse_action( 905 base_dir=base_dir, 906 path=path+[entry.get_longname()], 907 prompt=prompt+[" "] 908 ) 909 else: 910 if self.config.no_colors: 911 print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname())) 912 else: 913 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname())) 914 915 # Entrypoint 916 try: 917 if self.config.no_colors: 918 print("%s\\" % path) 919 else: 920 print("\x1b[1;96m%s\x1b[0m\\" % path) 921 recurse_action( 922 base_dir=self.smb_cwd, 923 path=[path], 924 prompt=[""] 925 ) 926 except (BrokenPipeError, KeyboardInterrupt) as e: 927 print("[!] Interrupted.") 928 self.close_smb_session() 929 self.init_smb_session() 930 931 # Setter / Getter 932 933 def set_share(self, shareName): 934 """ 935 Sets the current SMB share to the specified share name. 936 937 This method updates the SMB session to use the specified share name. It checks if the share name is valid 938 and updates the smb_share attribute of the SMBSession instance. 939 940 Parameters: 941 shareName (str): The name of the share to set as the current SMB share. 942 943 Raises: 944 ValueError: If the shareName is None or an empty string. 945 """ 946 947 if shareName is not None: 948 self.list_shares() 949 if shareName.lower() in self.available_shares.keys(): 950 # Doing this in order to keep the case of the share adevertised by the remote machine 951 self.smb_share = self.available_shares[shareName.lower()]["name"] 952 else: 953 print("[!] Could not set share '%s', it does not exist remotely." % shareName) 954 955 def set_cwd(self, path=None): 956 """ 957 Sets the current working directory on the SMB share to the specified path. 958 959 This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory. 960 If the specified path is not a directory, the cwd remains unchanged. 961 962 Parameters: 963 path (str): The path to set as the current working directory. 964 965 Raises: 966 ValueError: If the specified path is not a directory. 967 """ 968 969 if path is not None: 970 # Set path separators to ntpath sep 971 if '/' in path: 972 path = path.replace('/', ntpath.sep) 973 974 if path.startswith(ntpath.sep): 975 # Absolute path 976 path = path + ntpath.sep 977 else: 978 # Relative path to the CWD 979 if len(self.smb_cwd) == 0: 980 path = path + ntpath.sep 981 else: 982 path = self.smb_cwd + ntpath.sep + path 983 984 # Path normalization 985 path = ntpath.normpath(path) 986 path = re.sub(r'\\+', r'\\', path) 987 988 if path in ["", ".", ".."]: 989 self.smb_cwd = "" 990 else: 991 if self.path_isdir(pathFromRoot=path.strip(ntpath.sep)): 992 # Path exists on the remote 993 self.smb_cwd = ntpath.normpath(path) 994 else: 995 # Path does not exists or is not a directory on the remote 996 print("[!] Remote directory '%s' does not exist." % path)
19class SMBSession(object): 20 """ 21 Class SMBSession is designed to handle the session management for SMB (Server Message Block) protocol connections. 22 It provides functionalities to connect to an SMB server, authenticate using either NTLM or Kerberos, and manage SMB shares. 23 24 Attributes: 25 address (str): The IP address or hostname of the SMB server. 26 domain (str): The domain name for SMB server authentication. 27 username (str): The username for SMB server authentication. 28 password (str): The password for SMB server authentication. 29 lmhash (str): The LM hash of the user's password, if available. 30 nthash (str): The NT hash of the user's password, if available. 31 use_kerberos (bool): A flag to determine whether to use Kerberos for authentication. 32 kdcHost (str): The Key Distribution Center (KDC) host for Kerberos authentication. 33 debug (bool): A flag to enable debug output. 34 smbClient (object): The SMB client object used for the connection. 35 connected (bool): A flag to check the status of the connection. 36 smb_share (str): The current SMB share in use. 37 smb_path (str): The current path within the SMB share. 38 39 Methods: 40 __init__(address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, debug=False): 41 Initializes the SMBSession with the specified parameters. 42 init_smb_session(): 43 Initializes the SMB session by connecting to the server and authenticating using the specified method. 44 """ 45 46 def __init__(self, address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None): 47 super(SMBSession, self).__init__() 48 # Objects 49 self.config = config 50 51 # Target server 52 self.address = address 53 54 # Credentials 55 self.domain = domain 56 self.username = username 57 self.password = password 58 self.lmhash = lmhash 59 self.nthash = nthash 60 self.use_kerberos = use_kerberos 61 self.kdcHost = kdcHost 62 63 self.smbClient = None 64 self.connected = False 65 66 self.available_shares = {} 67 self.smb_share = None 68 self.smb_cwd = "" 69 70 self.list_shares() 71 72 # Connect and disconnect SMB session 73 74 def init_smb_session(self): 75 """ 76 Initializes and establishes a session with the SMB server. 77 78 This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration. 79 It attempts to connect to the SMB server specified by the `address` attribute and authenticate using the credentials provided during the object's initialization. 80 81 The method will print debug information if the `debug` attribute is set to True. Upon successful connection and authentication, it sets the `connected` attribute to True. 82 83 Returns: 84 bool: True if the connection and authentication are successful, False otherwise. 85 """ 86 87 self.connected = False 88 89 if self.config.debug: 90 print("[debug] [>] Connecting to remote SMB server '%s' ... " % self.address) 91 try: 92 self.smbClient = impacket.smbconnection.SMBConnection( 93 remoteName=self.address, 94 remoteHost=self.address, 95 sess_port=int(445) 96 ) 97 except OSError as err: 98 print("[!] %s" % err) 99 self.smbClient = None 100 101 if self.smbClient is not None: 102 if self.use_kerberos: 103 if self.config.debug: 104 print("[debug] [>] Authenticating as '%s\\%s' with kerberos ... " % (self.domain, self.username)) 105 try: 106 self.connected = self.smbClient.kerberosLogin( 107 user=self.username, 108 password=self.password, 109 domain=self.domain, 110 lmhash=self.lmhash, 111 nthash=self.nthash, 112 aesKey=self.aesKey, 113 kdcHost=self.kdcHost 114 ) 115 except impacket.smbconnection.SessionError as err: 116 if self.config.debug: 117 traceback.print_exc() 118 print("[!] Could not login: %s" % err) 119 self.connected = False 120 121 else: 122 if self.config.debug: 123 print("[debug] [>] Authenticating as '%s\\%s' with NTLM ... " % (self.domain, self.username)) 124 try: 125 self.connected = self.smbClient.login( 126 user=self.username, 127 password=self.password, 128 domain=self.domain, 129 lmhash=self.lmhash, 130 nthash=self.nthash 131 ) 132 except impacket.smbconnection.SessionError as err: 133 if self.config.debug: 134 traceback.print_exc() 135 print("[!] Could not login: %s" % err) 136 self.connected = False 137 138 if self.connected: 139 print("[+] Successfully authenticated to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username)) 140 else: 141 print("[!] Failed to authenticate to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username)) 142 143 return self.connected 144 145 def close_smb_session(self): 146 """ 147 Closes the current SMB session by disconnecting the SMB client. 148 149 This method ensures that the SMB client connection is properly closed. It checks if the client is connected 150 and if so, it closes the connection and resets the connection status. 151 152 Raises: 153 Exception: If the SMB client is not initialized or if there's an error during the disconnection process. 154 """ 155 156 if self.smbClient is not None: 157 if self.connected: 158 self.smbClient.close() 159 self.connected = False 160 if self.config.debug: 161 print("[+] SMB connection closed successfully.") 162 else: 163 if self.config.debug: 164 print("[!] No active SMB connection to close.") 165 else: 166 raise Exception("SMB client is not initialized.") 167 168 # Operations 169 170 def read_file(self, path=None): 171 if self.path_isfile(path=path): 172 tmp_file_path = self.smb_cwd + ntpath.sep + path 173 matches = self.smbClient.listPath( 174 shareName=self.smb_share, 175 path=tmp_file_path 176 ) 177 178 fh = io.BytesIO() 179 try: 180 # opening the files in streams instead of mounting shares allows 181 # for running the script from unprivileged containers 182 self.smbClient.getFile(self.smb_share, tmp_file_path, fh.write) 183 except impacket.smbconnection.SessionError as e: 184 return None 185 rawdata = fh.getvalue() 186 fh.close() 187 return rawdata 188 else: 189 print("[!] Remote path '%s' is not a file." % path) 190 191 def find(self, paths=[], callback=None): 192 def recurse_action(paths=[], depth=0, callback=None): 193 if callback is None: 194 return [] 195 next_directories_to_explore = [] 196 for path in paths: 197 remote_smb_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path) 198 entries = [] 199 200 try: 201 entries = self.smbClient.listPath( 202 shareName=self.smb_share, 203 path=(remote_smb_path + ntpath.sep + '*') 204 ) 205 except impacket.smbconnection.SessionError as err: 206 continue 207 # Remove dot names 208 entries = [e for e in entries if e.get_longname() not in [".", ".."]] 209 # Sort the entries ignoring case 210 entries = sorted(entries, key=lambda x:x.get_longname().lower()) 211 212 for entry in entries: 213 if entry.is_directory(): 214 callback(entry, path + entry.get_longname() + ntpath.sep, depth) 215 else: 216 callback(entry, path + entry.get_longname(), depth) 217 218 # Next directories to explore 219 for entry in entries: 220 if entry.is_directory(): 221 next_directories_to_explore.append(path + entry.get_longname() + ntpath.sep) 222 223 return next_directories_to_explore 224 # 225 if callback is not None: 226 depth = 0 227 while len(paths) != 0: 228 paths = recurse_action( 229 paths=paths, 230 depth=depth, 231 callback=callback 232 ) 233 depth = depth + 1 234 else: 235 print("[!] SMBSession.find(), callback function cannot be None.") 236 237 def get_file(self, path=None, keepRemotePath=False): 238 """ 239 Retrieves a file from the specified path on the SMB share. 240 241 This method attempts to retrieve a file from the given path within the currently connected SMB share. 242 If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local 243 file object and writing the contents of the remote file to it using the SMB client's getFile method. 244 245 Parameters: 246 path (str): The path of the file to retrieve. If None, uses the current smb_path. 247 248 Returns: 249 None 250 """ 251 252 tmp_file_path = self.smb_cwd + ntpath.sep + path 253 matches = self.smbClient.listPath( 254 shareName=self.smb_share, 255 path=tmp_file_path 256 ) 257 258 for entry in matches: 259 if entry.is_directory(): 260 print("[>] Skipping '%s' because it is a directory." % tmp_file_path) 261 else: 262 try: 263 if ntpath.sep in path: 264 outputfile = ntpath.dirname(path) + ntpath.sep + entry.get_longname() 265 else: 266 outputfile = entry.get_longname() 267 f = LocalFileIO( 268 mode="wb", 269 path=outputfile, 270 expected_size=entry.get_filesize(), 271 debug=self.config.debug, 272 keepRemotePath=keepRemotePath 273 ) 274 self.smbClient.getFile( 275 shareName=self.smb_share, 276 pathName=tmp_file_path, 277 callback=f.write 278 ) 279 f.close() 280 except (BrokenPipeError, KeyboardInterrupt) as e: 281 f.close() 282 print("\x1b[v\x1b[o\r[!] Interrupted.") 283 self.close_smb_session() 284 self.init_smb_session() 285 286 return None 287 288 def get_file_recursively(self, path=None): 289 """ 290 Recursively retrieves files from a specified path on the SMB share. 291 292 This method navigates through all directories starting from the given path, 293 and downloads all files found. It handles directories recursively, ensuring 294 that all nested files are retrieved. The method skips over directory entries 295 and handles errors gracefully, attempting to continue the operation where possible. 296 297 Parameters: 298 path (str): The initial directory path from which to start the recursive file retrieval. 299 If None, it starts from the root of the configured SMB share. 300 """ 301 302 def recurse_action(base_dir="", path=[]): 303 remote_smb_path = base_dir + ntpath.sep.join(path) 304 entries = self.smbClient.listPath( 305 shareName=self.smb_share, 306 path=remote_smb_path + '\\*' 307 ) 308 if len(entries) != 0: 309 files = [entry for entry in entries if not entry.is_directory()] 310 directories = [entry for entry in entries if entry.is_directory() and entry.get_longname() not in [".", ".."]] 311 312 # Files 313 if len(files) != 0: 314 print("[>] Retrieving files of '%s'" % remote_smb_path) 315 for entry_file in files: 316 if not entry_file.is_directory(): 317 f = LocalFileIO( 318 mode="wb", 319 path=remote_smb_path + ntpath.sep + entry_file.get_longname(), 320 expected_size=entry_file.get_filesize(), 321 debug=self.config.debug 322 ) 323 try: 324 self.smbClient.getFile( 325 shareName=self.smb_share, 326 pathName=remote_smb_path + ntpath.sep + entry_file.get_longname(), 327 callback=f.write 328 ) 329 f.close() 330 except BrokenPipeError as err: 331 f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err)) 332 f.close(remove=True) 333 break 334 except Exception as err: 335 f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err)) 336 f.close(remove=True) 337 338 # Directories 339 for entry_directory in directories: 340 if entry_directory.is_directory(): 341 recurse_action( 342 base_dir=self.smb_cwd, 343 path=path+[entry_directory.get_longname()] 344 ) 345 # Entrypoint 346 try: 347 recurse_action( 348 base_dir=self.smb_cwd, 349 path=[path] 350 ) 351 except (BrokenPipeError, KeyboardInterrupt) as e: 352 print("\x1b[v\x1b[o\r[!] Interrupted.") 353 self.close_smb_session() 354 self.init_smb_session() 355 356 def info(self, share=True, server=True): 357 """ 358 Displays information about the server and optionally the shares. 359 360 This method prints detailed information about the server's characteristics such as NetBIOS names, DNS details, OS information, and SMB capabilities. If the `share` parameter is set to True and a share is currently set, it will also attempt to display information about the share. 361 362 Parameters: 363 share (bool): If True, display information about the current share. 364 server (bool): If True, display information about the server. 365 366 Returns: 367 None 368 """ 369 370 if server: 371 if self.config.no_colors: 372 print("[+] Server:") 373 print(" ├─NetBIOS:") 374 print(" │ ├─ NetBIOS Hostname ──────── : %s" % (self.smbClient.getServerName())) 375 print(" │ └─ NetBIOS Domain ────────── : %s" % (self.smbClient.getServerDomain())) 376 print(" ├─DNS:") 377 print(" │ ├─ DNS Hostname ──────────── : %s" % (self.smbClient.getServerDNSHostName())) 378 print(" │ └─ DNS Domain ────────────── : %s" % (self.smbClient.getServerDNSDomainName())) 379 print(" ├─OS:") 380 print(" │ ├─ OS Name ───────────────── : %s" % (self.smbClient.getServerOS())) 381 print(" │ └─ OS Version ────────────── : %s.%s.%s" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild())) 382 print(" ├─Server:") 383 print(" │ ├─ Signing Required ──────── : %s" % (self.smbClient.isSigningRequired())) 384 print(" │ ├─ Login Required ────────── : %s" % (self.smbClient.isLoginRequired())) 385 print(" │ ├─ Supports NTLMv2 ───────── : %s" % (self.smbClient.doesSupportNTLMv2())) 386 MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"] 387 print(" │ ├─ Max size of read chunk ── : %d bytes (%s)" % (MaxReadSize, b_filesize(MaxReadSize))) 388 MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"] 389 print(" │ └─ Max size of write chunk ─ : %d bytes (%s)" % (MaxWriteSize, b_filesize(MaxWriteSize))) 390 print(" └─") 391 else: 392 print("[+] Server:") 393 print(" ├─NetBIOS:") 394 print(" │ ├─ \x1b[94mNetBIOS Hostname\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerName())) 395 print(" │ └─ \x1b[94mNetBIOS Domain\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDomain())) 396 print(" ├─DNS:") 397 print(" │ ├─ \x1b[94mDNS Hostname\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSHostName())) 398 print(" │ └─ \x1b[94mDNS Domain\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSDomainName())) 399 print(" ├─OS:") 400 print(" │ ├─ \x1b[94mOS Name\x1b[0m \x1b[90m─────────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerOS())) 401 print(" │ └─ \x1b[94mOS Version\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s.%s.%s\x1b[0m" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild())) 402 print(" ├─Server:") 403 print(" │ ├─ \x1b[94mSigning Required\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isSigningRequired())) 404 print(" │ ├─ \x1b[94mLogin Required\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isLoginRequired())) 405 print(" │ ├─ \x1b[94mSupports NTLMv2\x1b[0m \x1b[90m─────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.doesSupportNTLMv2())) 406 MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"] 407 print(" │ ├─ \x1b[94mMax size of read chunk\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxReadSize, b_filesize(MaxReadSize))) 408 MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"] 409 print(" │ └─ \x1b[94mMax size of write chunk\x1b[0m \x1b[90m─\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxWriteSize, b_filesize(MaxWriteSize))) 410 print(" └─") 411 412 if share and self.smb_share is not None: 413 share_name = self.available_shares.get(self.smb_share.lower(), "")["name"] 414 share_comment = self.available_shares.get(self.smb_share.lower(), "")["comment"] 415 share_type = self.available_shares.get(self.smb_share.lower(), "")["type"] 416 share_type =', '.join([s.replace("STYPE_","") for s in share_type]) 417 share_rawtype = self.available_shares.get(self.smb_share.lower(), "")["rawtype"] 418 if self.config.no_colors: 419 print("\n[+] Share:") 420 print(" ├─ Name ──────────── : %s" % (share_name)) 421 print(" ├─ Description ───── : %s" % (share_comment)) 422 print(" ├─ Type ──────────── : %s" % (share_type)) 423 print(" └─ Raw type value ── : %s" % (share_rawtype)) 424 else: 425 print("\n[+] Share:") 426 print(" ├─ \x1b[94mName\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_name)) 427 print(" ├─ \x1b[94mDescription\x1b[0m \x1b[90m─────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_comment)) 428 print(" ├─ \x1b[94mType\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_type)) 429 print(" └─ \x1b[94mRaw type value\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%s\x1b[0m" % (share_rawtype)) 430 431 def list_contents(self, path=None): 432 """ 433 Lists the contents of a specified directory on the SMB share. 434 435 This method retrieves the contents of a directory specified by `shareName` and `path`. If `shareName` or `path` 436 is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with 437 the long names of the files and directories as keys and their respective SMB entry objects as values. 438 439 Args: 440 shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None. 441 path (str, optional): The directory path to list contents from. Defaults to the current path if None. 442 443 Returns: 444 dict: A dictionary with file and directory names as keys and their SMB entry objects as values. 445 """ 446 447 if path is None or len(path) == 0: 448 path = self.smb_cwd 449 path = path.rstrip(ntpath.sep) + ntpath.sep + "*" 450 451 contents = {} 452 entries = self.smbClient.listPath( 453 shareName=self.smb_share, 454 path=path 455 ) 456 for entry in entries: 457 contents[entry.get_longname()] = entry 458 459 return contents 460 461 def list_shares(self): 462 """ 463 Lists all the shares available on the connected SMB server. 464 465 This method queries the SMB server to retrieve a list of all available shares. It populates the `shares` dictionary 466 with key-value pairs where the key is the share name and the value is a dictionary containing details about the share 467 such as its name, type, raw type, and any comments associated with the share. 468 469 Returns: 470 dict: A dictionary containing information about each share available on the server. 471 """ 472 473 self.available_shares = {} 474 475 if self.connected: 476 if self.smbClient is not None: 477 resp = self.smbClient.listShares() 478 479 for share in resp: 480 # SHARE_INFO_1 structure (lmshare.h) 481 # https://learn.microsoft.com/en-us/windows/win32/api/lmshare/ns-lmshare-share_info_1 482 sharename = share["shi1_netname"][:-1] 483 sharecomment = share["shi1_remark"][:-1] 484 sharetype = share["shi1_type"] 485 486 self.available_shares[sharename.lower()] = { 487 "name": sharename, 488 "type": STYPE_MASK(sharetype), 489 "rawtype": sharetype, 490 "comment": sharecomment 491 } 492 else: 493 print("[!] Error: SMBSession.smbClient is None.") 494 495 return self.available_shares 496 497 def mkdir(self, path=None): 498 """ 499 Creates a directory at the specified path on the SMB share. 500 501 This method takes a path and attempts to create the directory structure on the SMB share. If the path includes 502 nested directories, it will create each directory in the sequence. If a directory already exists, it will skip 503 the creation for that directory without raising an error. 504 505 Args: 506 path (str, optional): The full path of the directory to create on the SMB share. Defaults to None. 507 508 Note: 509 The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility. 510 """ 511 512 if path is not None: 513 # Prepare path 514 path = path.replace('/',ntpath.sep) 515 if ntpath.sep in path: 516 path = path.strip(ntpath.sep).split(ntpath.sep) 517 else: 518 path = [path] 519 520 # Create each dir in the path 521 for depth in range(1, len(path)+1): 522 tmp_path = ntpath.sep.join(path[:depth]) 523 try: 524 self.smbClient.createDirectory( 525 shareName=self.smb_share, 526 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + tmp_path + ntpath.sep) 527 ) 528 except impacket.smbconnection.SessionError as err: 529 if err.getErrorCode() == 0xc0000035: 530 # STATUS_OBJECT_NAME_COLLISION 531 # Remote directory already created, this is normal 532 # Src: https://github.com/fortra/impacket/blob/269ce69872f0e8f2188a80addb0c39fedfa6dcb8/impacket/nt_errors.py#L268C9-L268C19 533 pass 534 else: 535 print("[!] Failed to create directory '%s': %s" % (tmp_path, err)) 536 if self.config.debug: 537 traceback.print_exc() 538 else: 539 pass 540 541 def path_exists(self, path=None): 542 """ 543 Checks if the specified path exists on the SMB share. 544 545 This method determines if a given path exists on the SMB share by attempting to list the contents of the path. 546 If the path listing is successful and returns one or more entries, the path is considered to exist. 547 548 Args: 549 path (str, optional): The path to check on the SMB share. Defaults to None. 550 551 Returns: 552 bool: True if the path exists, False otherwise or if an error occurs. 553 """ 554 555 if path is not None: 556 path = path.replace('*','') 557 try: 558 contents = self.smbClient.listPath( 559 shareName=self.smb_share, 560 path=ntpath.normpath(self.smb_cwd + ntpath.sep + path + ntpath.sep) 561 ) 562 return (len(contents) != 0) 563 except Exception as e: 564 return False 565 else: 566 return False 567 568 def path_isdir(self, pathFromRoot=None): 569 """ 570 Checks if the specified path is a directory on the SMB share. 571 572 This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the 573 contents of the path and filtering for entries that match the basename of the path and are marked as directories. 574 575 Args: 576 path (str, optional): The path to check on the SMB share. Defaults to None. 577 578 Returns: 579 bool: True if the path is a directory, False otherwise or if an error occurs. 580 """ 581 582 if pathFromRoot is not None: 583 # Replace slashes if any 584 path = pathFromRoot.replace('/', ntpath.sep) 585 586 # Strip wildcards to avoid injections 587 path = path.replace('*','') 588 589 # Normalize path and strip leading backslash 590 path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep) 591 592 if path.strip() in ['', '.', '..']: 593 # By defininition they exist on the filesystem 594 return True 595 else: 596 try: 597 contents = self.smbClient.listPath( 598 shareName=self.smb_share, 599 path=path+'*' 600 ) 601 # Filter on directories 602 contents = [ 603 c for c in contents 604 if c.get_longname() == ntpath.basename(path) and c.is_directory() 605 ] 606 return (len(contents) != 0) 607 except Exception as e: 608 return False 609 else: 610 return False 611 612 def path_isfile(self, path=None): 613 """ 614 Checks if the specified path is a file on the SMB share. 615 616 This method determines if a given path corresponds to a file on the SMB share. It does this by listing the 617 contents of the path and filtering for entries that match the basename of the path and are not marked as directories. 618 619 Args: 620 path (str, optional): The path to check on the SMB share. Defaults to None. 621 622 Returns: 623 bool: True if the path is a file, False otherwise or if an error occurs. 624 """ 625 626 if path is not None: 627 path = path.replace('*','') 628 search_dir = ntpath.normpath(self.smb_cwd + ntpath.sep + path) 629 search_dir = ntpath.dirname(search_dir) + ntpath.sep + '*' 630 try: 631 contents = self.smbClient.listPath( 632 shareName=self.smb_share, 633 path=search_dir 634 ) 635 # Filter on files 636 contents = [ 637 c for c in contents 638 if c.get_longname() == ntpath.basename(path) and not c.is_directory() 639 ] 640 return (len(contents) != 0) 641 except Exception as e: 642 return False 643 else: 644 return False 645 646 def ping_smb_session(self): 647 """ 648 Tests the connectivity to the SMB server by sending an echo command. 649 650 This method attempts to send an echo command to the SMB server to check if the session is still active. 651 It updates the `connected` attribute of the class based on the success or failure of the echo command. 652 653 Returns: 654 bool: True if the echo command succeeds (indicating the session is active), False otherwise. 655 """ 656 657 try: 658 self.smbClient.getSMBServer().echo() 659 except Exception as e: 660 self.connected = False 661 return self.connected 662 663 def put_file(self, localpath=None): 664 """ 665 Uploads a single file to the SMB share. 666 667 This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path. 668 It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session. 669 General exceptions are caught and logged, with a traceback provided if debugging is enabled. 670 671 Args: 672 localpath (str, optional): The local file path of the file to be uploaded. Defaults to None. 673 """ 674 675 if os.path.exists(localpath): 676 if os.path.isfile(localpath): 677 try: 678 localfile = os.path.basename(localpath) 679 f = LocalFileIO( 680 mode="rb", 681 path=localpath, 682 debug=self.config.debug 683 ) 684 self.smbClient.putFile( 685 shareName=self.smb_share, 686 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + localfile + ntpath.sep), 687 callback=f.read 688 ) 689 f.close() 690 except (BrokenPipeError, KeyboardInterrupt) as err: 691 print("[!] Interrupted.") 692 self.close_smb_session() 693 self.init_smb_session() 694 except Exception as err: 695 print("[!] Failed to upload '%s': %s" % (localfile, err)) 696 if self.config.debug: 697 traceback.print_exc() 698 else: 699 print("[!] The specified localpath is a directory. Use 'put -r <directory>' instead.") 700 else: 701 print("[!] The specified localpath does not exist.") 702 703 def put_file_recursively(self, localpath=None): 704 """ 705 Recursively uploads files from a specified local directory to the SMB share. 706 707 This method walks through the given local directory and all its subdirectories, uploading each file to the 708 corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is, 709 it iterates over all files and directories within the local path, creating necessary directories on the SMB share 710 and uploading files. If the local path is not a directory, it prints an error message. 711 712 Args: 713 localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None. 714 """ 715 716 if os.path.exists(localpath): 717 if os.path.isfile(localpath): 718 # Iterate over all files and directories within the local path 719 local_files = {} 720 for root, dirs, files in os.walk(localpath): 721 if len(files) != 0: 722 local_files[root] = files 723 724 # Iterate over the found files 725 for local_dir_path in sorted(local_files.keys()): 726 print("[>] Putting files of '%s'" % local_dir_path) 727 728 # Create remote directory 729 remote_dir_path = local_dir_path.replace(os.path.sep, ntpath.sep) 730 self.mkdir( 731 path=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep) 732 ) 733 734 for local_file_path in local_files[local_dir_path]: 735 try: 736 f = LocalFileIO( 737 mode="rb", 738 path=local_dir_path + os.path.sep + local_file_path, 739 debug=self.config.debug 740 ) 741 self.smbClient.putFile( 742 shareName=self.smb_share, 743 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep + local_file_path), 744 callback=f.read 745 ) 746 f.close() 747 748 except BrokenPipeError as err: 749 f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err)) 750 f.close(remove=True) 751 break 752 except Exception as err: 753 f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err)) 754 f.close(remove=True) 755 else: 756 print("[!] The specified localpath is a file. Use 'put <file>' instead.") 757 else: 758 print("[!] The specified localpath does not exist.") 759 760 def rmdir(self, path=None): 761 """ 762 Removes a directory from the SMB share at the specified path. 763 764 This method attempts to delete a directory located at the given path on the SMB share. If the operation fails, 765 it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints 766 the stack trace of the exception. 767 768 Args: 769 path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None. 770 """ 771 try: 772 self.smbClient.deleteDirectory( 773 shareName=self.smb_share, 774 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 775 ) 776 except Exception as err: 777 print("[!] Failed to remove directory '%s': %s" % (path, err)) 778 if self.config.debug: 779 traceback.print_exc() 780 781 def rm(self, path=None): 782 """ 783 Removes a file from the SMB share at the specified path. 784 785 This method attempts to delete a file located at the given path on the SMB share. If the operation fails, 786 it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints 787 the stack trace of the exception. 788 789 Args: 790 path (str, optional): The path of the file to be removed on the SMB share. Defaults to None. 791 """ 792 try: 793 self.smbClient.deleteFile( 794 shareName=self.smb_share, 795 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 796 ) 797 except Exception as err: 798 print("[!] Failed to remove file '%s': %s" % (path, err)) 799 if self.config.debug: 800 traceback.print_exc() 801 802 def tree(self, path=None): 803 """ 804 Recursively lists the directory structure of the SMB share starting from the specified path. 805 806 This function prints a visual representation of the directory tree of the remote SMB share. It uses 807 recursion to navigate through directories and lists all files and subdirectories in each directory. 808 The output is color-coded and formatted to enhance readability, with directories highlighted in cyan. 809 810 Args: 811 path (str, optional): The starting path on the SMB share from which to begin listing the tree. 812 Defaults to the root of the current share. 813 """ 814 815 def recurse_action(base_dir="", path=[], prompt=[]): 816 bars = ["│ ", "├── ", "└── "] 817 818 remote_smb_path = ntpath.normpath(base_dir + ntpath.sep + ntpath.sep.join(path)) 819 820 entries = [] 821 try: 822 entries = self.smbClient.listPath( 823 shareName=self.smb_share, 824 path=remote_smb_path+'\\*' 825 ) 826 except impacket.smbconnection.SessionError as err: 827 code, const, text = err.getErrorCode(), err.getErrorString()[0], err.getErrorString()[1] 828 errmsg = "Error 0x%08x (%s): %s" % (code, const, text) 829 if self.config.no_colors: 830 print("%s%s" % (''.join(prompt+[bars[2]]), errmsg)) 831 else: 832 print("%s\x1b[1;91m%s\x1b[0m" % (''.join(prompt+[bars[2]]), errmsg)) 833 return 834 835 entries = [e for e in entries if e.get_longname() not in [".", ".."]] 836 entries = sorted(entries, key=lambda x:x.get_longname()) 837 838 # 839 if len(entries) > 1: 840 index = 0 841 for entry in entries: 842 index += 1 843 # This is the first entry 844 if index == 0: 845 if entry.is_directory(): 846 if self.config.no_colors: 847 print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 848 else: 849 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 850 recurse_action( 851 base_dir=base_dir, 852 path=path+[entry.get_longname()], 853 prompt=prompt+["│ "] 854 ) 855 else: 856 if self.config.no_colors: 857 print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname())) 858 else: 859 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname())) 860 861 # This is the last entry 862 elif index == len(entries): 863 if entry.is_directory(): 864 if self.config.no_colors: 865 print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 866 else: 867 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 868 recurse_action( 869 base_dir=base_dir, 870 path=path+[entry.get_longname()], 871 prompt=prompt+[" "] 872 ) 873 else: 874 if self.config.no_colors: 875 print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname())) 876 else: 877 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname())) 878 879 # These are entries in the middle 880 else: 881 if entry.is_directory(): 882 if self.config.no_colors: 883 print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 884 else: 885 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 886 recurse_action( 887 base_dir=base_dir, 888 path=path+[entry.get_longname()], 889 prompt=prompt+["│ "] 890 ) 891 else: 892 if self.config.no_colors: 893 print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname())) 894 else: 895 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname())) 896 897 # 898 elif len(entries) == 1: 899 entry = entries[0] 900 if entry.is_directory(): 901 if self.config.no_colors: 902 print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 903 else: 904 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 905 recurse_action( 906 base_dir=base_dir, 907 path=path+[entry.get_longname()], 908 prompt=prompt+[" "] 909 ) 910 else: 911 if self.config.no_colors: 912 print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname())) 913 else: 914 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname())) 915 916 # Entrypoint 917 try: 918 if self.config.no_colors: 919 print("%s\\" % path) 920 else: 921 print("\x1b[1;96m%s\x1b[0m\\" % path) 922 recurse_action( 923 base_dir=self.smb_cwd, 924 path=[path], 925 prompt=[""] 926 ) 927 except (BrokenPipeError, KeyboardInterrupt) as e: 928 print("[!] Interrupted.") 929 self.close_smb_session() 930 self.init_smb_session() 931 932 # Setter / Getter 933 934 def set_share(self, shareName): 935 """ 936 Sets the current SMB share to the specified share name. 937 938 This method updates the SMB session to use the specified share name. It checks if the share name is valid 939 and updates the smb_share attribute of the SMBSession instance. 940 941 Parameters: 942 shareName (str): The name of the share to set as the current SMB share. 943 944 Raises: 945 ValueError: If the shareName is None or an empty string. 946 """ 947 948 if shareName is not None: 949 self.list_shares() 950 if shareName.lower() in self.available_shares.keys(): 951 # Doing this in order to keep the case of the share adevertised by the remote machine 952 self.smb_share = self.available_shares[shareName.lower()]["name"] 953 else: 954 print("[!] Could not set share '%s', it does not exist remotely." % shareName) 955 956 def set_cwd(self, path=None): 957 """ 958 Sets the current working directory on the SMB share to the specified path. 959 960 This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory. 961 If the specified path is not a directory, the cwd remains unchanged. 962 963 Parameters: 964 path (str): The path to set as the current working directory. 965 966 Raises: 967 ValueError: If the specified path is not a directory. 968 """ 969 970 if path is not None: 971 # Set path separators to ntpath sep 972 if '/' in path: 973 path = path.replace('/', ntpath.sep) 974 975 if path.startswith(ntpath.sep): 976 # Absolute path 977 path = path + ntpath.sep 978 else: 979 # Relative path to the CWD 980 if len(self.smb_cwd) == 0: 981 path = path + ntpath.sep 982 else: 983 path = self.smb_cwd + ntpath.sep + path 984 985 # Path normalization 986 path = ntpath.normpath(path) 987 path = re.sub(r'\\+', r'\\', path) 988 989 if path in ["", ".", ".."]: 990 self.smb_cwd = "" 991 else: 992 if self.path_isdir(pathFromRoot=path.strip(ntpath.sep)): 993 # Path exists on the remote 994 self.smb_cwd = ntpath.normpath(path) 995 else: 996 # Path does not exists or is not a directory on the remote 997 print("[!] Remote directory '%s' does not exist." % path)
Class SMBSession is designed to handle the session management for SMB (Server Message Block) protocol connections. It provides functionalities to connect to an SMB server, authenticate using either NTLM or Kerberos, and manage SMB shares.
Attributes: address (str): The IP address or hostname of the SMB server. domain (str): The domain name for SMB server authentication. username (str): The username for SMB server authentication. password (str): The password for SMB server authentication. lmhash (str): The LM hash of the user's password, if available. nthash (str): The NT hash of the user's password, if available. use_kerberos (bool): A flag to determine whether to use Kerberos for authentication. kdcHost (str): The Key Distribution Center (KDC) host for Kerberos authentication. debug (bool): A flag to enable debug output. smbClient (object): The SMB client object used for the connection. connected (bool): A flag to check the status of the connection. smb_share (str): The current SMB share in use. smb_path (str): The current path within the SMB share.
Methods: __init__(address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, debug=False): Initializes the SMBSession with the specified parameters. init_smb_session(): Initializes the SMB session by connecting to the server and authenticating using the specified method.
46 def __init__(self, address, domain, username, password, lmhash, nthash, use_kerberos=False, kdcHost=None, config=None): 47 super(SMBSession, self).__init__() 48 # Objects 49 self.config = config 50 51 # Target server 52 self.address = address 53 54 # Credentials 55 self.domain = domain 56 self.username = username 57 self.password = password 58 self.lmhash = lmhash 59 self.nthash = nthash 60 self.use_kerberos = use_kerberos 61 self.kdcHost = kdcHost 62 63 self.smbClient = None 64 self.connected = False 65 66 self.available_shares = {} 67 self.smb_share = None 68 self.smb_cwd = "" 69 70 self.list_shares()
74 def init_smb_session(self): 75 """ 76 Initializes and establishes a session with the SMB server. 77 78 This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration. 79 It attempts to connect to the SMB server specified by the `address` attribute and authenticate using the credentials provided during the object's initialization. 80 81 The method will print debug information if the `debug` attribute is set to True. Upon successful connection and authentication, it sets the `connected` attribute to True. 82 83 Returns: 84 bool: True if the connection and authentication are successful, False otherwise. 85 """ 86 87 self.connected = False 88 89 if self.config.debug: 90 print("[debug] [>] Connecting to remote SMB server '%s' ... " % self.address) 91 try: 92 self.smbClient = impacket.smbconnection.SMBConnection( 93 remoteName=self.address, 94 remoteHost=self.address, 95 sess_port=int(445) 96 ) 97 except OSError as err: 98 print("[!] %s" % err) 99 self.smbClient = None 100 101 if self.smbClient is not None: 102 if self.use_kerberos: 103 if self.config.debug: 104 print("[debug] [>] Authenticating as '%s\\%s' with kerberos ... " % (self.domain, self.username)) 105 try: 106 self.connected = self.smbClient.kerberosLogin( 107 user=self.username, 108 password=self.password, 109 domain=self.domain, 110 lmhash=self.lmhash, 111 nthash=self.nthash, 112 aesKey=self.aesKey, 113 kdcHost=self.kdcHost 114 ) 115 except impacket.smbconnection.SessionError as err: 116 if self.config.debug: 117 traceback.print_exc() 118 print("[!] Could not login: %s" % err) 119 self.connected = False 120 121 else: 122 if self.config.debug: 123 print("[debug] [>] Authenticating as '%s\\%s' with NTLM ... " % (self.domain, self.username)) 124 try: 125 self.connected = self.smbClient.login( 126 user=self.username, 127 password=self.password, 128 domain=self.domain, 129 lmhash=self.lmhash, 130 nthash=self.nthash 131 ) 132 except impacket.smbconnection.SessionError as err: 133 if self.config.debug: 134 traceback.print_exc() 135 print("[!] Could not login: %s" % err) 136 self.connected = False 137 138 if self.connected: 139 print("[+] Successfully authenticated to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username)) 140 else: 141 print("[!] Failed to authenticate to '%s' as '%s\\%s'!" % (self.address, self.domain, self.username)) 142 143 return self.connected
Initializes and establishes a session with the SMB server.
This method sets up the SMB connection using either Kerberos or NTLM authentication based on the configuration.
It attempts to connect to the SMB server specified by the address attribute and authenticate using the credentials provided during the object's initialization.
The method will print debug information if the debug attribute is set to True. Upon successful connection and authentication, it sets the connected attribute to True.
Returns: bool: True if the connection and authentication are successful, False otherwise.
145 def close_smb_session(self): 146 """ 147 Closes the current SMB session by disconnecting the SMB client. 148 149 This method ensures that the SMB client connection is properly closed. It checks if the client is connected 150 and if so, it closes the connection and resets the connection status. 151 152 Raises: 153 Exception: If the SMB client is not initialized or if there's an error during the disconnection process. 154 """ 155 156 if self.smbClient is not None: 157 if self.connected: 158 self.smbClient.close() 159 self.connected = False 160 if self.config.debug: 161 print("[+] SMB connection closed successfully.") 162 else: 163 if self.config.debug: 164 print("[!] No active SMB connection to close.") 165 else: 166 raise Exception("SMB client is not initialized.")
Closes the current SMB session by disconnecting the SMB client.
This method ensures that the SMB client connection is properly closed. It checks if the client is connected and if so, it closes the connection and resets the connection status.
Raises: Exception: If the SMB client is not initialized or if there's an error during the disconnection process.
170 def read_file(self, path=None): 171 if self.path_isfile(path=path): 172 tmp_file_path = self.smb_cwd + ntpath.sep + path 173 matches = self.smbClient.listPath( 174 shareName=self.smb_share, 175 path=tmp_file_path 176 ) 177 178 fh = io.BytesIO() 179 try: 180 # opening the files in streams instead of mounting shares allows 181 # for running the script from unprivileged containers 182 self.smbClient.getFile(self.smb_share, tmp_file_path, fh.write) 183 except impacket.smbconnection.SessionError as e: 184 return None 185 rawdata = fh.getvalue() 186 fh.close() 187 return rawdata 188 else: 189 print("[!] Remote path '%s' is not a file." % path)
191 def find(self, paths=[], callback=None): 192 def recurse_action(paths=[], depth=0, callback=None): 193 if callback is None: 194 return [] 195 next_directories_to_explore = [] 196 for path in paths: 197 remote_smb_path = ntpath.normpath(self.smb_cwd + ntpath.sep + path) 198 entries = [] 199 200 try: 201 entries = self.smbClient.listPath( 202 shareName=self.smb_share, 203 path=(remote_smb_path + ntpath.sep + '*') 204 ) 205 except impacket.smbconnection.SessionError as err: 206 continue 207 # Remove dot names 208 entries = [e for e in entries if e.get_longname() not in [".", ".."]] 209 # Sort the entries ignoring case 210 entries = sorted(entries, key=lambda x:x.get_longname().lower()) 211 212 for entry in entries: 213 if entry.is_directory(): 214 callback(entry, path + entry.get_longname() + ntpath.sep, depth) 215 else: 216 callback(entry, path + entry.get_longname(), depth) 217 218 # Next directories to explore 219 for entry in entries: 220 if entry.is_directory(): 221 next_directories_to_explore.append(path + entry.get_longname() + ntpath.sep) 222 223 return next_directories_to_explore 224 # 225 if callback is not None: 226 depth = 0 227 while len(paths) != 0: 228 paths = recurse_action( 229 paths=paths, 230 depth=depth, 231 callback=callback 232 ) 233 depth = depth + 1 234 else: 235 print("[!] SMBSession.find(), callback function cannot be None.")
237 def get_file(self, path=None, keepRemotePath=False): 238 """ 239 Retrieves a file from the specified path on the SMB share. 240 241 This method attempts to retrieve a file from the given path within the currently connected SMB share. 242 If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local 243 file object and writing the contents of the remote file to it using the SMB client's getFile method. 244 245 Parameters: 246 path (str): The path of the file to retrieve. If None, uses the current smb_path. 247 248 Returns: 249 None 250 """ 251 252 tmp_file_path = self.smb_cwd + ntpath.sep + path 253 matches = self.smbClient.listPath( 254 shareName=self.smb_share, 255 path=tmp_file_path 256 ) 257 258 for entry in matches: 259 if entry.is_directory(): 260 print("[>] Skipping '%s' because it is a directory." % tmp_file_path) 261 else: 262 try: 263 if ntpath.sep in path: 264 outputfile = ntpath.dirname(path) + ntpath.sep + entry.get_longname() 265 else: 266 outputfile = entry.get_longname() 267 f = LocalFileIO( 268 mode="wb", 269 path=outputfile, 270 expected_size=entry.get_filesize(), 271 debug=self.config.debug, 272 keepRemotePath=keepRemotePath 273 ) 274 self.smbClient.getFile( 275 shareName=self.smb_share, 276 pathName=tmp_file_path, 277 callback=f.write 278 ) 279 f.close() 280 except (BrokenPipeError, KeyboardInterrupt) as e: 281 f.close() 282 print("\x1b[v\x1b[o\r[!] Interrupted.") 283 self.close_smb_session() 284 self.init_smb_session() 285 286 return None
Retrieves a file from the specified path on the SMB share.
This method attempts to retrieve a file from the given path within the currently connected SMB share. If the path points to a directory, it skips the retrieval. It handles file retrieval by creating a local file object and writing the contents of the remote file to it using the SMB client's getFile method.
Parameters: path (str): The path of the file to retrieve. If None, uses the current smb_path.
Returns: None
288 def get_file_recursively(self, path=None): 289 """ 290 Recursively retrieves files from a specified path on the SMB share. 291 292 This method navigates through all directories starting from the given path, 293 and downloads all files found. It handles directories recursively, ensuring 294 that all nested files are retrieved. The method skips over directory entries 295 and handles errors gracefully, attempting to continue the operation where possible. 296 297 Parameters: 298 path (str): The initial directory path from which to start the recursive file retrieval. 299 If None, it starts from the root of the configured SMB share. 300 """ 301 302 def recurse_action(base_dir="", path=[]): 303 remote_smb_path = base_dir + ntpath.sep.join(path) 304 entries = self.smbClient.listPath( 305 shareName=self.smb_share, 306 path=remote_smb_path + '\\*' 307 ) 308 if len(entries) != 0: 309 files = [entry for entry in entries if not entry.is_directory()] 310 directories = [entry for entry in entries if entry.is_directory() and entry.get_longname() not in [".", ".."]] 311 312 # Files 313 if len(files) != 0: 314 print("[>] Retrieving files of '%s'" % remote_smb_path) 315 for entry_file in files: 316 if not entry_file.is_directory(): 317 f = LocalFileIO( 318 mode="wb", 319 path=remote_smb_path + ntpath.sep + entry_file.get_longname(), 320 expected_size=entry_file.get_filesize(), 321 debug=self.config.debug 322 ) 323 try: 324 self.smbClient.getFile( 325 shareName=self.smb_share, 326 pathName=remote_smb_path + ntpath.sep + entry_file.get_longname(), 327 callback=f.write 328 ) 329 f.close() 330 except BrokenPipeError as err: 331 f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err)) 332 f.close(remove=True) 333 break 334 except Exception as err: 335 f.set_error(message="[bold red]Failed downloading '%s': %s" % (f.path, err)) 336 f.close(remove=True) 337 338 # Directories 339 for entry_directory in directories: 340 if entry_directory.is_directory(): 341 recurse_action( 342 base_dir=self.smb_cwd, 343 path=path+[entry_directory.get_longname()] 344 ) 345 # Entrypoint 346 try: 347 recurse_action( 348 base_dir=self.smb_cwd, 349 path=[path] 350 ) 351 except (BrokenPipeError, KeyboardInterrupt) as e: 352 print("\x1b[v\x1b[o\r[!] Interrupted.") 353 self.close_smb_session() 354 self.init_smb_session()
Recursively retrieves files from a specified path on the SMB share.
This method navigates through all directories starting from the given path, and downloads all files found. It handles directories recursively, ensuring that all nested files are retrieved. The method skips over directory entries and handles errors gracefully, attempting to continue the operation where possible.
Parameters: path (str): The initial directory path from which to start the recursive file retrieval. If None, it starts from the root of the configured SMB share.
356 def info(self, share=True, server=True): 357 """ 358 Displays information about the server and optionally the shares. 359 360 This method prints detailed information about the server's characteristics such as NetBIOS names, DNS details, OS information, and SMB capabilities. If the `share` parameter is set to True and a share is currently set, it will also attempt to display information about the share. 361 362 Parameters: 363 share (bool): If True, display information about the current share. 364 server (bool): If True, display information about the server. 365 366 Returns: 367 None 368 """ 369 370 if server: 371 if self.config.no_colors: 372 print("[+] Server:") 373 print(" ├─NetBIOS:") 374 print(" │ ├─ NetBIOS Hostname ──────── : %s" % (self.smbClient.getServerName())) 375 print(" │ └─ NetBIOS Domain ────────── : %s" % (self.smbClient.getServerDomain())) 376 print(" ├─DNS:") 377 print(" │ ├─ DNS Hostname ──────────── : %s" % (self.smbClient.getServerDNSHostName())) 378 print(" │ └─ DNS Domain ────────────── : %s" % (self.smbClient.getServerDNSDomainName())) 379 print(" ├─OS:") 380 print(" │ ├─ OS Name ───────────────── : %s" % (self.smbClient.getServerOS())) 381 print(" │ └─ OS Version ────────────── : %s.%s.%s" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild())) 382 print(" ├─Server:") 383 print(" │ ├─ Signing Required ──────── : %s" % (self.smbClient.isSigningRequired())) 384 print(" │ ├─ Login Required ────────── : %s" % (self.smbClient.isLoginRequired())) 385 print(" │ ├─ Supports NTLMv2 ───────── : %s" % (self.smbClient.doesSupportNTLMv2())) 386 MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"] 387 print(" │ ├─ Max size of read chunk ── : %d bytes (%s)" % (MaxReadSize, b_filesize(MaxReadSize))) 388 MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"] 389 print(" │ └─ Max size of write chunk ─ : %d bytes (%s)" % (MaxWriteSize, b_filesize(MaxWriteSize))) 390 print(" └─") 391 else: 392 print("[+] Server:") 393 print(" ├─NetBIOS:") 394 print(" │ ├─ \x1b[94mNetBIOS Hostname\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerName())) 395 print(" │ └─ \x1b[94mNetBIOS Domain\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDomain())) 396 print(" ├─DNS:") 397 print(" │ ├─ \x1b[94mDNS Hostname\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSHostName())) 398 print(" │ └─ \x1b[94mDNS Domain\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerDNSDomainName())) 399 print(" ├─OS:") 400 print(" │ ├─ \x1b[94mOS Name\x1b[0m \x1b[90m─────────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.getServerOS())) 401 print(" │ └─ \x1b[94mOS Version\x1b[0m \x1b[90m──────────────\x1b[0m : \x1b[93m%s.%s.%s\x1b[0m" % (self.smbClient.getServerOSMajor(), self.smbClient.getServerOSMinor(), self.smbClient.getServerOSBuild())) 402 print(" ├─Server:") 403 print(" │ ├─ \x1b[94mSigning Required\x1b[0m \x1b[90m────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isSigningRequired())) 404 print(" │ ├─ \x1b[94mLogin Required\x1b[0m \x1b[90m──────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.isLoginRequired())) 405 print(" │ ├─ \x1b[94mSupports NTLMv2\x1b[0m \x1b[90m─────────\x1b[0m : \x1b[93m%s\x1b[0m" % (self.smbClient.doesSupportNTLMv2())) 406 MaxReadSize = self.smbClient.getIOCapabilities()["MaxReadSize"] 407 print(" │ ├─ \x1b[94mMax size of read chunk\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxReadSize, b_filesize(MaxReadSize))) 408 MaxWriteSize = self.smbClient.getIOCapabilities()["MaxWriteSize"] 409 print(" │ └─ \x1b[94mMax size of write chunk\x1b[0m \x1b[90m─\x1b[0m : \x1b[93m%d bytes (%s)\x1b[0m" % (MaxWriteSize, b_filesize(MaxWriteSize))) 410 print(" └─") 411 412 if share and self.smb_share is not None: 413 share_name = self.available_shares.get(self.smb_share.lower(), "")["name"] 414 share_comment = self.available_shares.get(self.smb_share.lower(), "")["comment"] 415 share_type = self.available_shares.get(self.smb_share.lower(), "")["type"] 416 share_type =', '.join([s.replace("STYPE_","") for s in share_type]) 417 share_rawtype = self.available_shares.get(self.smb_share.lower(), "")["rawtype"] 418 if self.config.no_colors: 419 print("\n[+] Share:") 420 print(" ├─ Name ──────────── : %s" % (share_name)) 421 print(" ├─ Description ───── : %s" % (share_comment)) 422 print(" ├─ Type ──────────── : %s" % (share_type)) 423 print(" └─ Raw type value ── : %s" % (share_rawtype)) 424 else: 425 print("\n[+] Share:") 426 print(" ├─ \x1b[94mName\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_name)) 427 print(" ├─ \x1b[94mDescription\x1b[0m \x1b[90m─────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_comment)) 428 print(" ├─ \x1b[94mType\x1b[0m \x1b[90m────────────\x1b[0m : \x1b[93m%s\x1b[0m" % (share_type)) 429 print(" └─ \x1b[94mRaw type value\x1b[0m \x1b[90m──\x1b[0m : \x1b[93m%s\x1b[0m" % (share_rawtype))
Displays information about the server and optionally the shares.
This method prints detailed information about the server's characteristics such as NetBIOS names, DNS details, OS information, and SMB capabilities. If the share parameter is set to True and a share is currently set, it will also attempt to display information about the share.
Parameters: share (bool): If True, display information about the current share. server (bool): If True, display information about the server.
Returns: None
431 def list_contents(self, path=None): 432 """ 433 Lists the contents of a specified directory on the SMB share. 434 435 This method retrieves the contents of a directory specified by `shareName` and `path`. If `shareName` or `path` 436 is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with 437 the long names of the files and directories as keys and their respective SMB entry objects as values. 438 439 Args: 440 shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None. 441 path (str, optional): The directory path to list contents from. Defaults to the current path if None. 442 443 Returns: 444 dict: A dictionary with file and directory names as keys and their SMB entry objects as values. 445 """ 446 447 if path is None or len(path) == 0: 448 path = self.smb_cwd 449 path = path.rstrip(ntpath.sep) + ntpath.sep + "*" 450 451 contents = {} 452 entries = self.smbClient.listPath( 453 shareName=self.smb_share, 454 path=path 455 ) 456 for entry in entries: 457 contents[entry.get_longname()] = entry 458 459 return contents
Lists the contents of a specified directory on the SMB share.
This method retrieves the contents of a directory specified by shareName and path. If shareName or path
is not provided, it defaults to the instance's current SMB share or path. The method returns a dictionary with
the long names of the files and directories as keys and their respective SMB entry objects as values.
Args: shareName (str, optional): The name of the SMB share. Defaults to the current SMB share if None. path (str, optional): The directory path to list contents from. Defaults to the current path if None.
Returns: dict: A dictionary with file and directory names as keys and their SMB entry objects as values.
497 def mkdir(self, path=None): 498 """ 499 Creates a directory at the specified path on the SMB share. 500 501 This method takes a path and attempts to create the directory structure on the SMB share. If the path includes 502 nested directories, it will create each directory in the sequence. If a directory already exists, it will skip 503 the creation for that directory without raising an error. 504 505 Args: 506 path (str, optional): The full path of the directory to create on the SMB share. Defaults to None. 507 508 Note: 509 The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility. 510 """ 511 512 if path is not None: 513 # Prepare path 514 path = path.replace('/',ntpath.sep) 515 if ntpath.sep in path: 516 path = path.strip(ntpath.sep).split(ntpath.sep) 517 else: 518 path = [path] 519 520 # Create each dir in the path 521 for depth in range(1, len(path)+1): 522 tmp_path = ntpath.sep.join(path[:depth]) 523 try: 524 self.smbClient.createDirectory( 525 shareName=self.smb_share, 526 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + tmp_path + ntpath.sep) 527 ) 528 except impacket.smbconnection.SessionError as err: 529 if err.getErrorCode() == 0xc0000035: 530 # STATUS_OBJECT_NAME_COLLISION 531 # Remote directory already created, this is normal 532 # Src: https://github.com/fortra/impacket/blob/269ce69872f0e8f2188a80addb0c39fedfa6dcb8/impacket/nt_errors.py#L268C9-L268C19 533 pass 534 else: 535 print("[!] Failed to create directory '%s': %s" % (tmp_path, err)) 536 if self.config.debug: 537 traceback.print_exc() 538 else: 539 pass
Creates a directory at the specified path on the SMB share.
This method takes a path and attempts to create the directory structure on the SMB share. If the path includes nested directories, it will create each directory in the sequence. If a directory already exists, it will skip the creation for that directory without raising an error.
Args: path (str, optional): The full path of the directory to create on the SMB share. Defaults to None.
Note: The path should use forward slashes ('/') which will be converted to backslashes (ntpath.sep) for SMB compatibility.
541 def path_exists(self, path=None): 542 """ 543 Checks if the specified path exists on the SMB share. 544 545 This method determines if a given path exists on the SMB share by attempting to list the contents of the path. 546 If the path listing is successful and returns one or more entries, the path is considered to exist. 547 548 Args: 549 path (str, optional): The path to check on the SMB share. Defaults to None. 550 551 Returns: 552 bool: True if the path exists, False otherwise or if an error occurs. 553 """ 554 555 if path is not None: 556 path = path.replace('*','') 557 try: 558 contents = self.smbClient.listPath( 559 shareName=self.smb_share, 560 path=ntpath.normpath(self.smb_cwd + ntpath.sep + path + ntpath.sep) 561 ) 562 return (len(contents) != 0) 563 except Exception as e: 564 return False 565 else: 566 return False
Checks if the specified path exists on the SMB share.
This method determines if a given path exists on the SMB share by attempting to list the contents of the path. If the path listing is successful and returns one or more entries, the path is considered to exist.
Args: path (str, optional): The path to check on the SMB share. Defaults to None.
Returns: bool: True if the path exists, False otherwise or if an error occurs.
568 def path_isdir(self, pathFromRoot=None): 569 """ 570 Checks if the specified path is a directory on the SMB share. 571 572 This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the 573 contents of the path and filtering for entries that match the basename of the path and are marked as directories. 574 575 Args: 576 path (str, optional): The path to check on the SMB share. Defaults to None. 577 578 Returns: 579 bool: True if the path is a directory, False otherwise or if an error occurs. 580 """ 581 582 if pathFromRoot is not None: 583 # Replace slashes if any 584 path = pathFromRoot.replace('/', ntpath.sep) 585 586 # Strip wildcards to avoid injections 587 path = path.replace('*','') 588 589 # Normalize path and strip leading backslash 590 path = ntpath.normpath(path + ntpath.sep).lstrip(ntpath.sep) 591 592 if path.strip() in ['', '.', '..']: 593 # By defininition they exist on the filesystem 594 return True 595 else: 596 try: 597 contents = self.smbClient.listPath( 598 shareName=self.smb_share, 599 path=path+'*' 600 ) 601 # Filter on directories 602 contents = [ 603 c for c in contents 604 if c.get_longname() == ntpath.basename(path) and c.is_directory() 605 ] 606 return (len(contents) != 0) 607 except Exception as e: 608 return False 609 else: 610 return False
Checks if the specified path is a directory on the SMB share.
This method determines if a given path corresponds to a directory on the SMB share. It does this by listing the contents of the path and filtering for entries that match the basename of the path and are marked as directories.
Args: path (str, optional): The path to check on the SMB share. Defaults to None.
Returns: bool: True if the path is a directory, False otherwise or if an error occurs.
612 def path_isfile(self, path=None): 613 """ 614 Checks if the specified path is a file on the SMB share. 615 616 This method determines if a given path corresponds to a file on the SMB share. It does this by listing the 617 contents of the path and filtering for entries that match the basename of the path and are not marked as directories. 618 619 Args: 620 path (str, optional): The path to check on the SMB share. Defaults to None. 621 622 Returns: 623 bool: True if the path is a file, False otherwise or if an error occurs. 624 """ 625 626 if path is not None: 627 path = path.replace('*','') 628 search_dir = ntpath.normpath(self.smb_cwd + ntpath.sep + path) 629 search_dir = ntpath.dirname(search_dir) + ntpath.sep + '*' 630 try: 631 contents = self.smbClient.listPath( 632 shareName=self.smb_share, 633 path=search_dir 634 ) 635 # Filter on files 636 contents = [ 637 c for c in contents 638 if c.get_longname() == ntpath.basename(path) and not c.is_directory() 639 ] 640 return (len(contents) != 0) 641 except Exception as e: 642 return False 643 else: 644 return False
Checks if the specified path is a file on the SMB share.
This method determines if a given path corresponds to a file on the SMB share. It does this by listing the contents of the path and filtering for entries that match the basename of the path and are not marked as directories.
Args: path (str, optional): The path to check on the SMB share. Defaults to None.
Returns: bool: True if the path is a file, False otherwise or if an error occurs.
646 def ping_smb_session(self): 647 """ 648 Tests the connectivity to the SMB server by sending an echo command. 649 650 This method attempts to send an echo command to the SMB server to check if the session is still active. 651 It updates the `connected` attribute of the class based on the success or failure of the echo command. 652 653 Returns: 654 bool: True if the echo command succeeds (indicating the session is active), False otherwise. 655 """ 656 657 try: 658 self.smbClient.getSMBServer().echo() 659 except Exception as e: 660 self.connected = False 661 return self.connected
Tests the connectivity to the SMB server by sending an echo command.
This method attempts to send an echo command to the SMB server to check if the session is still active.
It updates the connected attribute of the class based on the success or failure of the echo command.
Returns: bool: True if the echo command succeeds (indicating the session is active), False otherwise.
663 def put_file(self, localpath=None): 664 """ 665 Uploads a single file to the SMB share. 666 667 This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path. 668 It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session. 669 General exceptions are caught and logged, with a traceback provided if debugging is enabled. 670 671 Args: 672 localpath (str, optional): The local file path of the file to be uploaded. Defaults to None. 673 """ 674 675 if os.path.exists(localpath): 676 if os.path.isfile(localpath): 677 try: 678 localfile = os.path.basename(localpath) 679 f = LocalFileIO( 680 mode="rb", 681 path=localpath, 682 debug=self.config.debug 683 ) 684 self.smbClient.putFile( 685 shareName=self.smb_share, 686 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + localfile + ntpath.sep), 687 callback=f.read 688 ) 689 f.close() 690 except (BrokenPipeError, KeyboardInterrupt) as err: 691 print("[!] Interrupted.") 692 self.close_smb_session() 693 self.init_smb_session() 694 except Exception as err: 695 print("[!] Failed to upload '%s': %s" % (localfile, err)) 696 if self.config.debug: 697 traceback.print_exc() 698 else: 699 print("[!] The specified localpath is a directory. Use 'put -r <directory>' instead.") 700 else: 701 print("[!] The specified localpath does not exist.")
Uploads a single file to the SMB share.
This method takes a local file path, opens the file, and uploads it to the SMB share at the specified path. It handles exceptions such as broken pipe errors or keyboard interrupts by closing and reinitializing the SMB session. General exceptions are caught and logged, with a traceback provided if debugging is enabled.
Args: localpath (str, optional): The local file path of the file to be uploaded. Defaults to None.
703 def put_file_recursively(self, localpath=None): 704 """ 705 Recursively uploads files from a specified local directory to the SMB share. 706 707 This method walks through the given local directory and all its subdirectories, uploading each file to the 708 corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is, 709 it iterates over all files and directories within the local path, creating necessary directories on the SMB share 710 and uploading files. If the local path is not a directory, it prints an error message. 711 712 Args: 713 localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None. 714 """ 715 716 if os.path.exists(localpath): 717 if os.path.isfile(localpath): 718 # Iterate over all files and directories within the local path 719 local_files = {} 720 for root, dirs, files in os.walk(localpath): 721 if len(files) != 0: 722 local_files[root] = files 723 724 # Iterate over the found files 725 for local_dir_path in sorted(local_files.keys()): 726 print("[>] Putting files of '%s'" % local_dir_path) 727 728 # Create remote directory 729 remote_dir_path = local_dir_path.replace(os.path.sep, ntpath.sep) 730 self.mkdir( 731 path=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep) 732 ) 733 734 for local_file_path in local_files[local_dir_path]: 735 try: 736 f = LocalFileIO( 737 mode="rb", 738 path=local_dir_path + os.path.sep + local_file_path, 739 debug=self.config.debug 740 ) 741 self.smbClient.putFile( 742 shareName=self.smb_share, 743 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + remote_dir_path + ntpath.sep + local_file_path), 744 callback=f.read 745 ) 746 f.close() 747 748 except BrokenPipeError as err: 749 f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err)) 750 f.close(remove=True) 751 break 752 except Exception as err: 753 f.set_error(message="[bold red]Failed uploading '%s': %s" % (f.path, err)) 754 f.close(remove=True) 755 else: 756 print("[!] The specified localpath is a file. Use 'put <file>' instead.") 757 else: 758 print("[!] The specified localpath does not exist.")
Recursively uploads files from a specified local directory to the SMB share.
This method walks through the given local directory and all its subdirectories, uploading each file to the corresponding directory structure on the SMB share. It first checks if the local path is a directory. If it is, it iterates over all files and directories within the local path, creating necessary directories on the SMB share and uploading files. If the local path is not a directory, it prints an error message.
Args: localpath (str, optional): The local directory path from which files will be uploaded. Defaults to None.
760 def rmdir(self, path=None): 761 """ 762 Removes a directory from the SMB share at the specified path. 763 764 This method attempts to delete a directory located at the given path on the SMB share. If the operation fails, 765 it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints 766 the stack trace of the exception. 767 768 Args: 769 path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None. 770 """ 771 try: 772 self.smbClient.deleteDirectory( 773 shareName=self.smb_share, 774 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 775 ) 776 except Exception as err: 777 print("[!] Failed to remove directory '%s': %s" % (path, err)) 778 if self.config.debug: 779 traceback.print_exc()
Removes a directory from the SMB share at the specified path.
This method attempts to delete a directory located at the given path on the SMB share. If the operation fails, it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints the stack trace of the exception.
Args: path (str, optional): The path of the directory to be removed on the SMB share. Defaults to None.
781 def rm(self, path=None): 782 """ 783 Removes a file from the SMB share at the specified path. 784 785 This method attempts to delete a file located at the given path on the SMB share. If the operation fails, 786 it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints 787 the stack trace of the exception. 788 789 Args: 790 path (str, optional): The path of the file to be removed on the SMB share. Defaults to None. 791 """ 792 try: 793 self.smbClient.deleteFile( 794 shareName=self.smb_share, 795 pathName=ntpath.normpath(self.smb_cwd + ntpath.sep + path), 796 ) 797 except Exception as err: 798 print("[!] Failed to remove file '%s': %s" % (path, err)) 799 if self.config.debug: 800 traceback.print_exc()
Removes a file from the SMB share at the specified path.
This method attempts to delete a file located at the given path on the SMB share. If the operation fails, it prints an error message indicating the failure and the reason. If debugging is enabled, it also prints the stack trace of the exception.
Args: path (str, optional): The path of the file to be removed on the SMB share. Defaults to None.
802 def tree(self, path=None): 803 """ 804 Recursively lists the directory structure of the SMB share starting from the specified path. 805 806 This function prints a visual representation of the directory tree of the remote SMB share. It uses 807 recursion to navigate through directories and lists all files and subdirectories in each directory. 808 The output is color-coded and formatted to enhance readability, with directories highlighted in cyan. 809 810 Args: 811 path (str, optional): The starting path on the SMB share from which to begin listing the tree. 812 Defaults to the root of the current share. 813 """ 814 815 def recurse_action(base_dir="", path=[], prompt=[]): 816 bars = ["│ ", "├── ", "└── "] 817 818 remote_smb_path = ntpath.normpath(base_dir + ntpath.sep + ntpath.sep.join(path)) 819 820 entries = [] 821 try: 822 entries = self.smbClient.listPath( 823 shareName=self.smb_share, 824 path=remote_smb_path+'\\*' 825 ) 826 except impacket.smbconnection.SessionError as err: 827 code, const, text = err.getErrorCode(), err.getErrorString()[0], err.getErrorString()[1] 828 errmsg = "Error 0x%08x (%s): %s" % (code, const, text) 829 if self.config.no_colors: 830 print("%s%s" % (''.join(prompt+[bars[2]]), errmsg)) 831 else: 832 print("%s\x1b[1;91m%s\x1b[0m" % (''.join(prompt+[bars[2]]), errmsg)) 833 return 834 835 entries = [e for e in entries if e.get_longname() not in [".", ".."]] 836 entries = sorted(entries, key=lambda x:x.get_longname()) 837 838 # 839 if len(entries) > 1: 840 index = 0 841 for entry in entries: 842 index += 1 843 # This is the first entry 844 if index == 0: 845 if entry.is_directory(): 846 if self.config.no_colors: 847 print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 848 else: 849 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 850 recurse_action( 851 base_dir=base_dir, 852 path=path+[entry.get_longname()], 853 prompt=prompt+["│ "] 854 ) 855 else: 856 if self.config.no_colors: 857 print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname())) 858 else: 859 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname())) 860 861 # This is the last entry 862 elif index == len(entries): 863 if entry.is_directory(): 864 if self.config.no_colors: 865 print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 866 else: 867 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 868 recurse_action( 869 base_dir=base_dir, 870 path=path+[entry.get_longname()], 871 prompt=prompt+[" "] 872 ) 873 else: 874 if self.config.no_colors: 875 print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname())) 876 else: 877 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname())) 878 879 # These are entries in the middle 880 else: 881 if entry.is_directory(): 882 if self.config.no_colors: 883 print("%s%s\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 884 else: 885 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[1]]), entry.get_longname())) 886 recurse_action( 887 base_dir=base_dir, 888 path=path+[entry.get_longname()], 889 prompt=prompt+["│ "] 890 ) 891 else: 892 if self.config.no_colors: 893 print("%s%s" % (''.join(prompt+[bars[1]]), entry.get_longname())) 894 else: 895 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[1]]), entry.get_longname())) 896 897 # 898 elif len(entries) == 1: 899 entry = entries[0] 900 if entry.is_directory(): 901 if self.config.no_colors: 902 print("%s%s\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 903 else: 904 print("%s\x1b[1;96m%s\x1b[0m\\" % (''.join(prompt+[bars[2]]), entry.get_longname())) 905 recurse_action( 906 base_dir=base_dir, 907 path=path+[entry.get_longname()], 908 prompt=prompt+[" "] 909 ) 910 else: 911 if self.config.no_colors: 912 print("%s%s" % (''.join(prompt+[bars[2]]), entry.get_longname())) 913 else: 914 print("%s\x1b[1m%s\x1b[0m" % (''.join(prompt+[bars[2]]), entry.get_longname())) 915 916 # Entrypoint 917 try: 918 if self.config.no_colors: 919 print("%s\\" % path) 920 else: 921 print("\x1b[1;96m%s\x1b[0m\\" % path) 922 recurse_action( 923 base_dir=self.smb_cwd, 924 path=[path], 925 prompt=[""] 926 ) 927 except (BrokenPipeError, KeyboardInterrupt) as e: 928 print("[!] Interrupted.") 929 self.close_smb_session() 930 self.init_smb_session()
Recursively lists the directory structure of the SMB share starting from the specified path.
This function prints a visual representation of the directory tree of the remote SMB share. It uses recursion to navigate through directories and lists all files and subdirectories in each directory. The output is color-coded and formatted to enhance readability, with directories highlighted in cyan.
Args: path (str, optional): The starting path on the SMB share from which to begin listing the tree. Defaults to the root of the current share.
956 def set_cwd(self, path=None): 957 """ 958 Sets the current working directory on the SMB share to the specified path. 959 960 This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory. 961 If the specified path is not a directory, the cwd remains unchanged. 962 963 Parameters: 964 path (str): The path to set as the current working directory. 965 966 Raises: 967 ValueError: If the specified path is not a directory. 968 """ 969 970 if path is not None: 971 # Set path separators to ntpath sep 972 if '/' in path: 973 path = path.replace('/', ntpath.sep) 974 975 if path.startswith(ntpath.sep): 976 # Absolute path 977 path = path + ntpath.sep 978 else: 979 # Relative path to the CWD 980 if len(self.smb_cwd) == 0: 981 path = path + ntpath.sep 982 else: 983 path = self.smb_cwd + ntpath.sep + path 984 985 # Path normalization 986 path = ntpath.normpath(path) 987 path = re.sub(r'\\+', r'\\', path) 988 989 if path in ["", ".", ".."]: 990 self.smb_cwd = "" 991 else: 992 if self.path_isdir(pathFromRoot=path.strip(ntpath.sep)): 993 # Path exists on the remote 994 self.smb_cwd = ntpath.normpath(path) 995 else: 996 # Path does not exists or is not a directory on the remote 997 print("[!] Remote directory '%s' does not exist." % path)
Sets the current working directory on the SMB share to the specified path.
This method updates the current working directory (cwd) of the SMB session to the given path if it is a valid directory. If the specified path is not a directory, the cwd remains unchanged.
Parameters: path (str): The path to set as the current working directory.
Raises: ValueError: If the specified path is not a directory.