Initial commit - ICEYOU v1.2 (production ready, sanitized)
This commit is contained in:
Subinacls
@@ -0,0 +1,3 @@
|
||||
"""ICEYOU - Personal System Intrusion Defense Monitor"""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@@ -0,0 +1,175 @@
|
||||
"""Action handlers triggered on intrusion detection: email, lock, etc."""
|
||||
|
||||
import smtplib
|
||||
import ssl
|
||||
import json
|
||||
from email.message import EmailMessage
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
import threading
|
||||
|
||||
from .config import Config
|
||||
from .utils import lock_workstation, get_timestamp
|
||||
|
||||
|
||||
class Actions:
|
||||
def __init__(self, config: Config, monitor_ref=None):
|
||||
self.config = config
|
||||
self.monitor = monitor_ref # for events.log writing via monitor._log_event
|
||||
|
||||
# ---------- public entry points ----------
|
||||
|
||||
def handle_intrusion(self, reason: str, snapshot_path: Optional[str] = None,
|
||||
snapshot_paths: Optional[list] = None) -> None:
|
||||
"""Main entry point called by Monitor on intrusion.
|
||||
|
||||
Note: does NOT lock the workstation. OS lock is reserved for repeated
|
||||
failed unlock attempts and is handled by the tray unlock dialog.
|
||||
|
||||
`snapshot_paths` (when supplied by multi-cam Monitor) attaches every
|
||||
captured frame; `snapshot_path` is the legacy single-attachment path.
|
||||
"""
|
||||
subject = reason
|
||||
body = (
|
||||
"ICEYOU detected potential unauthorized access.\n\n"
|
||||
f"Reason: {reason}\n"
|
||||
f"Time: {get_timestamp()}\n\n"
|
||||
"Check attached snapshot(s) (if available) and logs."
|
||||
)
|
||||
attachments = list(snapshot_paths) if snapshot_paths else (
|
||||
[snapshot_path] if snapshot_path else []
|
||||
)
|
||||
self.send_alert(subject, body, attachments)
|
||||
|
||||
def handle_unlock_attempt(self, record: dict) -> None:
|
||||
"""Email an alert for an interactive unlock attempt (password or face).
|
||||
|
||||
Includes the entered credential text on failed password attempts so
|
||||
you can review what the intruder typed.
|
||||
|
||||
Supports both `record["snapshot"]` (single path, back-compat) and
|
||||
`record["snapshots"]` (list of paths from multi-cam capture).
|
||||
"""
|
||||
method = record.get("method", "?")
|
||||
success = bool(record.get("success", False))
|
||||
snaps = record.get("snapshots")
|
||||
if not snaps:
|
||||
snap = record.get("snapshot")
|
||||
snaps = [snap] if snap else []
|
||||
if success and method == "password":
|
||||
# Don't send/save the real owner password.
|
||||
subject = f"Unlock success (password) at {get_timestamp()}"
|
||||
body = (
|
||||
"ICEYOU was unlocked successfully with the correct password.\n"
|
||||
f"Time: {get_timestamp()}\n"
|
||||
"Snapshot attached if available."
|
||||
)
|
||||
elif success and method == "face":
|
||||
subject = (
|
||||
f"Unlock success (face: {record.get('label')}, conf={record.get('confidence')}) "
|
||||
f"at {get_timestamp()}"
|
||||
)
|
||||
body = (
|
||||
"ICEYOU was unlocked by face recognition.\n"
|
||||
f"Label: {record.get('label')}\n"
|
||||
f"Confidence: {record.get('confidence')}\n"
|
||||
f"Time: {get_timestamp()}\n"
|
||||
)
|
||||
elif method == "password":
|
||||
entered = record.get("entered", "")
|
||||
subject = f"FAILED unlock attempt (password) at {get_timestamp()}"
|
||||
body = (
|
||||
"ICEYOU recorded a FAILED password unlock attempt.\n\n"
|
||||
f"Entered: {entered!r}\n"
|
||||
f"Time: {get_timestamp()}\n\n"
|
||||
"Snapshot of the person at the keyboard is attached."
|
||||
)
|
||||
else: # failed face
|
||||
subject = f"FAILED unlock attempt (face) at {get_timestamp()}"
|
||||
body = (
|
||||
"ICEYOU recorded a FAILED face unlock attempt.\n\n"
|
||||
f"Predicted label: {record.get('label')}\n"
|
||||
f"Confidence: {record.get('confidence')}\n"
|
||||
f"Time: {get_timestamp()}\n"
|
||||
)
|
||||
self.send_alert(subject, body, snaps)
|
||||
|
||||
def send_test_email(self) -> None:
|
||||
self.send_alert(
|
||||
"Test email",
|
||||
f"This is a test from ICEYOU at {get_timestamp()}. If you see this, SMTP works.",
|
||||
None,
|
||||
)
|
||||
|
||||
# ---------- internal ----------
|
||||
|
||||
def send_alert(self, subject: str, body: str, attachments) -> None:
|
||||
"""`attachments` may be None, a single path string, or a list of paths."""
|
||||
if not self.config.get("email.enabled", False):
|
||||
return
|
||||
if attachments is None:
|
||||
paths = []
|
||||
elif isinstance(attachments, (list, tuple)):
|
||||
paths = [p for p in attachments if p]
|
||||
else:
|
||||
paths = [attachments]
|
||||
threading.Thread(
|
||||
target=self._send_email,
|
||||
args=(subject, body, paths),
|
||||
daemon=True,
|
||||
).start()
|
||||
|
||||
def _log(self, event: str, details: str) -> None:
|
||||
if self.monitor is not None and hasattr(self.monitor, "_log_event"):
|
||||
try:
|
||||
self.monitor._log_event(event, details)
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
print(f"[Actions] {event}: {details}")
|
||||
|
||||
def _send_email(self, subject: str, body: str, snapshot_paths) -> None:
|
||||
email_cfg = self.config.get("email", {})
|
||||
prefix = email_cfg.get("subject_prefix", "[ICEYOU Alert]")
|
||||
# Normalize attachment paths to a list
|
||||
if snapshot_paths is None:
|
||||
paths = []
|
||||
elif isinstance(snapshot_paths, (list, tuple)):
|
||||
paths = list(snapshot_paths)
|
||||
else:
|
||||
paths = [snapshot_paths]
|
||||
try:
|
||||
msg = EmailMessage()
|
||||
msg["Subject"] = f"{prefix} {subject}"
|
||||
msg["From"] = email_cfg.get("from_addr", "")
|
||||
msg["To"] = email_cfg.get("to_addr", "")
|
||||
msg.set_content(body)
|
||||
|
||||
for p in paths:
|
||||
if not p:
|
||||
continue
|
||||
pp = Path(p)
|
||||
if not pp.exists():
|
||||
continue
|
||||
with open(pp, "rb") as f:
|
||||
img_data = f.read()
|
||||
msg.add_attachment(
|
||||
img_data,
|
||||
maintype="image",
|
||||
subtype="jpeg",
|
||||
filename=pp.name,
|
||||
)
|
||||
|
||||
context = ssl.create_default_context()
|
||||
host = email_cfg.get("smtp_server", "smtp.gmail.com")
|
||||
port = email_cfg.get("smtp_port", 587)
|
||||
with smtplib.SMTP(host, port, timeout=20) as server:
|
||||
server.ehlo()
|
||||
if email_cfg.get("use_tls", True):
|
||||
server.starttls(context=context)
|
||||
server.ehlo()
|
||||
server.login(email_cfg.get("username", ""), email_cfg.get("password", ""))
|
||||
server.send_message(msg)
|
||||
self._log("email_sent", f"{subject} (atts={len(paths)})")
|
||||
except Exception as e:
|
||||
self._log("email_failed", f"{type(e).__name__}: {e}")
|
||||
@@ -0,0 +1,243 @@
|
||||
"""Multi-camera manager with dead-camera detection.
|
||||
|
||||
Background
|
||||
----------
|
||||
On Windows DSHOW, `cv2.VideoCapture(idx, CAP_DSHOW)` for a phantom index
|
||||
(e.g. "OBS Virtual Camera" left registered with no provider, or an unplugged
|
||||
device) often returns `isOpened() == True` but the first `cap.read()` blocks
|
||||
forever waiting for data that will never arrive. To survive this we probe
|
||||
every requested index with a frame-read on a worker thread and a timeout;
|
||||
indices that don't deliver within `probe_timeout` are marked DEAD and
|
||||
permanently skipped.
|
||||
|
||||
Public API
|
||||
----------
|
||||
- `Camera(config)` multi-stream manager
|
||||
- `camera.read_frame()` primary stream frame (back-compat)
|
||||
- `camera.read_all_frames()` dict {index: frame}
|
||||
- `camera.capture_snapshot(reason)` single file from primary (back-compat)
|
||||
- `camera.capture_all_snapshots(reason)` one file per alive stream -> list[str]
|
||||
- `camera.is_vision_obscured()` True only if ALL alive streams are obscured
|
||||
- `camera.alive_streams` list of CameraStream (use for per-cam
|
||||
consumers such as MotionRecorder)
|
||||
- `camera.release()` close everything
|
||||
"""
|
||||
|
||||
import cv2
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import List, Optional, Dict
|
||||
|
||||
from .utils import get_timestamp
|
||||
from .config import Config
|
||||
|
||||
|
||||
class CameraStream:
|
||||
"""Single webcam wrapper that survives dead-device probing."""
|
||||
|
||||
def __init__(self, index: int):
|
||||
self.index = int(index)
|
||||
self._cap: Optional[cv2.VideoCapture] = None
|
||||
self._lock = threading.Lock()
|
||||
self.is_alive = False
|
||||
|
||||
def open(self, probe_timeout: float = 2.5) -> bool:
|
||||
"""Open the capture and verify it actually delivers a frame within
|
||||
`probe_timeout` seconds. Returns True on success, False if dead."""
|
||||
cap = cv2.VideoCapture(self.index, cv2.CAP_DSHOW)
|
||||
if not cap.isOpened():
|
||||
try:
|
||||
cap.release()
|
||||
except Exception:
|
||||
pass
|
||||
self.is_alive = False
|
||||
return False
|
||||
|
||||
# cap.read() can block forever on phantom DSHOW devices - probe in a
|
||||
# daemon thread and bail if it doesn't return in time.
|
||||
result = {"ok": False, "frame": None, "done": False}
|
||||
|
||||
def reader():
|
||||
try:
|
||||
ok, frame = cap.read()
|
||||
result["ok"] = bool(ok and frame is not None)
|
||||
result["frame"] = frame
|
||||
except Exception:
|
||||
result["ok"] = False
|
||||
finally:
|
||||
result["done"] = True
|
||||
|
||||
t = threading.Thread(target=reader, daemon=True)
|
||||
t.start()
|
||||
t.join(timeout=probe_timeout)
|
||||
|
||||
if not result["done"] or not result["ok"]:
|
||||
# Reader still blocked OR read returned no frame: dead camera.
|
||||
# We intentionally do NOT cap.release() here because the reader
|
||||
# may still be inside cap.read() and releasing concurrently can
|
||||
# crash the DSHOW driver. The leaked cap dies at process exit.
|
||||
self.is_alive = False
|
||||
return False
|
||||
|
||||
self._cap = cap
|
||||
self.is_alive = True
|
||||
return True
|
||||
|
||||
def read_frame(self):
|
||||
with self._lock:
|
||||
if self._cap is None:
|
||||
return None
|
||||
try:
|
||||
ok, frame = self._cap.read()
|
||||
if not ok or frame is None:
|
||||
return None
|
||||
return frame
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def release(self) -> None:
|
||||
with self._lock:
|
||||
if self._cap is not None:
|
||||
try:
|
||||
self._cap.release()
|
||||
except Exception:
|
||||
pass
|
||||
self._cap = None
|
||||
self.is_alive = False
|
||||
|
||||
|
||||
class Camera:
|
||||
"""Multi-camera manager. Backward compatible single-cam API + new
|
||||
multi-cam helpers."""
|
||||
|
||||
def __init__(self, config: Config):
|
||||
self.config = config
|
||||
self.snapshot_dir = Path(config.get("snapshot_dir", "snapshots"))
|
||||
self.snapshot_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Resolve requested indices: prefer the new list, fall back to legacy single
|
||||
indices = config.get("camera_device_indices")
|
||||
if not indices:
|
||||
indices = [int(config.get("camera_device_index", 0))]
|
||||
# De-duplicate while preserving order
|
||||
seen = set()
|
||||
ordered = []
|
||||
for i in indices:
|
||||
ii = int(i)
|
||||
if ii not in seen:
|
||||
seen.add(ii)
|
||||
ordered.append(ii)
|
||||
|
||||
# Camera obstruction detection settings
|
||||
obs = config.get("camera_obscured_detection", {}) or {}
|
||||
self.obscured_enabled = bool(obs.get("enabled", True))
|
||||
self.brightness_threshold = float(obs.get("brightness_threshold", 35))
|
||||
self.variance_threshold = float(obs.get("variance_threshold", 10))
|
||||
|
||||
probe_timeout = float(config.get("camera_probe_timeout_seconds", 2.5))
|
||||
|
||||
self.streams: List[CameraStream] = []
|
||||
for idx in ordered:
|
||||
stream = CameraStream(idx)
|
||||
if stream.open(probe_timeout=probe_timeout):
|
||||
print(f"[Camera] index {idx} OPEN")
|
||||
self.streams.append(stream)
|
||||
else:
|
||||
print(f"[Camera] index {idx} DEAD (skipped after {probe_timeout}s probe)")
|
||||
# keep the dead stream object out of the list entirely
|
||||
if not self.alive_streams:
|
||||
print("[Camera] WARNING: no working cameras detected")
|
||||
|
||||
# ---- discovery / accessors ----
|
||||
|
||||
@property
|
||||
def alive_streams(self) -> List[CameraStream]:
|
||||
return [s for s in self.streams if s.is_alive]
|
||||
|
||||
@property
|
||||
def alive_indices(self) -> List[int]:
|
||||
return [s.index for s in self.alive_streams]
|
||||
|
||||
@property
|
||||
def primary(self) -> Optional[CameraStream]:
|
||||
return self.alive_streams[0] if self.alive_streams else None
|
||||
|
||||
def is_camera_available(self) -> bool:
|
||||
return self.primary is not None
|
||||
|
||||
# ---- frame I/O ----
|
||||
|
||||
def read_frame(self):
|
||||
"""Back-compat: returns a frame from the PRIMARY (first alive) camera."""
|
||||
p = self.primary
|
||||
return p.read_frame() if p else None
|
||||
|
||||
def read_all_frames(self) -> Dict[int, object]:
|
||||
"""Returns {camera_index: frame} for every alive stream (None if a read failed)."""
|
||||
return {s.index: s.read_frame() for s in self.alive_streams}
|
||||
|
||||
# ---- snapshots ----
|
||||
|
||||
def capture_snapshot(self, reason: str = "trigger") -> Optional[str]:
|
||||
"""Back-compat single snapshot from PRIMARY camera. Returns path or None."""
|
||||
p = self.primary
|
||||
if p is None:
|
||||
return None
|
||||
frame = p.read_frame()
|
||||
if frame is None:
|
||||
return None
|
||||
ts = get_timestamp()
|
||||
suffix = f"cam{p.index}_" if len(self.alive_streams) > 1 else ""
|
||||
filename = f"snapshot_{ts}_{suffix}{reason}.jpg"
|
||||
filepath = self.snapshot_dir / filename
|
||||
return str(filepath) if cv2.imwrite(str(filepath), frame) else None
|
||||
|
||||
def capture_all_snapshots(self, reason: str = "trigger") -> List[str]:
|
||||
"""Snap from every alive camera. Returns list of saved file paths."""
|
||||
out: List[str] = []
|
||||
ts = get_timestamp()
|
||||
for s in self.alive_streams:
|
||||
frame = s.read_frame()
|
||||
if frame is None:
|
||||
continue
|
||||
filename = f"snapshot_{ts}_cam{s.index}_{reason}.jpg"
|
||||
filepath = self.snapshot_dir / filename
|
||||
if cv2.imwrite(str(filepath), frame):
|
||||
out.append(str(filepath))
|
||||
return out
|
||||
|
||||
# ---- obscured detection ----
|
||||
|
||||
def is_vision_obscured(self) -> bool:
|
||||
"""True only if EVERY alive camera reports obscured vision.
|
||||
|
||||
One unblocked camera = scene is visible to ICEYOU; don't trigger lockdown.
|
||||
"""
|
||||
if not self.obscured_enabled:
|
||||
return False
|
||||
alive = self.alive_streams
|
||||
if not alive:
|
||||
return False # No cameras = can't decide; don't escalate
|
||||
obscured_all = True
|
||||
for s in alive:
|
||||
frame = s.read_frame()
|
||||
if frame is None:
|
||||
continue
|
||||
try:
|
||||
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
||||
except Exception:
|
||||
continue
|
||||
mean_b = float(gray.mean())
|
||||
var = float(gray.std(ddof=0))
|
||||
if mean_b >= self.brightness_threshold and var >= self.variance_threshold:
|
||||
obscured_all = False
|
||||
break
|
||||
if obscured_all:
|
||||
print(f"[Camera] All {len(alive)} alive camera(s) report obscured")
|
||||
return obscured_all
|
||||
|
||||
# ---- lifecycle ----
|
||||
|
||||
def release(self) -> None:
|
||||
for s in self.streams:
|
||||
s.release()
|
||||
@@ -0,0 +1,81 @@
|
||||
"""Configuration loader for ICEYOU."""
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
|
||||
DEFAULT_CONFIG: Dict[str, Any] = {
|
||||
"idle_timeout_seconds": 300,
|
||||
"monitoring_enabled": True,
|
||||
"camera_device_index": 0,
|
||||
"snapshot_on_trigger": True,
|
||||
"snapshot_dir": "snapshots",
|
||||
"log_file": "events.log",
|
||||
"email": {
|
||||
"enabled": False,
|
||||
"smtp_server": "smtp.gmail.com",
|
||||
"smtp_port": 587,
|
||||
"use_tls": True,
|
||||
"username": "",
|
||||
"password": "",
|
||||
"from_addr": "",
|
||||
"to_addr": "",
|
||||
"subject_prefix": "[ICEYOU Alert]"
|
||||
},
|
||||
"lock_workstation": True,
|
||||
"device_monitoring": True,
|
||||
"white_screen_on_away": True,
|
||||
"unlock_password": "iceyou2026",
|
||||
"max_unlock_attempts": 3,
|
||||
"camera_obscured_detection": {
|
||||
"enabled": True,
|
||||
"brightness_threshold": 35,
|
||||
"variance_threshold": 10,
|
||||
"check_interval_seconds": 12
|
||||
},
|
||||
"hotkeys": {
|
||||
"force_away": "<ctrl>+<alt>+l",
|
||||
"unlock": "<ctrl>+<alt>+u"
|
||||
},
|
||||
"face_recognition_enabled": False
|
||||
}
|
||||
|
||||
class Config:
|
||||
def __init__(self, config_path: str = "config.json"):
|
||||
self.config_path = Path(config_path)
|
||||
self.data: Dict[str, Any] = DEFAULT_CONFIG.copy()
|
||||
self.load()
|
||||
|
||||
def load(self) -> None:
|
||||
if self.config_path.exists():
|
||||
try:
|
||||
with open(self.config_path, "r", encoding="utf-8") as f:
|
||||
user_config = json.load(f)
|
||||
# Deep merge
|
||||
self._merge(self.data, user_config)
|
||||
except Exception as e:
|
||||
print(f"Warning: Failed to load config: {e}. Using defaults.")
|
||||
else:
|
||||
print("No config.json found. Using defaults. Copy config.example.json to config.json")
|
||||
|
||||
def _merge(self, base: Dict, update: Dict) -> None:
|
||||
for k, v in update.items():
|
||||
if k in base and isinstance(base[k], dict) and isinstance(v, dict):
|
||||
self._merge(base[k], v)
|
||||
else:
|
||||
base[k] = v
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
keys = key.split(".")
|
||||
val = self.data
|
||||
for k in keys:
|
||||
if isinstance(val, dict) and k in val:
|
||||
val = val[k]
|
||||
else:
|
||||
return default
|
||||
return val
|
||||
|
||||
def save(self) -> None:
|
||||
with open(self.config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(self.data, f, indent=2)
|
||||
@@ -0,0 +1,66 @@
|
||||
"""Device monitoring for USB/peripheral connect/disconnect events.
|
||||
|
||||
Simple implementation using drive letter polling for removable media.
|
||||
For full WM_DEVICECHANGE support, a message window + WNDPROC is needed.
|
||||
"""
|
||||
|
||||
import os
|
||||
import string
|
||||
import threading
|
||||
import time
|
||||
from typing import Callable, Set, Optional
|
||||
|
||||
|
||||
class DeviceMonitor:
|
||||
def __init__(self, on_device_event: Callable[[str, str], None]):
|
||||
"""
|
||||
on_device_event: callback(event, details) e.g. ("device_added", "E:\\")
|
||||
"""
|
||||
self.on_device_event = on_device_event
|
||||
self._running = False
|
||||
self._thread: Optional[threading.Thread] = None
|
||||
self._known_drives: Set[str] = set()
|
||||
|
||||
def _get_removable_drives(self) -> Set[str]:
|
||||
drives = set()
|
||||
for letter in string.ascii_uppercase:
|
||||
drive = f"{letter}:"
|
||||
if os.path.exists(drive) and os.path.ismount(drive):
|
||||
# Heuristic: treat as removable if not C: and exists
|
||||
if letter != "C":
|
||||
drives.add(drive)
|
||||
return drives
|
||||
|
||||
def _poll(self) -> None:
|
||||
while self._running:
|
||||
current = self._get_removable_drives()
|
||||
added = current - self._known_drives
|
||||
removed = self._known_drives - current
|
||||
|
||||
for d in added:
|
||||
try:
|
||||
self.on_device_event("device_added", d)
|
||||
except Exception:
|
||||
pass
|
||||
for d in removed:
|
||||
try:
|
||||
self.on_device_event("device_removed", d)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
self._known_drives = current
|
||||
time.sleep(5) # Poll every 5 seconds
|
||||
|
||||
def start(self) -> None:
|
||||
if self._running:
|
||||
return
|
||||
self._running = True
|
||||
self._known_drives = self._get_removable_drives()
|
||||
self._thread = threading.Thread(target=self._poll, daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def stop(self) -> None:
|
||||
self._running = False
|
||||
if self._thread:
|
||||
self._thread.join(timeout=2)
|
||||
self._thread = None
|
||||
@@ -0,0 +1,200 @@
|
||||
"""Face recognition for ICEYOU unlock flow.
|
||||
|
||||
Loads the LBPH model trained by face_enrollment.py and verifies frames
|
||||
captured from the active camera. Designed to be safe even if the model
|
||||
or contrib opencv build is missing - methods just return False.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional, Tuple, List
|
||||
|
||||
import cv2
|
||||
|
||||
FACES_DIR = Path("faces")
|
||||
MODEL_FILE = FACES_DIR / "model.yml"
|
||||
LABELS_FILE = FACES_DIR / "labels.json"
|
||||
|
||||
|
||||
class FaceRecognizer:
|
||||
"""Verify a webcam frame against the enrolled LBPH model.
|
||||
|
||||
Lower LBPH confidence == better match. Typical thresholds:
|
||||
< 50 strong match
|
||||
< 70 acceptable
|
||||
> 90 poor / reject
|
||||
"""
|
||||
|
||||
def __init__(self, confidence_threshold: float = 70.0, allowed_labels: Optional[List[str]] = None, min_successful_matches: int = 2):
|
||||
self.confidence_threshold = float(confidence_threshold)
|
||||
self.allowed_labels = set(allowed_labels) if allowed_labels else None # None = any enrolled
|
||||
self.min_successful_matches = max(1, int(min_successful_matches))
|
||||
self._recognizer = None
|
||||
self._labels: dict = {}
|
||||
self._detectors: dict = {}
|
||||
self._loaded = False
|
||||
self._load()
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
return self._loaded and self._recognizer is not None
|
||||
|
||||
def _load(self) -> None:
|
||||
if not MODEL_FILE.exists() or not LABELS_FILE.exists():
|
||||
print("[FaceRecognizer] No trained model found - face unlock disabled.")
|
||||
return
|
||||
try:
|
||||
if not hasattr(cv2, "face"):
|
||||
print("[FaceRecognizer] cv2.face missing - install opencv-contrib-python.")
|
||||
return
|
||||
rec = cv2.face.LBPHFaceRecognizer_create()
|
||||
rec.read(str(MODEL_FILE))
|
||||
self._recognizer = rec
|
||||
with open(LABELS_FILE, "r", encoding="utf-8") as f:
|
||||
raw = json.load(f)
|
||||
self._labels = {int(k): v for k, v in raw.items()}
|
||||
self._detectors = self._load_detectors()
|
||||
self._loaded = True
|
||||
print(f"[FaceRecognizer] Loaded model with labels: {list(self._labels.values())}")
|
||||
print(f"[FaceRecognizer] Consensus requirement: {self.min_successful_matches} successful match(es) required")
|
||||
except Exception as e:
|
||||
print(f"[FaceRecognizer] Load failed: {e}")
|
||||
self._recognizer = None
|
||||
self._loaded = False
|
||||
|
||||
@staticmethod
|
||||
def _load_detectors() -> dict:
|
||||
base = Path(cv2.data.haarcascades)
|
||||
detectors = {}
|
||||
for key, name in (
|
||||
("frontal", "haarcascade_frontalface_default.xml"),
|
||||
("alt2", "haarcascade_frontalface_alt2.xml"),
|
||||
("profile", "haarcascade_profileface.xml"),
|
||||
):
|
||||
p = base / name
|
||||
if p.exists():
|
||||
detectors[key] = cv2.CascadeClassifier(str(p))
|
||||
return detectors
|
||||
|
||||
def _detect_faces(self, gray):
|
||||
if not self._detectors:
|
||||
return []
|
||||
eq = cv2.equalizeHist(gray)
|
||||
params = dict(scaleFactor=1.1, minNeighbors=4, minSize=(60, 60))
|
||||
for key in ("frontal", "alt2", "profile"):
|
||||
det = self._detectors.get(key)
|
||||
if det is None:
|
||||
continue
|
||||
faces = det.detectMultiScale(eq, **params)
|
||||
if len(faces) > 0:
|
||||
return faces
|
||||
# mirrored profile
|
||||
det = self._detectors.get("profile")
|
||||
if det is not None:
|
||||
flipped = cv2.flip(eq, 1)
|
||||
faces = det.detectMultiScale(flipped, **params)
|
||||
if len(faces) > 0:
|
||||
w = eq.shape[1]
|
||||
return [(w - x - fw, y, fw, fh) for (x, y, fw, fh) in faces]
|
||||
return []
|
||||
|
||||
def verify_frame(self, frame) -> Tuple[bool, Optional[str], Optional[float]]:
|
||||
"""Try to recognize an authorized user in `frame`.
|
||||
|
||||
Returns (matched, label, confidence). `matched` is True only if
|
||||
confidence is below threshold AND the predicted label is in
|
||||
allowed_labels (if configured).
|
||||
"""
|
||||
if not self.available or frame is None:
|
||||
return False, None, None
|
||||
try:
|
||||
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
||||
except Exception:
|
||||
return False, None, None
|
||||
faces = self._detect_faces(gray)
|
||||
if len(faces) == 0:
|
||||
return False, None, None
|
||||
|
||||
H, W = gray.shape[:2]
|
||||
best_label = None
|
||||
best_conf = None
|
||||
for (x, y, w, h) in faces:
|
||||
# Try a few crop variants - LBPH is sensitive to exact framing.
|
||||
# Enrollment saved with ~10% margin; we also try 5% and 15% to
|
||||
# tolerate small detector size differences.
|
||||
for margin_pct in (0.10, 0.05, 0.15):
|
||||
m = int(margin_pct * max(w, h))
|
||||
x0 = max(x - m, 0)
|
||||
y0 = max(y - m, 0)
|
||||
x1 = min(x + w + m, W)
|
||||
y1 = min(y + h + m, H)
|
||||
roi = gray[y0:y1, x0:x1]
|
||||
if roi.size == 0:
|
||||
continue
|
||||
try:
|
||||
# Match enrollment exactly: raw grayscale crop resized to 200x200.
|
||||
# NO equalizeHist (enrollment didn't apply it to saved images).
|
||||
roi_resized = cv2.resize(roi, (200, 200))
|
||||
except Exception:
|
||||
continue
|
||||
try:
|
||||
label_id, conf = self._recognizer.predict(roi_resized)
|
||||
except Exception:
|
||||
continue
|
||||
if best_conf is None or conf < best_conf:
|
||||
best_conf = conf
|
||||
best_label = self._labels.get(label_id)
|
||||
|
||||
if best_label is None or best_conf is None:
|
||||
return False, None, None
|
||||
if best_conf > self.confidence_threshold:
|
||||
return False, best_label, best_conf
|
||||
if self.allowed_labels is not None and best_label not in self.allowed_labels:
|
||||
return False, best_label, best_conf
|
||||
return True, best_label, best_conf
|
||||
|
||||
def verify_from_camera(self, camera, attempts: int = 3) -> Tuple[bool, Optional[str], Optional[float]]:
|
||||
"""Grab up to `attempts` frames across all alive cameras.
|
||||
|
||||
Requires at least `self.min_successful_matches` successful verifications
|
||||
(on allowed labels) before returning True. This greatly reduces the
|
||||
chance of a look-alike (child, sibling, etc.) triggering an unlock from
|
||||
a single lucky frame.
|
||||
|
||||
Multi-cam aware via alive_streams.
|
||||
"""
|
||||
if not self.available:
|
||||
return False, None, None
|
||||
sources = []
|
||||
alive = getattr(camera, "alive_streams", None)
|
||||
if alive:
|
||||
sources = list(alive)
|
||||
else:
|
||||
sources = [camera]
|
||||
|
||||
successes = 0
|
||||
matched_label = None
|
||||
matched_conf = None
|
||||
best = (False, None, None) # best non-success for reporting
|
||||
|
||||
for _ in range(max(1, attempts)):
|
||||
for src in sources:
|
||||
try:
|
||||
frame = src.read_frame()
|
||||
except Exception:
|
||||
continue
|
||||
if frame is None:
|
||||
continue
|
||||
ok, label, conf = self.verify_frame(frame)
|
||||
if ok and label is not None:
|
||||
successes += 1
|
||||
matched_label = label
|
||||
matched_conf = conf
|
||||
if successes >= self.min_successful_matches:
|
||||
return True, matched_label, matched_conf
|
||||
else:
|
||||
if conf is not None and (best[2] is None or conf < best[2]):
|
||||
best = (False, label, conf)
|
||||
|
||||
# Not enough consensus reached
|
||||
return best
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Global keyboard and mouse hooks using pynput."""
|
||||
|
||||
from pynput import keyboard, mouse
|
||||
from typing import Callable, Optional
|
||||
import threading
|
||||
|
||||
|
||||
class InputHooks:
|
||||
def __init__(self, on_activity: Callable[[str], None]):
|
||||
"""
|
||||
on_activity: callback(event_type) e.g. "keyboard", "mouse_move", "mouse_click"
|
||||
"""
|
||||
self.on_activity = on_activity
|
||||
self._keyboard_listener: Optional[keyboard.Listener] = None
|
||||
self._mouse_listener: Optional[mouse.Listener] = None
|
||||
self._running = False
|
||||
|
||||
def _on_key(self, key):
|
||||
if self._running:
|
||||
try:
|
||||
self.on_activity("keyboard")
|
||||
except Exception:
|
||||
pass
|
||||
# Do not suppress - let keys pass through
|
||||
|
||||
def _on_mouse_move(self, x, y):
|
||||
if self._running:
|
||||
try:
|
||||
self.on_activity("mouse_move")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _on_mouse_click(self, x, y, button, pressed):
|
||||
if self._running and pressed:
|
||||
try:
|
||||
self.on_activity("mouse_click")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def start(self) -> None:
|
||||
if self._running:
|
||||
return
|
||||
self._running = True
|
||||
self._keyboard_listener = keyboard.Listener(on_press=self._on_key)
|
||||
self._mouse_listener = mouse.Listener(
|
||||
on_move=self._on_mouse_move,
|
||||
on_click=self._on_mouse_click
|
||||
)
|
||||
self._keyboard_listener.start()
|
||||
self._mouse_listener.start()
|
||||
|
||||
def stop(self) -> None:
|
||||
self._running = False
|
||||
if self._keyboard_listener:
|
||||
self._keyboard_listener.stop()
|
||||
if self._mouse_listener:
|
||||
self._mouse_listener.stop()
|
||||
self._keyboard_listener = None
|
||||
self._mouse_listener = None
|
||||
|
||||
def pause(self) -> None:
|
||||
self._running = False
|
||||
|
||||
def resume(self) -> None:
|
||||
self._running = True
|
||||
@@ -0,0 +1,156 @@
|
||||
"""Low-level Windows keyboard hook to block escape combos while ICEYOU is locked.
|
||||
|
||||
Blocks (when active):
|
||||
- WIN keys (left/right) -> Start menu / Win+anything
|
||||
- CTRL+ESC -> Start menu
|
||||
- ALT+TAB, ALT+ESC -> task switching
|
||||
- ALT+F4 -> close window
|
||||
- ALT+SPACE -> window menu (Restore/Move/Close)
|
||||
- CTRL+SHIFT+ESC -> Task Manager hotkey
|
||||
|
||||
CANNOT block (by Windows design):
|
||||
- CTRL+ALT+DEL (Secure Attention Sequence)
|
||||
- Hard reset / power button
|
||||
These require Group Policy or admin / kernel privilege.
|
||||
"""
|
||||
|
||||
import ctypes
|
||||
import threading
|
||||
from ctypes import wintypes
|
||||
|
||||
|
||||
user32 = ctypes.WinDLL("user32", use_last_error=True)
|
||||
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
||||
|
||||
WH_KEYBOARD_LL = 13
|
||||
WM_KEYDOWN = 0x0100
|
||||
WM_SYSKEYDOWN = 0x0104
|
||||
WM_QUIT = 0x0012
|
||||
|
||||
VK_LWIN = 0x5B
|
||||
VK_RWIN = 0x5C
|
||||
VK_TAB = 0x09
|
||||
VK_ESCAPE = 0x1B
|
||||
VK_F4 = 0x73
|
||||
VK_SPACE = 0x20
|
||||
VK_MENU = 0x12 # ALT
|
||||
VK_CONTROL = 0x11
|
||||
VK_SHIFT = 0x10
|
||||
|
||||
LRESULT = ctypes.c_long
|
||||
ULONG_PTR = ctypes.c_size_t
|
||||
|
||||
|
||||
class KBDLLHOOKSTRUCT(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("vkCode", wintypes.DWORD),
|
||||
("scanCode", wintypes.DWORD),
|
||||
("flags", wintypes.DWORD),
|
||||
("time", wintypes.DWORD),
|
||||
("dwExtraInfo", ULONG_PTR),
|
||||
]
|
||||
|
||||
|
||||
HOOKPROC = ctypes.WINFUNCTYPE(LRESULT, ctypes.c_int, wintypes.WPARAM, wintypes.LPARAM)
|
||||
|
||||
user32.SetWindowsHookExW.restype = wintypes.HHOOK
|
||||
user32.SetWindowsHookExW.argtypes = [ctypes.c_int, HOOKPROC, wintypes.HINSTANCE, wintypes.DWORD]
|
||||
user32.UnhookWindowsHookEx.restype = wintypes.BOOL
|
||||
user32.UnhookWindowsHookEx.argtypes = [wintypes.HHOOK]
|
||||
user32.CallNextHookEx.restype = LRESULT
|
||||
user32.CallNextHookEx.argtypes = [wintypes.HHOOK, ctypes.c_int, wintypes.WPARAM, wintypes.LPARAM]
|
||||
user32.GetMessageW.argtypes = [ctypes.POINTER(wintypes.MSG), wintypes.HWND, wintypes.UINT, wintypes.UINT]
|
||||
user32.GetMessageW.restype = ctypes.c_int
|
||||
user32.TranslateMessage.argtypes = [ctypes.POINTER(wintypes.MSG)]
|
||||
user32.DispatchMessageW.argtypes = [ctypes.POINTER(wintypes.MSG)]
|
||||
user32.PostThreadMessageW.argtypes = [wintypes.DWORD, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM]
|
||||
user32.PostThreadMessageW.restype = wintypes.BOOL
|
||||
user32.GetAsyncKeyState.argtypes = [ctypes.c_int]
|
||||
user32.GetAsyncKeyState.restype = ctypes.c_short
|
||||
kernel32.GetCurrentThreadId.restype = wintypes.DWORD
|
||||
|
||||
|
||||
class KeyboardBlocker:
|
||||
"""Installs a WH_KEYBOARD_LL hook on its own thread with a message pump."""
|
||||
|
||||
def __init__(self):
|
||||
self._hook = None
|
||||
self._thread: threading.Thread = None
|
||||
self._thread_id: int = 0
|
||||
self._active = False
|
||||
self._hook_proc = None # keep ref so GC doesn't free callback
|
||||
self._ready = threading.Event()
|
||||
|
||||
@property
|
||||
def active(self) -> bool:
|
||||
return self._active
|
||||
|
||||
def start(self) -> None:
|
||||
if self._active:
|
||||
return
|
||||
self._ready.clear()
|
||||
self._thread = threading.Thread(target=self._run, daemon=True)
|
||||
self._thread.start()
|
||||
self._ready.wait(timeout=2.0)
|
||||
|
||||
def stop(self) -> None:
|
||||
if not self._active:
|
||||
return
|
||||
self._active = False
|
||||
if self._thread_id:
|
||||
user32.PostThreadMessageW(self._thread_id, WM_QUIT, 0, 0)
|
||||
if self._thread:
|
||||
self._thread.join(timeout=2.0)
|
||||
self._thread = None
|
||||
self._thread_id = 0
|
||||
|
||||
@staticmethod
|
||||
def _is_pressed(vk: int) -> bool:
|
||||
return (user32.GetAsyncKeyState(vk) & 0x8000) != 0
|
||||
|
||||
def _hook_callback(self, nCode, wParam, lParam):
|
||||
try:
|
||||
if nCode == 0 and wParam in (WM_KEYDOWN, WM_SYSKEYDOWN):
|
||||
kb = ctypes.cast(lParam, ctypes.POINTER(KBDLLHOOKSTRUCT))[0]
|
||||
vk = kb.vkCode
|
||||
|
||||
if vk in (VK_LWIN, VK_RWIN):
|
||||
return 1
|
||||
|
||||
alt = self._is_pressed(VK_MENU)
|
||||
ctrl = self._is_pressed(VK_CONTROL)
|
||||
|
||||
if alt and vk in (VK_TAB, VK_ESCAPE, VK_F4, VK_SPACE):
|
||||
return 1
|
||||
if ctrl and vk == VK_ESCAPE:
|
||||
return 1
|
||||
except Exception:
|
||||
pass
|
||||
return user32.CallNextHookEx(self._hook, nCode, wParam, lParam)
|
||||
|
||||
def _run(self) -> None:
|
||||
self._thread_id = kernel32.GetCurrentThreadId()
|
||||
self._hook_proc = HOOKPROC(self._hook_callback)
|
||||
self._hook = user32.SetWindowsHookExW(WH_KEYBOARD_LL, self._hook_proc, None, 0)
|
||||
if not self._hook:
|
||||
err = ctypes.get_last_error()
|
||||
print(f"[KeyboardBlocker] SetWindowsHookEx failed (error {err})")
|
||||
self._active = False
|
||||
self._ready.set()
|
||||
return
|
||||
self._active = True
|
||||
self._ready.set()
|
||||
|
||||
msg = wintypes.MSG()
|
||||
try:
|
||||
while True:
|
||||
rc = user32.GetMessageW(ctypes.byref(msg), None, 0, 0)
|
||||
if rc <= 0: # WM_QUIT (0) or error (-1)
|
||||
break
|
||||
user32.TranslateMessage(ctypes.byref(msg))
|
||||
user32.DispatchMessageW(ctypes.byref(msg))
|
||||
finally:
|
||||
if self._hook:
|
||||
user32.UnhookWindowsHookEx(self._hook)
|
||||
self._hook = None
|
||||
self._active = False
|
||||
@@ -0,0 +1,195 @@
|
||||
"""Core Monitor orchestrator: manages away state, hooks, camera, device monitor, and triggers actions."""
|
||||
|
||||
import threading
|
||||
import time
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any
|
||||
from queue import Queue, Empty
|
||||
|
||||
from .config import Config
|
||||
from .utils import get_idle_time_seconds, lock_workstation, get_timestamp, format_event
|
||||
from .camera import Camera
|
||||
from .input_hooks import InputHooks
|
||||
from .device_monitor import DeviceMonitor
|
||||
from .white_screen import WhiteScreen
|
||||
|
||||
|
||||
class Monitor:
|
||||
def __init__(self, config: Config):
|
||||
self.config = config
|
||||
self.idle_timeout = config.get("idle_timeout_seconds", 300)
|
||||
self.is_away = False
|
||||
self.manual_lock = False # True when away was forced (manual / intrusion); only unlock clears
|
||||
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.event_queue: Queue = Queue()
|
||||
self._running = False
|
||||
self._monitor_thread: Optional[threading.Thread] = None
|
||||
|
||||
self.camera = Camera(config)
|
||||
self.input_hooks = InputHooks(on_activity=self._on_input_activity)
|
||||
self.device_monitor = DeviceMonitor(on_device_event=self._on_device_event)
|
||||
|
||||
# WhiteScreen wired by TrayApp with verify_password/on_failed/on_unlocked callbacks
|
||||
dim_cfg = config.get("white_screen_dimming", {}) or {}
|
||||
self.white_screen = WhiteScreen(
|
||||
enabled=config.get("white_screen_on_away", True),
|
||||
max_attempts=config.get("max_unlock_attempts", 3),
|
||||
block_escape_keys=config.get("block_escape_keys", True),
|
||||
dimming_enabled=bool(dim_cfg.get("enabled", True)),
|
||||
dimming_duration_seconds=float(dim_cfg.get("duration_seconds", 60.0)),
|
||||
target_grey=dim_cfg.get("target_grey", "#1a1a1a"),
|
||||
)
|
||||
|
||||
self.log_file = Path(config.get("log_file", "events.log"))
|
||||
|
||||
# Actions will be injected later
|
||||
self.on_intrusion: Optional[callable] = None
|
||||
|
||||
def _log_event(self, event_type: str, details: str) -> None:
|
||||
event = {
|
||||
"timestamp": get_timestamp(),
|
||||
"type": event_type,
|
||||
"details": details,
|
||||
"away": self.is_away
|
||||
}
|
||||
try:
|
||||
with open(self.log_file, "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(event) + "\n")
|
||||
except Exception:
|
||||
pass
|
||||
print(format_event(event_type, details))
|
||||
|
||||
def _on_input_activity(self, activity_type: str) -> None:
|
||||
if not self.is_away:
|
||||
return
|
||||
# If the white screen / password gate is already up, this input is the
|
||||
# challenge response (owner typing the unlock phrase). Do NOT re-trigger
|
||||
# intrusion - that wipes the password field and spams alerts.
|
||||
if self.white_screen.is_shown:
|
||||
return
|
||||
now = time.time()
|
||||
if now - self.last_trigger_time < self.cooldown_seconds:
|
||||
return
|
||||
self.last_trigger_time = now
|
||||
self._log_event("input_activity", activity_type)
|
||||
self._trigger_intrusion(f"Input activity detected: {activity_type}")
|
||||
|
||||
def _on_device_event(self, event: str, details: str) -> None:
|
||||
self._log_event(event, details)
|
||||
if self.is_away:
|
||||
self._trigger_intrusion(f"Device event: {event} {details}")
|
||||
|
||||
def _trigger_intrusion(self, reason: str) -> None:
|
||||
if not self.config.get("monitoring_enabled", True):
|
||||
return
|
||||
self._log_event("intrusion_detected", reason)
|
||||
|
||||
# Intrusion locks the gate - only correct password can clear it
|
||||
self.manual_lock = True
|
||||
self.is_away = True
|
||||
|
||||
# Capture snapshot(s) from every working camera
|
||||
snapshot_path = None
|
||||
snapshot_paths = []
|
||||
if self.config.get("snapshot_on_trigger", True):
|
||||
snapshot_paths = self.camera.capture_all_snapshots(reason="intrusion") or []
|
||||
if snapshot_paths:
|
||||
snapshot_path = snapshot_paths[0]
|
||||
self._log_event(
|
||||
"snapshot_captured",
|
||||
json.dumps(snapshot_paths, ensure_ascii=False),
|
||||
)
|
||||
|
||||
# Call external handler (actions)
|
||||
if self.on_intrusion:
|
||||
try:
|
||||
self.on_intrusion(reason, snapshot_path, snapshot_paths)
|
||||
except TypeError:
|
||||
# back-compat: handler with old (reason, snapshot_path) signature
|
||||
try:
|
||||
self.on_intrusion(reason, snapshot_path)
|
||||
except Exception as e:
|
||||
self._log_event("action_error", str(e))
|
||||
except Exception as e:
|
||||
self._log_event("action_error", str(e))
|
||||
|
||||
def _state_machine(self) -> None:
|
||||
"""Background thread: update away state based on idle time + camera obstruction checks."""
|
||||
last_obscured_check = 0.0
|
||||
obs_cfg = self.config.get("camera_obscured_detection", {})
|
||||
check_interval = obs_cfg.get("check_interval_seconds", 12)
|
||||
|
||||
while self._running:
|
||||
idle = get_idle_time_seconds()
|
||||
now = time.time()
|
||||
snoozed = now < self.snooze_until
|
||||
should_be_away = (idle >= self.idle_timeout) and not snoozed
|
||||
|
||||
if should_be_away and not self.is_away:
|
||||
self.is_away = True
|
||||
self._log_event("state_change", f"User away (idle {idle:.1f}s)")
|
||||
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)
|
||||
if (self.is_away and
|
||||
self.white_screen.enabled and
|
||||
self.white_screen.is_shown and
|
||||
self.camera.obscured_enabled):
|
||||
now = time.time()
|
||||
if now - last_obscured_check >= check_interval:
|
||||
last_obscured_check = now
|
||||
if self.camera.is_vision_obscured():
|
||||
self._trigger_intrusion("Camera vision obscured (possible tampering)")
|
||||
|
||||
time.sleep(5)
|
||||
|
||||
def start(self) -> None:
|
||||
if self._running:
|
||||
return
|
||||
self._running = True
|
||||
self.input_hooks.start()
|
||||
if self.config.get("device_monitoring", True):
|
||||
self.device_monitor.start()
|
||||
self._monitor_thread = threading.Thread(target=self._state_machine, daemon=True)
|
||||
self._monitor_thread.start()
|
||||
self._log_event("monitor_started", "ICEYOU monitoring active")
|
||||
|
||||
def stop(self) -> None:
|
||||
self._running = False
|
||||
self.input_hooks.stop()
|
||||
self.device_monitor.stop()
|
||||
self.camera.release()
|
||||
if self.white_screen:
|
||||
self.white_screen.hide()
|
||||
if self._monitor_thread:
|
||||
self._monitor_thread.join(timeout=3)
|
||||
self._log_event("monitor_stopped", "Monitoring stopped")
|
||||
|
||||
def set_away(self, away: bool) -> None:
|
||||
self.is_away = away
|
||||
self._log_event("manual_state", f"Manually set away={away}")
|
||||
if away:
|
||||
# Manual force: lock until explicit unlock
|
||||
self.manual_lock = True
|
||||
self.snooze_until = 0.0
|
||||
if self.white_screen.enabled:
|
||||
self.white_screen.show()
|
||||
else:
|
||||
# Manual release (e.g. successful unlock): clear lock + snooze auto-away
|
||||
self.manual_lock = False
|
||||
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()
|
||||
@@ -0,0 +1,179 @@
|
||||
"""Motion detection + clip recording for the ICEYOU whiteout.
|
||||
|
||||
Runs only while the whiteout is shown. Continuously samples webcam frames
|
||||
into a rolling pre-buffer (default 5s). On motion: dumps the pre-buffer to
|
||||
an AVI clip, then keeps recording for `post_seconds` more. Fires `on_event`
|
||||
with the resulting clip path when finalized.
|
||||
|
||||
Uses simple frame-differencing for motion detection - cheap and effective
|
||||
in a controlled indoor scene.
|
||||
"""
|
||||
|
||||
import time
|
||||
import threading
|
||||
from collections import deque
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
import cv2
|
||||
|
||||
from .utils import get_timestamp
|
||||
|
||||
|
||||
class MotionRecorder:
|
||||
def __init__(
|
||||
self,
|
||||
camera, # anything with read_frame() - Camera OR CameraStream
|
||||
clips_dir: str = "motion_clips",
|
||||
fps: int = 15,
|
||||
pre_seconds: float = 5.0,
|
||||
post_seconds: float = 5.0,
|
||||
sensitivity: int = 2500, # min motion contour area (pixels)
|
||||
cooldown_seconds: float = 8.0,
|
||||
on_event: Optional[Callable[[str], None]] = None,
|
||||
cam_label: str = "", # appended to clip filename (e.g. "cam1") for multi-cam
|
||||
):
|
||||
self.camera = camera
|
||||
self.cam_label = str(cam_label or "").strip()
|
||||
self.clips_dir = Path(clips_dir)
|
||||
self.clips_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.fps = max(5, int(fps))
|
||||
self.pre_seconds = max(0.5, float(pre_seconds))
|
||||
self.post_seconds = max(0.5, float(post_seconds))
|
||||
self.sensitivity = int(sensitivity)
|
||||
self.cooldown_seconds = float(cooldown_seconds)
|
||||
self.on_event = on_event or (lambda path: None)
|
||||
|
||||
self._buffer = deque(maxlen=int(self.fps * self.pre_seconds))
|
||||
self._prev_gray = None
|
||||
self._running = False
|
||||
self._thread: Optional[threading.Thread] = None
|
||||
|
||||
self._writer = None
|
||||
self._writer_path: Optional[Path] = None
|
||||
self._recording_until = 0.0
|
||||
self._last_event_time = 0.0
|
||||
|
||||
@property
|
||||
def active(self) -> bool:
|
||||
return self._running
|
||||
|
||||
def start(self) -> None:
|
||||
if self._running:
|
||||
return
|
||||
self._running = True
|
||||
self._buffer.clear()
|
||||
self._prev_gray = None
|
||||
self._thread = threading.Thread(target=self._loop, daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def stop(self) -> None:
|
||||
if not self._running:
|
||||
return
|
||||
self._running = False
|
||||
if self._thread:
|
||||
self._thread.join(timeout=2.0)
|
||||
self._thread = None
|
||||
self._finalize_writer() # ensure any open clip is closed cleanly
|
||||
self._buffer.clear()
|
||||
self._prev_gray = None
|
||||
|
||||
# ---- internal ----
|
||||
|
||||
def _detect_motion(self, gray) -> bool:
|
||||
if self._prev_gray is None or self._prev_gray.shape != gray.shape:
|
||||
self._prev_gray = gray
|
||||
return False
|
||||
diff = cv2.absdiff(self._prev_gray, gray)
|
||||
self._prev_gray = gray
|
||||
_, thresh = cv2.threshold(diff, 25, 255, cv2.THRESH_BINARY)
|
||||
thresh = cv2.dilate(thresh, None, iterations=2)
|
||||
contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
|
||||
for c in contours:
|
||||
if cv2.contourArea(c) >= self.sensitivity:
|
||||
return True
|
||||
return False
|
||||
|
||||
def _begin_recording(self, sample_frame) -> None:
|
||||
if self._writer is not None:
|
||||
return
|
||||
h, w = sample_frame.shape[:2]
|
||||
suffix = f"_{self.cam_label}" if self.cam_label else ""
|
||||
path = self.clips_dir / f"motion_{get_timestamp()}{suffix}.avi"
|
||||
fourcc = cv2.VideoWriter_fourcc(*"MJPG")
|
||||
writer = cv2.VideoWriter(str(path), fourcc, self.fps, (w, h))
|
||||
if not writer.isOpened():
|
||||
print("[MotionRecorder] VideoWriter failed to open")
|
||||
return
|
||||
self._writer = writer
|
||||
self._writer_path = path
|
||||
# Dump pre-event buffer first so the clip starts BEFORE the motion
|
||||
for buffered in list(self._buffer):
|
||||
try:
|
||||
self._writer.write(buffered)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _finalize_writer(self) -> None:
|
||||
if self._writer is None:
|
||||
return
|
||||
try:
|
||||
self._writer.release()
|
||||
except Exception:
|
||||
pass
|
||||
path = self._writer_path
|
||||
self._writer = None
|
||||
self._writer_path = None
|
||||
self._recording_until = 0.0
|
||||
if path is not None:
|
||||
try:
|
||||
self.on_event(str(path))
|
||||
except Exception as e:
|
||||
print(f"[MotionRecorder] on_event error: {e}")
|
||||
|
||||
def _loop(self) -> None:
|
||||
period = 1.0 / self.fps
|
||||
# Small warm-up: grab a few frames before motion detection kicks in
|
||||
# so the previous-frame baseline isn't blank.
|
||||
warmup_until = time.time() + 0.7
|
||||
|
||||
while self._running:
|
||||
t0 = time.time()
|
||||
frame = self.camera.read_frame()
|
||||
if frame is None:
|
||||
time.sleep(period)
|
||||
continue
|
||||
|
||||
self._buffer.append(frame)
|
||||
|
||||
try:
|
||||
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
||||
gray = cv2.GaussianBlur(gray, (21, 21), 0)
|
||||
except Exception:
|
||||
gray = None
|
||||
|
||||
now = time.time()
|
||||
|
||||
# If currently recording, keep writing post-event frames
|
||||
if self._writer is not None:
|
||||
try:
|
||||
self._writer.write(frame)
|
||||
except Exception:
|
||||
pass
|
||||
if now >= self._recording_until:
|
||||
self._finalize_writer()
|
||||
|
||||
# Motion detection (skip during warmup + during active cooldown)
|
||||
if gray is not None and now > warmup_until:
|
||||
moved = self._detect_motion(gray)
|
||||
if moved:
|
||||
if now - self._last_event_time >= self.cooldown_seconds:
|
||||
self._last_event_time = now
|
||||
self._begin_recording(frame)
|
||||
# Extend post-roll while motion continues
|
||||
if self._writer is not None:
|
||||
self._recording_until = now + self.post_seconds
|
||||
|
||||
elapsed = time.time() - t0
|
||||
if elapsed < period:
|
||||
time.sleep(period - elapsed)
|
||||
@@ -0,0 +1,50 @@
|
||||
"""Helper to register ICEYOU to run at Windows startup."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import winreg
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def add_to_startup(app_name: str = "ICEYOU") -> bool:
|
||||
"""Add the current script to Windows startup via registry."""
|
||||
try:
|
||||
exe_path = sys.executable
|
||||
script_path = str(Path(__file__).parent.parent.parent / "main.py")
|
||||
command = f'"{exe_path}" "{script_path}"'
|
||||
|
||||
key = winreg.OpenKey(
|
||||
winreg.HKEY_CURRENT_USER,
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Run",
|
||||
0, winreg.KEY_SET_VALUE
|
||||
)
|
||||
winreg.SetValueEx(key, app_name, 0, winreg.REG_SZ, command)
|
||||
winreg.CloseKey(key)
|
||||
print(f"Added to startup: {command}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Failed to add to startup: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def remove_from_startup(app_name: str = "ICEYOU") -> bool:
|
||||
try:
|
||||
key = winreg.OpenKey(
|
||||
winreg.HKEY_CURRENT_USER,
|
||||
r"Software\Microsoft\Windows\CurrentVersion\Run",
|
||||
0, winreg.KEY_SET_VALUE
|
||||
)
|
||||
winreg.DeleteValue(key, app_name)
|
||||
winreg.CloseKey(key)
|
||||
print("Removed from startup.")
|
||||
return True
|
||||
except FileNotFoundError:
|
||||
print("Not registered.")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
add_to_startup()
|
||||
@@ -0,0 +1,368 @@
|
||||
"""System tray application for ICEYOU using pystray + Pillow."""
|
||||
|
||||
import threading
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from pystray import Icon, Menu, MenuItem
|
||||
from PIL import Image, ImageDraw
|
||||
from pynput import keyboard
|
||||
|
||||
from .config import Config
|
||||
from .monitor import Monitor
|
||||
from .actions import Actions
|
||||
from .utils import lock_workstation
|
||||
from .face_recognizer import FaceRecognizer
|
||||
from .motion_recorder import MotionRecorder
|
||||
|
||||
|
||||
def create_icon_image() -> Image.Image:
|
||||
"""Create a simple tray icon (red shield)."""
|
||||
img = Image.new("RGB", (64, 64), color=(30, 30, 30))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.polygon([(32, 4), (60, 16), (60, 40), (32, 60), (4, 40), (4, 16)], fill=(220, 53, 69))
|
||||
draw.ellipse([22, 18, 42, 38], fill=(255, 255, 255))
|
||||
draw.rectangle([28, 24, 36, 34], fill=(220, 53, 69))
|
||||
return img
|
||||
|
||||
|
||||
class TrayApp:
|
||||
def __init__(self):
|
||||
self.config = Config()
|
||||
self.monitor = Monitor(self.config)
|
||||
self.actions = Actions(self.config, self.monitor)
|
||||
self.monitor.on_intrusion = self._on_intrusion
|
||||
|
||||
# Optional face recognition
|
||||
self.face_recognizer: Optional[FaceRecognizer] = None
|
||||
face_cfg = self.config.get("face_unlock", {}) or {}
|
||||
face_enabled = bool(face_cfg.get("enabled", self.config.get("face_recognition_enabled", False)))
|
||||
if face_enabled:
|
||||
threshold = float(face_cfg.get("confidence_threshold", 70.0))
|
||||
allowed = face_cfg.get("allowed_labels") # None = any enrolled
|
||||
min_matches = int(face_cfg.get("min_successful_matches", 2))
|
||||
self.face_recognizer = FaceRecognizer(
|
||||
confidence_threshold=threshold,
|
||||
allowed_labels=allowed,
|
||||
min_successful_matches=min_matches,
|
||||
)
|
||||
if not self.face_recognizer.available:
|
||||
self.face_recognizer = None # Enrol first; fall back to password only
|
||||
|
||||
# Wire WhiteScreen callbacks
|
||||
ws = self.monitor.white_screen
|
||||
ws.verify_password = self._verify_password
|
||||
ws.on_failed_attempt = self._on_failed_attempt
|
||||
ws.on_unlocked = self._on_unlocked
|
||||
ws.on_attempt = self._on_unlock_attempt
|
||||
ws.max_attempts = self.config.get("max_unlock_attempts", 3)
|
||||
if self.face_recognizer is not None:
|
||||
ws.face_unlock = self._try_face_unlock
|
||||
ws.face_unlock_auto = bool(face_cfg.get("auto_attempt", True))
|
||||
ws.face_unlock_interval = float(face_cfg.get("attempt_interval_seconds", 4.0))
|
||||
|
||||
# Optional motion recorder (only active while whiteout shown).
|
||||
# Multi-cam: build one MotionRecorder per alive stream so each camera
|
||||
# gets its own clip file.
|
||||
motion_cfg = self.config.get("motion_recording", {}) or {}
|
||||
self.motion_clips_dir = motion_cfg.get("clips_dir", "motion_clips")
|
||||
if bool(motion_cfg.get("enabled", True)):
|
||||
recorders = []
|
||||
alive = self.monitor.camera.alive_streams
|
||||
multi = len(alive) > 1
|
||||
for stream in alive:
|
||||
label = f"cam{stream.index}" if multi else ""
|
||||
recorders.append(MotionRecorder(
|
||||
camera=stream,
|
||||
clips_dir=self.motion_clips_dir,
|
||||
fps=int(motion_cfg.get("fps", 15)),
|
||||
pre_seconds=float(motion_cfg.get("pre_seconds", 5.0)),
|
||||
post_seconds=float(motion_cfg.get("post_seconds", 5.0)),
|
||||
sensitivity=int(motion_cfg.get("sensitivity", 2500)),
|
||||
cooldown_seconds=float(motion_cfg.get("cooldown_seconds", 8.0)),
|
||||
on_event=self._on_motion_event,
|
||||
cam_label=label,
|
||||
))
|
||||
ws.motion_recorders = recorders
|
||||
|
||||
self.icon: Icon = None
|
||||
self._hotkey_listener: Optional[keyboard.GlobalHotKeys] = None
|
||||
self._create_icon()
|
||||
self._setup_hotkeys()
|
||||
|
||||
# ---- WhiteScreen callbacks ----
|
||||
|
||||
def _verify_password(self, entered: str) -> bool:
|
||||
correct = str(self.config.get("unlock_password", "iceyou2026")).strip()
|
||||
return entered.strip() == correct
|
||||
|
||||
def _try_face_unlock(self):
|
||||
"""Called by WhiteScreen on its background thread. Returns (matched, label, confidence)."""
|
||||
if self.face_recognizer is None or not self.face_recognizer.available:
|
||||
return False, None, None
|
||||
attempts = int(self.config.get("face_unlock", {}).get("frames_per_attempt", 4))
|
||||
return self.face_recognizer.verify_from_camera(self.monitor.camera, attempts=attempts)
|
||||
|
||||
def _on_motion_event(self, clip_path: str) -> None:
|
||||
"""Called by MotionRecorder after a clip is finalized.
|
||||
Also resets white-screen brightness tuner on motion.
|
||||
"""
|
||||
try:
|
||||
self.monitor._log_event("motion_recorded", clip_path)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
if self.icon:
|
||||
self.icon.notify(
|
||||
f"Motion event saved: {Path(clip_path).name}",
|
||||
"ICEYOU",
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
# Reset dimmer on any detected motion while locked
|
||||
try:
|
||||
ws = self.monitor.white_screen
|
||||
if ws.is_shown:
|
||||
ws.reset_brightness()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _on_unlock_attempt(self, record: dict) -> None:
|
||||
"""Fired by WhiteScreen after each interactive unlock attempt.
|
||||
|
||||
Captures a webcam snapshot of whoever is at the keyboard, and logs the
|
||||
attempt (including the entered text on failures) to events.log.
|
||||
"""
|
||||
# Run on a background thread so the Tk UI thread isn't blocked by
|
||||
# camera I/O.
|
||||
def worker():
|
||||
try:
|
||||
method = record.get("method", "?")
|
||||
success = bool(record.get("success", False))
|
||||
reason = f"unlock_{'success' if success else 'fail'}_{method}"
|
||||
# Snap every working camera for richer evidence
|
||||
snaps = self.monitor.camera.capture_all_snapshots(reason=reason)
|
||||
primary = snaps[0] if snaps else None
|
||||
entry = {
|
||||
"event": "unlock_attempt",
|
||||
"method": method,
|
||||
"success": success,
|
||||
"snapshot": primary,
|
||||
"snapshots": snaps,
|
||||
}
|
||||
if method == "password" and not success:
|
||||
entry["entered"] = record.get("entered", "")
|
||||
if method == "face":
|
||||
entry["label"] = record.get("label")
|
||||
entry["confidence"] = record.get("confidence")
|
||||
self.monitor._log_event("unlock_attempt", json.dumps(entry, ensure_ascii=False))
|
||||
|
||||
# Fire email alert with the entered credentials + all snapshots
|
||||
self.actions.handle_unlock_attempt({
|
||||
**record,
|
||||
"snapshot": primary,
|
||||
"snapshots": snaps,
|
||||
})
|
||||
except Exception as e:
|
||||
print(f"[TrayApp] unlock attempt logging failed: {e}")
|
||||
|
||||
threading.Thread(target=worker, daemon=True).start()
|
||||
|
||||
def _on_failed_attempt(self, remaining: int) -> None:
|
||||
if remaining == 0:
|
||||
try:
|
||||
lock_workstation()
|
||||
except Exception as e:
|
||||
print(f"Lock failed: {e}")
|
||||
|
||||
def _on_unlocked(self) -> None:
|
||||
self.monitor.set_away(False)
|
||||
try:
|
||||
if self.icon:
|
||||
self.icon.notify("Whiteout unlocked", "ICEYOU")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ---- Intrusion handler chained from Monitor ----
|
||||
|
||||
def _on_intrusion(self, reason: str, snapshot_path: Optional[str] = None,
|
||||
snapshot_paths: Optional[list] = None) -> None:
|
||||
"""Email + log + ensure white screen with password gate is showing."""
|
||||
self.actions.handle_intrusion(reason, snapshot_path, snapshot_paths)
|
||||
self.monitor.white_screen.enabled = True
|
||||
self.monitor.white_screen.show()
|
||||
|
||||
# ---- Tray UI ----
|
||||
|
||||
def _create_icon(self) -> None:
|
||||
menu = Menu(
|
||||
MenuItem("Start Monitoring", self._start_monitoring, default=True),
|
||||
MenuItem("Stop Monitoring", self._stop_monitoring),
|
||||
MenuItem("Toggle Away Mode", self._toggle_away),
|
||||
MenuItem("Toggle White Screen", self._toggle_white_screen),
|
||||
Menu.SEPARATOR,
|
||||
MenuItem("Open Snapshots Folder", self._open_snapshots),
|
||||
MenuItem("View Motion Clips", self._open_motion_clips),
|
||||
MenuItem("Open Log File", self._open_log),
|
||||
MenuItem("Send Test Email", self._send_test_email),
|
||||
MenuItem("Settings", self._open_settings),
|
||||
Menu.SEPARATOR,
|
||||
MenuItem("Exit", self._exit_app)
|
||||
)
|
||||
self.icon = Icon(
|
||||
"ICEYOU",
|
||||
create_icon_image(),
|
||||
"ICEYOU - Personal Monitor",
|
||||
menu
|
||||
)
|
||||
|
||||
def _start_monitoring(self, icon=None, item=None) -> None:
|
||||
self.monitor.start()
|
||||
self.icon.notify("Monitoring started", "ICEYOU")
|
||||
|
||||
def _stop_monitoring(self, icon=None, item=None) -> None:
|
||||
self.monitor.stop()
|
||||
self.icon.notify("Monitoring stopped", "ICEYOU")
|
||||
|
||||
def _toggle_away(self, icon=None, item=None) -> None:
|
||||
new_state = not self.monitor.is_away
|
||||
self.monitor.set_away(new_state)
|
||||
state = "away" if new_state else "present"
|
||||
self.icon.notify(f"Manually set to {state}", "ICEYOU")
|
||||
|
||||
def _toggle_white_screen(self, icon=None, item=None) -> None:
|
||||
ws = self.monitor.white_screen
|
||||
ws.enabled = not ws.enabled
|
||||
if not ws.enabled:
|
||||
ws.hide()
|
||||
self.icon.notify(f"White screen {'enabled' if ws.enabled else 'disabled'}", "ICEYOU")
|
||||
|
||||
# ---- Hotkeys ----
|
||||
|
||||
def _setup_hotkeys(self) -> None:
|
||||
hotkeys = self.config.get("hotkeys", {})
|
||||
force_away_combo = hotkeys.get("force_away", "<ctrl>+<alt>+l")
|
||||
unlock_combo = hotkeys.get("unlock", "<ctrl>+<alt>+u")
|
||||
|
||||
def on_force_away():
|
||||
self.monitor.set_away(True)
|
||||
try:
|
||||
self.icon.notify("Forced away mode + white screen", "ICEYOU")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def on_unlock():
|
||||
# Just ensure the white screen / password prompt is showing
|
||||
if self.monitor.is_away or self.monitor.white_screen.is_shown:
|
||||
self.monitor.white_screen.enabled = True
|
||||
self.monitor.white_screen.show()
|
||||
|
||||
try:
|
||||
self._hotkey_listener = keyboard.GlobalHotKeys({
|
||||
force_away_combo: on_force_away,
|
||||
unlock_combo: on_unlock,
|
||||
})
|
||||
except Exception as e:
|
||||
print(f"Hotkey registration failed (check combo syntax): {e}")
|
||||
|
||||
# ---- Tray menu actions ----
|
||||
|
||||
def _open_snapshots(self, icon=None, item=None) -> None:
|
||||
path = Path(self.config.get("snapshot_dir", "snapshots"))
|
||||
path.mkdir(exist_ok=True)
|
||||
os.startfile(str(path))
|
||||
|
||||
def _open_motion_clips(self, icon=None, item=None) -> None:
|
||||
"""Open the motion-clips folder. Reachable only via the tray icon -
|
||||
while the whiteout is up the escape-key hook prevents reaching the
|
||||
tray, so this is effectively a post-authentication action.
|
||||
"""
|
||||
path = Path(getattr(self, "motion_clips_dir", "motion_clips"))
|
||||
path.mkdir(exist_ok=True)
|
||||
os.startfile(str(path))
|
||||
|
||||
def _open_log(self, icon=None, item=None) -> None:
|
||||
log_path = Path(self.config.get("log_file", "events.log"))
|
||||
if log_path.exists():
|
||||
os.startfile(str(log_path))
|
||||
else:
|
||||
try:
|
||||
self.icon.notify("No log file yet.", "ICEYOU")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _send_test_email(self, icon=None, item=None) -> None:
|
||||
try:
|
||||
self.actions.send_test_email()
|
||||
self.icon.notify("Test email queued. Check inbox / spam.", "ICEYOU")
|
||||
except Exception as e:
|
||||
try:
|
||||
self.icon.notify(f"Test email failed: {e}", "ICEYOU")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _open_settings(self, icon=None, item=None) -> None:
|
||||
"""Open config.json in the default text editor.
|
||||
|
||||
We avoid creating a second Tk root here because WhiteScreen owns a
|
||||
persistent Tk root on its own thread, and multiple Tk interpreters
|
||||
across threads are unsafe and freeze the tray.
|
||||
"""
|
||||
cfg_path = Path("config.json")
|
||||
if not cfg_path.exists():
|
||||
example = Path("config.example.json")
|
||||
if example.exists():
|
||||
try:
|
||||
cfg_path.write_text(example.read_text(encoding="utf-8"), encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
if cfg_path.exists():
|
||||
try:
|
||||
subprocess.Popen(["notepad.exe", str(cfg_path)])
|
||||
except Exception as e:
|
||||
try:
|
||||
self.icon.notify(f"Open settings failed: {e}", "ICEYOU")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
self.icon.notify("Edit config.json then restart ICEYOU", "ICEYOU")
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
self.icon.notify("config.json not found", "ICEYOU")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _exit_app(self, icon=None, item=None) -> None:
|
||||
if self._hotkey_listener:
|
||||
try:
|
||||
self._hotkey_listener.stop()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
self.monitor.stop()
|
||||
except Exception:
|
||||
pass
|
||||
# Stop the tray icon; this causes icon.run() to return.
|
||||
# Do NOT call sys.exit here - it triggers a message handler error in pystray.
|
||||
self.icon.stop()
|
||||
|
||||
def run(self) -> None:
|
||||
# Pre-warm WhiteScreen Tk thread so the very first show() never blocks
|
||||
# the tray callback waiting for the interpreter to boot.
|
||||
try:
|
||||
self.monitor.white_screen._ensure_ui_thread()
|
||||
except Exception as e:
|
||||
print(f"WhiteScreen pre-warm failed: {e}")
|
||||
if self.config.get("monitoring_enabled", True):
|
||||
self.monitor.start()
|
||||
if self._hotkey_listener:
|
||||
self._hotkey_listener.start()
|
||||
self.icon.run()
|
||||
# icon.run() returned because _exit_app called icon.stop()
|
||||
sys.exit(0)
|
||||
@@ -0,0 +1,34 @@
|
||||
"""Windows utility functions: idle time, workstation lock, etc."""
|
||||
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
# Windows API: LASTINPUTINFO struct for idle detection
|
||||
class LASTINPUTINFO(ctypes.Structure):
|
||||
_fields_ = [("cbSize", wintypes.UINT), ("dwTime", wintypes.DWORD)]
|
||||
|
||||
user32 = ctypes.windll.user32
|
||||
kernel32 = ctypes.windll.kernel32
|
||||
|
||||
def get_idle_time_seconds() -> float:
|
||||
"""Return seconds since last user input (keyboard or mouse)."""
|
||||
last_input = LASTINPUTINFO()
|
||||
last_input.cbSize = ctypes.sizeof(LASTINPUTINFO)
|
||||
if user32.GetLastInputInfo(ctypes.byref(last_input)):
|
||||
millis = kernel32.GetTickCount() - last_input.dwTime
|
||||
return millis / 1000.0
|
||||
return 0.0
|
||||
|
||||
def lock_workstation() -> bool:
|
||||
"""Lock the current workstation (same as Win+L)."""
|
||||
return bool(user32.LockWorkStation())
|
||||
|
||||
def get_timestamp() -> str:
|
||||
"""ISO-like timestamp for filenames and logs."""
|
||||
return datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3]
|
||||
|
||||
def format_event(event_type: str, details: str) -> str:
|
||||
"""Simple event string."""
|
||||
return f"[{datetime.now().isoformat()}] {event_type}: {details}"
|
||||
Reference in New Issue
Block a user