"""Network scan screen. This screen combines two discovery activities into a single operator workflow: - Kerberos authority / KDC candidate detection (via `scan/network-agent`) - Ollama server discovery (via `etc/init_ollama.sh scan`) """ from __future__ import annotations import os import re import socket import threading import time from pathlib import Path import tkinter as tk from tkinter import ttk from knoe import screen as ui from knoe.core.env import ( DEFAULT_OLLAMA_PORT, get_resource_path, _format_ollama_host, resolve_knoe_home, ) from knoe.core.ollama_scan import run_ollama_scan from knoe.core.stream_exec import run_streaming_cmd def _get_global_scan_frames() -> list: """Lazy accessor to avoid circular import with __init__.py.""" import knoe.ui.screens as _pkg return _pkg.GLOBAL_SCAN_FRAMES class NetworkScreenMixin: """Network scan screen.""" def _render_network_scan_page(self): """Render combined Network + Ollama scan page.""" # 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, "knoe.dev", fill="#6e6e73", font=("SF Pro Text", 32, "bold"), anchor="ne", ) ui.canvas_text( self, right_margin, 85, "infrastructure.auto()", fill="#6e6e73", font=("SF Pro Text", 18), anchor="ne", ) self._render_title("Network Configuration + Ollama", y=150) self._render_paragraph( "Run a single combined scan to (1) identify Kerberos authority / KDC candidates " "and (2) discover Ollama servers on the subnet.", y=205, wrap=820, ) # Shared status x = 48 y = 260 self._combined_status_var = tk.StringVar(value="Ready") status_item = ui.canvas_text( self, x, y, "Status: Ready", fill="#6e6e73", font=("SF Pro Text", 10), ) self._canvas_items.append(status_item) def _update_status_text(*_args): try: self.bg_canvas.itemconfig( status_item, text=f"Status: {self._combined_status_var.get()}" ) except Exception: pass self._combined_status_var.trace_add("write", _update_status_text) # Start Combined Scan Button self.scan_btn = tk.Button( self.bg_canvas, text="Start Combined Scan", command=self._run_network_scan, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=14, pady=6, ) btn_window = self.bg_canvas.create_window(x + 160, y - 8, window=self.scan_btn, anchor="nw") self._canvas_items.append(btn_window) self._overlay_widgets.append(self.scan_btn) # Shared console — dark terminal with green text (Ollama-style) console_y = y + 28 console_frame = tk.Frame( self.bg_canvas, bg="#1e1e1e", highlightbackground="#444", highlightthickness=1, ) console_window = self.bg_canvas.create_window( x, console_y, window=console_frame, anchor="nw", width=900, height=130, ) self._canvas_items.append(console_window) self._overlay_widgets.append(console_frame) console_text = tk.Text( console_frame, bg="#1e1e1e", fg="#00ff00", font=("Menlo", 9), relief="flat", state="disabled", wrap="word", borderwidth=0, highlightthickness=0, insertbackground="#00ff00", ) console_scroll = ttk.Scrollbar( console_frame, orient="vertical", command=console_text.yview, ) console_text.configure(yscrollcommand=console_scroll.set) console_scroll.pack(side="right", fill="y") console_text.pack(side="left", fill="both", expand=True) self._overlay_widgets.append(console_text) self._overlay_widgets.append(console_scroll) self._combined_console = console_text # Result panels panels_y = console_y + 145 # --- KDC candidates (single-select) --- kdc_frame = tk.Frame( self.bg_canvas, bg="white", highlightbackground="#E0E0E0", highlightthickness=1, ) kdc_window = self.bg_canvas.create_window( x, panels_y, window=kdc_frame, anchor="nw", width=440, height=255, ) self._canvas_items.append(kdc_window) self._overlay_widgets.append(kdc_frame) kdc_title = ui.canvas_text( self, x, panels_y - 18, "Kerberos Authority / KDC Candidates", fill="#1d1d1f", font=("SF Pro Text", 12, "bold"), ) self._canvas_items.append(kdc_title) kdc_cols = ("host", "basis") kdc_tree = ttk.Treeview(kdc_frame, columns=kdc_cols, show="headings", height=7) kdc_tree.heading("host", text="Host / IP") kdc_tree.heading("basis", text="Basis") kdc_tree.column("host", width=170, anchor="w") kdc_tree.column("basis", width=240, anchor="w") kdc_tree.pack(side="left", fill="both", expand=True) kdc_scroll = ttk.Scrollbar(kdc_frame, orient="vertical", command=kdc_tree.yview) kdc_tree.configure(yscrollcommand=kdc_scroll.set) kdc_scroll.pack(side="right", fill="y") self._overlay_widgets.append(kdc_tree) self._overlay_widgets.append(kdc_scroll) self._kdc_tree = kdc_tree self._kdc_notice_item = ui.canvas_text( self, x, panels_y + 262, "", fill="#6e6e73", font=("SF Pro Text", 10), ) self._canvas_items.append(self._kdc_notice_item) use_kdc_btn = tk.Button( self.bg_canvas, text="Use Selected Authority", command=self._apply_selected_kdc_candidate, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 10), padx=10, pady=5, ) use_kdc_window = self.bg_canvas.create_window( x, panels_y + 290, window=use_kdc_btn, anchor="nw", ) self._canvas_items.append(use_kdc_window) self._overlay_widgets.append(use_kdc_btn) # --- Ollama servers (multi-select) --- ox = x + 460 ollama_frame = tk.Frame( self.bg_canvas, bg="white", highlightbackground="#E0E0E0", highlightthickness=1, ) ollama_window = self.bg_canvas.create_window( ox, panels_y, window=ollama_frame, anchor="nw", width=440, height=255, ) self._canvas_items.append(ollama_window) self._overlay_widgets.append(ollama_frame) ollama_title = ui.canvas_text( self, ox, panels_y - 18, "Ollama Servers (select zero or many)", fill="#1d1d1f", font=("SF Pro Text", 12, "bold"), ) self._canvas_items.append(ollama_title) o_cols = ("select", "host", "port", "models") o_tree = ttk.Treeview(ollama_frame, columns=o_cols, show="headings", height=7) o_tree.heading("select", text="Select") o_tree.heading("host", text="Host") o_tree.heading("port", text="Port") o_tree.heading("models", text="Models") o_tree.column("select", width=70, anchor="center") o_tree.column("host", width=150, anchor="w") o_tree.column("port", width=55, anchor="center") o_tree.column("models", width=150, anchor="w") o_tree.pack(side="left", fill="both", expand=True) o_scroll = ttk.Scrollbar(ollama_frame, orient="vertical", command=o_tree.yview) o_tree.configure(yscrollcommand=o_scroll.set) o_scroll.pack(side="right", fill="y") self._overlay_widgets.append(o_tree) self._overlay_widgets.append(o_scroll) self._ollama_tree = o_tree self._ollama_notice_item = ui.canvas_text( self, ox, panels_y + 262, "", fill="#6e6e73", font=("SF Pro Text", 10), ) self._canvas_items.append(self._ollama_notice_item) # Toggle selection by clicking the Select column def _on_ollama_click(event): try: item_id = o_tree.identify_row(event.y) col = o_tree.identify_column(event.x) if not item_id: return if col != "#1": return self._toggle_ollama_item(item_id) except Exception: pass o_tree.bind("", _on_ollama_click) # Load any existing selections into internal state self._ollama_selected_keys = set() self._ollama_primary_key = "" self._load_ollama_selection_from_config() # Seed KDC candidates from ansible topology / existing config self._kdc_candidates: dict[str, str] = {} self._seed_kdc_candidates_from_existing_state() self._render_kdc_candidates() # Seed console with topology summary (if present) if getattr(self, "ansible_topology_summary", ""): self._combined_console_write( "[network] " + self.ansible_topology_summary.replace("\n", "\n[network] ") ) self._combined_console_write( "[network] Ansible topology loaded; combined scan can refine detection.\n" ) # ---- Console helpers ------------------------------------------------- def _combined_console_write(self, content: str): def _do_write(): console = getattr(self, "_combined_console", None) if not console: return try: console.configure(state="normal") console.insert("end", content) console.see("end") console.configure(state="disabled") except Exception: pass self.safe_after(_do_write) def _combined_console_line(self, source: str, line: str): src = (source or "").strip().lower() or "scan" msg = (line or "").rstrip("\n") self._combined_console_write(f"[{src}] {msg}\n") def _run_network_scan(self): if getattr(self, "_scan_running", False): return self._action_flags["network_scan.run"] = True self._scan_running = True self._combined_status_var.set("Scanning...") # Disable button while running try: self.scan_btn.configure(state="disabled") except Exception: pass # Clear console console = getattr(self, "_combined_console", None) if console: try: console.configure(state="normal") console.delete("1.0", "end") console.configure(state="disabled") except Exception: pass self._combined_console_line("network", "Initializing scan/network-agent ...") self._combined_console_line("ollama", "Initializing subnet scan ...") # Track completion of both scan activities lock = threading.Lock() remaining = {"network": True, "ollama": True} outcomes: dict[str, tuple[bool, str]] = {} def _mark_done(kind: str, ok: bool, msg: str): nonlocal remaining with lock: remaining.pop(kind, None) outcomes[kind] = (ok, msg) if remaining: return def _finish(): self._scan_running = False try: self.scan_btn.configure(state="normal") except Exception: pass net_ok, _ = outcomes.get("network", (False, "")) ol_ok, _ = outcomes.get("ollama", (False, "")) if net_ok and ol_ok: self._combined_status_var.set("Scan complete") elif net_ok or ol_ok: self._combined_status_var.set("Scan complete (partial)") else: self._combined_status_var.set("Scan failed") self.safe_after(_finish) # --- Thread 1: network-agent scan (network) --- def network_worker(): ansible_kdc = "" try: ansible_kdc = (self.ansible_topology or {}).get("kdc_ip") or "" if not ansible_kdc: ansible_kdc = (self.knoe_cfg_data.get("Network", {}) or {}).get( "KDC_ANSIBLE_DETECTED", "" ) except Exception: ansible_kdc = "" try: scan_binary = get_resource_path("scan/network-agent") if not scan_binary.exists(): self._combined_console_line( "network", f"Scan binary not found at {scan_binary}" ) _mark_done("network", False, "missing binary") return knoe_home = resolve_knoe_home(env={}) scan_dir = knoe_home / "scan" scan_dir.mkdir(parents=True, exist_ok=True) env = os.environ.copy() env["PYTHONUNBUFFERED"] = "1" line_buffer = "" last_ui_update = 0.0 pending_lines: list[str] = [] kdc_re = re.compile( r"(?:KDC\s*(?:is|IP)?\s*[:=]|kerberos_kdc\s*=)\s*(?P[^\s\]]+)", re.IGNORECASE, ) def _maybe_add_kdc_candidate(host: str, basis: str): host = (host or "").strip().strip("[]():,") if not host: return if host in self._kdc_candidates: return self._kdc_candidates[host] = basis self.safe_after(self._render_kdc_candidates) def _maybe_set_kdc_primary(host: str): # Preserve current behavior: only auto-fill if ansible did not already set. if not host or ansible_kdc: return current = (self.kerberos_kdc.get() or "").strip() if current: return self.safe_after(lambda h=host: self.kerberos_kdc.set(h)) self.safe_after(lambda: self.kerberos_enabled.set(True)) try: self.safe_after(self._save_knoe_cfg) except Exception: pass def _flush_pending(force: bool = False): nonlocal pending_lines, last_ui_update if not pending_lines: return now = time.time() if not force: if now - last_ui_update <= 0.05 and sum( len(x) for x in pending_lines ) <= 2048: return chunk = "".join(pending_lines) pending_lines = [] for ln in chunk.splitlines(): if ln: self._combined_console_line("network", ln) last_ui_update = now def _parse_kdc_lines(text: str): nonlocal line_buffer line_buffer += text while "\n" in line_buffer: line, line_buffer = line_buffer.split("\n", 1) line = line.strip() if not line: continue m = kdc_re.search(line) if m: host = (m.group("host") or "").strip() _maybe_add_kdc_candidate(host, "network-agent") _maybe_set_kdc_primary(host) continue # Fallback heuristic: AD / port 88 lines containing IPs if "active directory" in line.lower() or " 88" in line or ":88" in line: for part in line.split(): cand = part.strip("[]():,") try: socket.inet_aton(cand) _maybe_add_kdc_candidate(cand, "heuristic") _maybe_set_kdc_primary(cand) break except Exception: continue def _on_chunk(text: str): pending_lines.append(text) _flush_pending() _parse_kdc_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_kdc_lines("\n") _mark_done("network", rc == 0, f"rc={rc}") except Exception as e: self._combined_console_line("network", f"Scan error: {e}") _mark_done("network", False, "exception") # --- Thread 2: Ollama subnet scan --- def ollama_worker(): try: rc, rows, notice = run_ollama_scan( Path(getattr(self, "project_root", Path.cwd())), on_progress_line=lambda line: self._combined_console_line( "ollama", line ), ) def _update(): self._ollama_rows = rows self._render_ollama_rows(rows) try: self.bg_canvas.itemconfig( self._ollama_notice_item, text=notice or "" ) except Exception: pass self.safe_after(_update) _mark_done("ollama", rc == 0, f"rc={rc}") except Exception as e: self._combined_console_line("ollama", f"Scan error: {e}") _mark_done("ollama", False, "exception") threading.Thread(target=network_worker, daemon=True).start() threading.Thread(target=ollama_worker, daemon=True).start() # ---- KDC candidate UI + persistence --------------------------------- def _seed_kdc_candidates_from_existing_state(self): # Existing config / ansible topology hints try: ansible_kdc = (self.ansible_topology or {}).get("kdc_ip") or "" if not ansible_kdc: ansible_kdc = (self.knoe_cfg_data.get("Network", {}) or {}).get( "KDC_ANSIBLE_DETECTED", "" ) if ansible_kdc: self._kdc_candidates[ansible_kdc] = "ansible topology" except Exception: pass try: current = (self.kerberos_kdc.get() or "").strip() if current: self._kdc_candidates.setdefault(current, "current selection") except Exception: pass def _render_kdc_candidates(self): tree = getattr(self, "_kdc_tree", None) if not tree: return try: for item in tree.get_children(): tree.delete(item) except Exception: pass # Stable ordering: basis priority then host items = sorted(self._kdc_candidates.items(), key=lambda kv: (kv[1] or "", kv[0])) for host, basis in items: try: tree.insert("", "end", values=(host, basis)) except Exception: pass # Preselect when unambiguous try: current = (self.kerberos_kdc.get() or "").strip() if current: for item in tree.get_children(): v = tree.item(item, "values") if v and v[0] == current: tree.selection_set(item) tree.see(item) break elif len(items) == 1: only = tree.get_children()[0] tree.selection_set(only) except Exception: pass try: self.bg_canvas.itemconfig( self._kdc_notice_item, text=( f"Selected: {(self.kerberos_kdc.get() or '').strip()}" if (self.kerberos_kdc.get() or '').strip() else "Selected: (none)" ), ) except Exception: pass def _apply_selected_kdc_candidate(self): tree = getattr(self, "_kdc_tree", None) if not tree: return sel = tree.selection() if not sel: return vals = tree.item(sel[0], "values") if not vals: return host = (vals[0] or "").strip() if not host: return try: self.kerberos_kdc.set(host) self.kerberos_enabled.set(True) except Exception: pass try: self._save_knoe_cfg() except Exception: pass self._render_kdc_candidates() # ---- Ollama multi-select UI + persistence ---------------------------- def _ollama_key(self, host: str, port: str) -> str: h = (host or "").strip() p = (port or "").strip() or DEFAULT_OLLAMA_PORT return f"{h}:{p}" if h else "" def _parse_ollama_key(self, key: str) -> tuple[str, str]: k = (key or "").strip() if ":" not in k: return k, DEFAULT_OLLAMA_PORT host, port = k.rsplit(":", 1) return host.strip(), (port.strip() or DEFAULT_OLLAMA_PORT) def _load_ollama_selection_from_config(self): data = self.knoe_cfg_data.get("Ollama", {}) or {} csv_val = (data.get("OLLAMA_SERVERS", "") or "").strip() selected: set[str] = set() if csv_val: for part in csv_val.split(","): part = part.strip() if part: selected.add(part) # Back-compat: if no list key exists, fall back to legacy single-host settings. if not selected: host = (data.get("OLLAMA_SERVER_HOST") or "").strip() or ( getattr(self, "ollama_server_host", tk.StringVar(value="")).get() or "" ).strip() port = (data.get("OLLAMA_SERVER_PORT") or "").strip() or ( getattr(self, "ollama_server_port", tk.StringVar(value="")).get() or "" ).strip() if host: selected.add(self._ollama_key(host, port)) self._ollama_selected_keys = selected # Primary is whatever the legacy host/port currently indicates (if selected); otherwise first selected. primary = self._ollama_key( (getattr(self, "ollama_server_host", tk.StringVar(value="")).get() or "").strip(), (getattr(self, "ollama_server_port", tk.StringVar(value="")).get() or "").strip(), ) if primary and primary in selected: self._ollama_primary_key = primary else: self._ollama_primary_key = next(iter(sorted(selected)), "") def _render_ollama_rows(self, rows: list[dict]): tree = getattr(self, "_ollama_tree", None) if not tree: return try: for item in tree.get_children(): tree.delete(item) except Exception: pass for r in rows or []: host = (r.get("host", "") or "").strip() port = str((r.get("port", "") or "")).strip() or DEFAULT_OLLAMA_PORT key = self._ollama_key(host, port) models = r.get("models", []) or [] models_text = ", ".join(models) if isinstance(models, list) else str(models) prefix = " [✓] " if key in self._ollama_selected_keys else " [ ] " try: tree.insert("", "end", values=(prefix, host, port, models_text)) except Exception: pass try: n = len(self._ollama_selected_keys) self.bg_canvas.itemconfig( self._ollama_notice_item, text=f"Selected: {n} server(s)" + ( f" (primary: {self._ollama_primary_key})" if self._ollama_primary_key else "" ), ) except Exception: pass def _toggle_ollama_item(self, item_id: str): tree = getattr(self, "_ollama_tree", None) if not tree: return vals = tree.item(item_id, "values") if not vals or len(vals) < 4: return host = (vals[1] or "").strip() port = (vals[2] or "").strip() or DEFAULT_OLLAMA_PORT key = self._ollama_key(host, port) if not key: return if key in self._ollama_selected_keys: self._ollama_selected_keys.remove(key) else: self._ollama_selected_keys.add(key) # Keep primary stable when possible if self._ollama_primary_key and self._ollama_primary_key not in self._ollama_selected_keys: self._ollama_primary_key = next(iter(sorted(self._ollama_selected_keys)), "") if not self._ollama_primary_key and self._ollama_selected_keys: self._ollama_primary_key = next(iter(sorted(self._ollama_selected_keys)), "") # Update row prefix and persist prefix = " [✓] " if key in self._ollama_selected_keys else " [ ] " try: new_vals = list(vals) new_vals[0] = prefix tree.item(item_id, values=new_vals) except Exception: pass self._persist_ollama_multi_selection() self._render_ollama_rows(getattr(self, "_ollama_rows", []) or []) def _persist_ollama_multi_selection(self): data = self.knoe_cfg_data.setdefault("Ollama", {}) selected = sorted(self._ollama_selected_keys) data["OLLAMA_SERVERS"] = ",".join(selected) # Derive legacy single-host keys from primary (if any) primary = self._ollama_primary_key if primary: host, port = self._parse_ollama_key(primary) if host: data["OLLAMA_SERVER_HOST"] = host data["OLLAMA_SERVER_PORT"] = port data["OLLAMA_HOST"] = _format_ollama_host(host, port) try: self.ollama_server_host.set(host) self.ollama_server_port.set(port) except Exception: pass else: # Selecting zero servers is valid. data.pop("OLLAMA_SERVER_HOST", None) data.pop("OLLAMA_SERVER_PORT", None) data.pop("OLLAMA_HOST", None) try: self.ollama_server_host.set("") self.ollama_server_port.set("") except Exception: pass try: self._save_knoe_cfg() except Exception: pass