Refresh host/cred/vuln lists from CSV files (max once per 5s).
(self)
| 338 | # ------------------------------------------------------------------ |
| 339 | |
| 340 | def refresh_list_data(self): |
| 341 | """Refresh host/cred/vuln lists from CSV files (max once per 5s).""" |
| 342 | now = time.time() |
| 343 | if now - self._last_data_refresh < 5: |
| 344 | return |
| 345 | self._last_data_refresh = now |
| 346 | |
| 347 | # Hosts from SQLite DB (primary) or netkb.csv (fallback) |
| 348 | self._hosts_data = [] |
| 349 | try: |
| 350 | if self.shared_data.db is not None: |
| 351 | for h in self.shared_data.db.get_all_hosts(): |
| 352 | if h.get('mac') == 'STANDALONE': |
| 353 | continue |
| 354 | self._hosts_data.append({ |
| 355 | 'ip': h.get('ip', '?'), |
| 356 | 'hostname': h.get('hostname', ''), |
| 357 | 'alive': h.get('status') == 'alive', |
| 358 | 'ports': h.get('ports', ''), |
| 359 | 'mac': h.get('mac', ''), |
| 360 | }) |
| 361 | elif os.path.exists(self.shared_data.netkbfile): |
| 362 | with open(self.shared_data.netkbfile, 'r') as f: |
| 363 | reader = csv.DictReader(f) |
| 364 | for row in reader: |
| 365 | if row.get("MAC Address") == "STANDALONE": |
| 366 | continue |
| 367 | self._hosts_data.append({ |
| 368 | 'ip': row.get('IPs', '?'), |
| 369 | 'hostname': row.get('Hostnames', ''), |
| 370 | 'alive': row.get('Alive', '0') == '1', |
| 371 | 'ports': row.get('Ports', ''), |
| 372 | 'mac': row.get('MAC Address', ''), |
| 373 | }) |
| 374 | except Exception as e: |
| 375 | logger.debug(f"Error reading hosts: {e}") |
| 376 | |
| 377 | # Credentials from crackedpwd/*.csv |
| 378 | self._creds_data = [] |
| 379 | try: |
| 380 | cred_files = glob.glob(f"{self.shared_data.crackedpwddir}/*.csv") |
| 381 | for filepath in cred_files: |
| 382 | service = os.path.basename(filepath).replace('.csv', '').upper() |
| 383 | try: |
| 384 | with open(filepath, 'r') as f: |
| 385 | reader = csv.reader(f) |
| 386 | header = next(reader, None) |
| 387 | for row in reader: |
| 388 | if len(row) >= 3: |
| 389 | self._creds_data.append({ |
| 390 | 'service': service, |
| 391 | 'host': row[0] if row else '', |
| 392 | 'user': row[1] if len(row) > 1 else '', |
| 393 | 'password': row[2] if len(row) > 2 else '', |
| 394 | }) |
| 395 | except Exception: |
| 396 | pass |
| 397 | except Exception as e: |
no test coverage detected