feat(docker): add docker-compose for grass-desktop service and update…

… grass version in Dockerfile also adding extra packages for #5 #6 mitigation, better grass-desktop main logic

grass-desktop.docker-compose.yaml

@@ -0,0 +1,21 @@
+services:
+  grass-desktop:
+    build:
+      context: .
+      dockerfile: grass-desktop.dockerfile
+    environment:
+      USER_EMAIL: ${USER_EMAIL}
+      USER_PASSWORD: ${USER_PASSWORD}
+      MAX_RETRY_MULTIPLIER: ${MAX_RETRY_MULTIPLIER}
+
+    ports:
+      - "5900:5900"
+      - "6080:6080"
+    tty: true
+    stdin_open: true
+    # develop:
+    #   watch:
+    #     - action: sync
+    #       path: ./
+    #       target: /app/
+

grass-desktop.dockerfile

@@ -1,9 +1,9 @@
 # --- Stage 1: Build Stage to Patch Grass Deb ---
     FROM debian:stable-slim AS grass-deb-patcher
 
-    ARG GRASS_VERSION="4.30.0"
+    ARG GRASS_VERSION="4.31.2"
     ARG GRASS_ARCH="amd64"
-    ARG GRASS_PACKAGE_URL="https://files.getgrass.io/file/grass-extension-upgrades/ubuntu-22.04/grass_${GRASS_VERSION}_${GRASS_ARCH}.deb"
+    ARG GRASS_PACKAGE_URL="https://files.getgrass.io/file/grass-extension-upgrades/ubuntu-22.04/Grass_${GRASS_VERSION}_${GRASS_ARCH}.deb"
 
     RUN apt-get update && apt-get install -y --no-install-recommends \
         binutils \
@@ -38,7 +38,10 @@
         apt-get install -y --no-install-recommends \
         xdotool \
         ca-certificates \
-        dpkg
+        dpkg \
+        libayatana-appindicator3-1 \
+        libwebkit2gtk-4.1-0 \ 
+        libgtk-3-0
 
     # Copy patched deb from builder stage
     COPY --from=grass-deb-patcher /tmp/grass.deb /tmp/grass.deb

grass-desktop_main.py

@@ -13,7 +13,6 @@ def setup_logging():
     Set up logging for the script.
 
     This function configures the root logger with an INFO level and a specific log format.
-    It does not take any parameters and does not return any value.
     """
     logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
@@ -59,113 +58,248 @@ def search_windows_by_name(window_name, max_attempts, max_retry_multiplier):
 
         attempt += 1
         if attempt < max_attempts:
-            logging.warning(f"{window_name} window not found. Retrying... (attempt {attempt + 1}/{max_attempts})")
-            # Backoff timing: random wait multiplied by attempt and multiplier
+            logging.warning(f"{window_name} window not found (attempt {attempt}/{max_attempts}). Retrying...")
             backoff_time = random.randint(11, 31) * attempt * max_retry_multiplier
-            logging.info(f"Backing off for {backoff_time} seconds before attempt {attempt + 1}/{max_attempts}...")
+            logging.info(f"Backing off for {backoff_time} seconds before next attempt...")
             time.sleep(backoff_time)
 
-    logging.error(f"Failed to find the {window_name} window after {max_attempts} attempts. Exiting with error.")
-    sys.exit(1)
+    logging.error(f"Failed to find the {window_name} window after {max_attempts} attempts.")
+    return None
 
 
-def main():
+def launch_grass_with_retries(max_attempts, wait_time):
+    """
+    Attempt to start the Grass application up to max_attempts times.
+    Returns a subprocess.Popen object if successful, otherwise None.
+    """
+    for attempt in range(max_attempts):
+        logging.info(f"Launching Grass desktop application... (attempt {attempt+1}/{max_attempts})")
+        try:
+            proc = subprocess.Popen(["/usr/bin/grass"])
+        except FileNotFoundError:
+            logging.error("Grass executable not found at /usr/bin/grass.")
+            return None
+
+        # Wait a little to see if the process remains active
+        time.sleep(wait_time)
+
+        if proc.poll() is not None:
+            # If the process ended prematurely, try again
+            logging.warning(f"Grass process ended prematurely on attempt {attempt+1}.")
+            if attempt < max_attempts - 1:
+                logging.info("Retrying Grass launch...")
+                continue
+            else:
+                logging.error(f"Failed to start Grass after {max_attempts} attempts.")
+                return None
+        else:
+            # Grass is still running
+            return proc
+    return None
+
+
+def send_xdotool_key(key):
+    """
+    Send a single key press using xdotool, return True if successful, False if not.
     """
-    Main function to run the Grass Desktop script.
+    ret = subprocess.run(["xdotool", "key", key], check=False)
+    return (ret.returncode == 0)
 
-    This function:
-    - Sets up logging.
-    - Reads environment variables for USER_EMAIL and USER_PASSWORD (and their alternatives).
-    - Launches the Grass application if not already configured.
-    - Uses xdotool to detect and focus the Grass window.
-    - Automates login steps using keystrokes sent via xdotool.
-    - Waits in the foreground until the Grass process exits.
 
-    Raises:
-        SystemExit: If credentials are not provided or if the Grass window cannot be found after the maximum retries.
+def kill_process(proc):
     """
-    setup_logging()
-    logging.info('Starting Grass Desktop script...')
-
-    # Read variables from environment for credentials
-    email_username = (os.getenv('USER_EMAIL') or os.getenv('GRASS_EMAIL')
-                      or os.getenv('GRASS_USER') or os.getenv('GRASS_USERNAME'))
-    password = os.getenv('USER_PASSWORD') or os.getenv('GRASS_PASSWORD') or os.getenv('GRASS_PASS')
+    Gracefully terminate a process, then forcibly kill if it doesn't exit.
+    """
+    if proc.poll() is None:
+        proc.terminate()
+        try:
+            proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            proc.kill()
 
-    # Retrieve retry multiplier
-    MAX_RETRY_MULTIPLIER = int(os.getenv('MAX_RETRY_MULTIPLIER', '3'))
 
-    # Check if credentials are provided
-    if not email_username or not password:
-        logging.error('No username or password provided. Please set the USER_EMAIL and USER_PASSWORD environment variables.')
-        sys.exit(1)
+def relaunch_grass(max_attempts, max_retry_multiplier):
+    """
+    Relaunch Grass and return the new process or None if failed.
+    """
+    wait_time = max_retry_multiplier
+    return launch_grass_with_retries(max_attempts, wait_time)
 
-    # Start the Grass application
-    logging.info("Launching Grass desktop application...")
-    grass_proc = subprocess.Popen(["/usr/bin/grass"])
 
-    # Check if Grass was previously configured
+def configure_grass(grass_proc, email_username, password, max_attempts, max_retry_multiplier):
+    """
+    Attempt to configure Grass (i.e., login and do initial setup) multiple times if the window disappears.
+    Returns True if configuration succeeded, False otherwise.
+    """
     configured_flag = os.path.expanduser("~/.grass-configured")
 
-    if not os.path.exists(configured_flag):
-        logging.info("Grass not configured yet. Waiting for Grass window to appear...")
+    # If already configured, nothing to do
+    if os.path.exists(configured_flag):
+        logging.info("Grass already configured.")
+        return True
 
-        MAX_ATTEMPTS = MAX_RETRY_MULTIPLIER
-        windows = search_windows_by_name("Grass", MAX_ATTEMPTS, MAX_RETRY_MULTIPLIER)
+    for attempt in range(max_attempts):
+        logging.info("Grass not configured yet. Waiting for Grass window to appear...")
+        windows = search_windows_by_name("Grass", max_attempts, max_retry_multiplier)
+        if windows is None:
+            # Window not found even after attempts
+            return False
 
-        delay = MAX_RETRY_MULTIPLIER * 5
+        delay = max_retry_multiplier * 5
         logging.info(f"Waiting {delay} seconds for Grass interface to load...")
+        time.sleep(delay)
+
+        # Re-check if window still exists before focusing
+        windows = search_windows_by_name("Grass", max_attempts, max_retry_multiplier)
+        if windows is None:
+            logging.error("Grass window disappeared before focusing. Restarting Grass process.")
+            kill_process(grass_proc)
+            grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+            if grass_proc is None:
+                return False
+            continue
 
-        # Focus the last found Grass window
         last_window = windows[-1]
         logging.info("Focusing the Grass main window...")
-        subprocess.run(["xdotool", "windowfocus", "--sync", last_window], check=True)
-        time.sleep(MAX_RETRY_MULTIPLIER *2 )  # Wait increased x2 to help slow devices
+        if subprocess.run(["xdotool", "windowfocus", "--sync", last_window], check=False).returncode != 0:
+            logging.error("Failed to focus Grass window. It might have disappeared. Restarting configuration...")
+            kill_process(grass_proc)
+            grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+            if grass_proc is None:
+                return False
+            continue
+
+        time.sleep(max_retry_multiplier * 2)
 
         logging.info("Performing Grass login steps...")
         # Press Tab x4, then Enter
         for _ in range(4):
-            subprocess.run(["xdotool", "key", "Tab"], check=True)
-        subprocess.run(["xdotool", "key", "Return"], check=True)
-        time.sleep(MAX_RETRY_MULTIPLIER *2)  # Wait increased x2 to help slow devices
+            if not send_xdotool_key("Tab"):
+                kill_process(grass_proc)
+                grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+                if grass_proc is None:
+                    return False
+                continue
+        if not send_xdotool_key("Return"):
+            kill_process(grass_proc)
+            grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+            if grass_proc is None:
+                return False
+            continue
+
+        time.sleep(max_retry_multiplier * 2)
 
         logging.info("Entering credentials...")
-        # Type the username and press Tab (with a x ms delay between keystrokes)
         if email_username:
-            subprocess.run(["xdotool", "type", "--delay", "125", email_username], check=True)
-        time.sleep(MAX_RETRY_MULTIPLIER)
-        subprocess.run(["xdotool", "key", "Tab"], check=True)
-
-        time.sleep(MAX_RETRY_MULTIPLIER)  # Wait added to help slow devices
-
-        # Type the password and press Return (with a x ms delay between keystrokes)
+            if subprocess.run(["xdotool", "type", "--delay", "125", email_username], check=False).returncode != 0:
+                kill_process(grass_proc)
+                grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+                if grass_proc is None:
+                    return False
+                continue
+
+        time.sleep(max_retry_multiplier)
+        if not send_xdotool_key("Tab"):
+            kill_process(grass_proc)
+            grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+            if grass_proc is None:
+                return False
+            continue
+
+        time.sleep(max_retry_multiplier)
         if password:
-            subprocess.run(["xdotool", "type", "--delay", "125", re.sub("^-", "\-", password)], check=True)
-        time.sleep(MAX_RETRY_MULTIPLIER)  # Wait added to help slow devices
+            # Use re.sub to escape leading dash if present
+            escaped_password = re.sub(r"^-", r"\-", password)
+            if subprocess.run(["xdotool", "type", "--delay", "125", escaped_password], check=False).returncode != 0:
+                kill_process(grass_proc)
+                grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+                if grass_proc is None:
+                    return False
+                continue
+
+        time.sleep(max_retry_multiplier)
 
         logging.info("Sending credentials...")
-        # Enter credentials and log in
-        subprocess.run(["xdotool", "key", "Return"], check=True)
+        if not send_xdotool_key("Return"):
+            kill_process(grass_proc)
+            grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+            if grass_proc is None:
+                return False
+            continue
 
         logging.info("Credentials sent. Waiting for login to complete...")
-        time.sleep(MAX_RETRY_MULTIPLIER*5)
+        time.sleep(max_retry_multiplier * 3)
 
-        # Enable auto updates 
+        # Enable auto updates (Tab x2, space x2)
         for _ in range(2):
-            subprocess.run(["xdotool", "key", "Tab"], check=True)
-            subprocess.run(["xdotool", "key", "space"], check=True)
-
-        time.sleep(MAX_RETRY_MULTIPLIER)  # Wait added to help slow devices
-
-        # Press Escape to leave submenu
-        subprocess.run(["xdotool", "key", "Escape"], check=True)
+            if not (send_xdotool_key("Tab") and send_xdotool_key("space")):
+                kill_process(grass_proc)
+                grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+                if grass_proc is None:
+                    return False
+                continue
+
+        time.sleep(max_retry_multiplier)
+
+        # Press Escape to leave Grass submenu
+        if not send_xdotool_key("Escape"):
+            kill_process(grass_proc)
+            grass_proc = relaunch_grass(max_attempts, max_retry_multiplier)
+            if grass_proc is None:
+                return False
+            continue
 
         logging.info("Grass configuration completed successfully. Marking as configured.")
         with open(configured_flag, "w") as f:
             f.write("")
+        return True
+
+    return False
+
+
+def main():
+    setup_logging()
+
+    MAX_RETRY_MULTIPLIER = int(os.getenv('MAX_RETRY_MULTIPLIER') or 3)
+
+    # Optional initial wait to help in slow environments
+    initial_wait = MAX_RETRY_MULTIPLIER * 5
+    logging.info(f"Initial wait of {initial_wait}s to allow the X server to stabilize on slow devices.")
+    time.sleep(initial_wait)
+
+    logging.info('Starting Grass Desktop script...')
+
+    # Retrieve credentials from env variables
+    email_username = (
+        os.getenv('USER_EMAIL') or os.getenv('GRASS_EMAIL')
+        or os.getenv('GRASS_USER') or os.getenv('GRASS_USERNAME')
+    )
+    password = (
+        os.getenv('USER_PASSWORD') or os.getenv('GRASS_PASSWORD')
+        or os.getenv('GRASS_PASS')
+    )
+
+    if not email_username or not password:
+        logging.error('No username or password provided. Please set the USER_EMAIL and USER_PASSWORD environment variables.')
+        sys.exit(1)
+
+    max_attempts = MAX_RETRY_MULTIPLIER
+    wait_time = MAX_RETRY_MULTIPLIER
+
+    # Launch Grass with retries
+    grass_proc = launch_grass_with_retries(max_attempts, wait_time)
+    if grass_proc is None:
+        sys.exit(1)
+
+    # Configure Grass (login etc.) if needed
+    if not configure_grass(grass_proc, email_username, password, max_attempts, MAX_RETRY_MULTIPLIER):
+        logging.error("Failed to configure Grass after multiple attempts. Exiting.")
+        kill_process(grass_proc)
+        sys.exit(1)
 
     logging.info("Keeping the Grass process in the foreground...")
     logging.info("Grass Desktop is earning...")
+
     # Keep the process running in the foreground until Grass exits
     grass_proc.wait()