| Server IP : 199.250.214.225 / Your IP : 3.16.212.214 Web Server : Apache System : Linux vps64074.inmotionhosting.com 3.10.0-1160.105.1.vz7.214.3 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : nicngo5 ( 1001) PHP Version : 7.4.33 Disable Function : exec,passthru,shell_exec,system MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /opt/dedrads/ |
Upload File : |
#!/opt/imh-python/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) 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 {self.filename}."
)
)
return True
else:
if pos_path.name != self.filename:
self.cms.main.print_debug(
red(
f"excluded: {pos_path.name} does not 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} 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, main, name, data):
self.main = main
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") as infile:
try:
cms_yaml = yaml.safe_load(infile)
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) 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()