check: bulletproof error handling in process checks

Author: Thomas Desveaux

nimp/base_commands/check.py

@@ -201,68 +201,91 @@ def _run_check(self, env: NimpEnvironment):
                     logging.debug("\t%s (%s) %s", ignored_process.exe(), ignored_process.pid, ignored_process.cmdline())
 
             for process in psutil.process_iter():
+                if not process.is_running():
+                    continue
+
                 logging.debug("Checking process %d", process.pid)
                 if process.pid in ignore_process_ids:
                     logging.debug("[Process(%d)] ignore process (self, parent or child)", process.pid)
                     continue
 
                 if current_user is not None:
-                    process_user = None
-                    try:
-                        process_user = process.username()
-                    except psutil.AccessDenied:
-                        logging.debug(
-                            "[Process(%d)] Failed to retrieve process user",
-                            process.pid,
-                        )
-                        continue
-
-                    if current_user != process_user:
-                        logging.debug(
-                            "[Process(%d)] ignore process from other user (self: %s, process user: %s)",
-                            process.pid,
-                            current_user,
-                            process_user,
-                        )
+                    if not _Processes._process_owned_by_user(process, current_user):
                         continue
 
                 checked_processes_count += 1
                 if not _Processes._process_matches_filters(process, env.filters):
                     continue
 
-                process_executable_path = process.exe()
-                process_basename = os.path.basename(process_executable_path)
-                if any(p.match(process_basename) for p in _Processes.PROCESS_IGNORE_PATTERNS):
-                    logging.info('[Process(%d)] process (%s) will be kept alive', process.pid, process_executable_path)
+                try:
+                    process_exe = process.exe()
+                except psutil.Error as exc:
+                    logging.debug('[Process(%d)] failed to retrieve process executable', process.pid, exc_info=exc)
+                    process_exe = "[UNKOWN]"
+
+                if _Processes._should_ignore_process(process):
+                    logging.info('[Process(%d)] process (%s) will be kept alive', process.pid, process_exe)
                     continue
 
                 problematic_processes.append(process)
-                logging.warning('[Process(%d)] Found problematic process (%s)', process.pid, process_executable_path)
-                if (parent_process := process.parent()) is not None:
-                    logging.warning('\tParent is %s (%s)', parent_process.pid, parent_process.exe())
+                logging.warning('[Process(%d)] Found problematic process (%s)', process.pid, process_exe)
+                try:
+                    if (parent_process := process.parent()) is not None:
+                        logging.warning('\tParent is %s (%s)', parent_process.pid, parent_process.exe())
+                except psutil.Error as exc:
+                    logging.debug(
+                        '[Process(%d)] failed to get parent process information for process "%s"',
+                        process.pid, process_exe, exc_info=exc
+                    )
 
             logging.info('%d processes checked.', checked_processes_count)
             if not problematic_processes:
                 # no problematic processes running, nothing to do.
                 return True
 
+            sleep_time = 5.0
             if not env.kill:
                 # Wait a bit, give a chance to problematic processes to end,
                 # even if not killed
-                sleep_time = 5.0
+
                 logging.debug("Wait %.2fs. Giving a chance to processes for a natural exit", sleep_time)
                 time.sleep(sleep_time)
             else:
                 for p in problematic_processes:
-                    logging.info('Requesting process %s termination', p.pid)
-                    p.terminate()
-                _, alive = psutil.wait_procs(problematic_processes, timeout=5)
+                    if p.is_running():
+                        logging.info('Requesting process %s termination', p.pid)
+                        p.terminate()
+                _, alive = psutil.wait_procs(problematic_processes, timeout=sleep_time)
                 for p in alive:
-                    logging.info('Process %s not terminated. Send kill.', p.pid)
-                    p.kill()
+                    if p.is_running():
+                        logging.info('Process %s not terminated. Send kill.', p.pid)
+                        p.kill()
 
         return False
 
+    @staticmethod
+    def _process_owned_by_user(process: psutil.Process, username: str):
+        try:
+            process_user = process.username()
+        except psutil.Error as exception:
+            logging.debug(
+                "[Process(%d)] Failed to retrieve process user",
+                process.pid,
+                exc_info=exception,
+            )
+            return False
+
+        is_same_user = username == process_user
+
+        logging.debug(
+            "[Process(%d)] ignore process from other user (self: %s, process user: %s)",
+            process.pid,
+            username,
+            process_user,
+        )
+
+        return is_same_user
+
     @staticmethod
     def _process_matches_filters(process: psutil.Process, filters: list[str]) -> bool:
         """Returns True if the process should be filtered out"""
@@ -293,14 +316,27 @@ def _process_matches_filters(process: psutil.Process, filters: list[str]) -> boo
                             popen_file.path,
                         )
                         return True
-        except psutil.AccessDenied as exc:
-            logging.debug("[Process(%d)] Access Denied!", process.pid, exc_info=exc)
+        except psutil.Error as exc:
+            logging.debug("[Process(%d)] Error!", process.pid, exc_info=exc)
             # failed to access a property of the process,
             # assume it does not match to be safe
             return False
 
         return False
 
+    @staticmethod
+    def _should_ignore_process(process: psutil.Process) -> bool:
+        try:
+            process_executable_path = process.exe()
+            process_basename = os.path.basename(process_executable_path)
+        except psutil.Error:
+            logging.debug(
+                "[Process(%d)] failed to retrieve process exe/basename",
+                process.pid,
+            )
+            return True
+        return any(p.match(process_basename) for p in _Processes.PROCESS_IGNORE_PATTERNS)
+
 
 class _Disks(CheckCommand):
     def __init__(self):