D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
dedrads
/
Filename :
check_software
back
Copy
#!/usr/lib/rads/venv/bin/python3 """customer software scanner""" from os import getuid, walk, cpu_count from pathlib import Path import argparse import sys from concurrent.futures import Future, ThreadPoolExecutor, as_completed import yaml from check_software_mods.template import BleachedColors, ModTemplate, HtmlColors import rads def read_urls() -> dict[str, str]: path = Path('/opt/sharedrads/etc/kb_urls.yaml') if not path.is_file(): path = Path('/opt/dedrads/etc/kb_urls.yaml') assert path.is_file() with open(path, encoding='ascii') as kbs: kb_urls = yaml.load(kbs, yaml.SafeLoader) key = 'hub' if rads.IMH_CLASS == 'hub' else 'imh' return kb_urls[key] KB_URLS = read_urls() def load_submodules() -> dict[str, type[ModTemplate]]: """Import all submodules from check_software_mods directory into a Module object""" mod_path = Path(__file__).parent / 'check_software_mods' mod_names = [] for entry in mod_path.iterdir(): if not entry.is_file(): continue if not entry.name.endswith('.py') or entry.name.startswith('_'): continue if entry.name == 'template.py': continue mod_names.append(entry.name[:-3]) # ensure we found at least one module; otherwise, something's missing assert mod_names check_software_mods = __import__( 'check_software_mods', globals(), locals(), mod_names, 0 ) mods_dict = {} for mod_name in mod_names: mods_dict[mod_name] = getattr(check_software_mods, mod_name).Module return mods_dict def parse_args(cms_args: list[str]): """Parse command arguments""" parser = argparse.ArgumentParser(description='customer software scanner') uid = getuid() format_group = parser.add_mutually_exclusive_group() # fmt: off format_group.add_argument( '--str', action='store_const', dest='style', const='str', help='for use by send_customer_str', ) format_group.add_argument( '--no-color', '--bleach', '-b', action='store_const', dest='style', const='bleach', help='Print with no colors', ) if uid == 0: parser.add_argument( '--root', '-r', dest='use_root', action='store_true', help="connect to MySQL as root", ) parser.add_argument( '--guides', '-g', action='store_true', dest='guide_bool', help="If no caching, will print a guide for WPSuperCache.", ) parser.add_argument( '--cms', '-c', dest='cms', choices=cms_args, help='Specific CMS to scan. If unspecified, scan all of them.', ) parser.add_argument( '--full', '-f', dest='full', action='store_true', help="Scan entire account, not just document roots and immediate " "subdirectories (slow)", ) threads = 4 if rads.vz.is_vps() else cpu_count() parser.add_argument( '--threads', type=int, default=threads, help=f'Max number of threads to scan with (default: {threads})', ) parser.add_argument( 'target', metavar='TARGET', type=str, nargs='+', help='username(s) or directory to scan (required)', ) # fmt: on args = parser.parse_args() if uid != 0: args.use_root = False if args.cms: args.cms = [args.cms] else: args.cms = cms_args if args.style == 'str': args.guide_bool = False # don't show it twice return args def get_doc_roots(username: str): """Get a list of directories which should be searched for websites""" try: udata = rads.UserData(username) except rads.CpuserError as exc: sys.exit(f'Could not scan {username} - {exc}') docroots = [Path(x) for x in udata.merged_roots] # make sure nothing dumb is in the list which would make us recursively # crawl the whole filesystem homedir = Path(f"~{username}").expanduser() docroots = [x for x in docroots if x.is_relative_to(homedir)] # got the doc roots, now get one level of subdirs, combine list subdirs = [] for docroot in docroots: try: for entry in docroot.iterdir(): if entry.is_dir() and not entry.is_symlink(): subdirs.append(entry) except Exception: pass # combine homedirs and subdirs lists to_crawl = docroots + subdirs # eliminate duplicates to_crawl = list(set(to_crawl)) to_crawl.sort() return to_crawl def find_configs( doc_roots: list[Path], mods: list[type[ModTemplate]] ) -> list[Path]: # collect a list of filenames which should be searched for to_find = target_configs(mods) # go through each document root and search for those files. Return them config_paths: list[Path] = [] for doc_root in doc_roots: try: for entry in doc_root.iterdir(): if not entry.is_file(): continue if entry.name in to_find: config_paths.append(entry) except OSError as exc: print(f"{doc_root} - {exc}", file=sys.stderr) return config_paths def target_configs(mods: list[type[ModTemplate]]) -> set[str]: to_find = set() for mod in mods: cfg = mod.config_file() if isinstance(cfg, list): to_find.update(cfg) else: to_find.add(cfg) return to_find def full_scan( user: str, path: Path | None, mods: list[type[ModTemplate]] ) -> list[Path]: home = Path(f"~{user}").expanduser().resolve() if path: # scanning a path provided from cli args topdir = path.resolve() else: # scanning a whole homedir topdir = home assert str(topdir).startswith('/home') # collect a list of filenames which should be searched for to_find = target_configs(mods) config_paths = [] for root, dirs, files in walk(str(topdir), topdown=True): if root == home: dirs[:] = [d for d in dirs if d not in ('mail', 'etc')] for name in files: if name in to_find: config_paths.append(Path(root, name)) return config_paths def get_targets(items: list[str], args) -> tuple[list[str], dict[str, Path]]: users = [] paths = {} if args.style == 'str': color = HtmlColors() elif args.style == 'bleach': color = BleachedColors() else: color = rads.color for item in items: if rads.is_cpuser(item): users.append(item) continue path = Path(item).expanduser().resolve() if not path.is_dir() or path.is_symlink(): print(color.red(f"{item} is not a cPanel user or directory")) continue if not str(path).startswith('/home'): # avoid Path.is_relative_to so dedis with /home2 will work print(color.red(f"{item} is an invalid user path")) continue owner = str(path).split('/', maxsplit=3)[2] if not rads.is_cpuser(owner): print(color.red(f"{item} is an invalid user path")) continue paths[owner] = path return users, paths def main(): """Main logic of customer software scanner: 1. Collect user input 2. Collect plugin info from repo 3. Scan installations""" mod_classes = load_submodules() args = parse_args(list(mod_classes.keys())) args.wp_found = False # wordpress.py sets this True later mods = [v for k, v in mod_classes.items() if k in args.cms] users, paths = get_targets(args.target, args) if not users and not paths: sys.exit(1) with ThreadPoolExecutor(args.threads) as pool: scans, tasks = [], [] for user in users: future = pool.submit(scan_user, tasks, pool, user, mods, args) scans.append(future) for user, path in paths.items(): future = pool.submit(scan_path, tasks, pool, user, path, mods, args) scans.append(future) for future in as_completed(scans): future.result() for future in as_completed(tasks): ret: ModTemplate = future.result() out = ret.out.getvalue() if args.style == 'str': out = out.replace('\n', '<br />\n') out = out.replace(' ', ' ') print('<div style="font-family: monospace">') print(out, end='') if args.style == 'str': print('</div>') print("\n\n\n", end='') if args.guide_bool and args.wp_found: print(f"Caching instructions found here:\n{KB_URLS['w3_total']}") def scan_user( tasks: list[Future], pool: ThreadPoolExecutor, user: str, mods: list[type[ModTemplate]], args, ) -> None: susp = Path('/var/cpanel/suspended', user).is_file() if args.full: site_configs = full_scan(user=user, path=None, mods=mods) else: site_configs = find_configs(get_doc_roots(user), mods) run_mods(tasks, pool, susp, site_configs, mods, args) def scan_path( tasks: list[Future], pool: ThreadPoolExecutor, user: str, path: Path, mods: list[type[ModTemplate]], args, ) -> list[Future]: susp = Path('/var/cpanel/suspended', user).is_file() site_configs = full_scan(user=user, path=path, mods=mods) run_mods(tasks, pool, susp, site_configs, mods, args) def run_mods( tasks: list[Future], pool: ThreadPoolExecutor, susp: bool, site_configs, mods: list[type[ModTemplate]], args, ) -> None: for mod in mods: for path in site_configs: if not mod.is_config(path): continue future = pool.submit(mod, args, path, susp, KB_URLS) tasks.append(future) if __name__ == "__main__": main()