1#@!#!123s

: / opt / dedrads /
Filename : quarantine
back
#!/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.")