/
opt
/
imh-scan
/
/opt/imh-scan
mkdir
upload
Name
Size
Mode
Actions
procwatch/
-
0755
rm
sigs/
-
0755
rm
clamlib.py
22692
0644
edit
dl
rm
imh-procwatch
25993
0755
edit
dl
rm
imh-scan
14685
0755
edit
dl
rm
mkdef
4876
0755
edit
dl
rm
Edit:
/opt/imh-scan/imh-procwatch
(25993B)
#!/usr/lib/imh-scan/venv/bin/python3 """Check of the current running processes on the server for malware""" import os import platform import re import shlex import sys import time import pwd import configparser import subprocess import argparse import signal from dataclasses import dataclass import socket from pathlib import Path # import io from typing import Union import yaml import rads import rads.color as c import psutil from clamlib import ScanResult, Scanner, ask_prompt, jail_files if rads.IMH_ROLE == 'shared': QUARANTINE = '/opt/sharedrads/quarantine' else: QUARANTINE = '/opt/dedrads/quarantine' sys.stdout.reconfigure(errors="surrogateescape", line_buffering=True) sys.stderr.reconfigure(errors="surrogateescape", line_buffering=True) PROC_RE = re.compile(r'php\d?(?:-fpm|-cgi)?|perl$') INI_FILE = '/opt/imh-scan/procwatch/settings.ini' Q_DIR = Path('/opt/imh-scan/procwatch/q') SCANLOG_DIR = Path('/opt/imh-scan/procwatch/scanlog') IGNORE_DIR = Path('/opt/imh-scan/ignore') @dataclass class ProcData: pid: int username: str cwd: str cmdline: str uid: int class Config(configparser.ConfigParser): def __init__(self): super().__init__() if not self.read(INI_FILE): sys.exit(f"Could not read {INI_FILE}") @property def rolloff(self) -> int: return self.getint('tolerance', 'rolloff') @property def cooldown(self) -> int: return self.getint('tolerance', 'cooldown') @property def ignore_duration(self) -> int: return self.getint('tolerance', 'ignore_duration') @property def admin_email(self) -> str: return self.get('notify', 'admin_email') class ProcWatch: def __init__(self, args, conf: Config): self.conf = conf self._verbose: bool = args.verbose self.ignore: Union[int, None] = args.ignore self.quarantine: bool = args.quarantine self.user: Union[str, None] = args.user self.ignore_run: bool = args.ignore_run self.hostname = socket.gethostname() self.time = int(time.time()) self.lock_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) self.scanner = Scanner( disable_default=True, disable_freshclam=True, verbose=False, heuristic=True, exclude=[], extra_heuri=False, install=False, update=False, phishing=False, disable_media=False, disable_excludes=False, enable_maldetect=True, disable_new_yara=False, ) def verbose(self, *args, **kwargs): """extra stdout for the -v flag""" if self._verbose: print(*args, **kwargs) @staticmethod def _scannable_proc(proc: psutil.Process) -> bool: if proc.info['uids'].effective == 0: return False return PROC_RE.match(proc.info['name']) def audit_procs(self): """looks at existing procs, gets their open files, and scans them""" self.verbose('running audit') procs = filter( self._scannable_proc, psutil.process_iter(['pid', 'uids', 'name', 'username', 'cwd']), ) procs = list(procs) # generator -> list path_to_procs: dict[Path, list[ProcData]] = {} self.verbose('procs found:', procs) for proc in procs: try: cmdline = proc.cmdline() if cmdline[0].startswith('php-fpm: pool'): continue except Exception: continue p_data = ProcData( pid=proc.info['pid'], username=proc.info['username'], cwd=proc.info['cwd'], cmdline=shlex.join(cmdline), uid=proc.info['uids'].effective, ) if self.should_ignore(proc.info['username']): self.verbose("ignoring", proc.info['username']) continue try: # guessing file path based on cmdline path_guess = Path(p_data.cwd, cmdline[-1].lstrip('/')) self.verbose(f'path guess: {path_guess}\n cwd: {p_data.cwd}') if path_guess.is_file() and self.check_file( path_guess, p_data.uid ): if path_to_procs.get(path_guess): path_to_procs[path_guess].append(p_data) else: path_to_procs[path_guess] = [p_data] # proc_list[path_guess] = p_data except Exception as exc: self.verbose(f'failed to guess path of original file:\n{exc}') try: for open_file in proc.open_files(): file_path = Path(open_file[0]) if self.check_file(file_path, p_data['uid']): if not path_to_procs.get(file_path): path_to_procs[file_path] = [p_data] else: path_to_procs[file_path].append(p_data) except Exception as exc: self.verbose(type(exc).__name__, exc, sep=': ') scan_paths = list(map(str, path_to_procs)) self.verbose('scan paths:', *scan_paths, sep='\n') if not scan_paths: return result = self.run_scan(paths=scan_paths) if not result: return # run_scan already printed the error self.verbose('results:', *result.all_found, sep='\n') if not result: self.verbose('no malware procs detected') return user_list: dict[str, list[ProcData]] = {} pid_list: list[int] = [] for file_path in result.all_found.keys(): for proc in path_to_procs[file_path]: pid = proc.pid if pid in pid_list: continue if self.quarantine: self.verbose(f'killing {pid}') os.kill(pid, signal.SIGKILL) if not user_list.get(proc.username): user_list[proc.username] = [proc] else: user_list[proc.username].append(proc) pid_list.append(pid) for user, p_datas in user_list.items(): self.queue_scan(user, p_datas) def check_file(self, file_path: Path, uid: int): """checks if a file should be scanned by audit_procs""" self.verbose('checking', file_path) if not file_path.is_file(): return False try: stat = file_path.stat() except OSError as exc: self.verbose(type(exc).__name__, exc, sep=': ') return False if stat.st_uid != uid: self.verbose('proc owner doesnt own the file:', file_path) return False if stat.st_uid == 0: self.verbose('skipping root file:', file_path) return False if stat.st_size > 25 * 2**20: # file is > 25 MiB return False return True def run_scan( self, paths: Union[list[str], None] = None, user: Union[str, None] = None, ) -> Union[ScanResult, None]: """runs a scan on targeted files found during audit queues a full user scan if something is detected""" if not paths and not user: self.verbose('error: no scan targets') sys.exit(1) scan_paths = paths or [] if user: try: homedir = rads.get_homedir(user) except Exception as exc: print(f"{user}: {exc}", file=sys.stderr) return None scan_paths.append(homedir) self.scanner.cpu_wait() self.verbose('about to scan') result: ScanResult = self.scanner.scan( scan_paths=scan_paths, log_tuples=[] ) if user: self.ignore_user(user, self.conf.cooldown) self.verbose(result.summary) return result def queue_scan(self, user: str, p_datas: list[ProcData]): """queues a scan after finding a malicious file during audit""" queue_user = Q_DIR / user if queue_user.exists(): # scan already queued self.verbose('scan already queued for', user) return # dump to queue full user scan self.yaml_dump(list(map(vars, p_datas)), queue_user) def full_quarantine(self, user: str): """full user quarantine using seanc rad""" try: subprocess.call([QUARANTINE, '-f', user], timeout=600) except FileNotFoundError: print('Error:', QUARANTINE, 'command not found', file=sys.stderr) return False except subprocess.TimeoutExpired: print('Error: Quarantining', user, 'timed out.', file=sys.stderr) return False return True def should_ignore(self, user: str): """function for checking if users are on the manual ignore list if the current time passes the mtime of the file, stop ignoring""" ignore_file = IGNORE_DIR / user try: mtime = ignore_file.stat().st_mtime except FileNotFoundError: return False if mtime > self.time: self.verbose(f'ignoring {user}, {ignore_file} detected ') return True return False def iter_queue(self): """parent function to handle checking history then execute actions mostly just runs the user scanning function""" for entry in Q_DIR.iterdir(): user = entry.name try: pwd.getpwnam(user) except KeyError: # user doesnt exist self.verbose(f"user {user} doesn't exist") entry.unlink() continue actions = self.check_history(user) if not actions: self.verbose('doing nothing') continue yield user, actions def check_history(self, user: str) -> list[str]: """checks history file for previous offenses mostly unused logic as all offenses are manually reviewed unless auto quarantine mode is enabled""" try: homedir = Path(pwd.getpwnam(user).pw_dir) except KeyError: print('Error: user', user, 'does not exist', file=sys.stderr) return [] if not homedir.exists(): print('Error:', homedir, 'does not exist', file=sys.stderr) return [] path = homedir / '.imh/.audit_history' path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) if not path.exists(): self.verbose('first offense, new history file') self.yaml_dump([self.time], path, 0o600) return ['notify_admin'] try: history_data = self.yaml_load(path) except Exception as exc: print(f'resetting {path} due to error: {exc}') history_data = None else: if not isinstance(history_data, list): print('history_data not a list, resetting') history_data = None for entry in history_data: if not isinstance(entry, int): print('history_data entry not an integer, resetting') history_data = None break if history_data is None: self.yaml_dump([self.time], path, 0o600) history_data = [self.time] return ['notify_admin'] historic_ct = 0 repeat_ct = 0 day_ct = 0 history_data: list[int] for stamp in history_data: time_diff = self.time - stamp if time_diff < 0: self.verbose('error: negative time diff') # rolloff time # 6 months if time_diff > self.conf.rolloff: self.verbose('old offender') historic_ct += 1 # 24 hours elif time_diff > 86400: self.verbose('offense about 24 hours ago') day_ct += 1 # 3 hours elif time_diff > 10800: self.verbose('offense about 3 hours ago') repeat_ct += 1 # cooldown elif time_diff < self.conf.cooldown: self.verbose('user on cooldown') return [] if repeat_ct: # repeat within 3 hours self.verbose('offense repeat within 3 hours, frequent_offense') elif day_ct: # repeat within 24 hours self.verbose('offense repeat within 24 hours, daily_offense') elif not historic_ct: # normal first offence self.verbose('first_offense normal') elif historic_ct <= 3: # occasional offender self.verbose('historic offender within 182 days') elif historic_ct > 3 and not repeat_ct: # chronic offender self.verbose('first_offense historic') history_data.append(self.time) self.yaml_dump(history_data, path) return ['notify_admin'] def handle_user(self, user: str, actions: list[str]): """executes the user scan handles auto quarantine""" # The only behavior implemented is to scan the user and notify. # This assert forces a crash if someone later edits check_history and # doesn't adjust here too. assert actions[0] == 'notify_admin' queue_file = Q_DIR / user self.verbose(f'initiating scan on {user}') result = self.run_scan(user=user) if not result or not result.all_found: self.verbose('nothing detected') queue_file.unlink(missing_ok=True) return data = vars(result) data['p_datas'] = self.yaml_load(queue_file) p_datas = [ProcData(**x) for x in data['p_datas']] self.yaml_dump(data, SCANLOG_DIR / user) self.send_notice(user, result, p_datas, SCANLOG_DIR / user) queue_file.unlink(missing_ok=True) def yaml_load(self, file_name): """basic yaml loading function, uses SafeLoader""" try: with open(file_name, encoding='utf-8') as file: return yaml.load(file, Loader=yaml.SafeLoader) except Exception as exc: print(f'error failed to load yaml file: {file_name}\n{exc}') sys.exit(1) def yaml_dump( self, data: Union[dict, list], path: Path, mode: Union[int, None] = None ): """basic yaml dump function""" try: with open(path, 'w', encoding='utf-8') as file: yaml.dump(data, file) except Exception as exc: sys.exit(f'error failed to dump yaml file: {path}\n{exc}') if mode is not None: path.chmod(mode) def send_notice( self, user: str, result: ScanResult, p_datas: list[ProcData], log_path: Path, ): """sends email notices on positive user scans""" proc_info = 'Original procwatch detected process info:\n' try: for p_data in p_datas: if psutil.pid_exists(p_data.pid): proc_status = '(currently running!!!)' else: proc_status = '(not currently running)' proc_info += ( f'\npid: {p_data.pid} {proc_status}\n' f'cwd: {p_data.cwd}\n' f'cmdline: {p_data.cmdline}\n' ) except Exception as exc: self.verbose(f'error getting process data:\n{exc}') detections = '' for path, sig in result.all_found.items(): detections += f'{path}: {sig}\n' host = platform.node().split('.')[0] message = f""" imh-procwatch has detected a running malicious processes for {user}. A full user scan has been completed. The command to view, quarantine, and kill processes is below: [root@{host} ~]# imh-procwatch -u {user} {proc_info} Hostname: {self.hostname} Log path: {log_path} Detections: {detections} {result.summary} """ self.verbose(f'trying to send email to {self.conf.admin_email}') try: rads.send_email( to_addr=self.conf.admin_email, subject=f'imh-procwatch scan results for {user}', body=message, ssl=True, server=('localhost', 465), errs=True, ) except Exception as exc: self.verbose(type(exc).__name__, exc, sep=': ') def ignore_user(self, user: str, duration: int): """Marks a user to be ignored for x days. Theres an automatic 24 hour cooldown for all users, so it should only be used for longer ignore periods""" ignore_mtime = self.time + duration ignore_file = IGNORE_DIR / user ignore_file.touch(mode=0o644, exist_ok=True) os.utime(ignore_file, (ignore_mtime, ignore_mtime)) self.verbose(f'setting procwatch to ignore {user} for {self.ignore}s') def get_lock(self): """domain socket locking to avoid stacking procs""" try: self.lock_socket.bind('\0imh-procwatch') return True except OSError: return False @staticmethod def is_owner(user, file_path): try: owner = Path(file_path).owner() except Exception: print(c.yellow('WARN: skipping missing file:'), file_path) return False if owner != user: print( c.yellow('WARN: file not owned by the user, skipping it:'), file_path, ) return False return True def q_prompt(self): """function for automating the detection review process imh-procwatch -u {user}""" assert isinstance(self.user, str) scanlog = SCANLOG_DIR / self.user if not scanlog.exists(): sys.exit(f'error: no scanlog found for user at {scanlog}') loaded = self.yaml_load(scanlog) result = ScanResult( command=loaded['command'], hits_found={Path(k): v for k, v in loaded['hits_found'].items()}, heur_found={Path(k): v for k, v in loaded['heur_found'].items()}, summary=loaded['summary'], ) for path, sig in result.hits_found.copy().items(): if self.is_owner(self.user, path): print(path, c.red(sig), sep=':') else: result.hits_found.pop(path) for path, sig in result.heur_found.copy().items(): if self.is_owner(self.user, path): print(path, c.yellow(sig), sep=':') else: result.heur_found.pop(path) if result.all_found: user_input = ask_prompt( 'Would you like to quarantine the detected files? (y|n|a|f)', c.yellow('y = excludes heuristics'), c.green('n = no quarantine'), c.red('a = quarantines all detections'), c.red('f = runs full user quarantine using rad'), chars=('y', 'n', 'a', 'f'), ) if user_input == 'a': jail_files(list(result.all_found)) elif user_input == 'y': jail_files(list(result.hits_found)) elif user_input == 'f': self.full_quarantine(self.user) else: print('skipping quarante due to user input') else: print('WARN: no files to quarantine') p_datas = [ProcData(**x) for x in loaded['p_datas']] for p_data in p_datas: if not psutil.pid_exists(p_data.pid): print(f'pid no longer running: {p_data.pid}') continue proc_obj = psutil.Process(p_data.pid) cmdline = proc_obj.cmdline() user_input = ask_prompt( c.red("WARNING: process is still running"), f"pid: {p_data.pid}", f"cmdline: {cmdline}", c.bold('Would you like to kill the detected process? (y|n|a)'), c.yellow('y = only kills the pid (SIGTERM)'), c.green('n = no kill'), c.red('a = runs "pkill -9 -u {self.user}"'), chars=('y', 'n', 'a'), ) if user_input == 'a': cmd = ['pkill', '-9', '-u', self.user] cmd_str = ' '.join(cmd) print(f'running cmd: {cmd_str}') subprocess.check_output(cmd) time.sleep(1) elif user_input == 'y': print(f"killing pid: {p_data.pid}") try: proc_obj.kill() except Exception as exc: print(f'error killing process:\n{exc}') elif user_input == 'n': print('not killing process') self.reset_cpanel(self.user) cron_path = Path('/var/spool/cron', self.user) if not cron_path.is_file(): print(f'No cron file at {cron_path}') return print(c.green('Showing contents of'), f"{c.red(cron_path)}:") with open(cron_path, encoding='utf-8') as file: for line in file: if line.strip(): # skip printing blank lines print(line, end='') def reset_cpanel(self, user, prompt=True): shared_path = '/opt/sharedrads/reset_cpanel' ded_path = '/opt/dedrads/reset_cpanel' if Path(shared_path).is_file(): reset_path = shared_path elif Path(ded_path).is_file(): reset_path = ded_path else: self.verbose('reset_cpanel not found') return if not prompt: return user_input = ask_prompt( 'Would you like to reset the cpanel user password? (y|n)', c.yellow('y = runs reset'), c.green('n = doesnt reset'), chars=('y', 'n'), ) if user_input == 'n': print('ok: skipping cpanel password reset') return assert user_input == 'y' cmd = [reset_path, user, '-m', 'Active malware detected'] print(f'running: {" ".join(cmd)}') try: out = subprocess.check_output(cmd) print(out) except Exception as exc: print(f'error: failed cpanel_reset:\n{exc}') def check_cpu(): """Exit if load is higher than cpu count""" cpu_limit = os.cpu_count() loadavg = os.getloadavg()[0] if loadavg >= cpu_limit: sys.exit( f'error: load exceeds cpu count, aborting. {loadavg} > {cpu_limit}' ) def get_args(conf: Config): """gets arguments""" parser = argparse.ArgumentParser(description=__doc__) # fmt: off parser.add_argument( '-a', '--audit', action='store_true', help='aduit processes for malware', ) parser.add_argument( '-c', '--check-queue', action='store_true', help='check procwatch queue and run scans', ) parser.add_argument( '-q', '--quarantine', action='store_true', help='automatically quarantine malware', ) parser.add_argument( '-i', '--ignore', type=int, nargs='?', default=None, const='', help='has procwatch ignore user for int(N) days, use with -u [USER]', ) parser.add_argument( '-v', '--verbose', action='store_true', help='enables verbose output to stdout', ) parser.add_argument( '-A', '--all', action='store_true', help='runs the audit then checks the queue', ) parser.add_argument( '-u', '--user', help='checks for completed user scanlogs' ) # fmt: on args = parser.parse_args() if args.user: if not rads.is_cpuser(args.user): parser.print_help() sys.exit('error: invalid username, exiting') if args.user == 'root': sys.exit('error: procwatch doesnt support root audits') if args.ignore == '': # -i flag with no input args.ignore = conf.ignore_duration args.ignore_run = True elif args.ignore is None: # no -i flag args.ignore_run = False else: # -i flag with input args.ignore *= 86400 # days -> secs args.ignore_run = True if args.ignore_run and not args.user: sys.exit('error: ignore flag needs a user specified with -u') if args.check_queue: if not args.quarantine and '@' not in conf.admin_email: sys.exit( 'error: email address invalid and is required ' 'when not using auto quarantine' ) return args def main(): conf = Config() args = get_args(conf) if os.getuid() != 0: sys.exit("This script must run as root") IGNORE_DIR.mkdir(mode=0o755, parents=True, exist_ok=True) Q_DIR.mkdir(mode=0o755, parents=True, exist_ok=True) SCANLOG_DIR.mkdir(mode=0o700, parents=True, exist_ok=True) check_cpu() proc_watch = ProcWatch(args, conf) if args.user and not proc_watch.ignore_run: proc_watch.q_prompt() return if proc_watch.ignore_run: proc_watch.ignore_user(args.user, args.ignore) return if args.audit or args.all: proc_watch.audit_procs() if args.check_queue or args.all: if not proc_watch.get_lock(): proc_watch.verbose('scanner already running the queue') sys.exit(1) for user, actions in proc_watch.iter_queue(): proc_watch.handle_user(user, actions) if __name__ == '__main__': main()
Save
cmd:
run