Update the shared data with the latest system information.
(self)
| 267 | df = pd.DataFrame(columns=["IP", "Hostname", "MAC Address", "Port", "Vulnerabilities"]) |
| 268 | df.to_csv(self.shared_data.vuln_summary_file, index=False) |
| 269 | self.shared_data.vulnnbr = 0 |
| 270 | logger.info("Vulnerability summary file created.") |
| 271 | else: |
| 272 | # Get alive hosts from SQLite database instead of CSV |
| 273 | try: |
| 274 | db_stats = self.shared_data.db.get_stats() |
| 275 | alive_hosts = self.shared_data.db.get_all_hosts() |
| 276 | alive_macs = { |
| 277 | h['mac'] for h in alive_hosts |
| 278 | if h.get('status') == 'alive' and h.get('mac') != 'STANDALONE' |
| 279 | } |
| 280 | logger.debug(f"Loaded {len(alive_macs)} alive MACs from database") |
| 281 | except Exception as e: |
| 282 | logger.warning(f"Could not get alive MACs from database: {e}") |
| 283 | alive_macs = set() |
| 284 | |
| 285 | |
| 286 | try: |
| 287 | # Check if file is not empty and has content |
| 288 | if os.path.getsize(self.shared_data.vuln_summary_file) > 0: |
| 289 | with open(self.shared_data.vuln_summary_file, 'r') as file: |
| 290 | df = pd.read_csv(file) |
| 291 | else: |
| 292 | logger.debug("vuln_summary file is empty, initializing with empty DataFrame") |
| 293 | df = pd.DataFrame(columns=["IP", "Hostname", "MAC Address", "Port", "Vulnerabilities"]) |
| 294 | except (pd.errors.EmptyDataError, pd.errors.ParserError) as e: |
| 295 | logger.warning(f"Could not parse vuln_summary file: {e}, creating new one") |
| 296 | df = pd.DataFrame(columns=["IP", "Hostname", "MAC Address", "Port", "Vulnerabilities"]) |
| 297 | all_vulnerabilities = set() |
| 298 | |
| 299 | for index, row in df.iterrows(): |
| 300 | mac_address = row["MAC Address"] |
| 301 | if mac_address in alive_macs and mac_address != "STANDALONE": |
| 302 | vulnerabilities = row["Vulnerabilities"] |
| 303 | if pd.isna(vulnerabilities) or not isinstance(vulnerabilities, str): |
| 304 | continue |
| 305 | |
| 306 | if vulnerabilities and isinstance(vulnerabilities, str): |
| 307 | all_vulnerabilities.update(vulnerabilities.split("; ")) |
| 308 | |
| 309 | self.shared_data.vulnnbr = len(all_vulnerabilities) |
| 310 | logger.debug(f"Updated vulnerabilities count: {self.shared_data.vulnnbr}") |
| 311 | |
| 312 | if os.path.exists(self.shared_data.livestatusfile): |
| 313 | try: |
| 314 | # Check if file is not empty and has content |
| 315 | if os.path.getsize(self.shared_data.livestatusfile) > 0: |
| 316 | with open(self.shared_data.livestatusfile, 'r+') as livestatus_file: |
| 317 | livestatus_df = pd.read_csv(livestatus_file) |
| 318 | if not livestatus_df.empty: |
| 319 | livestatus_df.loc[0, 'Vulnerabilities Count'] = self.shared_data.vulnnbr |
| 320 | livestatus_df.to_csv(self.shared_data.livestatusfile, index=False) |
| 321 | logger.debug(f"Updated livestatusfile with vulnerability count: {self.shared_data.vulnnbr}") |
| 322 | else: |
| 323 | logger.debug("livestatus file is empty, skipping update") |
| 324 | except (pd.errors.EmptyDataError, pd.errors.ParserError) as e: |
| 325 | logger.warning(f"Could not parse livestatus file: {e}") |
| 326 | else: |
no test coverage detected