Initial commit - ICEYOU v1.2 (changes to settings and overrun camera monitoring))

This commit is contained in:
Subinacls
2026-06-05 06:17:07 -04:00
parent 24cd01e46c
commit e3313dbc1b
8 changed files with 742 additions and 191 deletions
+51 -4
View File
@@ -24,6 +24,7 @@ class Monitor:
self.last_trigger_time = 0.0
self.cooldown_seconds = 30 # Avoid spam triggers
self.snooze_until = 0.0 # When > now, skip auto-away based on idle
self.paused = False # When True, monitoring (idle detection, hooks, motion) is suspended
self.event_queue: Queue = Queue()
self._running = False
@@ -124,6 +125,16 @@ class Monitor:
check_interval = obs_cfg.get("check_interval_seconds", 12)
while self._running:
if self.paused:
# While paused, check if we should auto-resume on long idle
idle = get_idle_time_seconds()
if idle >= self.idle_timeout:
self.resume_monitoring()
# After resuming, fall through to normal away logic
else:
time.sleep(5)
continue
idle = get_idle_time_seconds()
now = time.time()
snoozed = now < self.snooze_until
@@ -135,17 +146,17 @@ class Monitor:
if self.white_screen.enabled:
self.white_screen.show()
elif not should_be_away and self.is_away and not snoozed and not self.manual_lock:
# Only auto-leave away if there was actual recent activity AND we are not in a manual/intrusion lock
if idle < self.idle_timeout:
self.is_away = False
self._log_event("state_change", "User returned")
if self.white_screen.enabled:
self.white_screen.hide()
# Camera obstruction detection (only when away + white screen active)
# Camera obstruction detection only before the white screen is shown.
# Once the whiteout is active, obstruction is no longer relevant.
if (self.is_away and
self.white_screen.enabled and
self.white_screen.is_shown and
not self.white_screen.is_shown and
self.camera.obscured_enabled):
now = time.time()
if now - last_obscured_check >= check_interval:
@@ -192,4 +203,40 @@ class Monitor:
self.snooze_until = time.time() + max(self.idle_timeout, 60)
self.last_trigger_time = time.time()
if self.white_screen.enabled:
self.white_screen.hide()
self.white_screen.hide()
def pause_monitoring(self) -> None:
"""Temporarily suspend monitoring (no idle detection, no motion, no intrusions).
Used when the user is actively present.
"""
if self.paused:
return
self.paused = True
self._log_event("monitoring_paused", "Monitoring paused (user present)")
# Stop hooks and device monitoring
self.input_hooks.stop()
self.device_monitor.stop()
# Release camera when monitoring is paused
self.camera.release()
# Hide white screen + stop motion recorders if active
if self.white_screen.is_shown:
self.white_screen.hide()
def resume_monitoring(self) -> None:
"""Resume monitoring after being paused."""
if not self.paused:
return
self.paused = False
self._log_event("monitoring_resumed", "Monitoring resumed")
# Restart hooks
self.input_hooks.start()
if self.config.get("device_monitoring", True):
self.device_monitor.start()
# Reset idle tracking so we don't immediately go away
self.snooze_until = time.time() + max(self.idle_timeout, 60)
self.last_trigger_time = time.time()
+39 -4
View File
@@ -181,9 +181,11 @@ class TrayApp:
def _on_unlocked(self) -> None:
self.monitor.set_away(False)
# Automatically pause monitoring when user is present
self.monitor.pause_monitoring()
try:
if self.icon:
self.icon.notify("Whiteout unlocked", "ICEYOU")
self.icon.notify("Whiteout unlocked - Monitoring paused", "ICEYOU")
except Exception:
pass
@@ -200,8 +202,8 @@ class TrayApp:
def _create_icon(self) -> None:
menu = Menu(
MenuItem("Start Monitoring", self._start_monitoring, default=True),
MenuItem("Stop Monitoring", self._stop_monitoring),
MenuItem("Resume Monitoring", self._resume_monitoring),
MenuItem("Toggle Away Mode", self._toggle_away),
MenuItem("Toggle White Screen", self._toggle_white_screen),
Menu.SEPARATOR,
@@ -225,8 +227,16 @@ class TrayApp:
self.icon.notify("Monitoring started", "ICEYOU")
def _stop_monitoring(self, icon=None, item=None) -> None:
self.monitor.stop()
self.icon.notify("Monitoring stopped", "ICEYOU")
self.monitor.pause_monitoring()
self.icon.notify("Monitoring stopped (paused)", "ICEYOU")
def _pause_monitoring(self, icon=None, item=None) -> None:
self.monitor.pause_monitoring()
self.icon.notify("Monitoring paused", "ICEYOU")
def _resume_monitoring(self, icon=None, item=None) -> None:
self.monitor.resume_monitoring()
self.icon.notify("Monitoring resumed", "ICEYOU")
def _toggle_away(self, icon=None, item=None) -> None:
new_state = not self.monitor.is_away
@@ -247,6 +257,8 @@ class TrayApp:
hotkeys = self.config.get("hotkeys", {})
force_away_combo = hotkeys.get("force_away", "<ctrl>+<alt>+l")
unlock_combo = hotkeys.get("unlock", "<ctrl>+<alt>+u")
toggle_monitor_combo = hotkeys.get("toggle_monitoring", "<ctrl>+<alt>+m")
stop_monitor_combo = hotkeys.get("stop_monitoring", "<ctrl>+<alt>+s")
def on_force_away():
self.monitor.set_away(True)
@@ -261,10 +273,33 @@ class TrayApp:
self.monitor.white_screen.enabled = True
self.monitor.white_screen.show()
def on_toggle_monitoring():
if self.monitor.paused:
self.monitor.resume_monitoring()
try:
self.icon.notify("Monitoring resumed via hotkey", "ICEYOU")
except Exception:
pass
else:
self.monitor.pause_monitoring()
try:
self.icon.notify("Monitoring paused via hotkey", "ICEYOU")
except Exception:
pass
def on_stop_monitoring():
self.monitor.pause_monitoring()
try:
self.icon.notify("Monitoring stopped via hotkey", "ICEYOU")
except Exception:
pass
try:
self._hotkey_listener = keyboard.GlobalHotKeys({
force_away_combo: on_force_away,
unlock_combo: on_unlock,
toggle_monitor_combo: on_toggle_monitoring,
stop_monitor_combo: on_stop_monitoring,
})
except Exception as e:
print(f"Hotkey registration failed (check combo syntax): {e}")
+567
View File
@@ -0,0 +1,567 @@
"""Persistent fullscreen white-screen lockout with embedded password unlock.
Uses a SINGLE Tk root for the lifetime of the app to avoid Tcl async-delete
crashes when show/hide is called rapidly from multiple threads. The Tk root
and mainloop live on a dedicated UI thread. All other threads communicate
with the UI via `root.after()` (which is thread-safe).
"""
import threading
import time
import tkinter as tk
from tkinter import StringVar
from typing import Callable, Optional
from pynput import keyboard, mouse
from .keyboard_block import KeyboardBlocker
class WhiteScreen:
def __init__(
self,
enabled: bool = True,
verify_password: Optional[Callable[[str], bool]] = None,
on_failed_attempt: Optional[Callable[[int], None]] = None,
on_unlocked: Optional[Callable[[], None]] = None,
max_attempts: int = 3,
face_unlock: Optional[Callable[[], tuple]] = None,
face_unlock_auto: bool = False,
face_unlock_interval: float = 4.0,
block_escape_keys: bool = True,
on_attempt: Optional[Callable[[dict], None]] = None,
motion_recorder=None,
motion_recorders=None,
dimming_enabled: bool = True,
dimming_duration_seconds: float = 60.0,
target_grey: str = "#000000",
):
self.enabled = enabled
self.verify_password = verify_password or (lambda s: False)
self.on_failed_attempt = on_failed_attempt or (lambda r: None)
self.on_unlocked = on_unlocked or (lambda: None)
self.max_attempts = max_attempts
# face_unlock() -> (matched: bool, label: Optional[str], confidence: Optional[float])
self.face_unlock = face_unlock
self.face_unlock_auto = face_unlock_auto
self.face_unlock_interval = max(1.5, float(face_unlock_interval))
self.block_escape_keys = block_escape_keys
# on_attempt(record: dict) - fired after each interactive unlock attempt
# record keys: method ('password'|'face'), success (bool),
# entered (str, only present for failed password),
# label (str|None, face only), confidence (float|None, face only)
self.on_attempt = on_attempt or (lambda rec: None)
self._key_blocker = KeyboardBlocker() if block_escape_keys else None
# Accept either a list (new multi-cam API) or a single recorder (back-compat)
recs = []
if motion_recorders:
recs = list(motion_recorders)
elif motion_recorder is not None:
recs = [motion_recorder]
self.motion_recorders = recs
# Dimming / brightness tuner settings (color only - no transparency)
self.dimming_enabled = dimming_enabled
self.dimming_duration = max(10.0, float(dimming_duration_seconds))
self.target_grey = target_grey or "#000000"
self._root: Optional[tk.Tk] = None
self._thread: Optional[threading.Thread] = None
self._ready_event = threading.Event()
self._shown = False
self._failed_count = 0
self._face_busy = False
self._status_var: Optional[StringVar] = None
self._pw_var: Optional[StringVar] = None
self._entry: Optional[tk.Entry] = None
self._face_btn: Optional[tk.Button] = None
self._title_label: Optional[tk.Label] = None
self._status_label: Optional[tk.Label] = None
# Dimmer state
self._dimmer_running = False
self._dimmer_thread: Optional[threading.Thread] = None
self._last_activity = time.time()
self._activity_listener_kb = None
self._activity_listener_mouse = None
@property
def is_shown(self) -> bool:
return self._shown
def _ensure_ui_thread(self) -> None:
"""Start the persistent Tk thread if not already running."""
if self._thread and self._thread.is_alive():
return
self._ready_event.clear()
self._thread = threading.Thread(target=self._run_ui, daemon=True)
self._thread.start()
# Wait briefly for root to initialize
self._ready_event.wait(timeout=3.0)
def show(self) -> None:
if not self.enabled:
return
self._ensure_ui_thread()
if not self._root:
return
self._failed_count = 0
try:
self._root.after(0, self._show_window)
except Exception:
pass
def hide(self) -> None:
if not self._root:
self._shown = False
return
try:
self._root.after(0, self._hide_window)
except Exception:
pass
# ---- methods that run on the Tk thread ----
def _show_window(self) -> None:
if not self._root:
return
try:
already_shown = self._shown
sw = self._root.winfo_screenwidth()
sh = self._root.winfo_screenheight()
self._root.geometry(f"{sw}x{sh}+0+0")
self._root.deiconify()
self._root.attributes("-topmost", True)
self._root.lift()
if not already_shown:
# Only reset field & status on a fresh show, otherwise we'd wipe
# the password the user is currently typing.
if self._pw_var:
self._pw_var.set("")
if self._status_var:
self._status_var.set("Enter unlock phrase to dismiss whiteout:")
if self._entry:
self._entry.focus_force()
# Install OS-level keyboard blocker for escape combos
if self._key_blocker is not None:
try:
self._key_blocker.start()
except Exception as e:
print(f"[WhiteScreen] keyboard blocker start failed: {e}")
# Start motion recording while locked
for rec in self.motion_recorders:
try:
rec.start()
except Exception as e:
print(f"[WhiteScreen] motion recorder start failed: {e}")
# Start brightness dimmer + activity listeners
if self.dimming_enabled:
self._last_activity = time.time()
self._start_dimmer()
self._start_activity_listeners()
self._shown = True
except Exception as e:
print(f"[WhiteScreen] show error: {e}")
def _hide_window(self) -> None:
if not self._root:
return
try:
self._root.withdraw()
self._shown = False
if self._key_blocker is not None:
try:
self._key_blocker.stop()
except Exception as e:
print(f"[WhiteScreen] keyboard blocker stop failed: {e}")
if self.motion_recorders:
for rec in self.motion_recorders:
try:
rec.stop()
except Exception as e:
print(f"[WhiteScreen] motion recorder stop failed: {e}")
# Stop dimmer and activity listeners, restore full white background
self._stop_dimmer()
self._stop_activity_listeners()
try:
self._root.configure(bg="white")
except Exception:
pass
except Exception as e:
print(f"[WhiteScreen] hide error: {e}")
# ---- brightness dimmer / tuner ----
def reset_brightness(self) -> None:
"""Reset to full brightness (pure white) and restart the dimming timer.
Called on any keyboard/mouse activity or camera motion while locked.
"""
self._last_activity = time.time()
if self._root:
try:
self._root.after(0, lambda: self._apply_brightness("#ffffff"))
except Exception:
pass
def _start_dimmer(self) -> None:
if self._dimmer_running:
return
self._dimmer_running = True
self._dimmer_thread = threading.Thread(target=self._dimmer_loop, daemon=True)
self._dimmer_thread.start()
def _stop_dimmer(self) -> None:
self._dimmer_running = False
if self._dimmer_thread:
self._dimmer_thread.join(timeout=1.0)
self._dimmer_thread = None
def _dimmer_loop(self) -> None:
"""Gradually shifts background color from white toward a darker grey.
No transparency is used — the window stays fully opaque.
"""
step = 0.15 # check ~6-7 times per second
while self._dimmer_running and self._root:
try:
now = time.time()
elapsed = now - self._last_activity
progress = min(1.0, elapsed / self.dimming_duration)
# Interpolate background color (white -> target_grey)
grey = self._interpolate_color("#ffffff", self.target_grey, progress)
if self._root:
self._root.after(0, lambda g=grey: self._apply_brightness(g))
except Exception:
pass
time.sleep(step)
def _interpolate_color(self, start_hex: str, end_hex: str, t: float) -> str:
"""Simple linear interpolation between two #rrggbb colors."""
def hex_to_rgb(h):
h = h.lstrip("#")
return tuple(int(h[i:i+2], 16) for i in (0, 2, 4))
def rgb_to_hex(rgb):
return "#{:02x}{:02x}{:02x}".format(*rgb)
s = hex_to_rgb(start_hex)
e = hex_to_rgb(end_hex)
r = int(s[0] + (e[0] - s[0]) * t)
g = int(s[1] + (e[1] - s[1]) * t)
b = int(s[2] + (e[2] - s[2]) * t)
return rgb_to_hex((r, g, b))
def _apply_brightness(self, bg_color: str) -> None:
"""Apply background color and invert text for readability from white to black."""
if not self._root:
return
try:
is_dark = self._is_dark_color(bg_color)
text_color = "#eeeeee" if is_dark else "#222222"
title_color = "#ff4444" if is_dark else "#b00020" # keep ICEYOU title prominent
self._root.configure(bg=bg_color)
for child in self._root.winfo_children():
if isinstance(child, tk.Frame):
child.configure(bg=bg_color)
for sub in child.winfo_children():
if isinstance(sub, tk.Label):
if sub is getattr(self, "_title_label", None):
sub.configure(bg=bg_color, fg=title_color)
elif sub is getattr(self, "_status_label", None):
sub.configure(bg=bg_color, fg=text_color)
else:
sub.configure(bg=bg_color, fg=text_color)
elif isinstance(sub, tk.Button):
sub.configure(bg=bg_color, fg=text_color)
elif isinstance(sub, tk.Entry):
sub.configure(bg=bg_color, fg=text_color,
insertbackground=text_color)
except Exception:
pass
def _is_dark_color(self, hex_color: str) -> bool:
"""Return True if the color is dark enough to need light text."""
try:
h = hex_color.lstrip("#")
r, g, b = int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)
luminance = (r * 0.299 + g * 0.587 + b * 0.114) / 255
return luminance < 0.5
except Exception:
return True # default to dark treatment if parsing fails
def _start_activity_listeners(self) -> None:
"""Install lightweight pynput listeners that reset brightness on any input."""
self._stop_activity_listeners()
def on_any_activity(*_):
if self._shown:
self.reset_brightness()
try:
self._activity_listener_kb = keyboard.Listener(on_press=on_any_activity)
self._activity_listener_kb.start()
except Exception as e:
print(f"[WhiteScreen] keyboard activity listener failed: {e}")
try:
self._activity_listener_mouse = mouse.Listener(
on_move=on_any_activity,
on_click=on_any_activity,
on_scroll=on_any_activity,
)
self._activity_listener_mouse.start()
except Exception as e:
print(f"[WhiteScreen] mouse activity listener failed: {e}")
def _stop_activity_listeners(self) -> None:
if self._activity_listener_kb:
try:
self._activity_listener_kb.stop()
except Exception:
pass
self._activity_listener_kb = None
if self._activity_listener_mouse:
try:
self._activity_listener_mouse.stop()
except Exception:
pass
self._activity_listener_mouse = None
def _run_ui(self) -> None:
"""Runs on dedicated thread. Creates root + mainloop, never destroys until app exit."""
try:
root = tk.Tk()
self._root = root
root.title("ICEYOU")
root.configure(bg="white")
root.overrideredirect(True)
root.attributes("-topmost", True)
root.attributes("-alpha", 1.0) # always fully opaque
root.protocol("WM_DELETE_WINDOW", lambda: None)
# Pre-size to full screen (used by _show_window via geometry too)
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
root.geometry(f"{sw}x{sh}+0+0")
root.withdraw() # Start hidden
# UI elements (built once)
card = tk.Frame(root, bg="white", bd=2, relief="solid",
highlightbackground="#cccccc", highlightthickness=1)
card.place(relx=0.5, rely=0.5, anchor="center", width=440, height=240)
self._title_label = tk.Label(
card, text="ICEYOU LOCKED",
bg="white", fg="#b00020",
font=("Segoe UI", 18, "bold")
)
self._title_label.pack(pady=(18, 4))
self._status_var = StringVar(value="Enter unlock phrase to dismiss whiteout:")
self._status_label = tk.Label(
card, textvariable=self._status_var,
bg="white", fg="#333333",
font=("Segoe UI", 10), wraplength=400, justify="center"
)
self._status_label.pack(pady=4)
self._pw_var = StringVar()
self._entry = tk.Entry(
card, textvariable=self._pw_var, show="*",
width=30, font=("Segoe UI", 12), justify="center"
)
self._entry.pack(pady=10)
tk.Button(
card, text="Unlock", command=self._on_submit,
width=14, font=("Segoe UI", 10, "bold")
).pack(pady=(6, 2))
if self.face_unlock is not None:
self._face_btn = tk.Button(
card, text="Unlock by Face", command=self._on_face_click,
width=18, font=("Segoe UI", 9)
)
self._face_btn.pack(pady=(0, 6))
root.bind("<Return>", lambda e: self._on_submit())
# Periodic re-assert topmost; only steal focus if entry doesn't have it
def keep_focus():
try:
if self._shown and self._root:
self._root.attributes("-topmost", True)
self._root.lift()
if self._entry:
try:
focused = self._root.focus_get()
except Exception:
focused = None
if focused is not self._entry:
self._entry.focus_force()
if self._root:
self._root.after(4000, keep_focus)
except Exception:
pass
root.after(4000, keep_focus)
# Periodic face unlock attempt while shown (if enabled)
def auto_face():
try:
if (self.face_unlock_auto and self._shown
and self.face_unlock is not None and not self._face_busy):
self._attempt_face_unlock(silent=True)
if self._root:
self._root.after(int(self.face_unlock_interval * 1000), auto_face)
except Exception:
pass
if self.face_unlock is not None and self.face_unlock_auto:
root.after(int(self.face_unlock_interval * 1000), auto_face)
self._ready_event.set()
root.mainloop()
except Exception as e:
print(f"[WhiteScreen] UI thread error: {e}")
self._ready_event.set()
finally:
self._root = None
self._shown = False
self._status_var = None
self._pw_var = None
self._entry = None
self._title_label = None
self._status_label = None
def _on_submit(self) -> None:
if not self._pw_var:
return
entered = self._pw_var.get().strip()
self._pw_var.set("")
try:
ok = self.verify_password(entered)
except Exception as e:
print(f"[WhiteScreen] verify error: {e}")
ok = False
# Fire attempt callback (entered only included on failure to avoid logging real password)
try:
record = {"method": "password", "success": ok}
if not ok:
record["entered"] = entered
self.on_attempt(record)
except Exception as e:
print(f"[WhiteScreen] on_attempt error: {e}")
if ok:
self._failed_count = 0
try:
self.on_unlocked()
except Exception as e:
print(f"[WhiteScreen] on_unlocked error: {e}")
self._hide_window()
return
self._failed_count += 1
remaining = max(0, self.max_attempts - self._failed_count)
if remaining == 0:
self._failed_count = 0
if self._status_var:
self._status_var.set("Too many failed attempts. Locking workstation.")
try:
self.on_failed_attempt(0)
except Exception as e:
print(f"[WhiteScreen] on_failed_attempt error: {e}")
if self._root and self._status_var:
status_var = self._status_var
self._root.after(
1500,
lambda: status_var.set("Enter unlock phrase to dismiss whiteout:")
)
else:
if self._status_var:
self._status_var.set(
f"Incorrect. {remaining} attempt(s) remaining before OS lock."
)
try:
self.on_failed_attempt(remaining)
except Exception:
pass
if self._entry:
self._entry.focus_force()
# ---- face unlock ----
def _on_face_click(self) -> None:
self._attempt_face_unlock(silent=False, manual=True)
def _attempt_face_unlock(self, silent: bool = False, manual: bool = False) -> None:
"""Run face_unlock callback in a background thread; on success unlock."""
if self.face_unlock is None or self._face_busy:
return
self._face_busy = True
if not silent and self._status_var:
self._status_var.set("Looking for your face...")
if self._face_btn:
try:
self._face_btn.config(state="disabled")
except Exception:
pass
def worker():
matched = False
label = None
conf = None
try:
result = self.face_unlock()
if isinstance(result, tuple):
matched = bool(result[0])
label = result[1] if len(result) > 1 else None
conf = result[2] if len(result) > 2 else None
else:
matched = bool(result)
except Exception as e:
print(f"[WhiteScreen] face_unlock error: {e}")
# Hand result back to Tk thread
if self._root:
try:
self._root.after(0, lambda: self._face_unlock_result(matched, label, conf, silent, manual))
except Exception:
self._face_busy = False
threading.Thread(target=worker, daemon=True).start()
def _face_unlock_result(self, matched: bool, label, conf, silent: bool, manual: bool = False) -> None:
self._face_busy = False
if self._face_btn:
try:
self._face_btn.config(state="normal")
except Exception:
pass
# Fire attempt log: always for manual clicks, only on success for auto (avoid flooding)
if manual or matched:
try:
self.on_attempt({
"method": "face",
"success": matched,
"label": label,
"confidence": float(conf) if conf is not None else None,
})
except Exception as e:
print(f"[WhiteScreen] on_attempt error: {e}")
if matched:
self._failed_count = 0
try:
self.on_unlocked()
except Exception as e:
print(f"[WhiteScreen] on_unlocked error: {e}")
self._hide_window()
return
if not silent and self._status_var:
if conf is not None:
self._status_var.set(
f"Face not recognised (best={label or '?'}, conf={conf:.1f}). Try the password."
)
else:
self._status_var.set("No face detected. Try the password.")