mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 16:04:31 +00:00
ec50f82cc8
Separation of concerns: merge silent/UI actions & modularize screens. ProleInstallerBase (actions.py): Created shared base class with 37 deduplicated methods previously duplicated between ProleSilentInstaller and ProleInstaller. Namespace, environment, secret, deployment, port-forward, authority/repair, image, and logging helpers now defined once. Subclasses override _get_input() to bridge their data-access layers. screens.py -> screens/ package (18 mixin modules): Split 10,234-line monolithic screens.py into focused mixin modules: base, navigation, welcome, dependencies, network, environment, database, cluster, services, security, ollama, supabase, docker, build, packaging, deploy, validate, cfg. __init__.py composes ProleInstaller from all mixins and re-exports has_display(), main() for full backward compatibility. All 37 tests pass with no regressions.
242 lines
11 KiB
Python
242 lines
11 KiB
Python
"""Network scan screen."""
|
|
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
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
|
|
|
|
|
|
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
|
|
if not GLOBAL_SCAN_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_SCAN_FRAMES) < max_frames:
|
|
# Create a copy and resize
|
|
frame = gif.copy().convert('RGBA')
|
|
frame.thumbnail((24, 24), Image.LANCZOS)
|
|
GLOBAL_SCAN_FRAMES.append(ImageTk.PhotoImage(frame))
|
|
gif.seek(len(GLOBAL_SCAN_FRAMES))
|
|
except EOFError:
|
|
pass
|
|
except Exception as e:
|
|
print(f"Error loading scan animation: {e}")
|
|
self._scan_frames = GLOBAL_SCAN_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 self._scan_frames:
|
|
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 10s timeout and summary analysis
|
|
# Set bufsize=0 for truly unbuffered binary stream reading
|
|
# Ensure PYTHONUNBUFFERED=1 is set in case prole-agent is/uses Python
|
|
env = os.environ.copy()
|
|
env['PYTHONUNBUFFERED'] = '1'
|
|
process = subprocess.Popen([str(scan_binary), "-t", "10"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
bufsize=0,
|
|
env=env,
|
|
cwd=str(scan_dir))
|
|
|
|
# Capture output in real-time using larger reads to avoid overhead
|
|
fd = process.stdout.fileno()
|
|
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))
|
|
|
|
while True:
|
|
try:
|
|
# Use a smaller chunk size to encourage more frequent reads
|
|
chunk_bytes = os.read(fd, 1024)
|
|
except (EOFError, OSError):
|
|
chunk_bytes = b''
|
|
|
|
if not chunk_bytes:
|
|
if process.poll() is not None:
|
|
break
|
|
# If we have pending output, flush it before sleeping
|
|
if pending_output:
|
|
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 = time.time()
|
|
time.sleep(0.01)
|
|
continue
|
|
|
|
text = chunk_bytes.decode('utf-8', errors='replace')
|
|
pending_output.append(text)
|
|
|
|
# Update more frequently if we have enough data or enough time has passed
|
|
now = time.time()
|
|
if now - last_ui_update > 0.05 or sum(len(x) for x in pending_output) > 2048:
|
|
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
|
|
|
|
# Accumulate for line parsing (for AD/KDC detection)
|
|
for char in text:
|
|
line_buffer.append(char)
|
|
if char == '\n':
|
|
line = "".join(line_buffer)
|
|
line_buffer = []
|
|
|
|
# Parse "KDC is: IP" from the new 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
|
|
|
|
process.wait()
|
|
|
|
# Final flush of any remaining output
|
|
if pending_output:
|
|
text_to_flush = "".join(pending_output)
|
|
self.safe_after(lambda t=text_to_flush: self.scan_results_console.write(t))
|
|
|
|
if process.returncode == 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 {process.returncode})"))
|
|
|
|
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()
|