Add internet detector utility

scripts/utils/internet_detect_tool.py

@@ -0,0 +1,265 @@
+import threading
+import requests
+import win32api
+import win32gui
+import win32con
+import urllib3
+import hashlib
+import winreg
+import struct
+import signal
+import ctypes
+import time
+import os
+import re
+
+
+# Define constants
+CHECK_INTERVAL = 2  # Seconds
+CONNECT_TEST_URLS_AND_RESPONSES = {
+    "http://www.google.com": 'ce770667e5f9b0d8f55367bb79419689d90c48451bb33f079f3a9a72ae132de8',        # HTTP Test
+    "https://www.wikipedia.com": 'd38b38a2dd476e045c299e8ee5d6466834456d97bd592a71746b423a6a05f386',    # HTTPS Test #2
+    "https://www.youtube.com": 'fb7accfff8c6f8ea9b03c91ee5576d0d08080e9ba35918d801aaeb6020dbc88c'       # HTTPS Test #3
+    }
+CONNECT_TEST_URL = "https://www.msftconnecttest.com/connecttest.txt"
+EXPECTED_RESPONSE = "Microsoft Connect Test"
+SPI_SETDESKWALLPAPER = 20
+SPIF_UPDATEINIFILE = 0x01
+SPIF_SENDWININICHANGE = 0x02
+COLOR_DESKTOP = 1
+ICON_RED_X = os.path.join(os.environ.get('VM_COMMON_DIR'), "red_circle.ico")
+ICON_INDICATOR = os.path.join(os.environ.get('VM_COMMON_DIR'), "indicator.ico")
+DEFAULT_BACKGROUND = os.path.join(os.environ.get('VM_COMMON_DIR'), "background.png")
+INTERNET_BACKGROUND = os.path.join(os.environ.get('VM_COMMON_DIR'),"background-internet.png")
+
+# Global variables
+tray_icon = None
+stop_event = threading.Event()  # To signal the background thread to stop
+hwnd = None     # We'll assign the window handle here later
+check_thread = None
+tray_icon_thread = None
+# Win32 API icon handles
+hicon_indicator = None
+hicon_red_x = None
+
+def signal_handler(sig, frame):
+    global check_thread, tray_icon_thread, tray_icon
+    print("Ctrl+C detected. Exiting...")
+    stop_event.set()    # Signal the background thread to stop
+    if check_thread:
+        check_thread.join()
+    if tray_icon_thread:
+        tray_icon_thread.join()
+    if tray_icon:
+        del tray_icon
+    exit(0)
+
+def load_icon(icon_path):
+    try:
+        return win32gui.LoadImage(None, icon_path, win32con.IMAGE_ICON, 0, 0, win32con.LR_LOADFROMFILE)
+    except Exception as e:
+        print(f"Error loading indicator icon: {e}")
+        return None
+
+class SysTrayIcon:
+    def __init__(self, hwnd, icon, tooltip):
+        self.hwnd = hwnd
+        self.icon = icon
+        self.tooltip = tooltip
+
+        # System tray icon data structure
+        self.nid = (self.hwnd, 0, win32gui.NIF_ICON | win32gui.NIF_MESSAGE | win32gui.NIF_TIP, 
+                    win32con.WM_USER + 20, self.icon, self.tooltip)
+
+        # Add the icon to the system tray
+        win32gui.Shell_NotifyIcon(win32gui.NIM_ADD, self.nid)
+
+    def set_icon(self, icon):
+        self.icon = icon
+        self.nid = (self.hwnd, 0, win32gui.NIF_ICON | win32gui.NIF_MESSAGE | win32gui.NIF_TIP, 
+                    win32con.WM_USER + 20, self.icon, self.tooltip)
+        win32gui.Shell_NotifyIcon(win32gui.NIM_MODIFY, self.nid)
+
+    def show_balloon_tip(self, title, msg):
+        # Display a balloon tip notification
+        win32gui.Shell_NotifyIcon(win32gui.NIM_MODIFY, (
+            self.hwnd,
+            0,
+            win32gui.NIF_INFO,
+            win32con.WM_USER + 20,
+            self.icon,
+            self.tooltip,
+            msg,
+            200,
+            title,
+        ))
+
+    def __del__(self):
+        # Remove the icon from the system tray when the object is destroyed
+        win32gui.Shell_NotifyIcon(win32gui.NIM_DELETE, self.nid)
+
+def extract_title(data):
+    match = re.search(r'<title>(.*?)</title', data)
+    if match:
+        return match.group(1)
+    else:
+        return None
+
+def check_internet_secondary():
+    # Backup internet connectivity tests
+    for url, hash in CONNECT_TEST_URLS_AND_RESPONSES.items():
+        try:
+            response = requests.get(url, timeout=5, verify=False)
+            if hash == hashlib.sha256(extract_title(response.text).encode()).hexdigest():
+                print("Internet connectivity detected via URL: {}".format(url))
+                return True
+        except:
+            pass
+    return False
+
+def check_internet_primary():
+    try:
+        # Primary internet connectivity test (common test Microsoft uses)
+        response = requests.get(CONNECT_TEST_URL, timeout=5, verify=False)
+        if EXPECTED_RESPONSE in response.text:
+            return True
+        else:
+            return check_internet_secondary()
+
+    except requests.exceptions.RequestException:
+        return check_internet_secondary()
+
+def update_tray_icon():
+    global tray_icon, hicon_indicator, hicon_red_x
+    if check_internet_primary():
+        tray_icon.set_icon(hicon_red_x)
+        # Set the background to internet connection background
+        if get_wallpaper_path() != INTERNET_BACKGROUND:     # Checked so program isn't continuously setting the wallpaper
+            set_wallpaper(INTERNET_BACKGROUND)
+    else:  # Reset background when internet is not detected
+        tray_icon.set_icon(hicon_indicator)
+        # Reset background to default
+        if get_wallpaper_path() != DEFAULT_BACKGROUND:      # Checked so program isn't continuously setting the wallpaper
+            reset_background()
+
+def check_internet_loop():
+    while not stop_event.is_set():
+        update_tray_icon()
+        time.sleep(CHECK_INTERVAL)
+
+def tray_icon_loop():
+    global hwnd, tray_icon, hicon_indicator, hicon_red_x, stop_event
+    # Load icons
+    hicon_indicator = load_icon(ICON_INDICATOR)
+    hicon_red_x = load_icon(ICON_RED_X)
+
+    # Wait for hwnd to be initialized
+    while hwnd is None:
+        time.sleep(0.1)
+
+    if hicon_indicator is None or hicon_red_x is None:
+        print("Error: Failed to load icons. Exiting TrayIconThread.")
+        return
+
+    tray_icon = SysTrayIcon(hwnd, hicon_indicator, "Internet Detector")
+
+    while not stop_event.is_set():
+        msg = win32gui.PeekMessage(hwnd, 0, 0, 0)
+        if msg and len(msg) == 6:
+            win32gui.TranslateMessage(msg)
+            win32gui.DispatchMessage(msg)
+        time.sleep(0.1)
+
+def enable_transparency_effects():
+  try:
+      key = winreg.OpenKey(winreg.HKEY_CURRENT_USER,
+                           r"Software\Microsoft\Windows\CurrentVersion\Themes\Personalize",
+                           0, winreg.KEY_ALL_ACCESS)
+
+      winreg.SetValueEx(key, "EnableTransparency", 0, winreg.REG_DWORD, 1)
+      winreg.CloseKey(key)
+  except WindowsError as e:
+      print(f"Error accessing or modifying registry: {e}")
+
+def get_wallpaper_path():
+    """Attempts to retrieve the path to the current wallpaper image."""
+    # Try to get the path from the registry (for wallpapers set through Windows settings)
+    try:
+        key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Control Panel\Desktop", 0, winreg.KEY_READ)
+        value, _ = winreg.QueryValueEx(key, "Wallpaper")
+        winreg.CloseKey(key)
+        if value:
+            return value
+    except WindowsError:
+        pass
+
+    # Check for cached wallpaper files (if the above fails)
+    cached_files_dir = os.path.join(os.getenv("APPDATA"), r"Microsoft\Windows\Themes\CachedFiles")
+    transcoded_wallpaper_path = os.path.join(os.getenv("APPDATA"), r"Microsoft\Windows\Themes\TranscodedWallpaper")
+
+    for file in os.listdir(cached_files_dir):
+        if file.endswith(('.jpg', '.jpeg', '.bmp', '.png')):
+            return os.path.join(cached_files_dir, file)
+
+    if os.path.exists(transcoded_wallpaper_path):
+        return transcoded_wallpaper_path
+
+    # If all else fails, return None
+    return None
+
+def set_wallpaper(image_path):
+    """Sets the desktop wallpaper to the image at the specified path."""
+    print("Setting wallpaper to: {}".format(image_path))
+    result = ctypes.windll.user32.SystemParametersInfoW(
+        SPI_SETDESKWALLPAPER, 0, image_path, SPIF_UPDATEINIFILE | SPIF_SENDWININICHANGE
+    )
+    if not result:
+        print("Error setting wallpaper. Make sure the image path is correct.")
+
+def reset_background():
+    # Set background back to default wallpaper
+    set_wallpaper(DEFAULT_BACKGROUND)
+
+def main_loop():
+    global stop_event, check_thread, tray_icon_thread, tray_icon
+    # Create and start the threads
+    tray_icon_thread = threading.Thread(target=tray_icon_loop)
+    check_thread = threading.Thread(target=check_internet_loop)
+
+    tray_icon_thread.start()
+    # Wait for the tray icon to finish initializing
+    while tray_icon is None:
+        time.sleep(0.1)
+
+    check_thread.start()
+
+    while not stop_event.is_set():
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    signal.signal(signal.SIGINT, signal_handler)
+    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+    enable_transparency_effects()
+
+    # Create a hidden window to receive messages (required for system tray icons)
+    def wndProc(hwnd, msg, wparam, lparam):
+        if lparam == win32con.WM_LBUTTONDBLCLK:
+            print("Left button double clicked")
+        elif msg == win32con.WM_COMMAND:
+            if wparam == 1023:  # Example menu item ID
+                print("Exit selected")
+                win32gui.DestroyWindow(hwnd)
+                # app.quit()  # Exit the PyQt application
+        return win32gui.DefWindowProc(hwnd, msg, wparam, lparam)
+
+    wc = win32gui.WNDCLASS()
+    hinst = wc.hInstance = win32api.GetModuleHandle(None)
+    wc.lpszClassName = "Internet Detector"
+    wc.lpfnWndProc = wndProc
+    classAtom = win32gui.RegisterClass(wc)
+    hwnd = win32gui.CreateWindow(classAtom, "Internet Detector", 0, 0, 0, 0, 0, 0, 0, hinst, None)
+
+    print("Current wallpaper: {}".format(get_wallpaper_path()))
+
+    main_loop()