Export database to CSV format (for backward compatibility). Args: csv_path: Path to CSV file (default: netkb.csv) Returns: bool: True if successful
(self, csv_path: str = None)
| 1367 | delim = ',' if ',' in ports_str else ';' |
| 1368 | for token in ports_str.split(delim): |
| 1369 | token = token.strip() |
| 1370 | if token and token != '0' and token.isdigit(): |
| 1371 | total_ports += 1 |
| 1372 | snapshot['total_ports'] = total_ports |
| 1373 | |
| 1374 | return snapshot |
| 1375 | except Exception as e: |
| 1376 | logger.warning(f"get_count_snapshot failed: {e}") |
| 1377 | return snapshot |
| 1378 | |
| 1379 | def export_to_csv(self, csv_path: str = None) -> bool: |
| 1380 | """ |
| 1381 | Export database to CSV format (for backward compatibility). |
| 1382 | |
| 1383 | Args: |
| 1384 | csv_path: Path to CSV file (default: netkb.csv) |
| 1385 | |
| 1386 | Returns: |
| 1387 | bool: True if successful |
| 1388 | """ |
| 1389 | if csv_path is None: |
| 1390 | csv_path = self.netkb_csv |
| 1391 | |
| 1392 | try: |
| 1393 | hosts = self.get_all_hosts() |
| 1394 | |
| 1395 | if not hosts: |
| 1396 | logger.warning("No hosts to export") |
| 1397 | return False |
| 1398 | |
| 1399 | # Define CSV columns |
| 1400 | fieldnames = [ |
| 1401 | 'MAC', 'IP', 'Hostname', 'Vendor', 'Ports', 'Services', |
| 1402 | 'Nmap Vulnerabilities', 'Alive Count', 'Network Profile', |
| 1403 | 'Scanner', 'ssh_connector', 'rdp_connector', 'ftp_connector', |
| 1404 | 'smb_connector', 'telnet_connector', 'sql_connector', |
| 1405 | 'steal_files_ssh', 'steal_files_rdp', 'steal_files_ftp', |
| 1406 | 'steal_files_smb', 'steal_files_telnet', 'steal_data_sql', |
| 1407 | 'nmap_vuln_scanner', 'Notes', 'First Seen', 'Last Seen', 'Status' |
| 1408 | ] |
| 1409 | |
| 1410 | with open(csv_path, 'w', newline='', encoding='utf-8') as csvfile: |
| 1411 | writer = csv.DictWriter(csvfile, fieldnames=fieldnames) |
| 1412 | writer.writeheader() |
| 1413 | |
| 1414 | for host in hosts: |
| 1415 | row = { |
| 1416 | 'MAC': host.get('mac', ''), |
| 1417 | 'IP': host.get('ip', ''), |
| 1418 | 'Hostname': host.get('hostname', ''), |
| 1419 | 'Vendor': host.get('vendor', ''), |
| 1420 | 'Ports': host.get('ports', ''), |
| 1421 | 'Services': host.get('services', ''), |
| 1422 | 'Nmap Vulnerabilities': host.get('vulnerabilities', ''), |
| 1423 | 'Alive Count': host.get('alive_count', 0), |
| 1424 | 'Network Profile': host.get('network_profile', ''), |
| 1425 | 'Scanner': host.get('scanner_status', ''), |
| 1426 | 'ssh_connector': host.get('ssh_connector', ''), |
nothing calls this directly
no test coverage detected