Coordinates the three recon runners and tracks scan state.
| 132 | "started_at": self.started_at.isoformat() if self.started_at else None, |
| 133 | "completed_at": self.completed_at.isoformat() if self.completed_at else None, |
| 134 | "results": {r.value: result.to_dict() for r, result in self.results.items()}, |
| 135 | "handed_off": self.handed_off, |
| 136 | "handoff_zap_scan_id": self.handoff_zap_scan_id, |
| 137 | "error_message": self.error_message, |
| 138 | "duration_seconds": ( |
| 139 | (self.completed_at or datetime.now()) - self.started_at |
| 140 | ).total_seconds() if self.started_at else 0, |
| 141 | } |
| 142 | |
| 143 | |
| 144 | class ReconEngine: |
| 145 | """Coordinates the three recon runners and tracks scan state.""" |
| 146 | |
| 147 | def __init__(self, shared_data=None): |
| 148 | self.shared_data = shared_data |
| 149 | self._lock = threading.Lock() |
| 150 | self.active_scans: Dict[str, ReconScanState] = {} |
| 151 | self._scan_history: deque = deque(maxlen=100) |
| 152 | self._tool_paths: Dict[str, str] = {} |
| 153 | self._detect_tools() |
| 154 | threading.Thread(target=self._reaper_loop, daemon=True, name="recon-reaper").start() |
| 155 | |
| 156 | def is_available(self) -> bool: |
| 157 | # Pre-flight recon is lightweight network work — a TLS handshake, DNS |
| 158 | # lookups and (optionally) an ffuf content sweep — the same class as |
| 159 | # the CLI vuln scanners, which run on any board. It used to require |
| 160 | # server mode (8GB RAM), which was the wrong bar and hid it on smaller |
| 161 | # boards where the rest of the Adv Scan tab now works. It runs anywhere; |
| 162 | # content discovery degrades to status=error if ffuf isn't installed. |
| 163 | return True |
| 164 | |
| 165 | def _detect_tools(self) -> None: |
| 166 | for tool in ("ffuf",): |
| 167 | path = shutil.which(tool) |
| 168 | if path: |
| 169 | self._tool_paths[tool] = path |
| 170 | logger.info(f"Found {tool} at {path}") |
| 171 | else: |
| 172 | logger.warning(f"{tool} not found in PATH; CONTENT_DISCOVERY will report status=error") |
| 173 | |
| 174 | def start_scan( |
| 175 | self, |
| 176 | target: str, |
| 177 | recon_types: List[ReconType], |
| 178 | timeout: int = DEFAULT_ENGINE_TIMEOUT, |
| 179 | ) -> str: |
| 180 | if not self.is_available(): |
| 181 | raise RuntimeError("Recon engine is not available") |
| 182 | if not recon_types: |
| 183 | raise ValueError("recon_types must not be empty") |
| 184 | |
| 185 | scan_id = f"RECON-{uuid.uuid4().hex[:12]}-{int(time.time())}" |
| 186 | state = ReconScanState(scan_id=scan_id, target=target, recon_types=list(recon_types)) |
| 187 | with self._lock: |
| 188 | self.active_scans[scan_id] = state |
| 189 | |
| 190 | threading.Thread( |
| 191 | target=self._run_scan, |
no outgoing calls
no test coverage detected