Initiates the network scan, updates the netkb file, and displays the results. Now also stores results in memory for immediate orchestrator access.
(self, job=None)
| 1637 | self.logger.error(f"Error in clean_scan_results: {e}") |
| 1638 | |
| 1639 | def scan(self, job=None): |
| 1640 | """ |
| 1641 | Initiates the network scan, updates the netkb file, and displays the results. |
| 1642 | Now also stores results in memory for immediate orchestrator access. |
| 1643 | """ |
| 1644 | interface_override = getattr(job, 'interface', None) if job else None |
| 1645 | network_hint = getattr(job, 'network_cidr', None) if job else None |
| 1646 | if not network_hint and job and getattr(job, 'ip_address', None) and getattr(job, 'cidr', None): |
| 1647 | network_hint = f"{job.ip_address}/{job.cidr}" |
| 1648 | |
| 1649 | previous_interface = self.arp_scan_interface |
| 1650 | previous_network_hint = self._active_scan_network |
| 1651 | |
| 1652 | if interface_override: |
| 1653 | self.arp_scan_interface = interface_override |
| 1654 | self._active_scan_network = network_hint |
| 1655 | |
| 1656 | try: |
| 1657 | self.shared_data.ragnarorch_status = "NetworkScanner" |
| 1658 | job_descriptor = "" |
| 1659 | if job and getattr(job, 'ssid', None): |
| 1660 | job_descriptor = f" for {job.ssid} ({self.arp_scan_interface})" |
| 1661 | self.logger.info(f"Starting Network Scanner{job_descriptor}") |
| 1662 | network = self.get_network() |
| 1663 | self.get_gateway_info() |
| 1664 | self.shared_data.bjornstatustext2 = str(network) |
| 1665 | portstart = self.shared_data.portstart |
| 1666 | portend = self.shared_data.portend |
| 1667 | extra_ports = self.shared_data.portlist |
| 1668 | scanner = self.ScanPorts(self, network, portstart, portend, extra_ports) |
| 1669 | ip_data, open_ports, all_ports, csv_result_file, netkbfile, alive_ips = scanner.start() |
| 1670 | |
| 1671 | # Convert alive MACs to use pseudo-MACs for hosts without real MAC addresses |
| 1672 | alive_macs = set() |
| 1673 | for i, mac in enumerate(ip_data.mac_list): |
| 1674 | if mac == "00:00:00:00:00:00" and i < len(ip_data.ip_list): |
| 1675 | # Convert to pseudo-MAC using the same logic as update_netkb |
| 1676 | ip = ip_data.ip_list[i] |
| 1677 | ip_parts = ip.split('.') |
| 1678 | if len(ip_parts) == 4: |
| 1679 | pseudo_mac = f"00:00:{int(ip_parts[0]):02x}:{int(ip_parts[1]):02x}:{int(ip_parts[2]):02x}:{int(ip_parts[3]):02x}" |
| 1680 | alive_macs.add(pseudo_mac) |
| 1681 | self.logger.debug(f"Added pseudo-MAC {pseudo_mac} to alive_macs for IP {ip}") |
| 1682 | else: |
| 1683 | alive_macs.add(mac) |
| 1684 | |
| 1685 | table = Table(title="Scan Results", show_lines=True) if Table else None |
| 1686 | if table: |
| 1687 | table.add_column("IP", style="cyan", no_wrap=True) |
| 1688 | table.add_column("Hostname", style="cyan", no_wrap=True) |
| 1689 | table.add_column("Alive", style="cyan", no_wrap=True) |
| 1690 | table.add_column("MAC Address", style="cyan", no_wrap=True) |
| 1691 | for port in all_ports: |
| 1692 | table.add_column(f"{port}", style="green") |
| 1693 | |
| 1694 | netkb_data = [] |
| 1695 | for ip, ports, hostname, mac in zip(ip_data.ip_list, open_ports.values(), ip_data.hostname_list, ip_data.mac_list): |
| 1696 | if self.blacklistcheck and (mac in self.mac_scan_blacklist or ip in self.ip_scan_blacklist): |
no test coverage detected