mirror of
https://github.com/dredx/prole.git
synced 2026-09-27 15:24:30 +00:00
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""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()
|