D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
proc
/
3
/
root
/
opt
/
dedrads
/
Filename :
cms_check.py
back
Copy
#!/usr/lib/rads/venv/bin/python3 from pathlib import Path import argparse import json import subprocess import os import time import re import yaml from rads.color import green, yellow, bold, blue, red class Fingerprint: """Fingerprint to find a CMS installation.""" def __init__(self, cms, fp_data): self.cms = cms self.fp_data = fp_data self.filename = fp_data.get("file") self.signature = fp_data.get("signature") self.excludes = {2: ["softaculous"], 3: ["quarantine"]} self.paths_found = [] def search(self): """Uses mlocate to search for the file signature.""" cmd = [ "/bin/locate", "-d", "/var/lib/mlocate/mlocate.db", self.filename, ] with subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, encoding='utf-8', ) as proc: for out_line in proc.stdout: line = out_line.strip() if not os.path.exists(line) or self.should_exclude(line): continue if self.signature: with open(line, encoding="utf-8") as stream: match = re.search(self.signature, stream.read()) if match: self.paths_found.append(line) else: self.cms.main.print_debug( red(f"{self.signature} not found within {line}") ) else: self.paths_found.append(line) def should_exclude(self, path): """Determines whether or not the path should be excluded. Args: path (str): found path Returns: bool: True if should be excluded, else False """ self.cms.main.print_debug(f"Checking excludes on {path}") pos_path = Path(path) if "/" in self.filename: if self.filename not in path: self.cms.main.print_debug( red( f"excluded: {pos_path.name} does not match " f"{self.filename}." ) ) return True else: if pos_path.name != self.filename: self.cms.main.print_debug( red( f"excluded: {pos_path.name} does not " f"match {self.filename}." ) ) return True for index, part in enumerate(pos_path.parts): if index in self.excludes: if part in self.excludes[index]: self.cms.main.print_debug( red( f"excluded: excluded part, {part} " f"in {pos_path.parts}" ) ) return True if not self.cms.main.args.include_all: for docroot_path in self.cms.main.document_roots: if docroot_path in pos_path.parents: return False else: return False self.cms.main.print_debug( red(f"excluded: didn't determine if we shouldn't exclude: {path}") ) return True class CMS: def __init__(self, cms_search: 'CMSSearch', name, data): self.main = cms_search self.name = name self.data = data self.actual_name = name self.fingerprints = [] self.versions = [] def load(self): """Loads the actual name and the fingerprints from the CMS Signature data. """ self.actual_name = self.data.get("name") for fingerprint in self.data.get("fingerprints", []): fp = Fingerprint(self, fingerprint) fp.search() self.fingerprints.append(fp) def found_installations(self): """Returns all installation paths found for the first fingerprint that found an installation. Returns: dict: key, value dictionary of path and docroot domain. """ all_paths = {} for fp in self.fingerprints: if len(fp.paths_found) > 0: for path in fp.paths_found: all_paths[path] = self.main.find_docroot_domain(path) return all_paths return all_paths class CMSSearch: def __init__(self, args): self.document_roots = {} self.args = args self.installs = {} def run(self): """Primary run entry to begin CMS search.""" self.collect_docroots() self.load_cms_data() self.display_results() def print_debug(self, message): """Prints a debug message if verbose is enabled. Args: message (str): the message to print """ if self.args.verbose: print(blue(message)) def display_results(self): """Lists all found installations in a specific format based on given flags. """ out_json = {"cms": {}} for cms_name, cms_obj in self.installs.items(): cms_obj: 'CMS' installs = cms_obj.found_installations() actual_name = cms_obj.actual_name if len(installs) > 0: if self.args.output == "simple": print(cms_name, installs) elif self.args.output == "normal": print(bold(yellow(actual_name))) for install in installs: domain = green(installs[install]) print(f"{domain}: {install}") print() elif self.args.output == "json": for install in installs: domain = installs[install] if actual_name not in out_json["cms"]: out_json["cms"][actual_name] = {} if domain not in out_json["cms"][actual_name]: out_json["cms"][actual_name][domain] = [] out_json["cms"][actual_name][domain].append(domain) if self.args.output == "json": print(json.dumps(out_json)) def load_cms_data(self): """Safely loads the CMS signatures from a yaml file.""" with open( "/opt/dedrads/extras/cms_signatures.yaml", encoding="utf-8" ) as infile: try: cms_yaml = yaml.load(infile, Loader=yaml.SafeLoader) for cms_name in cms_yaml: cms = CMS(self, cms_name, cms_yaml[cms_name]) cms.load() self.installs[cms_name] = cms except yaml.YAMLError as err: print(err) def find_docroot_domain(self, path): """Returns a domain for the document root that the provided path belongs to. Args: path (str): path of the installation Returns: str: domain of the document root """ for docroot_path, domain in self.document_roots.items(): if docroot_path in Path(path).parents: return domain return "no-domain" def collect_docroots(self): """Collects all document roots from cPanel.""" for user_folder in Path("/var/cpanel/userdata/").glob("*"): for user_item in Path(user_folder).glob("*"): with open(user_item, encoding="utf-8") as stream: docroot = "" servername = "" for line in stream.readlines(): if line.startswith("documentroot:"): docroot = Path(line.split()[1].strip()) if line.startswith("servername:"): servername = line.split()[1].strip() if docroot and servername: self.document_roots[docroot] = servername def check_db_age(self): """ Checks the mlocate database modification age and runs update if older than 24 hours. """ mtime = os.path.getmtime("/var/lib/mlocate/mlocate.db") diff = time.time() - mtime if diff > 86400: if self.args.output != "json": print(yellow("Locate DB is too old, running update.")) self.update_db() def update_db(self): """Updates the mlocate database.""" ret_code = subprocess.call(["/bin/updatedb"]) if self.args.output == "json": return if ret_code: print(red("Failed to update mlocate database.")) else: print(green("Updated mlocate database.")) def parse_args(): args = argparse.ArgumentParser( description=( "Uses signatures to determine what CMS installations are " "installed on the server." ) ) output_choices = ["json", "simple", "normal"] args.add_argument( "--output", nargs="?", help="type of output", choices=output_choices, default="normal", ) args.add_argument( "--include-all", action="store_true", help=( "Includes all installations, including those not in a " "document root." ), ) args.add_argument("--verbose", action="store_true", help="Verbose output.") return args.parse_args() def main(): args = parse_args() cms_searcher = CMSSearch(args) cms_searcher.check_db_age() cms_searcher.run() if __name__ == "__main__": main()