(self, target: str)
| 279 | logger.error(f"Recon runner {rt.value} raised in {scan_id}: {exc}\n{traceback.format_exc()}") |
| 280 | result = ReconResult( |
| 281 | recon_type=rt, |
| 282 | target=target, |
| 283 | status="error", |
| 284 | error_message=str(exc), |
| 285 | ) |
| 286 | state.results[rt] = result |
| 287 | except TimeoutError: |
| 288 | state.error_message = f"engine timeout after {timeout}s" |
| 289 | logger.warning(f"Recon scan {scan_id} hit engine timeout") |
| 290 | except Exception as exc: |
| 291 | import traceback |
| 292 | state.error_message = str(exc) |
| 293 | logger.error(f"Recon scan {scan_id} engine failed: {exc}\n{traceback.format_exc()}") |
| 294 | |
| 295 | state.status = "completed" |
| 296 | state.completed_at = datetime.now() |
| 297 | self._scan_history.append(scan_id) |
| 298 | |
| 299 | def _run_port_scan(self, target: str) -> ReconResult: |
| 300 | """Discover open web/management ports on the target and classify each as |
| 301 | http vs https (via a TLS handshake). The result feeds the handoff gate, |
| 302 | so the operator can point ZAP at the port(s) that are actually up instead |
| 303 | of guessing 80/443. Lightweight parallel TCP connect — no nmap needed.""" |
| 304 | result = ReconResult(recon_type=ReconType.PORT_SCAN, target=target) |
| 305 | started = time.monotonic() |
| 306 | host, _ = _split_host_port(target, default_port=80) |
| 307 | try: |
| 308 | server_hostname = None if _is_ip(host) else host |
| 309 | except Exception: |
| 310 | server_hostname = host |
| 311 | |
| 312 | open_ports: List[Dict[str, Any]] = [] |
| 313 | lock = threading.Lock() |
| 314 | |
| 315 | def probe(port: int): |
| 316 | try: |
| 317 | raw = socket.create_connection((host, port), timeout=2.5) |
| 318 | except Exception: |
| 319 | return # closed / filtered |
| 320 | scheme = "http" |
| 321 | try: |
| 322 | ctx = ssl.create_default_context() |
| 323 | ctx.check_hostname = False |
| 324 | ctx.verify_mode = ssl.CERT_NONE |
| 325 | tls = ctx.wrap_socket(raw, server_hostname=server_hostname) |
| 326 | scheme = "https" |
| 327 | try: |
| 328 | tls.close() |
| 329 | except Exception: |
| 330 | pass |
| 331 | except Exception: |
| 332 | try: |
| 333 | raw.close() |
| 334 | except Exception: |
| 335 | pass |
| 336 | with lock: |
| 337 | open_ports.append({"port": port, "scheme": scheme, |
| 338 | "url": f"{scheme}://{host}:{port}"}) |
nothing calls this directly
no test coverage detected