MCPcopy Create free account
hub / github.com/PierreGode/Ragnar / update_shared_data

Method update_shared_data

pager/pager_display.py:1174–1226  ·  view source on GitHub ↗
(self)

Source from the content-addressed store, hash-verified

1172 logger.error(f"Error in update_vuln_count: {e}")
1173
1174 def update_shared_data(self):
1175 with self.semaphore:
1176 try:
1177 # Read stats from SQLite DB (primary source)
1178 if self.shared_data.db is not None:
1179 try:
1180 db_stats = self.shared_data.db.get_stats()
1181 hosts = self.shared_data.db.get_all_hosts()
1182 alive = [h for h in hosts if h.get('status') == 'alive']
1183 self.shared_data.targetnbr = len(alive)
1184 self.shared_data.networkkbnbr = db_stats.get('total_hosts', 0)
1185 self.shared_data.portnbr = sum(
1186 len(h['ports'].split(',')) for h in alive if h.get('ports')
1187 )
1188 self.shared_data.vulnnbr = db_stats.get('hosts_with_vulns', 0)
1189 except Exception as e:
1190 logger.debug(f"DB stats read failed: {e}")
1191 elif os.path.exists(self.shared_data.livestatusfile):
1192 # Fallback: read from CSV
1193 with open(self.shared_data.livestatusfile, 'r') as file:
1194 reader = csv.DictReader(file)
1195 for row in reader:
1196 self.shared_data.portnbr = int(row.get('Total Open Ports', 0) or 0)
1197 self.shared_data.targetnbr = int(row.get('Alive Hosts Count', 0) or 0)
1198 self.shared_data.networkkbnbr = int(row.get('All Known Hosts Count', 0) or 0)
1199 self.shared_data.vulnnbr = int(row.get('Vulnerabilities Count', 0) or 0)
1200 break
1201
1202 crackedpw_files = glob.glob(f"{self.shared_data.crackedpwddir}/*.csv")
1203 total_passwords = 0
1204 for filepath in crackedpw_files:
1205 try:
1206 with open(filepath, 'r') as f:
1207 reader = csv.reader(f)
1208 next(reader, None)
1209 total_passwords += sum(1 for _ in reader)
1210 except Exception:
1211 pass
1212 self.shared_data.crednbr = total_passwords
1213
1214 total_data = sum([len(files) for r, d, files in os.walk(self.shared_data.datastolendir)])
1215 self.shared_data.datanbr = total_data
1216
1217 total_zombies = sum([len(files) for r, d, files in os.walk(self.shared_data.zombiesdir)])
1218 self.shared_data.zombiesnbr = total_zombies
1219
1220 self.shared_data.update_stats()
1221 self.shared_data.wifi_connected = self.is_wifi_connected()
1222
1223 except FileNotFoundError as e:
1224 logger.debug(f"Data file not ready: {e}")
1225 except Exception as e:
1226 logger.error(f"Error updating shared data: {e}")
1227
1228 def display_comment(self, status):
1229 comment = self.commentaire_ia.get_commentaire(status)

Callers 1

Calls 7

is_wifi_connectedMethod · 0.95
get_all_hostsMethod · 0.80
debugMethod · 0.80
update_statsMethod · 0.80
errorMethod · 0.80
get_statsMethod · 0.45
getMethod · 0.45

Tested by

no test coverage detected