FM / dedrads

/opt/dedrads

Name	Size	Mode	Actions
account_review/	-	0555	rm
check_software_mods/	-	0555	rm
cms_tools/	-	0555	rm
etc/	-	0555	rm
extras/	-	0555	rm
mailparse/	-	0555	rm
mysql/	-	0555	rm
nlp_scripts/	-	0555	rm
oldrads/	-	0555	rm
perl/	-	0555	rm
provision/	-	0555	rm
python/	-	0555	rm
suspended/	-	0555	rm
__pycache__/	-	0555	rm
account-review	799	0555	edit dl rm
allfw	6259	0555	edit dl rm
alp.py	16553	0555	edit dl rm
autossl_sync	6267	0555	edit dl rm
check_apache	9205	0555	edit dl rm
check_bandwidth	4546	0555	edit dl rm
check_boxtrapper	1684	0555	edit dl rm
check_cpu	12212	0555	edit dl rm
check_crons	1827	0555	edit dl rm
check_dcpumon	1973	0555	edit dl rm
check_dns	1827	0555	edit dl rm
check_domlogs	61	0555	edit dl rm
check_exim	3731	0555	edit dl rm
check_hacks	22604	0555	edit dl rm
check_imap	5249	0555	edit dl rm
check_io	6635	0555	edit dl rm
check_max_children	16341	0555	edit dl rm
check_mem	1660	0555	edit dl rm
check_misc	1510	0555	edit dl rm
check_pacct	5201	0555	edit dl rm
check_php	4344	0555	edit dl rm
check_pop3	2295	0555	edit dl rm
check_prov.py	4714	0555	edit dl rm
check_server	696	0555	edit dl rm
check_size	490	0555	edit dl rm
check_software	9849	0555	edit dl rm
check_spamd	2426	0555	edit dl rm
check_traffic	1933	0555	edit dl rm
check_user	1887	0555	edit dl rm
clean_exim.py	6390	0555	edit dl rm
clusterfix	5343	0555	edit dl rm
cmspass.py	3970	0555	edit dl rm
cms_check.py	9825	0555	edit dl rm
cms_counter.py	36734	0555	edit dl rm
cms_creds	713	0555	edit dl rm
cms_dumpdb	4485	0555	edit dl rm
cms_pw	1416	0555	edit dl rm
cpanel-api	7040	0555	edit dl rm
cpanel_postgres_manager.py	29004	0555	edit dl rm
cpmerge	4631	0555	edit dl rm
cpumon	2195	0555	edit dl rm
dcpumon.pl	1239	0555	edit dl rm
dedcheck	18127	0555	edit dl rm
dns-sync	14386	0555	edit dl rm
docroot.py	4425	0555	edit dl rm
du-tree	7450	0555	edit dl rm
envinfo.py	6633	0555	edit dl rm
exclude_rbl.py	5249	0555	edit dl rm
extract-vhost	2691	0555	edit dl rm
filescan	3566	0555	edit dl rm
find_warez	1460	0555	edit dl rm
first_setup.py	19120	0555	edit dl rm
fixwpcron.py	4822	0555	edit dl rm
fix_dns_cluster	353	0555	edit dl rm
goaccess	1217720	0555	edit dl rm
hostsfilemods	4467	0555	edit dl rm
imap_io	2323	0555	edit dl rm
innodb_converter.py	9343	0555	edit dl rm
killall911	530	0555	edit dl rm
killdns	941	0555	edit dl rm
lastcommcache.sh	1374	0555	edit dl rm
legal_lock_down.sh	7450	0555	edit dl rm
lil-cpanel	52002	0555	edit dl rm
listacct	4575	0555	edit dl rm
mailscan	4544	0555	edit dl rm
mail_sources.py	7407	0555	edit dl rm
megaclisas-status	38664	0555	edit dl rm
migrate2central.sh	14418	0555	edit dl rm
modify-account	40631	0555	edit dl rm
modsec_disable.py	11461	0555	edit dl rm
monarxctl	8912	0555	edit dl rm
msp.pl	27714	0555	edit dl rm
mysql_dstat	474	0555	edit dl rm
mysql_selector.py	73102	0555	edit dl rm
nlp	4582	0555	edit dl rm
procscrape	1863	0555	edit dl rm
quarantine	11083	0555	edit dl rm
radsfunctions.sh	863	0555	edit dl rm
recent-cp	16148	0555	edit dl rm
remote_dump	18347	0555	edit dl rm
reset_cpanel	3205	0555	edit dl rm
server-load	2192	0555	edit dl rm
show-conns	20105	0555	edit dl rm
show-conns-adv.py	54667	0555	edit dl rm
sparta.py	382498	0555	edit dl rm
sqltop	27129	0555	edit dl rm
telcheck	2411	0555	edit dl rm
temprootreset	186	0555	edit dl rm
unsuspend_user	1307	0555	edit dl rm
unsusprunner.sh	1840	0555	edit dl rm
updatednsadmin	526	0555	edit dl rm
update_spf	14972	0555	edit dl rm
upgrade-check	9216	0555	edit dl rm
vhost_data.py	6221	0555	edit dl rm
wp-xray.sh	85816	0555	edit dl rm

Edit: /opt/dedrads/quarantine (11083B)

#!/usr/lib/rads/venv/bin/python3
"""Quarantine tool"""
from pathlib import Path
from os import kill, getuid, chown
import re
import pwd
import shlex
import sys
import time
import shutil
import argparse
from argparse import ArgumentTypeError as BadArg
import logging
import logging.handlers
import psutil
import rads

logger = logging.getLogger('quarantine')

USER_RE = re.compile(r'/home[0-9]?/(\w+)')
# EPOCH truncates the minute from the unix timestamp so running this in a loop
# won't create numerous quarantine directories
EPOCH = int(time.time())
EPOCH -= EPOCH % 60
USAGE = """
    - PARTIAL: quarantine item1 item2 item3 ...
    - FULL:    quarantine (--full|-f) user1 user2 ...
    - *BOTH:   quarantine item1 item2 ... (--full|-f) user1 user2 ...

* Specify items first or else each item will be considered a user"""
DESCRIPTION = """
Relative Paths: If you do not specify the absolute path for a given item, then
    it will attempt to create a full path for you. So, the current working
    directory is very important.  This is because the relative path will be
    joined with your current working for directory. For example, running
    quarantine public_html from /root will assume that you are trying to
    quarantine /root/public_html.

Full Quarantines: The docroots used with full quarantines are found by using the
    rads.UserData module. If this fails for some reason, then an error will be
    logged. You'll then have to proceed by manually providing the specified
    items to be quarantined.

Docroot Recreation: If the actions taken by the quarantine script happen to
    quarantine a document root for any domain, then the program will
    automatically recreate it.

Specifying Items/Users: There is no limitation on what can be specified as an
    item. The script will try to iterate through each item, determine the user,
    and quarantine the item as as necessary. If the user or item is invalid, a
    warning will be logged and the item will not be quarantined.

Quarantining Subdirectories: When quarantining files, the script creates the
    quarantine destination recursively. So, providing the script a subdirectory
    item first will cause a File exists error when attempting to quarantine the
    top-level directory. Ex: don't do the following:
        `quarantine public_html/addondomain.com public_html`

Examples:
    quarantine --full userna5
    quarantine ~userna5/public_html/addondomain.com ~userna5/subdomain.com
"""
RUN_FIXPERMS = []

def setup_logging(clevel=logging.INFO):
    """Setup logging"""
    logger.setLevel(logging.DEBUG)
    # stderr
    con = logging.StreamHandler(stream=sys.stderr)
    con.setLevel(clevel)
    con_format = logging.Formatter("%(levelname)s: %(message)s")
    con.setFormatter(con_format)
    logger.addHandler(con)
    # syslog
    try:
        flog = logging.handlers.SysLogHandler(address='/dev/log')
        # syslog is configured to only log INFO and above, so use the same
        flog.setLevel(logging.INFO)
        flog_format = logging.Formatter("[%(name)s:%(levelname)s] %(message)s")
        flog.setFormatter(flog_format)
        logger.addHandler(flog)
    except Exception as e:
        logger.warning("Failed to open logfile: %s", str(e))

def quarantine(path: Path, user_home: Path):
    """Quarantine file/directory"""
    # Ex: item = "/home/userna5/public_html"
    # Ex: qdest = "/home/userna5/quarantine/quarantine_1538571067/home/userna5/"
    # Ex: "/home/userna5/public_html" -> \
    # "/home/userna5/quarantine/quarantine_1538571067/home/userna5/public_html"
    qdir = user_home / f"quarantine/quarantine_{EPOCH}"
    qdest = qdir / str(path.parent).lstrip('/')
    try:
        qdest.mkdir(mode=0o700, parents=True, exist_ok=True)
    except PermissionError:
        logger.critical(
            "Permission denied to %s: Run fixperms and retry", qdest
        )
        return
    except OSError as exc:
        logger.critical("%s: %s", type(exc).__name__, exc)
        return
    logger.debug("Successfully created quarantine directory (%s).", qdest)
    try:
        shutil.move(path, qdest)
    except Exception as exc:
        logger.critical("Error occurred quarantining %s: %s", path, exc)
        return
    logger.info(
        "Successfully quarantined '%s' -> '%s'", path, qdest / path.name
    )

def killall(user):
    """Kills all active processes under the passed user"""
    procs: dict[int, str] = {}
    for pid in psutil.pids():
        try:
            proc = psutil.Process(pid)
        except psutil.NoSuchProcess:
            continue
        if proc.username() == user:
            procs[proc.pid] = shlex.join(proc.cmdline())
    killed = []
    for pid, cmd in procs.items():
        try:
            kill(pid, 9)
        except OSError as exc:
            logging.warning(
                "Could not kill %s - %s: %s", cmd, type(exc).__name__, exc
            )
            continue
        killed.append(cmd)
    if killed:
        print("Killed user processes:", *killed, sep='\n')

def cleanup(user: str, docroots: list[Path]):
    """Miscellaneous cleanup tasks"""
    uid = pwd.getpwnam(user).pw_uid
    nobody_uid = pwd.getpwnam('nobody').pw_uid
    # Iterate through each document root and if it does not
    # exist, then re-create it.  Since the only root will
    # have privileges to set ownership uid:nobody_uid, logic
    # exists to skip that step.  If the step was skipped, then
    # the a WARNING message is shown advising a run of fixperms.
    for path in sorted(docroots):
        if path.exists():
            continue
        try:
            path.mkdir(parents=True, mode=0o750)
        except OSError as exc:
            logger.error("Error recreating '%s': %s", path, exc)
            continue
        logger.debug("Successfully recreated '%s'", path)
        # If root, then set proper ownership (user:nobody)
        # Else, set flag to notify user they must run fixperms
        if UID != 0:
            RUN_FIXPERMS.append(user)
            continue
        try:
            chown(path, uid, nobody_uid)
        except OSError as e:
            logger.error("Error setting ownership for '%s': %s", path, e)
            continue
        logger.debug("Successfully set ownership for '%s'", path)

def valid_user(user: str):
    if not rads.cpuser_safe(user):
        raise BadArg(f"{user} is an invalid user")
    return user

def valid_path(item: str) -> Path:
    try:
        path = Path(item)
        if not path.resolve().is_relative_to('/home'):
            logger.warning("%s is outside /home, skipping.", path)
            return None
        if not path.exists():
            logger.warning("%s does not exist, skipping.", path)
            return None
        return path
    except Exception as e:
        logger.warning("Error validating %s: %s, skipping.", item, e)
        return None

def item_validation(path: Path, user_home: Path):
    """Test for valid item"""
    if not path.is_relative_to(user_home):
        logger.warning("Not quarantining item (%s) outside homedir", path)
    elif not path.exists():
        logger.warning("%s wasn't found", path)
    elif path == user_home:
        logger.warning("Not quarantining user's homedir (%s)", path)
    elif not str(path).startswith("/home"):
        logger.warning("Quarantining is restricted to /home")
    else:
        return True
    return False

def get_userdata(user: str, all_roots: bool = False) -> list[Path]:
    """Gets docroots from rads.Userdata"""
    try:
        if all_roots:
            docroots = map(Path, rads.UserData(user).all_roots)
        else:
            docroots = map(Path, rads.UserData(user).merged_roots)
        home = Path(f"~{user}").expanduser()
    except Exception as exc:
        logger.error("Error retrieving %s's cPanel userdata: %s", user, exc)
        return []
    return [
        x
        for x in docroots
        if x.is_relative_to(home) and str(x).startswith('/home')
    ]

def parse_args() -> tuple[set[Path], bool]:
    """Parse Arguments"""
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawTextHelpFormatter,
        usage=USAGE,
        description=DESCRIPTION,
    )
    parser.set_defaults(loglevel=logging.INFO, items=[], users=[])
    parser.add_argument(
        "items",
        nargs='*',
        help="files and directories to quarantine",
        type=valid_path,
    )
    parser.add_argument(
        "-f",
        "--full",
        dest='users',
        nargs="+",
        type=valid_user,
        help="Perform a full quarantine of all docroots for given users",
    )
    parser.add_argument(
        "-d",
        '--debug',
        dest='loglevel',
        action='store_const',
        const=logging.DEBUG,
        help="Enable debug output",
    )
    parser.add_argument(
        "-k",
        '--kill',
        dest='kill',
        action='store_true',
        help="Kill all processes under the detected user after quarantine",
    )
    args = parser.parse_args()
    setup_logging(args.loglevel)
    items: list[Path] = [item for item in args.items if item is not None]
    for user in args.users:  # for any user supplied with --full
        items.extend(get_userdata(user))
    items = {x.resolve() for x in items}
    return items, args.kill

def main():
    """Main Logic"""
    paths, do_kill = parse_args()
    user_docroots: dict[str, list[Path]] = {}
    # For each item: parse user, validate user, store userdata, set
    # homedir, validate item, quarantine item. The contents of items can include
    # files/directories across multiple users
    for path in paths:
        if user_match := USER_RE.match(str(path)):
            user = user_match.group(1)
        else:
            logger.warning("Unable to determine user for %s", path)
            continue
        if not rads.cpuser_safe(user):
            logger.warning('Skipping invalid user %s', user)
            continue

if user not in user_docroots:
            user_docroots[user] = get_userdata(user, all_roots=True)
        try:
            user_home = Path(f"~{user}").expanduser()
        except Exception:
            logger.warning("Could not get homedir for %s", user)
            continue
        if not item_validation(path, user_home):
            logger.warning("Skipping %s", path)
            continue
        quarantine(path, user_home)
        if do_kill:
            killall(user)

# This block of code calls the cleanup function.
    # It currently just ensures that all the user's docroots are present
    for user, docroots in user_docroots.items():
        cleanup(user, docroots)

if not RUN_FIXPERMS:
        return
    logger.warning(
        "During the quarantine process some document roots were quarantined. "
        "You must run fixperms to correct permissions/ownership of these "
        "document roots. Please run: fixperms %s",
        shlex.join(RUN_FIXPERMS),
    )

UID = getuid()
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print("Canceled.")

cmd: