Skip to content

check: bulletproof error handling in process checks #107

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Oct 22, 2024
Merged

check: bulletproof error handling in process checks #107

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 68 additions & 32 deletions nimp/base_commands/check.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"""
Expand Down Expand Up @@ -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):
Expand Down