Manages network-based intelligence for vulnerabilities and credentials
| 23 | |
| 24 | |
| 25 | class NetworkIntelligence: |
| 26 | """Manages network-based intelligence for vulnerabilities and credentials""" |
| 27 | |
| 28 | def __init__(self, shared_data): |
| 29 | self.shared_data = shared_data |
| 30 | self.logger = Logger(name="NetworkIntelligence", level=logging.INFO) |
| 31 | |
| 32 | # Network context tracking |
| 33 | self.current_network = None |
| 34 | self.previous_network = None |
| 35 | self.network_change_time = None |
| 36 | self.network_history = {} |
| 37 | |
| 38 | # Intelligence storage paths |
| 39 | self.intelligence_dir = self._resolve_intelligence_dir() |
| 40 | self._rebuild_file_paths() |
| 41 | |
| 42 | # State management |
| 43 | self.active_vulnerabilities = {} # network_id -> {vuln_id: vuln_data} |
| 44 | self.active_credentials = {} # network_id -> {cred_id: cred_data} |
| 45 | self.resolved_vulnerabilities = {} |
| 46 | self.resolved_credentials = {} |
| 47 | self.network_profiles = {} |
| 48 | |
| 49 | # Configuration |
| 50 | self.resolution_timeout = shared_data.config.get('network_resolution_timeout', 3600) # 1 hour |
| 51 | self.confirmation_scans = shared_data.config.get('network_confirmation_scans', 3) |
| 52 | self.network_change_grace_period = shared_data.config.get('network_change_grace', 300) # 5 minutes |
| 53 | |
| 54 | # Initialize intelligence system |
| 55 | self.setup_intelligence_directory() |
| 56 | self.load_intelligence_data() |
| 57 | |
| 58 | def _resolve_intelligence_dir(self) -> str: |
| 59 | candidate = getattr(self.shared_data, 'network_intelligence_dir', None) |
| 60 | if candidate: |
| 61 | return candidate |
| 62 | return os.path.join(self.shared_data.datadir, 'intelligence') |
| 63 | |
| 64 | def _rebuild_file_paths(self): |
| 65 | self.network_profiles_file = os.path.join(self.intelligence_dir, 'network_profiles.json') |
| 66 | self.active_findings_file = os.path.join(self.intelligence_dir, 'active_findings.json') |
| 67 | self.resolved_findings_file = os.path.join(self.intelligence_dir, 'resolved_findings.json') |
| 68 | |
| 69 | def set_storage_root(self, intelligence_dir: str): |
| 70 | """Switch to a new intelligence directory (per-network).""" |
| 71 | if not intelligence_dir or intelligence_dir == self.intelligence_dir: |
| 72 | return |
| 73 | self.save_intelligence_data() |
| 74 | self.intelligence_dir = intelligence_dir |
| 75 | self._rebuild_file_paths() |
| 76 | self.setup_intelligence_directory() |
| 77 | self.load_intelligence_data() |
| 78 | self.logger.info(f"Network intelligence storage switched to: {self.intelligence_dir}") |
| 79 | |
| 80 | def setup_intelligence_directory(self): |
| 81 | """Create intelligence directory structure""" |
| 82 | try: |
no outgoing calls
no test coverage detected