"""Network scan screen.""" import os import socket import threading import time from pathlib import Path import tkinter as tk from installer import screen as ui from installer.core.env import get_resource_path from installer.core.stream_exec import run_streaming_cmd def _get_global_scan_frames() -> list: """Lazy accessor to avoid circular import with __init__.py.""" import installer.ui.screens as _pkg return _pkg.GLOBAL_SCAN_FRAMES class NetworkScreenMixin: """Network scan screen.""" def _render_network_scan_page(self): # Letterhead at top right (matching welcome screen theme) content_width = self.bg_canvas.winfo_width() or 975 right_margin = content_width - 48 ui.canvas_text( self, right_margin, 40, "Prole", fill="#6e6e73", font=("SF Pro Text", 32, "bold"), anchor="ne", ) ui.canvas_text( self, right_margin, 85, "Infrastructure Automated.", fill="#6e6e73", font=("SF Pro Text", 18), anchor="ne", ) self._render_title("Network Configuration Scan", y=150) self._render_paragraph( "The network prole-agent will detect Kerberos services, Active Directory controllers, and other configuration details required for deployment.", y=210, ) y = 280 self.scan_status_var = tk.StringVar(value="Ready to prole-agent") # Ready to scan text (drawn on canvas) status_item = ui.canvas_text( self, 48, y, "Ready to prole-agent", fill="black", font=("SF Pro Text", 12) ) # Link variable to canvas text def update_status_text(*args): try: self.bg_canvas.itemconfig(status_item, text=self.scan_status_var.get()) except Exception: pass self.scan_status_var.trace_add("write", update_status_text) # Standardized Console Output self.scan_results_console = self._create_console_output( y=320, title="Scan Output", width=880, height=380 ) self.scan_results_text = self.scan_results_console.text if getattr(self, "ansible_topology_summary", ""): self.scan_results_console.write(self.ansible_topology_summary + "\n") self.scan_results_console.write( "Ansible topology loaded; network scan can refine detection.\n" ) y = 740 # Start Network Scan Button (placed on canvas) self.scan_btn = tk.Button( self.bg_canvas, text="Start Network Scan", command=self._run_network_scan, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 11), padx=20, pady=10, ) self.btn_window = self.bg_canvas.create_window( 48, y, window=self.scan_btn, anchor="nw" ) self._overlay_widgets.append(self.scan_btn) self._canvas_items.append(self.btn_window) # Prepare animation frames global_frames = _get_global_scan_frames() if not global_frames: try: from PIL import Image, ImageTk gif_path = get_resource_path("img/prole-type.gif") if gif_path.exists(): gif = Image.open(str(gif_path)) max_frames = 120 try: while len(global_frames) < max_frames: # Create a copy and resize frame = gif.copy().convert("RGBA") frame.thumbnail((24, 24), Image.LANCZOS) global_frames.append(ImageTk.PhotoImage(frame)) gif.seek(len(global_frames)) except EOFError: pass except Exception as e: print(f"Error loading scan animation: {e}") self._scan_frames = global_frames y += 60 # Hint about auto-fill autofill_msg = "Scan results will auto-fill Kerberos and Environment settings." self._canvas_items.append( ui.canvas_text( self, 48, y, autofill_msg, fill="#6e6e73", font=("SF Pro Text", 10, "italic"), ) ) def _animate_scan_button(self, frame_idx=0): if not getattr(self, "_scan_running", False) or not self._scan_frames: if hasattr(self, "scan_btn"): self.scan_btn.config(image="", compound="none") return self.scan_btn.config(image=self._scan_frames[frame_idx], compound="left") next_idx = (frame_idx + 1) % len(self._scan_frames) self.root.after(100, lambda: self._animate_scan_button(next_idx)) def _run_network_scan(self): if getattr(self, "_scan_running", False): return self._action_flags["network_scan.run"] = True self._scan_running = True self._scan_kdc_value = None ansible_kdc = "" try: ansible_kdc = (self.ansible_topology or {}).get("kdc_ip") or "" if not ansible_kdc: ansible_kdc = (self.prole_cfg_data.get("Network", {}) or {}).get( "KDC_ANSIBLE_DETECTED", "" ) except Exception: ansible_kdc = "" self.scan_results_console.clear() self.scan_results_console.write( "Initializing network scan using prole-net/prole-agent ...\n" ) self.scan_status_var.set("Scanning...") # Start animation if getattr(self, "_scan_frames", None): self._animate_scan_button() def worker(): try: # Use the new scan binary scan_binary = get_resource_path("prole-net/prole-agent") if not scan_binary.exists(): self.safe_after( lambda: self.scan_results_console.write( f"Scan binary not found at {scan_binary}\n" ) ) self.safe_after(lambda: self.scan_status_var.set("Scan failed")) self._scan_running = False return # Create writable scan directory for output/cache # This avoids issues with PyInstaller's read-only _MEIPASS directory prole_home = Path.home() / ".prole" scan_dir = prole_home / "scan" scan_dir.mkdir(parents=True, exist_ok=True) # Run scan from writable directory with summary analysis # Ensure PYTHONUNBUFFERED=1 is set in case prole-agent is/uses Python env = os.environ.copy() env["PYTHONUNBUFFERED"] = "1" line_buffer = "" last_ui_update = 0 pending_output = [] def maybe_set_kdc(ip: str): if not ip: return if ansible_kdc: return if self._scan_kdc_value: return self._scan_kdc_value = ip self.safe_after(lambda i=ip: self.kerberos_kdc.set(i)) self.safe_after(lambda: self.kerberos_enabled.set(True)) def _flush_pending(force: bool = False): nonlocal pending_output, last_ui_update if not pending_output: return now = time.time() if not force: if now - last_ui_update <= 0.05 and sum( len(x) for x in pending_output ) <= 2048: return text_to_flush = "".join(pending_output) pending_output = [] self.safe_after(lambda t=text_to_flush: self.scan_results_console.write(t)) last_ui_update = now def _parse_lines(text: str): nonlocal line_buffer line_buffer += text while "\n" in line_buffer: line, line_buffer = line_buffer.split("\n", 1) line = line + "\n" # Parse "KDC is: IP" from the summary analysis format if "KDC is:" in line: try: ip_part = line.split("KDC is:")[1].strip() ip = ip_part.split()[0].strip("[]():,") if ip: maybe_set_kdc(ip) except (IndexError, ValueError): pass elif "Active Directory" in line or "88" in line: parts = line.split() for part in parts: try: socket.inet_aton(part.strip("[]():,")) ip = part.strip("[]():,") maybe_set_kdc(ip) break except socket.error: continue def _on_chunk(text: str): pending_output.append(text) _flush_pending() _parse_lines(text) rc = run_streaming_cmd( [str(scan_binary)], env=env, cwd=str(scan_dir), on_stdout=_on_chunk, on_stderr=_on_chunk, ) _flush_pending(force=True) if line_buffer: _parse_lines("\n") # Final flush of any remaining output if rc == 0: self.safe_after(lambda: self.scan_status_var.set("Scan complete")) else: self.safe_after( lambda: self.scan_status_var.set( f"Scan failed (code {rc})" ) ) self._scan_running = False except Exception as e: self.safe_after( lambda: self.scan_results_console.write(f"Scan error: {str(e)}\n") ) self.safe_after(lambda: self.scan_status_var.set("Scan failed")) self._scan_running = False threading.Thread(target=worker, daemon=True).start()